[PATCH 5/6] amstream: Implement audio stream sample Update.

[Anton Baskanov]

Thanks for the suggestion. I've reworked the event queue, now it only stores 
the received samples.

On 22/4/20 4:36, you wrote:
> On 4/21/20 2:35 PM, Anton Baskanov wrote:
> > Hello Zeb, thanks for the comments, I've replied inline.
> > 
> > On 4/20/20 5:07 AM, you wrote:
> >> Hello Anton, nice work on the patch! It mostly looks good to me; I have
> >> some comments and questions inlined.
> >> 
> >> On 4/18/20 9:34 AM, Anton Baskanov wrote:
> >>> +static void process_update(IAudioStreamSampleImpl *sample, struct
> >>> queued_event *event) +{
> >>> +    DWORD advance;
> >>> +
> >>> +    if (event->type == QET_END_OF_STREAM)
> >>> +    {
> >>> +        sample->update_hr = sample->position ? S_OK : MS_S_ENDOFSTREAM;
> >>> +        return;
> >>> +    }
> >>> +
> >>> +    advance = min(event->length - event->position, sample->length -
> >>> sample->position); +    CopyMemory(&sample->pointer[sample->position],
> >>> &event->pointer[event->position], advance);
> >> 
> >> memcpy()?
> > 
> > Done.
> > 
> >>> +
> >>> +    event->position += advance;
> >>> +    sample->position += advance;
> >>> +
> >>> +    sample->update_hr = (sample->position == sample->length) ? S_OK :
> >>> MS_S_PENDING; +}
> >>> +
> >>> +static void process_updates(struct audio_stream *stream)
> >>> +{
> >>> +    while (!list_empty(&stream->update_queue) &&
> >>> !list_empty(&stream->event_queue)) +    {
> >>> +        IAudioStreamSampleImpl *sample =
> >>> LIST_ENTRY(list_head(&stream->update_queue), IAudioStreamSampleImpl,
> >>> entry); +        struct queued_event *event =
> >>> LIST_ENTRY(list_head(&stream->event_queue), struct queued_event, entry);
> >>> +
> >>> +        process_update(sample, event);
> >>> +
> >>> +        if (MS_S_PENDING != sample->update_hr)
> >>> +            remove_queued_update(sample);
> >>> +        if ((event->type != QET_END_OF_STREAM) && (event->position ==
> >>> event->length)) +            remove_queued_event(event);
> >> 
> >> It kind of feels weird to me to queue EOS events the same way as
> >> samples, given they apply to all streams at once, can't be interleaved,
> >> can't be sent multiple times, and don't get removed from the queue until
> >> the stream is stopped (or flushed, presumably?).
> >> 
> >> Maybe it would be clearer not to queue EOS events at all, but instead to
> >> loop through all queued samples in audio_sink_EndOfStream(), completing
> >> them with MS_E_ENDOFSTREAM, and similarly to check for stream->eos in
> >> process_updates(). I imagine you could get rid of the queued_event_type
> >> enumeration then, unless there's another kind of event that I'm not
> >> anticipating?
> > 
> > Queuing EOS events is actually the correct way of doing things, otherwise
> > the final samples would get lost. Consider the following scenario:
> > 
> > Receive()
> > EndOfStream()
> > Update() -> S_OK
> > Update() -> MS_S_ENDOFSTREAM
> > 
> > It is expected that Update retrieves all available data before returning
> > MS_S_ENDOFSTREAM. The tests confirm this behavior.
> 
> Sure, I'd expect that. I guess as far as I see, it would still be
> simpler this way, you'd just need to make sure you consume all queued
> samples *before* checking stream->eos in process_updates().
> 
> >>> +    }
> >>> +}
> >>> +
> >>> 
> >>>   static inline struct audio_stream
> >>>   *impl_from_IAMMediaStream(IAMMediaStream *iface) {
> >>>   
> >>>       return CONTAINING_RECORD(iface, struct audio_stream,
> >>>       IAMMediaStream_iface);
> >>> 
> >>> @@ -449,6 +503,85 @@ static inline struct audio_stream
> >>> *impl_from_IAudioMediaStream(IAudioMediaStream>
> >>> 
> >>>       return CONTAINING_RECORD(iface, struct audio_stream,
> >>>       IAudioMediaStream_iface);>
> >>>   
> >>>   }
> >>> 
> >>> +static HRESULT WINAPI IAudioStreamSampleImpl_Update(IAudioStreamSample
> >>> *iface, DWORD flags, HANDLE event, +        PAPCFUNC func_APC, DWORD
> >>> APC_data)
> >>> +{
> >>> +    IAudioStreamSampleImpl *sample =
> >>> impl_from_IAudioStreamSample(iface);
> >>> +    struct audio_stream *stream;
> >>> +    DWORD length;
> >>> +    BYTE *pointer;
> >>> +    HRESULT hr;
> >>> +
> >>> +    TRACE("(%p)->(%x,%p,%p,%u)\n", iface, flags, event, func_APC,
> >>> APC_data); +
> >>> +    stream = impl_from_IAudioMediaStream(sample->parent);
> >> 
> >> Perhaps easier would be to take patch 2/6 a step further and store
> >> sample->parent as a pointer to struct audio_stream. Then you wouldn't
> >> need to move this function out of place.
> > 
> > Done.
> > 
> >>> +
> >>> +    hr = IAudioData_GetInfo(sample->audio_data, &length, &pointer,
> >>> NULL);
> >>> +    if (FAILED(hr))
> >>> +        return hr;
> >>> +
> >>> +    if (event && func_APC)
> >>> +        return E_INVALIDARG;
> >>> +
> >>> +    if (func_APC)
> >>> +    {
> >>> +        FIXME("APC support is not implemented!\n");
> >>> +        return E_NOTIMPL;
> >>> +    }
> >>> +
> >>> +    if (event)
> >>> +    {
> >>> +        FIXME("Event parameter support is not implemented!\n");
> >>> +        return E_NOTIMPL;
> >>> +    }
> >>> +
> >>> +    if (flags & ~SSUPDATE_ASYNC)
> >>> +    {
> >>> +        FIXME("Unsupported flags: %x\n", flags);
> >>> +        return E_NOTIMPL;
> >>> +    }
> >>> +
> >>> +    EnterCriticalSection(&stream->cs);
> >>> +
> >>> +    if (stream->state == State_Stopped)
> >>> +    {
> >>> +        LeaveCriticalSection(&stream->cs);
> >>> +        return MS_E_NOTRUNNING;
> >>> +    }
> >> 
> >> Do you have tests for how Update() behaves while paused? If it's as
> >> simple as "identical to stopped or running", it'd be nice to add a quick
> >> test below.
> >> 
> >>> +    if (!stream->peer)
> >>> +    {
> >>> +        LeaveCriticalSection(&stream->cs);
> >>> +        return MS_S_ENDOFSTREAM;
> >>> +    }
> >>> +    if (MS_S_PENDING == sample->update_hr)
> >>> +    {
> >>> +        LeaveCriticalSection(&stream->cs);
> >>> +        return MS_E_BUSY;
> >>> +    }
> >>> +
> >>> +    sample->length = length;
> >>> +    sample->pointer = pointer;
> >>> +    sample->position = 0;
> >>> +    sample->update_hr = MS_S_PENDING;
> >>> +    ResetEvent(sample->update_event);
> >>> +    list_add_tail(&stream->update_queue, &sample->entry);
> >>> +
> >>> +    process_updates(stream);
> >>> +
> >>> +    hr = sample->update_hr;
> >>> +    if (hr != MS_S_PENDING || (flags & SSUPDATE_ASYNC))
> >>> +    {
> >>> +        LeaveCriticalSection(&stream->cs);
> >>> +        return hr;
> >>> +    }
> >>> +
> >>> +    LeaveCriticalSection(&stream->cs);
> >>> +
> >>> +    WaitForSingleObject(sample->update_event, INFINITE);
> >>> +
> >>> +    return sample->update_hr;
> >>> +}
> >>> +