winegstreamer: Always run gstreamer callbacks on a Wine thread.

dlls/winegstreamer/Makefile.in

@@ -4,7 +4,7 @@ EXTRAINCL = $(GSTREAMER_CFLAGS)
 EXTRALIBS = $(GSTREAMER_LIBS) $(PTHREAD_LIBS)
 
 C_SRCS = \
-	glibthread.c \
+	gst_cbs.c \
 	gstdemux.c \
 	gsttffilter.c \
 	main.c

dlls/winegstreamer/gst_cbs.c

+/*
+ * Copyright 2015 Andrew Eikum for CodeWeavers
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
+ */
+
+#include "config.h"
+
+#include <gst/app/gstappsink.h>
+#include <gst/app/gstappsrc.h>
+#include <gst/app/gstappbuffer.h>
+#include <gst/gstutils.h>
+
+#include "wine/list.h"
+
+#include "gst_cbs.h"
+
+/* gstreamer calls our callbacks from threads that Wine did not create. Some
+ * callbacks execute code which requires Wine to have created the thread
+ * (critical sections, debug logging, dshow client code). Since gstreamer can't
+ * provide an API to override its thread creation, we have to intercept all
+ * callbacks in code which avoids the Wine thread requirement, and then
+ * dispatch those callbacks on a thread that is known to be created by Wine.
+ *
+ * This file must not contain any code that depends on the Wine TEB!
+ */
+
+static void call_cb(struct cb_data *cbdata)
+{
+    pthread_mutex_init(&cbdata->lock, NULL);
+    pthread_cond_init(&cbdata->cond, NULL);
+    cbdata->finished = 0;
+
+    if(is_wine_thread()){
+        /* The thread which triggered gstreamer to call this callback may
+         * already hold a critical section. If so, executing the callback on a
+         * worker thread can cause a deadlock. If  we are already on a Wine
+         * thread, then there is no need to run this callback on a worker
+         * thread anyway, which avoids the deadlock issue. */
+        perform_cb(NULL, cbdata);
+
+        pthread_cond_destroy(&cbdata->cond);
+        pthread_mutex_destroy(&cbdata->lock);
+
+        return;
+    }
+
+    pthread_mutex_lock(&cb_list_lock);
+
+    list_add_tail(&cb_list, &cbdata->entry);
+    pthread_cond_broadcast(&cb_list_cond);
+
+    pthread_mutex_lock(&cbdata->lock);
+
+    pthread_mutex_unlock(&cb_list_lock);
+
+    while(!cbdata->finished)
+        pthread_cond_wait(&cbdata->cond, &cbdata->lock);
+
+    pthread_mutex_unlock(&cbdata->lock);
+
+    pthread_cond_destroy(&cbdata->cond);
+    pthread_mutex_destroy(&cbdata->lock);
+}
+
+GstBusSyncReply watch_bus_wrapper(GstBus *bus, GstMessage *msg, gpointer user)
+{
+    struct cb_data cbdata = { WATCH_BUS };
+
+    cbdata.u.watch_bus_data.bus = bus;
+    cbdata.u.watch_bus_data.msg = msg;
+    cbdata.u.watch_bus_data.user = user;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.watch_bus_data.ret;
+}
+
+void existing_new_pad_wrapper(GstElement *bin, GstPad *pad, gboolean last,
+        gpointer user)
+{
+    struct cb_data cbdata = { EXISTING_NEW_PAD };
+
+    cbdata.u.existing_new_pad_data.bin = bin;
+    cbdata.u.existing_new_pad_data.pad = pad;
+    cbdata.u.existing_new_pad_data.last = last;
+    cbdata.u.existing_new_pad_data.user = user;
+
+    call_cb(&cbdata);
+}
+
+gboolean check_get_range_wrapper(GstPad *pad)
+{
+    struct cb_data cbdata = { CHECK_GET_RANGE };
+
+    cbdata.u.check_get_range_data.pad = pad;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.check_get_range_data.ret;
+}
+
+gboolean query_function_wrapper(GstPad *pad, GstQuery *query)
+{
+    struct cb_data cbdata = { QUERY_FUNCTION };
+
+    cbdata.u.query_function_data.pad = pad;
+    cbdata.u.query_function_data.query = query;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.query_function_data.ret;
+}
+
+gboolean activate_push_wrapper(GstPad *pad, gboolean activate)
+{
+    struct cb_data cbdata = { ACTIVATE_PUSH };
+
+    cbdata.u.activate_push_data.pad = pad;
+    cbdata.u.activate_push_data.activate = activate;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.activate_push_data.ret;
+}
+
+void no_more_pads_wrapper(GstElement *decodebin, gpointer user)
+{
+    struct cb_data cbdata = { NO_MORE_PADS };
+
+    cbdata.u.no_more_pads_data.decodebin = decodebin;
+    cbdata.u.no_more_pads_data.user = user;
+
+    call_cb(&cbdata);
+}
+
+GstFlowReturn request_buffer_src_wrapper(GstPad *pad, guint64 ofs, guint len,
+        GstBuffer **buf)
+{
+    struct cb_data cbdata = { REQUEST_BUFFER_SRC };
+
+    cbdata.u.request_buffer_src_data.pad = pad;
+    cbdata.u.request_buffer_src_data.ofs = ofs;
+    cbdata.u.request_buffer_src_data.len = len;
+    cbdata.u.request_buffer_src_data.buf = buf;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.request_buffer_src_data.ret;
+}
+
+gboolean event_src_wrapper(GstPad *pad, GstEvent *event)
+{
+    struct cb_data cbdata = { EVENT_SRC };
+
+    cbdata.u.event_src_data.pad = pad;
+    cbdata.u.event_src_data.event = event;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.event_src_data.ret;
+}
+
+gboolean event_sink_wrapper(GstPad *pad, GstEvent *event)
+{
+    struct cb_data cbdata = { EVENT_SINK };
+
+    cbdata.u.event_sink_data.pad = pad;
+    cbdata.u.event_sink_data.event = event;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.event_sink_data.ret;
+}
+
+GstFlowReturn request_buffer_sink_wrapper(GstPad *pad, guint64 ofs, guint size,
+        GstCaps *caps, GstBuffer **buf)
+{
+    struct cb_data cbdata = { REQUEST_BUFFER_SINK };
+
+    cbdata.u.request_buffer_sink_data.pad = pad;
+    cbdata.u.request_buffer_sink_data.ofs = ofs;
+    cbdata.u.request_buffer_sink_data.size = size;
+    cbdata.u.request_buffer_sink_data.caps = caps;
+    cbdata.u.request_buffer_sink_data.buf = buf;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.request_buffer_sink_data.ret;
+}
+
+gboolean accept_caps_sink_wrapper(GstPad *pad, GstCaps *caps)
+{
+    struct cb_data cbdata = { ACCEPT_CAPS_SINK };
+
+    cbdata.u.accept_caps_sink_data.pad = pad;
+    cbdata.u.accept_caps_sink_data.caps = caps;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.accept_caps_sink_data.ret;
+}
+
+gboolean setcaps_sink_wrapper(GstPad *pad, GstCaps *caps)
+{
+    struct cb_data cbdata = { SETCAPS_SINK };
+
+    cbdata.u.setcaps_sink_data.pad = pad;
+    cbdata.u.setcaps_sink_data.caps = caps;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.setcaps_sink_data.ret;
+}
+
+GstFlowReturn got_data_sink_wrapper(GstPad *pad, GstBuffer *buf)
+{
+    struct cb_data cbdata = { GOT_DATA_SINK };
+
+    cbdata.u.got_data_sink_data.pad = pad;
+    cbdata.u.got_data_sink_data.buf = buf;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.got_data_sink_data.ret;
+}
+
+GstFlowReturn got_data_wrapper(GstPad *pad, GstBuffer *buf)
+{
+    struct cb_data cbdata = { GOT_DATA };
+
+    cbdata.u.got_data_data.pad = pad;
+    cbdata.u.got_data_data.buf = buf;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.got_data_data.ret;
+}
+
+GstFlowReturn request_buffer_wrapper(GstPad *pad, guint64 ofs, guint size,
+        GstCaps *caps, GstBuffer **buf)
+{
+    struct cb_data cbdata = { REQUEST_BUFFER };
+
+    cbdata.u.request_buffer_data.pad = pad;
+    cbdata.u.request_buffer_data.ofs = ofs;
+    cbdata.u.request_buffer_data.size = size;
+    cbdata.u.request_buffer_data.caps = caps;
+    cbdata.u.request_buffer_data.buf = buf;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.request_buffer_data.ret;
+}
+
+void removed_decoded_pad_wrapper(GstElement *bin, GstPad *pad, gpointer user)
+{
+    struct cb_data cbdata = { REMOVED_DECODED_PAD };
+
+    cbdata.u.removed_decoded_pad_data.bin = bin;
+    cbdata.u.removed_decoded_pad_data.pad = pad;
+    cbdata.u.removed_decoded_pad_data.user = user;
+
+    call_cb(&cbdata);
+}
+
+GstAutoplugSelectResult autoplug_blacklist_wrapper(GstElement *bin, GstPad *pad,
+        GstCaps *caps, GstElementFactory *fact, gpointer user)
+{
+    struct cb_data cbdata = { AUTOPLUG_BLACKLIST };
+
+    cbdata.u.autoplug_blacklist_data.bin = bin;
+    cbdata.u.autoplug_blacklist_data.pad = pad;
+    cbdata.u.autoplug_blacklist_data.caps = caps;
+    cbdata.u.autoplug_blacklist_data.fact = fact;
+    cbdata.u.autoplug_blacklist_data.user = user;
+
+    call_cb(&cbdata);
+
+    return cbdata.u.autoplug_blacklist_data.ret;
+}
+
+void unknown_type_wrapper(GstElement *bin, GstPad *pad, GstCaps *caps, gpointer user)
+{
+    struct cb_data cbdata = { UNKNOWN_TYPE };
+
+    cbdata.u.unknown_type_data.bin = bin;
+    cbdata.u.unknown_type_data.pad = pad;
+    cbdata.u.unknown_type_data.caps = caps;
+    cbdata.u.unknown_type_data.user = user;
+
+    call_cb(&cbdata);
+}
+
+void release_sample_wrapper(gpointer data)
+{
+    struct cb_data cbdata = { RELEASE_SAMPLE };
+
+    cbdata.u.release_sample_data.data = data;
+
+    call_cb(&cbdata);
+}
+
+void Gstreamer_transform_pad_added_wrapper(GstElement *filter, GstPad *pad, gpointer user)
+{
+    struct cb_data cbdata = { TRANSFORM_PAD_ADDED };
+
+    cbdata.u.transform_pad_added_data.filter = filter;
+    cbdata.u.transform_pad_added_data.pad = pad;
+    cbdata.u.transform_pad_added_data.user = user;
+
+    call_cb(&cbdata);
+}

