[PATCH v2 3/3] winegstreamer: Replace source pad interface with GstAppSrc.

[Derek Lesho]

On 9/13/21 1:57 PM, Zebediah Figura wrote:
>> @@ -541,15 +536,63 @@ static bool CDECL 
>> wg_parser_get_next_read_offset(struct wg_parser *parser,
>>   static void CDECL wg_parser_push_data(struct wg_parser *parser,
>>           const void *data, uint32_t size)
>>   {
>> +    GstBuffer *buffer;
>> +    GstFlowReturn ret;
>> +    GError *error;
>> +    GstMessage *message;
>> +
>> +    if (!data)
>> +    {
>> +        pthread_mutex_lock(&parser->mutex);
>> +
>> +        error = g_error_new(G_FILE_ERROR, G_FILE_ERROR_FAILED, 
>> "WG-Parser client failed to read data at offset %" G_GUINT64_FORMAT, 
>> parser->next_pull_offset);
>> +        message = gst_message_new_error(NULL, error, "");
>> +        if (!gst_bus_post(parser->bus, message))
>> +        {
>> +            GST_ERROR("Failed to post error message to bus!\n");
>> +            gst_message_unref(message);
>> +        }
>
> Is there a point in posting a message to the bus?
The point would be to have a unified error path when something goes 
wrong.  (whether the error be GStreamer internal or a read failure)
>
>> + parser->read_request.pending_read = false;
>
> This is fine temporarily. Ultimately I think we should be terminating 
> the client-side thread on error, and if possible not calling 
> wg_parser_push_data() at all.
It would be a little tricky to do so given that this thread exists 
throughout initialization, and not signaling the error to GStreamer 
would mean we have to add yet another shutdown path that doesn't 
interfere with the others.
>
>> +
>> +        pthread_mutex_unlock(&parser->mutex);
>> +        return;
>> +    }
>> +
>> +    if (!size)
>> +    {
>> +        pthread_mutex_lock(&parser->mutex);
>> +
>> +        g_signal_emit_by_name(G_OBJECT(parser->appsrc), 
>> "end-of-stream", &ret);
>> +        parser->read_request.pending_read = false;
>> +
>> +        pthread_mutex_unlock(&parser->mutex);
>> +        return;
>> +    }
>> +
>> +    /*  We could avoid this extra copy using gst_buffer_new_wrapped.
>> +        However, PE wouldn't know when to release the buffer 
>> allocations as the buffer
>> +        objects are queued, so we'd have to create a ring buffer the 
>> size of the gstappsrc
>> +        queue on the PE side and validate that we don't overrun on 
>> the unix side.  I'm not
>> +        yet convinced that trying to reduce copies of compressed 
>> data is worth the
>> +        complexity.  */
>> +    buffer = gst_buffer_new_and_alloc(size);
>> +    gst_buffer_fill(buffer, 0, data, size);
>> +
>>       pthread_mutex_lock(&parser->mutex);
>> -    parser->read_request.size = size;
>> -    parser->read_request.done = true;
>> -    parser->read_request.ret = !!data;
>> -    if (data)
>> -        memcpy(parser->read_request.data, data, size);
>> -    parser->read_request.data = NULL;
>> +    assert(parser->read_request.pending_read);
>> +
>> +    GST_BUFFER_OFFSET(buffer) = parser->next_pull_offset;
>> +    g_signal_emit_by_name(G_OBJECT(parser->appsrc), "push-buffer", 
>> buffer, &ret);
>> +
>> +    /* In random-access mode, GST_FLOW_EOS shouldn't be returned */
>> +    assert(ret == GST_FLOW_OK || ret == GST_FLOW_FLUSHING);
>> +    if (ret == GST_FLOW_OK)
>> +        parser->next_pull_offset += size;
>> +    else
>> +        gst_buffer_unref(buffer);
>
> According to my reading of the source code, GstAppSrc will take the 
> reference even if it returns GST_FLOW_FLUSHING. That's a bit 
> unintuitive and should probably be clearly documented on the GStreamer 
> side, but anyway...
Good catch, thanks.
>
>> +
>> +    parser->read_request.pending_read = false;
>>       pthread_mutex_unlock(&parser->mutex);
>> -    pthread_cond_signal(&parser->read_done_cond);
>>   }
>>     static void CDECL wg_parser_set_unlimited_buffering(struct 
>> wg_parser *parser)
>