API Core: Fix race in 'BackgroundConsumer._thread_main'.

[tseaver]

See #7817.

In addition to passing all api_core tests locally, I have verified that the firestore and pubsub system tests all run cleanly with this patch (an earlier version caused them to hang).

[plamut]

Looks good, it solves one part of the issue.

There is still a possibility that Consumer gets paused right after we release the lock, and just before we fetch a new response from the server. However, I don't think there is a straightforward way of addressing that

Since the self._bidi_rpc.recv() might block, we must release the _wake lock, otherwise shutting down the background consumer could get blocked because of it (details). This was nicely caught by system tests with an earlier version of the fix.

All in all this PR represents an improvement overall, thus approving.

[tseaver]

@plamut I updated the PR description / commit message to avoid closing #7817. Can you please clarify how we might reproduce / fix the remaining part?

[plamut]

Reproducing the issue

I did not manage to reproduce it with a real application (the time window is narrow), but I came up with a test that detects if recv() is called in BackgroundConsumer when the latter is in paused state (should not happen) - https://gist.github.com/plamut/8f996fdc9113e8de2b2c050befb36ff6

The idea is to constantly pause/resume the consumer and check if the latter is indeed paused while recv() is being executed.

The test consistently fails on my machine, but also passes if one comments out the following:

pause_resume_thread.start()

Can anyone confirm my findings and proof-read the test?

Fixing the issue

Apparently, we must be holding the self._wake lock when receiving messages, otherwise another thread can pause the consumer in the meantime, and the messages will be received and delivered in a paused state.

However, simply indenting the following block into the with self._wake: context would introduce other, more serious, problems:

_LOGGER.debug("waiting for recv.")
response = self._bidi_rpc.recv()
_LOGGER.debug("recved response.")
self._on_response(response)

Since self._bidi_rpc.recv() might block, holding the self._wake lock would also block stopping the consumer, for instance, because that operation would also try to obtain the very same self._wake lock (details).

Is there a good way to make the recv() method non-blocking? Probably not?

Estimation of the issue impact

I'm not familiar with Firestore, thus I can only say for PubSub.

The background consumer gets paused when the streaming pull manager determines that the client currently has more than MAX_LOAD messages on its hands, and resumes the consumer when there is enough capacity.

If the consumer main loop detects the paused state too late, one extra batch of server messages will be received before pausing in the next iteration.

For the PubSub client, this would mean receiving one extra batch of messages that would sit in the holding buffer until they can be processed. The extra memory usage is thus bounded by the maximum size of a single response from the server - probably not a showstopper.