dlls/winegstreamer/gst_cbs.h

+/*
+ * Copyright 2015 Andrew Eikum for CodeWeavers
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
+ */
+
+#ifndef GST_CBS_H
+#define GST_CBS_H
+
+#include "wine/list.h"
+#include "windef.h"
+#include <pthread.h>
+
+typedef enum {
+  GST_AUTOPLUG_SELECT_TRY,
+  GST_AUTOPLUG_SELECT_EXPOSE,
+  GST_AUTOPLUG_SELECT_SKIP
+} GstAutoplugSelectResult;
+
+enum CB_TYPE {
+    WATCH_BUS,
+    EXISTING_NEW_PAD,
+    CHECK_GET_RANGE,
+    QUERY_FUNCTION,
+    ACTIVATE_PUSH,
+    NO_MORE_PADS,
+    REQUEST_BUFFER_SRC,
+    EVENT_SRC,
+    EVENT_SINK,
+    REQUEST_BUFFER_SINK,
+    ACCEPT_CAPS_SINK,
+    SETCAPS_SINK,
+    GOT_DATA_SINK,
+    GOT_DATA,
+    REQUEST_BUFFER,
+    REMOVED_DECODED_PAD,
+    AUTOPLUG_BLACKLIST,
+    UNKNOWN_TYPE,
+    RELEASE_SAMPLE,
+    TRANSFORM_PAD_ADDED
+};
+
+struct cb_data {
+    enum CB_TYPE type;
+    union {
+        struct watch_bus_data {
+            GstBus *bus;
+            GstMessage *msg;
+            gpointer user;
+            GstBusSyncReply ret;
+        } watch_bus_data;
+        struct existing_new_pad_data {
+            GstElement *bin;
+            GstPad *pad;
+            gboolean last;
+            gpointer user;
+        } existing_new_pad_data;
+        struct check_get_range_data {
+            GstPad *pad;
+            gboolean ret;
+        } check_get_range_data;
+        struct query_function_data {
+            GstPad *pad;
+            GstQuery *query;
+            gboolean ret;
+        } query_function_data;
+        struct activate_push_data {
+            GstPad *pad;
+            gboolean activate;
+            gboolean ret;
+        } activate_push_data;
+        struct no_more_pads_data {
+            GstElement *decodebin;
+            gpointer user;
+        } no_more_pads_data;
+        struct request_buffer_src_data {
+            GstPad *pad;
+            guint64 ofs;
+            guint len;
+            GstBuffer **buf;
+            GstFlowReturn ret;
+        } request_buffer_src_data;
+        struct event_src_data {
+            GstPad *pad;
+            GstEvent *event;
+            gboolean ret;
+        } event_src_data;
+        struct event_sink_data {
+            GstPad *pad;
+            GstEvent *event;
+            gboolean ret;
+        } event_sink_data;
+        struct request_buffer_sink_data {
+            GstPad *pad;
+            guint64 ofs;
+            guint size;
+            GstCaps *caps;
+            GstBuffer **buf;
+            GstFlowReturn ret;
+        } request_buffer_sink_data;
+        struct accept_caps_sink_data {
+            GstPad *pad;
+            GstCaps *caps;
+            gboolean ret;
+        } accept_caps_sink_data;
+        struct setcaps_sink_data {
+            GstPad *pad;
+            GstCaps *caps;
+            gboolean ret;
+        } setcaps_sink_data;
+        struct got_data_sink_data {
+            GstPad *pad;
+            GstBuffer *buf;
+            GstFlowReturn ret;
+        } got_data_sink_data;
+        struct got_data_data {
+            GstPad *pad;
+            GstBuffer *buf;
+            GstFlowReturn ret;
+        } got_data_data;
+        struct request_buffer_data {
+            GstPad *pad;
+            guint64 ofs;
+            guint size;
+            GstCaps *caps;
+            GstBuffer **buf;
+            GstFlowReturn ret;
+        } request_buffer_data;
+        struct removed_decoded_pad_data {
+            GstElement *bin;
+            GstPad *pad;
+            gpointer user;
+        } removed_decoded_pad_data;
+        struct autoplug_blacklist_data {
+            GstElement *bin;
+            GstPad *pad;
+            GstCaps *caps;
+            GstElementFactory *fact;
+            gpointer user;
+            GstAutoplugSelectResult ret;
+        } autoplug_blacklist_data;
+        struct unknown_type_data {
+            GstElement *bin;
+            GstPad *pad;
+            GstCaps *caps;
+            gpointer user;
+        } unknown_type_data;
+        struct release_sample_data {
+            gpointer data;
+        } release_sample_data;
+        struct transform_pad_added_data {
+            GstElement *filter;
+            GstPad *pad;
+            gpointer user;
+        } transform_pad_added_data;
+    } u;
+
+    int finished;
+    pthread_mutex_t lock;
+    pthread_cond_t cond;
+    struct list entry;
+};
+
+extern pthread_mutex_t cb_list_lock DECLSPEC_HIDDEN;
+extern pthread_cond_t cb_list_cond DECLSPEC_HIDDEN;
+extern struct list cb_list DECLSPEC_HIDDEN;
+void CALLBACK perform_cb(TP_CALLBACK_INSTANCE *instance, void *user) DECLSPEC_HIDDEN;
+BOOL is_wine_thread(void) DECLSPEC_HIDDEN;
+void mark_wine_thread(void) DECLSPEC_HIDDEN;
+
+GstBusSyncReply watch_bus_wrapper(GstBus *bus, GstMessage *msg, gpointer user) DECLSPEC_HIDDEN;
+void existing_new_pad_wrapper(GstElement *bin, GstPad *pad, gboolean last, gpointer user) DECLSPEC_HIDDEN;
+gboolean check_get_range_wrapper(GstPad *pad) DECLSPEC_HIDDEN;
+gboolean query_function_wrapper(GstPad *pad, GstQuery *query) DECLSPEC_HIDDEN;
+gboolean activate_push_wrapper(GstPad *pad, gboolean activate) DECLSPEC_HIDDEN;
+void no_more_pads_wrapper(GstElement *decodebin, gpointer user) DECLSPEC_HIDDEN;
+GstFlowReturn request_buffer_src_wrapper(GstPad *pad, guint64 ofs, guint len, GstBuffer **buf) DECLSPEC_HIDDEN;
+gboolean event_src_wrapper(GstPad *pad, GstEvent *event) DECLSPEC_HIDDEN;
+gboolean event_sink_wrapper(GstPad *pad, GstEvent *event) DECLSPEC_HIDDEN;
+GstFlowReturn request_buffer_sink_wrapper(GstPad *pad, guint64 ofs, guint size, GstCaps *caps, GstBuffer **buf) DECLSPEC_HIDDEN;
+gboolean accept_caps_sink_wrapper(GstPad *pad, GstCaps *caps) DECLSPEC_HIDDEN;
+gboolean setcaps_sink_wrapper(GstPad *pad, GstCaps *caps) DECLSPEC_HIDDEN;
+GstFlowReturn got_data_sink_wrapper(GstPad *pad, GstBuffer *buf) DECLSPEC_HIDDEN;
+GstFlowReturn got_data_wrapper(GstPad *pad, GstBuffer *buf) DECLSPEC_HIDDEN;
+GstFlowReturn request_buffer_wrapper(GstPad *pad, guint64 ofs, guint size, GstCaps *caps, GstBuffer **buf) DECLSPEC_HIDDEN;
+void removed_decoded_pad_wrapper(GstElement *bin, GstPad *pad, gpointer user) DECLSPEC_HIDDEN;
+GstAutoplugSelectResult autoplug_blacklist_wrapper(GstElement *bin, GstPad *pad, GstCaps *caps, GstElementFactory *fact, gpointer user) DECLSPEC_HIDDEN;
+void unknown_type_wrapper(GstElement *bin, GstPad *pad, GstCaps *caps, gpointer user) DECLSPEC_HIDDEN;
+void release_sample_wrapper(gpointer data) DECLSPEC_HIDDEN;
+void Gstreamer_transform_pad_added_wrapper(GstElement *filter, GstPad *pad, gpointer user) DECLSPEC_HIDDEN;
+
+#endif

