[PATCH v2 0/3] MR4479: winegstreamer: Implement wg_muxer_finalize.

[Ziqing Hui (＠zhui)]

--
  v2: winegstreamer: Implement wg_muxer_finalize.
      winestreamer: Implement {Begin,End}Finalize for media sink.
      winegstreamer: Avoid command leaking in media_sink_queue_command.

https://gitlab.winehq.org/wine/wine/-/merge_requests/4479

[PATCH v2 1/3] winegstreamer: Avoid command leaking in media_sink_queue_command.

[Ziqing Hui]

From: Ziqing Hui

---
 dlls/winegstreamer/media_sink.c | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/dlls/winegstreamer/media_sink.c b/dlls/winegstreamer/media_sink.c
index 6996d689e83..ab383d670b3 100644
--- a/dlls/winegstreamer/media_sink.c
+++ b/dlls/winegstreamer/media_sink.c
@@ -525,7 +525,11 @@ static HRESULT media_sink_queue_command(struct media_sink *media_sink, enum asyn
     if (FAILED(hr = async_command_create(op, &command)))
         return hr;

-    return MFPutWorkItem(MFASYNC_CALLBACK_QUEUE_STANDARD, &media_sink->async_callback, &command->IUnknown_iface);
+    if (FAILED(hr = MFPutWorkItem(MFASYNC_CALLBACK_QUEUE_STANDARD,
+            &media_sink->async_callback, &command->IUnknown_iface)))
+        IUnknown_Release(&command->IUnknown_iface);
+
+    return hr;
 }

static HRESULT media_sink_queue_stream_event(struct media_sink *media_sink, MediaEventType type)

[PATCH v2 2/3] winestreamer: Implement {Begin,End}Finalize for media sink.

[Ziqing Hui]

From: Ziqing Hui

---
 dlls/winegstreamer/gst_private.h  |  1 +
 dlls/winegstreamer/main.c         | 15 +++++++
 dlls/winegstreamer/media_sink.c   | 72 +++++++++++++++++++++++++++++--
 dlls/winegstreamer/unix_private.h |  1 +
 dlls/winegstreamer/unixlib.h      |  1 +
 dlls/winegstreamer/wg_muxer.c     |  6 +++
 dlls/winegstreamer/wg_parser.c    |  2 +
 7 files changed, 94 insertions(+), 4 deletions(-)

diff --git a/dlls/winegstreamer/gst_private.h b/dlls/winegstreamer/gst_private.h
index 26873edb74b..a7f2b8b1e6b 100644
--- a/dlls/winegstreamer/gst_private.h
+++ b/dlls/winegstreamer/gst_private.h
@@ -116,6 +116,7 @@ HRESULT wg_muxer_add_stream(wg_muxer_t muxer, UINT32 stream_id, const struct wg_
 HRESULT wg_muxer_start(wg_muxer_t muxer);
 HRESULT wg_muxer_push_sample(wg_muxer_t muxer, struct wg_sample *sample, UINT32 stream_id);
 HRESULT wg_muxer_read_data(wg_muxer_t muxer, void *buffer, UINT32 *size, UINT64 *offset);
+HRESULT wg_muxer_finalize(wg_muxer_t muxer);

unsigned int wg_format_get_bytes_for_uncompressed(wg_video_format format, unsigned int width, unsigned int height);
 unsigned int wg_format_get_max_size(const struct wg_format *format);
diff --git a/dlls/winegstreamer/main.c b/dlls/winegstreamer/main.c
index 900905be0d9..d264b9a77e0 100644
--- a/dlls/winegstreamer/main.c
+++ b/dlls/winegstreamer/main.c
@@ -567,6 +567,21 @@ HRESULT wg_muxer_read_data(wg_muxer_t muxer, void *buffer, UINT32 *size, UINT64
     return HRESULT_FROM_NT(status);
 }

+HRESULT wg_muxer_finalize(wg_muxer_t muxer)
+{
+    NTSTATUS status;
+
+    TRACE("muxer %#I64x.\n", muxer);
+
+    if ((status = WINE_UNIX_CALL(unix_wg_muxer_finalize, &muxer)))
+    {
+        WARN("Failed to finalize, status %#lx.\n", status);
+        return HRESULT_FROM_NT(status);
+    }
+
+    return S_OK;
+}
+
 #define ALIGN(n, alignment) (((n) + (alignment) - 1) & ~((alignment) - 1))

unsigned int wg_format_get_stride(const struct wg_format *format)
diff --git a/dlls/winegstreamer/media_sink.c b/dlls/winegstreamer/media_sink.c
index ab383d670b3..c3a21053c38 100644
--- a/dlls/winegstreamer/media_sink.c
+++ b/dlls/winegstreamer/media_sink.c
@@ -32,6 +32,7 @@ enum async_op
     ASYNC_STOP,
     ASYNC_PAUSE,
     ASYNC_PROCESS,
+    ASYNC_FINALIZE,
 };

