[gst-devel] Leaking buffers

[Will Newton]

Hi all,

I have an app that streams audio using the neon source plugin, the
icydemux plugin and a fakesink. Occasionally the source plugin will
throw an error when the http connection resets. I try to handle this
by dismantling and rebuilding the pipeline. Unfortunately it seems
that the way I do this is leaking buffers.

I diagnosed this by running under valgrind massif and the space used
by gst_buffer_new_and_alloc takes a little uptick every time the
stream resets.

I have attached the app I am using. Could anyone help me find where
these buffers should be getting unreffed but presumably aren't?

Thanks,
-------------- next part --------------

#include <stdlib.h>

#include <glib.h>
#include <gst/gst.h>

typedef struct stream {
    char *name;
    char *url;
} stream_t;

/* I think we can get the station name out of the metadata of the stream
 * by doing a get_property on the http src plugin but by then it's too late
 * to be useful.
 */
stream_t streams[] = {
    { "Smooth Jazz",     "http://82.197.167.134:80" },
    { "Swiss Classical", "http://82.197.167.137:80" },
    { "RadioCrazy Jazz", "http://82.197.165.138:80" },
    { "Smooth Jazz",     "http://82.197.167.134:80" },
    /* { "HotMixRadio",     "http://91.121.6.37:80" }, */
    { NULL, NULL },
};

static int current_stream = 0;

static GstElement *http_src;
static GstElement *icydemux_filter;
static GstElement *fake_sink;

static GstElement *pipeline;
static GMainLoop *loop;
static GstBus *bus;

static void next_stream(void);


static gboolean
bus_call (GstBus     *bus,
	  GstMessage *msg,
	  gpointer    data)
{
    GMainLoop *loop = (GMainLoop *)data;

    switch (GST_MESSAGE_TYPE(msg)) {
    case GST_MESSAGE_TAG: {
	GstTagList *tag_list;
	gchar *title;

	gst_message_parse_tag(msg, &tag_list);

	if (gst_tag_list_get_string (tag_list, GST_TAG_TITLE, &title)) {
	    g_print("Title: %s\n", title);
	    g_free(title);
	}

	gst_tag_list_free(tag_list);
	break;
    }
    case GST_MESSAGE_EOS:
	g_print("End-of-stream\n");
	next_stream();
	break;
    case GST_MESSAGE_ERROR: {
	gchar *debug;
	GError *err;

	gst_message_parse_error(msg, &err, &debug);

	g_print("Error: %s (%s)\n", err->message, debug);
	g_error_free(err);
	g_free(debug);

	if (msg->src == (GstObject *)http_src)
	    next_stream();
	else
	    g_main_loop_quit(loop);
	break;
    }
    default:
	break;
    }
    return TRUE;
}

static void
new_pad (GstElement *element,
	 GstPad     *pad,
	 gpointer    data)
{
    GstPad *sinkpad;
    /* We can now link this pad with the audio decoder */
    g_print("Dynamic pad created, linking icydemux to sink\n");

    sinkpad = gst_element_get_pad(fake_sink, "sink");
    gst_pad_link(pad, sinkpad);

    gst_object_unref(sinkpad);
}

static void
init_http_src(void)
{
    g_object_set(G_OBJECT(http_src), "location",
		 streams[current_stream].url, NULL);

    g_object_set(G_OBJECT(http_src), "user-agent",
		 "Metagence-Zappa/1.0", NULL);
    g_object_set(G_OBJECT(http_src), "iradio-mode", TRUE, NULL);
}

static void
print_stream_name(void)
{
    g_print("Playing stream \"%s\".\n", streams[current_stream].name);
}

static void
shutdown_stream(void)
{
    //gst_debug_set_default_threshold(GST_LEVEL_DEBUG);

    g_print("Stopping stream\n");
    g_assert(gst_element_set_state(pipeline, GST_STATE_NULL) == 1);

    //gst_debug_set_default_threshold(GST_LEVEL_DEFAULT);

    g_print("Unlinking elements\n");
    gst_element_unlink(http_src, icydemux_filter);

    gst_element_unlink(icydemux_filter, fake_sink);

    g_print("Destroying icydemux\n");
    /* This will unref and delete the object */
    g_assert(gst_bin_remove(GST_BIN(pipeline), icydemux_filter));

    g_print("Destroying http source\n");
    /* This will unref and delete the object */
    g_assert(gst_bin_remove(GST_BIN(pipeline), http_src));
}

static void
start_stream(void)
{
    g_print("Creating new http source\n");
    http_src = gst_element_factory_make("neonhttpsrc", "neonhttpsrc");
    g_assert(http_src);

    gst_bin_add(GST_BIN(pipeline), http_src);

    g_print("Creating new icydemux\n");
    icydemux_filter = gst_element_factory_make("icydemux", "icydemux");
    g_assert(icydemux_filter);

    gst_bin_add(GST_BIN(pipeline), icydemux_filter);

    g_print("Initting http source\n");
    init_http_src();

    g_assert(gst_element_link(http_src, icydemux_filter));

    gst_element_link(http_src, icydemux_filter);
    g_signal_connect(icydemux_filter, "pad-added", G_CALLBACK(new_pad), NULL);

    g_print("Setting pipeline to playing\n");
    gst_element_set_state(pipeline, GST_STATE_PLAYING);

    print_stream_name();
}

static void
next_stream(void)
{
    int new_stream = current_stream;

    if (streams[new_stream].url == NULL)
	new_stream = 0;

    current_stream = new_stream;

    shutdown_stream();
    start_stream();
}

int main(void)
{
    g_print("Initting gstreamer\n");
    gst_init(NULL, NULL);
    loop = g_main_loop_new(NULL, FALSE);

    g_print("Creating pipeline elements\n");
    pipeline = gst_pipeline_new("pipeline");

    fake_sink = gst_element_factory_make("fakesink", "fakesink");
    g_assert(fake_sink);

    bus = gst_pipeline_get_bus(GST_PIPELINE(pipeline));
    gst_bus_add_watch(bus, bus_call, loop);
    gst_object_unref(bus);

    gst_bin_add(GST_BIN(pipeline), fake_sink);

    start_stream();

    g_main_loop_run(loop);

    g_print("Returned, stopping playback\n");
    gst_element_set_state(pipeline, GST_STATE_NULL);

    gst_object_unref(GST_OBJECT(pipeline));

    return 0;
}