[gst-devel] Proper usage of fdsink and fdsrc (out-of-process processing)

[Andrzej K. Haczewski]

2010/7/23 Andrzej K. Haczewski <ahaczewski at gmail.com>:
> Hi,
>
> I have very strange problem that I can't understand and maybe some
> Linux hacker here can give me a hint about what might be wrong.
>
> I'm creating a pipeline that generates sound (audiotestsrc), puts it
> through queue to fdsink, reads whatever comes out of fdsrc (through
> queue) to the fakesink.
>
> It blocks, and me and my colleagues have no idea as to what may cause
> the deadlock. The reason I'm doing such a strange pipeline is that I
> have a commercial encoder that comes as a executable (no SDK, no libs
> to link to) that is able to get input from stdin and write output to
> stdout. To link it with a pipeline I'm spawning it within my GStreamer
> application and I build pipeline like that redirecting pipes of the
> encoder to fdsink and fdsrc.
>
> Can anyone explain why there is a deadlock?
>
> Here's the example code I'm talking about (first is python GStreamer
> application, next is the "testprocessing" code that you might switch
> the fdsink/fdsrc to):

Here's another try, that works. It uses 2 pipelines, one for each end.
Why I can't have fdsink and fdsrc in the middle of the pipeline then?
What am I getting wrong about the pipeline functionality?

(pipe2.py)
#!/usr/bin/env python

import glib, gobject

glib.threads_init()

import pygst, gst
import sys, os
import subprocess, signal, shlex

def on_handoff(sink, buffer, pad):
    print "handoff: %d" % (len(buffer),)

def on_message(bus, message):
    print "BUS %s: %s" % (bus.get_name(), message)

if __name__ == "__main__":

#    argv = "testprocessing"

#    process = subprocess.Popen(shlex.split(argv), close_fds=True,
#                               stdin=subprocess.PIPE, stdout=subprocess.PIPE)

    pipeline = gst.Pipeline("test")
    pipeline2 = gst.Pipeline("test2")

    src = gst.element_factory_make("audiotestsrc")
    q1 = gst.element_factory_make("queue", "sink-queue")
    fsink = gst.element_factory_make("fdsink")
    fsrc = gst.element_factory_make("fdsrc")
    q2 = gst.element_factory_make("queue", "src-queue")
    sink = gst.element_factory_make("fakesink")

    sink.set_property("signal-handoffs", True)
    sink.connect("handoff", on_handoff);

    readfd, writefd = os.pipe()

    os.write(writefd, "1")
    s = os.read(readfd, 1)
    assert s == "1"
    print "================****************+++++++++++++++++", readfd, writefd

    fsink.set_property("fd", writefd)
    fsrc.set_property("fd", readfd)
#    fsink.set_property("fd", process.stdin.fileno())
#    fsrc.set_property("fd", process.stdout.fileno())

    pipeline.add(src, q1, fsink)
    pipeline2.add(fsrc, q2, sink)

    gst.element_link_many(src, q1, fsink)
    gst.element_link_many(fsrc, q2, sink)

    bus = pipeline.get_bus()
    bus.add_signal_watch()
    bus.connect("message", on_message)

    bus2 = pipeline2.get_bus()
    bus2.add_signal_watch()
    bus2.connect("message", on_message)

    mainloop = glib.MainLoop()

    pipeline.set_state(gst.STATE_PLAYING)
    pipeline2.set_state(gst.STATE_PLAYING)

    try:
        mainloop.run()
    except:
        pass

    print "Finishing"

    #process.send_signal(signal.SIGINT)
    #process.wait()

    print "Stopping pipeline"

    pipeline.set_state(gst.STATE_NULL)
    pipeline2.set_state(gst.STATE_NULL)
    del pipeline
    del pipeline2