struct async_command
@@ -48,6 +49,10 @@ struct async_command
             IMFSample *sample;
             UINT32 stream_id;
         } process;
+        struct
+        {
+            IMFAsyncResult *result;
+        } finalize;
     } u;
 };

@@ -80,6 +85,7 @@ struct media_sink
         STATE_STARTED,
         STATE_STOPPED,
         STATE_PAUSED,
+        STATE_FINALIZED,
         STATE_SHUTDOWN,
     } state;

@@ -155,6 +161,8 @@ static ULONG WINAPI async_command_Release(IUnknown *iface)
     {
         if (command->op == ASYNC_PROCESS && command->u.process.sample)
             IMFSample_Release(command->u.process.sample);
+        else if (command->op == ASYNC_FINALIZE && command->u.finalize.result)
+            IMFAsyncResult_Release(command->u.finalize.result);
         free(command);
     }

@@ -630,6 +638,51 @@ static HRESULT media_sink_process(struct media_sink *media_sink, IMFSample *samp
     return hr;
 }

+static HRESULT media_sink_begin_finalize(struct media_sink *media_sink, IMFAsyncCallback *callback, IUnknown *state)
+{
+    struct async_command *command;
+    IMFAsyncResult *result;
+    HRESULT hr;
+
+    if (media_sink->state == STATE_SHUTDOWN)
+        return MF_E_SHUTDOWN;
+    if (!callback)
+        return S_OK;
+
+    if (FAILED(hr = async_command_create(ASYNC_FINALIZE, &command)))
+        return hr;
+
+    if (FAILED(hr = MFCreateAsyncResult(NULL, callback, state, &result)))
+    {
+        free(command);
+        return hr;
+    }
+    IMFAsyncResult_AddRef((command->u.finalize.result = result));
+
+    if (FAILED(hr = MFPutWorkItem(MFASYNC_CALLBACK_QUEUE_STANDARD,
+            &media_sink->async_callback, &command->IUnknown_iface)))
+        IUnknown_Release(&command->IUnknown_iface);
+
+    return hr;
+}
+
+static HRESULT media_sink_finalize(struct media_sink *media_sink, IMFAsyncResult *result)
+{
+    HRESULT hr;
+
+    media_sink->state = STATE_FINALIZED;
+
+    hr = wg_muxer_finalize(media_sink->muxer);
+
+    if (SUCCEEDED(hr))
+        hr = media_sink_write_stream(media_sink);
+
+    IMFAsyncResult_SetStatus(result, hr);
+    MFInvokeCallback(result);
+
+    return hr;
+}
+
 static HRESULT WINAPI media_sink_QueryInterface(IMFFinalizableMediaSink *iface, REFIID riid, void **obj)
 {
     struct media_sink *media_sink = impl_from_IMFFinalizableMediaSink(iface);
@@ -869,16 +922,23 @@ static HRESULT WINAPI media_sink_Shutdown(IMFFinalizableMediaSink *iface)

static HRESULT WINAPI media_sink_BeginFinalize(IMFFinalizableMediaSink *iface, IMFAsyncCallback *callback, IUnknown *state)
 {
-    FIXME("iface %p, callback %p, state %p stub!\n", iface, callback, state);
+    struct media_sink *media_sink = impl_from_IMFFinalizableMediaSink(iface);
+    HRESULT hr;

-    return E_NOTIMPL;
+    TRACE("iface %p, callback %p, state %p.\n", iface, callback, state);
+
+    EnterCriticalSection(&media_sink->cs);
+    hr =  media_sink_begin_finalize(media_sink, callback, state);
+    LeaveCriticalSection(&media_sink->cs);
+
+    return hr;
 }

static HRESULT WINAPI media_sink_EndFinalize(IMFFinalizableMediaSink *iface, IMFAsyncResult *result)
 {
-    FIXME("iface %p, result %p stub!\n", iface, result);
+    TRACE("iface %p, result %p.\n", iface, result);

-    return E_NOTIMPL;
+    return result ? IMFAsyncResult_GetStatus(result) : E_INVALIDARG;
 }

static const IMFFinalizableMediaSinkVtbl media_sink_vtbl =
@@ -1135,6 +1195,10 @@ static HRESULT WINAPI media_sink_callback_Invoke(IMFAsyncCallback *iface, IMFAsy
             if (FAILED(hr = media_sink_process(media_sink, command->u.process.sample, command->u.process.stream_id)))
                 WARN("Failed to process sample, hr %#lx.\n", hr);
             break;
+        case ASYNC_FINALIZE:
+            if (FAILED(hr = media_sink_finalize(media_sink, command->u.finalize.result)))
+                WARN("Failed to finalize, hr %#lx.\n", hr);
+            break;
         default:
             WARN("Unsupported op %u.\n", command->op);
             break;
diff --git a/dlls/winegstreamer/unix_private.h b/dlls/winegstreamer/unix_private.h
index 070e1285b38..5162c8d5863 100644
--- a/dlls/winegstreamer/unix_private.h
+++ b/dlls/winegstreamer/unix_private.h
@@ -71,6 +71,7 @@ extern NTSTATUS wg_muxer_add_stream(void *args) DECLSPEC_HIDDEN;
 extern NTSTATUS wg_muxer_start(void *args) DECLSPEC_HIDDEN;
 extern NTSTATUS wg_muxer_push_sample(void *args) DECLSPEC_HIDDEN;
 extern NTSTATUS wg_muxer_read_data(void *args) DECLSPEC_HIDDEN;
+extern NTSTATUS wg_muxer_finalize(void *args) DECLSPEC_HIDDEN;

/* wg_allocator.c */

diff --git a/dlls/winegstreamer/unixlib.h b/dlls/winegstreamer/unixlib.h
index 7a2a4a8da2f..86bd380c351 100644
--- a/dlls/winegstreamer/unixlib.h
+++ b/dlls/winegstreamer/unixlib.h
@@ -446,6 +446,7 @@ enum unix_funcs
     unix_wg_muxer_start,
     unix_wg_muxer_push_sample,
     unix_wg_muxer_read_data,
+    unix_wg_muxer_finalize,

unix_wg_funcs_count,
 };
diff --git a/dlls/winegstreamer/wg_muxer.c b/dlls/winegstreamer/wg_muxer.c
index c0614807459..bbe4d4d1f9d 100644
--- a/dlls/winegstreamer/wg_muxer.c
+++ b/dlls/winegstreamer/wg_muxer.c
@@ -523,3 +523,9 @@ NTSTATUS wg_muxer_read_data(void *args)

return STATUS_SUCCESS;
 }