dlls/winegstreamer/gst_private.h

@@ -42,6 +42,12 @@ IUnknown * CALLBACK Gstreamer_Mp3_create(IUnknown *pUnkOuter, HRESULT *phr);
 IUnknown * CALLBACK Gstreamer_YUV_create(IUnknown *pUnkOuter, HRESULT *phr);
 IUnknown * CALLBACK Gstreamer_Splitter_create(IUnknown *pUnkOuter, HRESULT *phr);
 
-void g_thread_impl_init(void);
 DWORD Gstreamer_init(void);
+
+GstFlowReturn got_data(GstPad *pad, GstBuffer *buf) DECLSPEC_HIDDEN;
+GstFlowReturn request_buffer(GstPad *pad, guint64 ofs, guint size, GstCaps *caps, GstBuffer **buf) DECLSPEC_HIDDEN;
+void Gstreamer_transform_pad_added(GstElement *filter, GstPad *pad, gpointer user) DECLSPEC_HIDDEN;
+
+void start_dispatch_thread(void) DECLSPEC_HIDDEN;
+
 #endif /* __GST_PRIVATE_INCLUDED__ */

dlls/winegstreamer/gsttffilter.c

@@ -28,6 +28,7 @@
 
 #include "gst_private.h"
 #include "gst_guids.h"
