[pulseaudio-discuss] [PATCH v9] pipe-source: [DRAFT] implement autosuspend behavior

[Raman Shishniou]

On 02/22/2018 10:18 PM, Georg Chini wrote:
>> -        /* Hmm, nothing to do. Let's sleep */
>> -        pollfd->events = (short) (u->source->thread_info.state == PA_SOURCE_RUNNING ? POLLIN : 0);
>> +        /* Post data to source, discard data or wait for state transition to be complete */
>> +        if (PA_SOURCE_IS_OPENED(u->source->thread_info.state)) {
>> +
>> +            if (u->writer_connected && u->corkfd >= 0) {
>> +                pa_assert_se(pa_close(u->corkfd) == 0);
>> +                u->corkfd = -1;
>> +            }
>> +
>> +            if (u->memchunk.length > 0) {
>> +                pa_source_post(u->source, &u->memchunk);
>> +                pa_memblock_unref(u->memchunk.memblock);
>> +                u->memchunk.memblock = pa_memblock_new(u->core->mempool, u->pipe_size);
>> +                u->memchunk.length = 0;
>> +            }
>> +
>> +        } else if (u->suspended_by_user)
> 
> You have to open the corkfd here as well to avoid POLLHUP spam if the writer
> disconnects during user suspend.
> 

During user suspend we are doing normal polling. If the writer will be disconnected
we'll see it by reading 0 bytes and corkfd will be opened.

>> +            u->memchunk.length = 0;
>> +
>> +
>> +        pollfd->events = u->memchunk.length ? 0 : POLLIN;
> 
> How do you wake up from autosuspend? When the writer disconnects,
> memchunk.length will be 0 and so you do no longer listen for POLLIN.
> I guess you need to know the current suspend cause here:
	> user suspended -> events = 0
> (only auto suspended + no writer connected) or running -> events = POLLIN
> only auto suspended + writer connected (=wake up transition) -> events = 0
> 

Yes. This is what I do, keep data in memchunk and wait while source will be resumed.

- Source autosuspended, corkfd opened, rtpoll wait for POLLIN
- Someone connected and send us data.
- pa_rtpoll_run() returned and pollfd->revents = POLLIN

First iteration:

- pollfd->revents == POLLIN
  - l = read()
  - l > 0:
    - memchunk.length = l
    - writer_connected is false
      - send PIPE_SOURCE_RESUME message
      - set writer_connected = true,
      ! do not close corkfd here to avoid POLLHUP spam !

- Data cannot be posted because source still suspended, i.e. we just send
  resume message to main thread and thread_info.state not changed yet.
- Data cannot be discarded - source auto suspended, not suspended by user.

- So we need to wait for source to be resumed (and keep data in memchunk):
  - events = 0

- pa_rtpoll_run(): corkfd still opened, but events = 0 - we are waiting for our
  resume message will be processed.

Next iteration

- resume message was processed

- pollfd->events = 0 (pellfb->events was set to 0 in previous iteration)

- Source was resumed:
  - writer_connected is true and corkfd still opened:
     -close corkfd
  - post data to source
  - allocate new buffer
  - memchunk.length = 0

- We are processed pending data (memchunk.length == 0)
  - set events = POLLIN

- pa_rtpoll_run(): corkfd closed, events = POLLIN - we are waiting for new data

>>             if ((ret = pa_rtpoll_run(u->rtpoll)) < 0)
>>               goto fail;
>> @@ -188,8 +281,13 @@ static void thread_func(void *userdata) {
>>             pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
>>   -        if (pollfd->revents & ~POLLIN) {
>> -            pa_log("FIFO shutdown.");
>> +        if (pollfd->revents & ~(POLLIN|POLLHUP)) {
>> +            pa_log("FD error: %s%s%s%s",
>> +                   pollfd->revents & POLLOUT ? "POLLOUT " : "",
>> +                   pollfd->revents & POLLERR ? "POLLERR " : "",
>> +                   pollfd->revents & POLLPRI ? "POLLPRI " : "",
>> +                   pollfd->revents & POLLNVAL ? "POLLNVAL " : "");
>> +
>>               goto fail;
>>           }
>>       }
>> @@ -238,13 +336,26 @@ int pa__init(pa_module *m) {
>>           goto fail;
>>       }
>>   +    if (!(u->msg = pa_msgobject_new(pipe_source_msg)))
>> +        goto fail;
>> +
>> +    u->msg->parent.process_msg = pipe_source_process_msg;
>> +
>>       u->filename = pa_runtime_path(pa_modargs_get_value(ma, "file", DEFAULT_FILE_NAME));
>>         if (mkfifo(u->filename, 0666) < 0) {
>>           pa_log("mkfifo('%s'): %s", u->filename, pa_cstrerror(errno));
>>           goto fail;
>>       }
>> -    if ((u->fd = pa_open_cloexec(u->filename, O_RDWR, 0)) < 0) {
>> +
>> +    if ((u->corkfd = pa_open_cloexec(u->filename, O_RDWR, 0)) < 0) {
>> +        pa_log("open('%s'): %s", u->filename, pa_cstrerror(errno));
>> +        goto fail;
>> +    }
>> +
>> +    u->writer_connected = true;
>> +
>> +    if ((u->fd = pa_open_cloexec(u->filename, O_RDONLY, 0)) < 0) {
>>           pa_log("open('%s'): %s", u->filename, pa_cstrerror(errno));
>>           goto fail;
>>       }
>> @@ -289,13 +400,18 @@ int pa__init(pa_module *m) {
>>         pa_source_set_asyncmsgq(u->source, u->thread_mq.inq);
>>       pa_source_set_rtpoll(u->source, u->rtpoll);
>> -    pa_source_set_fixed_latency(u->source, pa_bytes_to_usec(pa_pipe_buf(u->fd), &u->source->sample_spec));
>> +
>> +    u->pipe_size = pa_pipe_buf(u->fd);
>> +    u->pipe_size = PA_MIN(pa_mempool_block_size_max(m->core->mempool), u->pipe_size);
>> +    pa_source_set_fixed_latency(u->source, pa_bytes_to_usec(u->pipe_size, &u->source->sample_spec));
>>         u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
>>       pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
>>       pollfd->fd = u->fd;
>>       pollfd->events = pollfd->revents = 0;
>>   +    u->suspended_by_user = false;
>> +
>>       if (!(u->thread = pa_thread_new("pipe-source", thread_func, u))) {
>>           pa_log("Failed to create thread.");
>>           goto fail;
>> @@ -360,6 +476,12 @@ void pa__done(pa_module *m) {
>>           pa_xfree(u->filename);
>>       }
>>   +    if (u->msg)
>> +        pa_xfree(u->msg);
>> +
>> +    if (u->corkfd >= 0)
>> +        pa_assert_se(pa_close(u->corkfd) == 0);
>> +
>>       if (u->fd >= 0)
>>           pa_assert_se(pa_close(u->fd) == 0);
>>   
> 
> I believe you need some user_unsuspended and/or auto_unsuspended
> flag that is set in the SET_STATE handler to handle the first data read
> after suspend correctly.
> 
> When you unsuspend from auto suspend, the first chunk you already read
> from the pipe should be posted - as far as I can see this happens.
> 
> However, when you unsuspend from user suspend, the data in the pipe may
> be old - you did not read anything during suspend. Therefore the first chunk
> you read should be discarded. That does not happen as far as I can tell. Or
> am I overlooking something (again)?
> 
> The general approach of the patch looks OK now.
> 
--
Raman