[PATCH 0/4] MR676: mfreadwrite/writer: Add sample/marker queue for each stream.
Signed-off-by: Nikolay Sivov <nsivov(a)codeweavers.com> -- https://gitlab.winehq.org/wine/wine/-/merge_requests/676
From: Nikolay Sivov <nsivov(a)codeweavers.com> Signed-off-by: Nikolay Sivov <nsivov(a)codeweavers.com> --- dlls/mfreadwrite/writer.c | 298 ++++++++++++++++++++++++++++++++++++-- 1 file changed, 289 insertions(+), 9 deletions(-) diff --git a/dlls/mfreadwrite/writer.c b/dlls/mfreadwrite/writer.c index ed05b0d3191..a157bccd785 100644 --- a/dlls/mfreadwrite/writer.c +++ b/dlls/mfreadwrite/writer.c @@ -27,6 +27,7 @@ #include "mf_private.h" #include "wine/debug.h" +#include "wine/list.h" WINE_DEFAULT_DEBUG_CHANNEL(mfplat); @@ -36,11 +37,30 @@ enum writer_state SINK_WRITER_STATE_WRITING, }; +struct marker_context +{ + IUnknown IUnknown_iface; + LONG refcount; + + unsigned int marker_type; + void *user_context; +}; + +struct pending_item +{ + struct list entry; + unsigned int marker_type; + IMFSample *sample; + LONGLONG timestamp; + void *user_context; +}; + struct stream { IMFStreamSink *stream_sink; IMFTransform *encoder; MF_SINK_WRITER_STATISTICS stats; + struct list queue; }; struct sink_writer @@ -76,6 +96,74 @@ static struct sink_writer *impl_from_events_callback_IMFAsyncCallback(IMFAsyncCa return CONTAINING_RECORD(iface, struct sink_writer, events_callback); } +static struct marker_context *impl_from_marker_context_IUnknown(IUnknown *iface) +{ + return CONTAINING_RECORD(iface, struct marker_context, IUnknown_iface); +} + +static HRESULT WINAPI marker_context_QueryInterface(IUnknown *iface, REFIID riid, void **out) +{ + if (IsEqualIID(riid, &IID_IUnknown)) + { + *out = iface; + IUnknown_AddRef(iface); + return S_OK; + } + + *out = NULL; + return E_NOINTERFACE; +} + +static ULONG WINAPI marker_context_AddRef(IUnknown *iface) +{ + struct marker_context *context = impl_from_marker_context_IUnknown(iface); + return InterlockedIncrement(&context->refcount); +} + +static ULONG WINAPI marker_context_Release(IUnknown *iface) +{ + struct marker_context *context = impl_from_marker_context_IUnknown(iface); + LONG refcount = InterlockedDecrement(&context->refcount); + + if (!refcount) + free(context); + + return refcount; +} + +static const IUnknownVtbl marker_context_vtbl = +{ + marker_context_QueryInterface, + marker_context_AddRef, + marker_context_Release, +}; + +static HRESULT create_marker_context(unsigned int marker_type, void *user_context, + IUnknown **ret) +{ + struct marker_context *object; + + if (!(object = calloc(1, sizeof(*object)))) + return E_OUTOFMEMORY; + + object->IUnknown_iface.lpVtbl = &marker_context_vtbl; + object->refcount = 1; + object->marker_type = marker_type; + object->user_context = user_context; + + *ret = &object->IUnknown_iface; + + return S_OK; +} + +static void sink_writer_release_pending_item(struct pending_item *item) +{ + list_remove(&item->entry); + if (item->sample) + IMFSample_Release(item->sample); + free(item); +} + static HRESULT WINAPI sink_writer_QueryInterface(IMFSinkWriter *iface, REFIID riid, void **out) { TRACE("%p, %s, %p.\n", iface, debugstr_guid(riid), out); @@ -103,6 +191,16 @@ static ULONG WINAPI sink_writer_AddRef(IMFSinkWriter *iface) return refcount; } +static void sink_writer_drop_pending_items(struct stream *stream) +{ + struct pending_item *item, *next; + + LIST_FOR_EACH_ENTRY_SAFE(item, next, &stream->queue, struct pending_item, entry) + { + sink_writer_release_pending_item(item); + } +} + static ULONG WINAPI sink_writer_Release(IMFSinkWriter *iface) { struct sink_writer *writer = impl_from_IMFSinkWriter(iface); @@ -125,6 +223,7 @@ static ULONG WINAPI sink_writer_Release(IMFSinkWriter *iface) IMFStreamSink_Release(stream->stream_sink); if (stream->encoder) IMFTransform_Release(stream->encoder); + sink_writer_drop_pending_items(stream); } DeleteCriticalSection(&writer->cs); free(writer); @@ -148,12 +247,14 @@ static HRESULT sink_writer_add_stream(struct sink_writer *writer, IMFStreamSink if (FAILED(hr = IMFStreamSink_GetIdentifier(stream_sink, &id))) return hr; *index = writer->streams.count++; - stream = &writer->streams.items[*index]; + stream = &writer->streams.items[*index]; + memset(stream, 0, sizeof(*stream)); stream->stream_sink = stream_sink; IMFStreamSink_AddRef(stream_sink); - memset(&stream->stats, 0, sizeof(stream->stats)); stream->stats.cb = sizeof(stream->stats); + list_init(&stream->queue); + writer->streams.next_id = max(writer->streams.next_id, id); return hr; @@ -283,6 +384,147 @@ static HRESULT sink_writer_get_buffer_length(IMFSample *sample, LONGLONG *timest return hr; } +static HRESULT sink_writer_place_marker(struct sink_writer *writer, struct stream *stream, unsigned int marker_type, + LONGLONG timestamp, void *user_context) +{ + PROPVARIANT value, context; + IUnknown *context_obj; + HRESULT hr; + + if (FAILED(hr = create_marker_context(marker_type, user_context, &context_obj))) return hr; + + context.vt = VT_UNKNOWN; + context.punkVal = context_obj; + value.vt = VT_I8; + value.hVal.QuadPart = timestamp; + hr = IMFStreamSink_PlaceMarker(stream->stream_sink, marker_type, marker_type == MFSTREAMSINK_MARKER_TICK ? &value : NULL, + &context); + IUnknown_Release(context_obj); + return hr; +} + +static HRESULT sink_writer_queue_marker(struct sink_writer *writer, struct stream *stream, unsigned int marker_type, + LONGLONG timestamp, void *user_context) +{ + struct pending_item *item; + + if (list_empty(&stream->queue)) + return sink_writer_place_marker(writer, stream, marker_type, timestamp, user_context); + + if (!(item = calloc(1, sizeof(*item)))) + return E_OUTOFMEMORY; + + item->marker_type = marker_type; + item->timestamp = timestamp; + list_add_tail(&stream->queue, &item->entry); + + return S_OK; +} + +static HRESULT sink_writer_send_stream_tick(struct sink_writer *writer, unsigned int index, LONGLONG timestamp) +{ + struct stream *stream; + + if (!(stream = sink_writer_get_stream(writer, index))) return MF_E_INVALIDSTREAMNUMBER; + + writer->stats.llLastStreamTickReceived = timestamp; + writer->stats.qwNumStreamTicksReceived++; + stream->stats.llLastStreamTickReceived = timestamp; + stream->stats.qwNumStreamTicksReceived++; + + return sink_writer_queue_marker(writer, stream, MFSTREAMSINK_MARKER_TICK, timestamp, NULL); +} + +static HRESULT sink_writer_notify_end_of_segment(struct sink_writer *writer, unsigned int index) +{ + struct stream *stream; + + if (!(stream = sink_writer_get_stream(writer, index))) return MF_E_INVALIDSTREAMNUMBER; + + return sink_writer_queue_marker(writer, stream, MFSTREAMSINK_MARKER_ENDOFSEGMENT, 0, NULL); +} + +static HRESULT sink_writer_process_sample(struct sink_writer *writer, struct stream *stream) +{ + struct pending_item *item, *next; + LONGLONG timestamp; + IMFSample *sample; + HRESULT hr; + + if (list_empty(&stream->queue)) return S_OK; + item = LIST_ENTRY(list_head(&stream->queue), struct pending_item, entry); + if (!item->sample) return S_OK; + + IMFSample_AddRef((sample = item->sample)); + sink_writer_release_pending_item(item); + + writer->stats.dwNumOutstandingSinkSampleRequests--; + stream->stats.dwNumOutstandingSinkSampleRequests--; + + if (FAILED(hr = IMFSample_GetSampleTime(sample, ×tamp))) + { + IMFSample_Release(sample); + writer->status = hr; + return S_OK; + } + + writer->stats.llLastTimestampProcessed = timestamp; + stream->stats.llLastTimestampProcessed = timestamp; + + hr = IMFStreamSink_ProcessSample(stream->stream_sink, sample); + IMFSample_Release(sample); + if (FAILED(hr)) return hr; + + LIST_FOR_EACH_ENTRY_SAFE(item, next, &stream->queue, struct pending_item, entry) + { + if (item->sample) break; + sink_writer_place_marker(writer, stream, item->marker_type, item->timestamp, item->user_context); + sink_writer_release_pending_item(item); + } + + return hr; +} + +static HRESULT sink_writer_encode_sample(struct sink_writer *writer, struct stream *stream, IMFSample *sample) +{ + struct pending_item *item; + + /* FIXME: call the encoder, queue its output */ + + if (!(item = calloc(1, sizeof(*item)))) + return E_OUTOFMEMORY; + + item->sample = sample; + IMFSample_AddRef(item->sample); + list_add_tail(&stream->queue, &item->entry); + + return S_OK; +} + +static HRESULT sink_writer_write_sample(struct sink_writer *writer, struct stream *stream, IMFSample *sample) +{ + LONGLONG timestamp; + DWORD length; + HRESULT hr; + + if (FAILED(hr = sink_writer_get_buffer_length(sample, ×tamp, &length))) return hr; + + stream->stats.llLastTimestampReceived = timestamp; + stream->stats.qwNumSamplesReceived++; + stream->stats.dwByteCountQueued += length; + + writer->stats.llLastTimestampReceived = timestamp; + writer->stats.qwNumSamplesReceived++; + writer->stats.dwByteCountQueued += length; + + if (FAILED(hr = sink_writer_encode_sample(writer, stream, sample))) return hr; + + if (stream->stats.dwNumOutstandingSinkSampleRequests) + hr = sink_writer_process_sample(writer, stream); + + return hr; +} + static HRESULT WINAPI sink_writer_WriteSample(IMFSinkWriter *iface, DWORD index, IMFSample *sample) { struct sink_writer *writer = impl_from_IMFSinkWriter(iface); @@ -306,8 +548,6 @@ static HRESULT WINAPI sink_writer_WriteSample(IMFSinkWriter *iface, DWORD index, } else if (SUCCEEDED(hr = sink_writer_get_buffer_length(sample, ×tamp, &length))) { - /* FIXME: queue sample */ - stream->stats.llLastTimestampReceived = timestamp; stream->stats.qwNumSamplesReceived++; stream->stats.dwByteCountQueued += length; @@ -315,18 +555,32 @@ static HRESULT WINAPI sink_writer_WriteSample(IMFSinkWriter *iface, DWORD index, writer->stats.llLastTimestampReceived = timestamp; writer->stats.qwNumSamplesReceived++; writer->stats.dwByteCountQueued += length; + + hr = sink_writer_write_sample(writer, stream, sample); } - EnterCriticalSection(&writer->cs); + LeaveCriticalSection(&writer->cs); return hr; } static HRESULT WINAPI sink_writer_SendStreamTick(IMFSinkWriter *iface, DWORD index, LONGLONG timestamp) { - FIXME("%p, %lu, %s.\n", iface, index, wine_dbgstr_longlong(timestamp)); + struct sink_writer *writer = impl_from_IMFSinkWriter(iface); + HRESULT hr; - return E_NOTIMPL; + TRACE("%p, %lu, %s.\n", iface, index, wine_dbgstr_longlong(timestamp)); + + EnterCriticalSection(&writer->cs); + + if (writer->state != SINK_WRITER_STATE_WRITING) + hr = MF_E_INVALIDREQUEST; + else + hr = sink_writer_send_stream_tick(writer, index, timestamp); + + LeaveCriticalSection(&writer->cs); + + return hr; } static HRESULT WINAPI sink_writer_PlaceMarker(IMFSinkWriter *iface, DWORD index, void *context) @@ -338,9 +592,33 @@ static HRESULT WINAPI sink_writer_PlaceMarker(IMFSinkWriter *iface, DWORD index, static HRESULT WINAPI sink_writer_NotifyEndOfSegment(IMFSinkWriter *iface, DWORD index) { - FIXME("%p, %lu.\n", iface, index); + struct sink_writer *writer = impl_from_IMFSinkWriter(iface); + HRESULT hr = S_OK; + unsigned int i; - return E_NOTIMPL; + TRACE("%p, %lu.\n", iface, index); + + EnterCriticalSection(&writer->cs); + + if (writer->state != SINK_WRITER_STATE_WRITING) + hr = MF_E_INVALIDREQUEST; + else if (index == MF_SINK_WRITER_ALL_STREAMS) + { + for (i = 0; i < writer->streams.count; ++i) + { + if (FAILED(hr = sink_writer_notify_end_of_segment(writer, index))) + { + WARN("Failed to place a marker for stream %u.\n", i); + break; + } + } + } + else + hr = sink_writer_notify_end_of_segment(writer, index); + + LeaveCriticalSection(&writer->cs); + + return hr; } static HRESULT WINAPI sink_writer_Flush(IMFSinkWriter *iface, DWORD index) @@ -539,6 +817,8 @@ static HRESULT WINAPI sink_writer_events_callback_Invoke(IMFAsyncCallback *iface stream->stats.llLastSinkSampleRequest = timestamp; stream->stats.dwNumOutstandingSinkSampleRequests++; + sink_writer_process_sample(writer, stream); + break; default: ; -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/676
From: Nikolay Sivov <nsivov(a)codeweavers.com> (cherry picked from commit b44bfb7645d78973495241eca53a9da9023a46fb) Signed-off-by: Nikolay Sivov <nsivov(a)codeweavers.com> --- dlls/mfreadwrite/writer.c | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/dlls/mfreadwrite/writer.c b/dlls/mfreadwrite/writer.c index a157bccd785..4e4bc988c9b 100644 --- a/dlls/mfreadwrite/writer.c +++ b/dlls/mfreadwrite/writer.c @@ -83,6 +83,8 @@ struct sink_writer HRESULT status; MF_SINK_WRITER_STATISTICS stats; + IMFSinkWriterCallback *callback; + CRITICAL_SECTION cs; }; @@ -215,6 +217,8 @@ static ULONG WINAPI sink_writer_Release(IMFSinkWriter *iface) IMFPresentationClock_Release(writer->clock); if (writer->sink) IMFMediaSink_Release(writer->sink); + if (writer->callback) + IMFSinkWriterCallback_Release(writer->callback); for (i = 0; i < writer->streams.count; ++i) { struct stream *stream = &writer->streams.items[i]; @@ -885,6 +889,12 @@ HRESULT create_sink_writer_from_sink(IMFMediaSink *sink, IMFAttributes *attribut object->stats.cb = sizeof(object->stats); InitializeCriticalSection(&object->cs); + if (attributes) + { + IMFAttributes_GetUnknown(attributes, &MF_SINK_WRITER_ASYNC_CALLBACK, + &IID_IMFSinkWriterCallback, (void **)&object->callback); + } + if (FAILED(hr = sink_writer_initialize_existing_streams(object, sink))) { IMFSinkWriter_Release(&object->IMFSinkWriter_iface); -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/676
From: Nikolay Sivov <nsivov(a)codeweavers.com> (cherry picked from commit 97e481b4929e862c825d4c3febc3fb4c9f31c8e2) Signed-off-by: Nikolay Sivov <nsivov(a)codeweavers.com> --- dlls/mfreadwrite/writer.c | 61 +++++++++++++++++++++++++++++++++++---- 1 file changed, 55 insertions(+), 6 deletions(-) diff --git a/dlls/mfreadwrite/writer.c b/dlls/mfreadwrite/writer.c index 4e4bc988c9b..10806f30cf5 100644 --- a/dlls/mfreadwrite/writer.c +++ b/dlls/mfreadwrite/writer.c @@ -16,6 +16,8 @@ * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA */ +#include <assert.h> + #define COBJMACROS #define NONAMELESSUNION @@ -140,6 +142,14 @@ static const IUnknownVtbl marker_context_vtbl = marker_context_Release, }; +static struct marker_context *unsafe_impl_from_marker_context_IUnknown(IUnknown *iface) +{ + if (!iface) + return NULL; + assert(iface->lpVtbl == &marker_context_vtbl); + return CONTAINING_RECORD(iface, struct marker_context, IUnknown_iface); +} + static HRESULT create_marker_context(unsigned int marker_type, void *user_context, IUnknown **ret) { @@ -587,11 +597,28 @@ static HRESULT WINAPI sink_writer_SendStreamTick(IMFSinkWriter *iface, DWORD ind return hr; } -static HRESULT WINAPI sink_writer_PlaceMarker(IMFSinkWriter *iface, DWORD index, void *context) +static HRESULT WINAPI sink_writer_PlaceMarker(IMFSinkWriter *iface, DWORD index, void *user_context) { - FIXME("%p, %lu, %p.\n", iface, index, context); + struct sink_writer *writer = impl_from_IMFSinkWriter(iface); + struct stream *stream; + HRESULT hr; - return E_NOTIMPL; + TRACE("%p, %lu, %p.\n", iface, index, user_context); + + EnterCriticalSection(&writer->cs); + + if (!writer->callback) + hr = MF_E_INVALIDREQUEST; + else if (writer->state != SINK_WRITER_STATE_WRITING) + hr = MF_E_INVALIDREQUEST; + else if (!(stream = sink_writer_get_stream(writer, index))) + hr = MF_E_INVALIDSTREAMNUMBER; + else + hr = sink_writer_queue_marker(writer, stream, MFSTREAMSINK_MARKER_DEFAULT, 0, user_context); + + LeaveCriticalSection(&writer->cs); + + return hr; } static HRESULT WINAPI sink_writer_NotifyEndOfSegment(IMFSinkWriter *iface, DWORD index) @@ -768,14 +795,18 @@ static HRESULT WINAPI sink_writer_callback_GetParameters(IMFAsyncCallback *iface return E_NOTIMPL; } -static struct stream *sink_writer_get_stream_for_stream_sink(struct sink_writer *writer, IMFStreamSink *stream_sink) +static struct stream *sink_writer_get_stream_for_stream_sink(struct sink_writer *writer, + IMFStreamSink *stream_sink, DWORD *index) { - size_t i; + unsigned int i; for (i = 0; i < writer->streams.count; ++i) { if (writer->streams.items[i].stream_sink == stream_sink) + { + *index = i; return &writer->streams.items[i]; + } } return NULL; @@ -784,12 +815,15 @@ static struct stream *sink_writer_get_stream_for_stream_sink(struct sink_writer static HRESULT WINAPI sink_writer_events_callback_Invoke(IMFAsyncCallback *iface, IMFAsyncResult *result) { struct sink_writer *writer = impl_from_events_callback_IMFAsyncCallback(iface); + struct marker_context *context; IMFStreamSink *stream_sink; MediaEventType event_type; struct stream *stream; IMFMediaEvent *event; LONGLONG timestamp; HRESULT status, hr; + PROPVARIANT value; + DWORD index; TRACE("%p, %p.\n", iface, result); @@ -803,12 +837,14 @@ static HRESULT WINAPI sink_writer_events_callback_Invoke(IMFAsyncCallback *iface TRACE("Got event %lu.\n", event_type); + PropVariantInit(&value); + EnterCriticalSection(&writer->cs); if (writer->status == S_OK && FAILED(status)) writer->status = status; - if (writer->status == S_OK && (stream = sink_writer_get_stream_for_stream_sink(writer, stream_sink))) + if (writer->status == S_OK && (stream = sink_writer_get_stream_for_stream_sink(writer, stream_sink, &index))) { switch (event_type) { @@ -824,6 +860,17 @@ static HRESULT WINAPI sink_writer_events_callback_Invoke(IMFAsyncCallback *iface sink_writer_process_sample(writer, stream); break; + + case MEStreamSinkMarker: + if (FAILED(hr = IMFMediaEvent_GetValue(event, &value))) break; + if (value.vt != VT_UNKNOWN || !(context = unsafe_impl_from_marker_context_IUnknown(value.punkVal))) break; + + /* This relies on the fact that default marker type is only used for PlaceMarker(). */ + if (context->marker_type == MFSTREAMSINK_MARKER_DEFAULT) + IMFSinkWriterCallback_OnMarker(writer->callback, index, context->user_context); + + break; + default: ; } @@ -831,6 +878,8 @@ static HRESULT WINAPI sink_writer_events_callback_Invoke(IMFAsyncCallback *iface LeaveCriticalSection(&writer->cs); + PropVariantClear(&value); + IMFMediaEvent_Release(event); IMFStreamSink_BeginGetEvent(stream_sink, iface, (IUnknown *)stream_sink); -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/676
From: Nikolay Sivov <nsivov(a)codeweavers.com> Signed-off-by: Nikolay Sivov <nsivov(a)codeweavers.com> --- dlls/mfreadwrite/writer.c | 44 +++++++++++++++++++++++++++++++++++++-- 1 file changed, 42 insertions(+), 2 deletions(-) diff --git a/dlls/mfreadwrite/writer.c b/dlls/mfreadwrite/writer.c index 10806f30cf5..aaf972734c4 100644 --- a/dlls/mfreadwrite/writer.c +++ b/dlls/mfreadwrite/writer.c @@ -458,6 +458,22 @@ static HRESULT sink_writer_notify_end_of_segment(struct sink_writer *writer, uns return sink_writer_queue_marker(writer, stream, MFSTREAMSINK_MARKER_ENDOFSEGMENT, 0, NULL); } +static HRESULT sink_writer_flush(struct sink_writer *writer, unsigned int index) +{ + struct stream *stream; + + if (!(stream = sink_writer_get_stream(writer, index))) return MF_E_INVALIDSTREAMNUMBER; + + sink_writer_drop_pending_items(stream); + + IMFStreamSink_Flush(stream->stream_sink); + + if (stream->encoder) + IMFTransform_ProcessMessage(stream->encoder, MFT_MESSAGE_COMMAND_FLUSH, 0); + + return sink_writer_place_marker(writer, stream, MFSTREAMSINK_MARKER_ENDOFSEGMENT, 0, NULL); +} + static HRESULT sink_writer_process_sample(struct sink_writer *writer, struct stream *stream) { struct pending_item *item, *next; @@ -654,9 +670,33 @@ static HRESULT WINAPI sink_writer_NotifyEndOfSegment(IMFSinkWriter *iface, DWORD static HRESULT WINAPI sink_writer_Flush(IMFSinkWriter *iface, DWORD index) { - FIXME("%p, %lu.\n", iface, index); + struct sink_writer *writer = impl_from_IMFSinkWriter(iface); + HRESULT hr = S_OK; + unsigned int i; - return E_NOTIMPL; + TRACE("%p, %lu.\n", iface, index); + + EnterCriticalSection(&writer->cs); + + if (writer->state != SINK_WRITER_STATE_WRITING) + hr = MF_E_INVALIDREQUEST; + else if (index == MF_SINK_WRITER_ALL_STREAMS) + { + for (i = 0; i < writer->streams.count; ++i) + { + if (FAILED(hr = sink_writer_flush(writer, i))) + { + WARN("Failed to flush stream %u.\n", i); + break; + } + } + } + else + hr = sink_writer_flush(writer, index); + + LeaveCriticalSection(&writer->cs); + + return hr; } static HRESULT WINAPI sink_writer_Finalize(IMFSinkWriter *iface) -- GitLab https://gitlab.winehq.org/wine/wine/-/merge_requests/676
participants (2)
-
Nikolay Sivov -
Nikolay Sivov (＠nsivov)