+#include "gst_cbs.h"
 
 #include "uuids.h"
 #include "mmreg.h"
@@ -125,6 +126,8 @@ static HRESULT WINAPI Gstreamer_transform_ProcessBegin(TransformFilter *iface) {
     GstTfImpl *This = (GstTfImpl*)iface;
     int ret;
 
+    mark_wine_thread();
+
     ret = gst_element_set_state(This->filter, GST_STATE_PLAYING);
     TRACE("Returned: %i\n", ret);
     return S_OK;
@@ -146,12 +149,7 @@ static HRESULT WINAPI Gstreamer_transform_DecideBufferSize(TransformFilter *tf,
     return IMemAllocator_SetProperties(pAlloc, ppropInputRequest, &actual);
 }
 
-static void release_sample(void *data) {
-    TRACE("Releasing %p\n", data);
-    IMediaSample_Release((IMediaSample *)data);
-}
-
-static GstFlowReturn got_data(GstPad *pad, GstBuffer *buf) {
+GstFlowReturn got_data(GstPad *pad, GstBuffer *buf) {
     GstTfImpl *This = gst_pad_get_element_private(pad);
     IMediaSample *sample = GST_APP_BUFFER(buf)->priv;
     REFERENCE_TIME tStart, tStop;
@@ -188,7 +186,7 @@ static GstFlowReturn got_data(GstPad *pad, GstBuffer *buf) {
     return GST_FLOW_OK;
 }
 
-static GstFlowReturn request_buffer(GstPad *pad, guint64 ofs, guint size, GstCaps *caps, GstBuffer **buf) {
+GstFlowReturn request_buffer(GstPad *pad, guint64 ofs, guint size, GstCaps *caps, GstBuffer **buf) {
     GstTfImpl *This = gst_pad_get_element_private(pad);
     IMediaSample *sample;
     BYTE *ptr;
@@ -202,7 +200,7 @@ static GstFlowReturn request_buffer(GstPad *pad, guint64 ofs, guint size, GstCap
     }
     IMediaSample_SetActualDataLength(sample, size);
     IMediaSample_GetPointer(sample, &ptr);
-    *buf = gst_app_buffer_new(ptr, size, release_sample, sample);
+    *buf = gst_app_buffer_new(ptr, size, release_sample_wrapper, sample);
 
     if (!*buf) {
         IMediaSample_Release(sample);
@@ -224,9 +222,11 @@ static HRESULT WINAPI Gstreamer_transform_ProcessData(TransformFilter *iface, IM
     int ret;
     TRACE("Reading %p\n", sample);
 
+    mark_wine_thread();
+
     EnterCriticalSection(&This->tf.csReceive);
     IMediaSample_GetPointer(sample, &data);
-    buf = gst_app_buffer_new(data, IMediaSample_GetActualDataLength(sample), release_sample, sample);
+    buf = gst_app_buffer_new(data, IMediaSample_GetActualDataLength(sample), release_sample_wrapper, sample);
     if (!buf) {
         LeaveCriticalSection(&This->tf.csReceive);
         return S_OK;
@@ -267,6 +267,8 @@ static HRESULT WINAPI Gstreamer_transform_ProcessEnd(TransformFilter *iface) {
     GstTfImpl *This = (GstTfImpl*)iface;
     int ret;
 
+    mark_wine_thread();
+
     LeaveCriticalSection(&This->tf.csReceive);
     ret = gst_element_set_state(This->filter, GST_STATE_READY);
     EnterCriticalSection(&This->tf.csReceive);
@@ -274,8 +276,9 @@ static HRESULT WINAPI Gstreamer_transform_ProcessEnd(TransformFilter *iface) {
     return S_OK;
 }
 
-static void Gstreamer_transform_pad_added(GstElement *filter, GstPad *pad, GstTfImpl *This)
+void Gstreamer_transform_pad_added(GstElement *filter, GstPad *pad, gpointer user)
 {
+    GstTfImpl *This = (GstTfImpl*)user;
     int ret;
     if (!GST_PAD_IS_SRC(pad))
         return;
@@ -294,6 +297,8 @@ static HRESULT Gstreamer_transform_ConnectInput(GstTfImpl *This, const AM_MEDIA_
     BOOL done = FALSE, found = FALSE;
     int ret;
 
+    mark_wine_thread();
+
     This->filter = gst_element_factory_make(This->gstreamer_name, NULL);
     if (!This->filter) {
         FIXME("Could not make %s filter\n", This->gstreamer_name);
@@ -303,8 +308,8 @@ static HRESULT Gstreamer_transform_ConnectInput(GstTfImpl *This, const AM_MEDIA_
     gst_pad_set_element_private (This->my_src, This);
 
     This->my_sink = gst_pad_new(NULL, GST_PAD_SINK);
-    gst_pad_set_chain_function(This->my_sink, got_data);
-    gst_pad_set_bufferalloc_function(This->my_sink, request_buffer);
+    gst_pad_set_chain_function(This->my_sink, got_data_wrapper);
+    gst_pad_set_bufferalloc_function(This->my_sink, request_buffer_wrapper);
     gst_pad_set_element_private (This->my_sink, This);
 
     ret = gst_pad_set_caps(This->my_src, capsin);
@@ -362,7 +367,7 @@ static HRESULT Gstreamer_transform_ConnectInput(GstTfImpl *This, const AM_MEDIA_
     gst_iterator_free(it);
     found = !!This->their_src;
     if (!found)
-        g_signal_connect(This->filter, "pad-added", G_CALLBACK(Gstreamer_transform_pad_added), This);
+        g_signal_connect(This->filter, "pad-added", G_CALLBACK(Gstreamer_transform_pad_added_wrapper), This);
     ret = gst_pad_link(This->my_src, This->their_sink);
     if (ret < 0) {
         WARN("Failed to link with %i\n", ret);
@@ -382,6 +387,8 @@ static HRESULT Gstreamer_transform_ConnectInput(GstTfImpl *This, const AM_MEDIA_
 static HRESULT WINAPI Gstreamer_transform_Cleanup(TransformFilter *tf, PIN_DIRECTION dir) {
     GstTfImpl *This = (GstTfImpl*)tf;
 
+    mark_wine_thread();
+
     if (dir == PINDIR_INPUT)
     {
         if (This->filter) {
@@ -405,6 +412,7 @@ static HRESULT WINAPI Gstreamer_transform_Cleanup(TransformFilter *tf, PIN_DIREC
 static HRESULT WINAPI Gstreamer_transform_EndOfStream(TransformFilter *iface) {
     GstTfImpl *This = (GstTfImpl*)iface;
     TRACE("%p\n", This);
+    mark_wine_thread();
 
     gst_pad_push_event(This->my_src, gst_event_new_eos());
     return S_OK;
@@ -413,6 +421,7 @@ static HRESULT WINAPI Gstreamer_transform_EndOfStream(TransformFilter *iface) {
 static HRESULT WINAPI Gstreamer_transform_BeginFlush(TransformFilter *iface) {
     GstTfImpl *This = (GstTfImpl*)iface;
     TRACE("%p\n", This);
+    mark_wine_thread();
 
     gst_pad_push_event(This->my_src, gst_event_new_flush_start());
     return S_OK;
@@ -421,6 +430,7 @@ static HRESULT WINAPI Gstreamer_transform_BeginFlush(TransformFilter *iface) {
 static HRESULT WINAPI Gstreamer_transform_EndFlush(TransformFilter *iface) {
     GstTfImpl *This = (GstTfImpl*)iface;
     TRACE("%p\n", This);
+    mark_wine_thread();
 
     gst_pad_push_event(This->my_src, gst_event_new_flush_stop());
     return S_OK;
@@ -429,6 +439,7 @@ static HRESULT WINAPI Gstreamer_transform_EndFlush(TransformFilter *iface) {
 static HRESULT WINAPI Gstreamer_transform_NewSegment(TransformFilter *iface, REFERENCE_TIME tStart, REFERENCE_TIME tStop, double dRate) {
     GstTfImpl *This = (GstTfImpl*)iface;
     TRACE("%p\n", This);
+    mark_wine_thread();
 
     gst_pad_push_event(This->my_src, gst_event_new_new_segment_full(1,
                        1.0, dRate, GST_FORMAT_TIME, 0, tStop <= tStart ? -1 : tStop * 100, tStart*100));
@@ -438,6 +449,7 @@ static HRESULT WINAPI Gstreamer_transform_NewSegment(TransformFilter *iface, REF
 static HRESULT WINAPI Gstreamer_transform_QOS(TransformFilter *iface, IBaseFilter *sender, Quality qm) {
     GstTfImpl *This = (GstTfImpl*)iface;
     REFERENCE_TIME late = qm.Late;
+    mark_wine_thread();
     if (qm.Late < 0 && -qm.Late > qm.TimeStamp)
         late = -qm.TimeStamp;
     gst_pad_push_event(This->my_sink, gst_event_new_qos(1000. / qm.Proportion, late * 100, qm.TimeStamp * 100));
@@ -480,6 +492,8 @@ static HRESULT WINAPI Gstreamer_Mp3_SetMediaType(TransformFilter *tf, PIN_DIRECT
     HRESULT hr;
     int layer;
 
+    mark_wine_thread();
+
     if (dir != PINDIR_INPUT)
         return S_OK;
 
@@ -572,6 +586,7 @@ IUnknown * CALLBACK Gstreamer_Mp3_create(IUnknown *punkout, HRESULT *phr)
         *phr = E_FAIL;
         return NULL;
     }
+    mark_wine_thread();
     plugin = Gstreamer_FindMatch("audio/mpeg, mpegversion=(int) 1");
     if (!plugin)
     {
@@ -620,6 +635,8 @@ static HRESULT WINAPI Gstreamer_YUV_SetMediaType(TransformFilter *tf, PIN_DIRECT
     int avgtime;
     LONG width, height;
 
+    mark_wine_thread();
+
     if (dir != PINDIR_INPUT)
         return S_OK;
 
@@ -737,6 +754,8 @@ static HRESULT WINAPI Gstreamer_AudioConvert_SetMediaType(TransformFilter *tf, P
     BOOL inisfloat = FALSE;
     int indepth;
 
+    mark_wine_thread();
+
     if (dir != PINDIR_INPUT)
         return S_OK;
 

dlls/winegstreamer/main.c

@@ -254,7 +254,6 @@ DWORD Gstreamer_init(void) {
         argv[0] = argv0;
         argv[1] = argv1;
         argv[2] = NULL;
-        g_thread_impl_init();
         inited = gst_init_check(&argc, &argv, &err);
         HeapFree(GetProcessHeap(), 0, argv);
         if (err) {
@@ -269,6 +268,8 @@ DWORD Gstreamer_init(void) {
                                (LPCWSTR)hInst, &newhandle);
             if (!newhandle)
                 ERR("Could not pin module %p\n", hInst);
+
+            start_dispatch_thread();
         }
     }
     return inited;