+
+NTSTATUS wg_muxer_finalize(void *args)
+{
+    GST_FIXME("Not implemented.");
+    return STATUS_NOT_IMPLEMENTED;
+}
diff --git a/dlls/winegstreamer/wg_parser.c b/dlls/winegstreamer/wg_parser.c
index 84bb65dbccd..b73dde18d55 100644
--- a/dlls/winegstreamer/wg_parser.c
+++ b/dlls/winegstreamer/wg_parser.c
@@ -1922,6 +1922,7 @@ const unixlib_entry_t __wine_unix_call_funcs[] =
     X(wg_muxer_start),
     X(wg_muxer_push_sample),
     X(wg_muxer_read_data),
+    X(wg_muxer_finalize),
 };

C_ASSERT(ARRAYSIZE(__wine_unix_call_funcs) == unix_wg_funcs_count);
@@ -2253,6 +2254,7 @@ const unixlib_entry_t __wine_unix_call_wow64_funcs[] =
     X(wg_muxer_start),
     X64(wg_muxer_push_sample),
     X64(wg_muxer_read_data),
+    X(wg_muxer_finalize),
 };

C_ASSERT(ARRAYSIZE(__wine_unix_call_wow64_funcs) == unix_wg_funcs_count);

[PATCH v2 3/3] winegstreamer: Implement wg_muxer_finalize.

[Ziqing Hui]

From: Ziqing Hui

---
 dlls/winegstreamer/wg_muxer.c | 33 +++++++++++++++++++++++++++++++--
 1 file changed, 31 insertions(+), 2 deletions(-)

diff --git a/dlls/winegstreamer/wg_muxer.c b/dlls/winegstreamer/wg_muxer.c
index bbe4d4d1f9d..0055d9ef16c 100644
--- a/dlls/winegstreamer/wg_muxer.c
+++ b/dlls/winegstreamer/wg_muxer.c
@@ -63,6 +63,9 @@ struct wg_muxer
     GstBuffer *buffer;

pthread_mutex_t mutex;
+    pthread_cond_t cond;
+
+    bool eos;
     guint64 offset; /* Write offset of the output buffer generated by muxer. */

