[PATCH v1 3/3] mfreadwrite: Implement MF_SOURCE_READER_ASYNC_CALLBACK.

[Nikolay Sivov]

On 3/17/20 7:48 PM, Derek Lesho wrote:
> @@ -587,6 +594,8 @@ static ULONG WINAPI src_reader_Release(IMFSourceReader *iface)
>                   list_remove(&ptr->entry);
>                   heap_free(ptr);
>               }
> +
> +            MFUnlockWorkQueue(stream->read_samples_queue);
>           }
>           heap_free(reader->streams);
>           DeleteCriticalSection(&reader->cs);
Is additional queue necessary for synchronous case?

> +    if (FAILED(hr = media_stream_get_id(state, &id)))
> +    {
> +        WARN("Bad stream %p, hr %#x.\n", state, hr);
> +    }
> +
> +    for (unsigned int i = 0; i < reader->stream_count; ++i)
> +    {
> +        if (id == reader->streams[i].id)
> +        {
> +            struct media_stream *stream = &reader->streams[i];
> +            IMFSample *sample = NULL;
> +            DWORD stream_flags;
> +            LONGLONG timestamp = 0;
> +
> +            hr = next_sample(stream, &sample, &stream_flags, FALSE);
> +            if (sample)
> +            {
> +                IMFSample_GetSampleTime(sample, &timestamp);
> +            }
> +
> +            TRACE("Invoking read sample callback %p with (hr = %#x, stream_idx = %u, flags = %#x, timestamp %lu, sample %p)\n", reader->async_callback, hr, i, stream_flags, timestamp, sample);
> +            hr = IMFSourceReaderCallback_OnReadSample(reader->async_callback, hr, i, stream_flags, timestamp, sample);
> +            IMFSample_Release(sample);
> +            return hr;
> +        }
> +    }
> +
> +    return S_OK;
We already talked about this part - blocking in next_sample(), even on 
dedicated thread, is unnecessary.

What will happen is MENewSample from stream queue -> wake thread in 
dedicated per-stream queue -> process -> call OnReadSample.
Without blocking you'll have MENewSample -> process on same event 
handling thread -> call OnReadSample.

I don't see a need for additional complexity.

> +    switch (index)
> +    {
> +        case MF_SOURCE_READER_FIRST_VIDEO_STREAM:
> +            stream_index = reader->first_video_stream_index;
> +            break;
> +        case MF_SOURCE_READER_FIRST_AUDIO_STREAM:
> +            stream_index = reader->first_audio_stream_index;
> +            break;
> +        case MF_SOURCE_READER_ANY_STREAM:
> +            FIXME("Non-specific requests are not supported.\n");
> +            return E_NOTIMPL;
> +        default:
> +            stream_index = index;
> +    }
> +
> +    /* Can't read from deselected streams. */
> +    if (FAILED(hr = source_reader_get_stream_selection(reader, stream_index, &selected)) && !selected)
> +        return hr;
If that's how it works, this part applies to both sync and async cases.

> +    EnterCriticalSection(&stream->cs);
> +    while(!stream->stream)
> +    {
> +        SleepConditionVariableCS(&stream->sample_event, &stream->cs, INFINITE);
> +    }
> +    LeaveCriticalSection(&stream->cs);
> +
> +    TRACE("Dispatching read sample callback for stream %p\n", stream->stream);
> +    if (FAILED(hr = MFPutWorkItem(stream->read_samples_queue, &reader->read_samples_callback, (IUnknown*)stream->stream)))
> +    {
> +        WARN("Failed to submit item hr = %#x\n", hr);
> +        return E_FAIL;
> +    }
This blocks for indefinite amount of time, for ReadSample method that's 
supposed to be non-blocking. It's correct of course to start the source 
here,
which also could be potentially shared with sync case, but waiting for 
new samples, using 'sample_event' that's not meant for it, is wrong.

It should start source, and keep a note of number of times each stream 
was requested, later on MENewStream/MEUpdatedStream you can issue that 
many requests at once.