struct list streams;
@@ -175,6 +178,13 @@ static gboolean muxer_sink_event_cb(GstPad *pad, GstObject *parent, GstEvent *ev

switch (event->type)
     {
+        case GST_EVENT_EOS:
+            pthread_mutex_lock(&muxer->mutex);
+            muxer->eos = true;
+            pthread_mutex_unlock(&muxer->mutex);
+            pthread_cond_signal(&muxer->cond);
+            break;
+
         case GST_EVENT_SEGMENT:
             pthread_mutex_lock(&muxer->mutex);

@@ -248,6 +258,7 @@ NTSTATUS wg_muxer_create(void *args)
     list_init(&muxer->streams);
     muxer->offset = GST_BUFFER_OFFSET_NONE;
     pthread_mutex_init(&muxer->mutex, NULL);
+    pthread_cond_init(&muxer->cond, NULL);
     if (!(muxer->container = gst_bin_new("wg_muxer")))
         goto out;
     if (!(muxer->output_queue = gst_atomic_queue_new(8)))
@@ -287,6 +298,7 @@ out:
         gst_atomic_queue_unref(muxer->output_queue);
     if (muxer->container)
         gst_object_unref(muxer->container);
+    pthread_cond_destroy(&muxer->cond);
     pthread_mutex_destroy(&muxer->mutex);
     free(muxer);

@@ -317,6 +329,7 @@ NTSTATUS wg_muxer_destroy(void *args)
     gst_element_set_state(muxer->container, GST_STATE_NULL);
     gst_object_unref(muxer->container);

+    pthread_cond_destroy(&muxer->cond);
     pthread_mutex_destroy(&muxer->mutex);

free(muxer);
@@ -526,6 +539,22 @@ NTSTATUS wg_muxer_read_data(void *args)

NTSTATUS wg_muxer_finalize(void *args)
 {
-    GST_FIXME("Not implemented.");
-    return STATUS_NOT_IMPLEMENTED;
+    struct wg_muxer *muxer = get_muxer(*(wg_muxer_t *)args);
+    struct wg_muxer_stream *stream;
+
+    /* Notify each stream of EOS. */
+    LIST_FOR_EACH_ENTRY(stream, &muxer->streams, struct wg_muxer_stream, entry)
+    {
+        if (!push_event(stream->my_src, gst_event_new_segment_done(GST_FORMAT_BYTES, -1))
+                || !push_event(stream->my_src, gst_event_new_eos()))
+            return STATUS_UNSUCCESSFUL;
+    }
+
+    /* Wait for muxer EOS. */
+    pthread_mutex_lock(&muxer->mutex);
+    while (!muxer->eos)
+        pthread_cond_wait(&muxer->cond, &muxer->mutex);
+    pthread_mutex_unlock(&muxer->mutex);
+
+    return STATUS_SUCCESS;
 }

[Ziqing Hui (＠zhui)]

Introducing a helper begin_finalize() function avoids duplicating LeaveCriticalSection() calls.

[Rémi Bernon (＠rbernon)]

Rémi Bernon (@rbernon) commented about dlls/winegstreamer/media_sink.c:
>  if (FAILED(hr = async_command_create(op, &command)))
>      return hr;
>
> - return MFPutWorkItem(MFASYNC_CALLBACK_QUEUE_STANDARD, &media_sink->async_callback, &command->IUnknown_iface);
> - if (FAILED(hr = MFPutWorkItem(MFASYNC_CALLBACK_QUEUE_STANDARD,
> -
>        &media_sink->async_callback, &command->IUnknown_iface)))
> -
>    IUnknown_Release(&command->IUnknown_iface);
> -
> - return hr;

Eh sorry, actually it's not only when it fails, it's always leaking it. This should be:

```suggestion:-4+0
    hr = MFPutWorkItem(MFASYNC_CALLBACK_QUEUE_STANDARD,
            &media_sink->async_callback, &command->IUnknown_iface));
    IUnknown_Release(&command->IUnknown_iface);

return hr;
```

[Rémi Bernon (＠rbernon)]

Rémi Bernon (@rbernon) commented about dlls/winegstreamer/media_sink.c:
> -
> - if (FAILED(hr = async_command_create(ASYNC_FINALIZE, &command)))
> -
>    return hr;
> -
> - if (FAILED(hr = MFCreateAsyncResult(NULL, callback, state, &result)))
> - {
> -
>    free(command);
> -
>    return hr;
> - }
> - IMFAsyncResult_AddRef((command->u.finalize.result = result));
> -
> - if (FAILED(hr = MFPutWorkItem(MFASYNC_CALLBACK_QUEUE_STANDARD,
> -
>        &media_sink->async_callback, &command->IUnknown_iface)))
> -
>    IUnknown_Release(&command->IUnknown_iface);
> -
> - return hr;

```suggestion:-14+0
    if (FAILED(hr = async_command_create(ASYNC_FINALIZE, &command)))
        return hr;

if (FAILED(hr = MFCreateAsyncResult(NULL, callback, state, &result)))
    {
        IUnknown_Release(&command->IUnknown_iface);
        return hr;
    }
    IMFAsyncResult_AddRef((command->u.finalize.result = result));

hr = MFPutWorkItem(MFASYNC_CALLBACK_QUEUE_STANDARD,
            &media_sink->async_callback, &command->IUnknown_iface));
    IUnknown_Release(&command->IUnknown_iface);

return hr;
```