Skip to content

Logging: Clarify worker thread implementation. #8228

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Jun 6, 2019
Merged

Logging: Clarify worker thread implementation. #8228

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5bef636
Clarify worker thread implementation.
tseaver Jun 5, 2019
ae95819
Lint.
tseaver Jun 5, 2019
e55db0c
Moar lint.
tseaver Jun 6, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 6 additions & 12 deletions logging/google/cloud/logging/handlers/transports/background_thread.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
import threading
import time

from six.moves import range
from six.moves import queue

from google.cloud.logging.handlers.transports.base import Transport
Expand Down Expand Up @@ -56,8 +55,8 @@ def _get_many(queue_, max_items=None, max_latency=0):
item from a queue. This number includes the time required to retrieve
the first item.

:rtype: Sequence
:returns: A sequence of items retrieved from the queue.
:rtype: list
:returns: items retrieved from the queue.
"""
start = time.time()
# Always return at least one item.
Expand Down Expand Up @@ -132,8 +131,8 @@ def _thread_main(self):
"""
_LOGGER.debug("Background thread started.")

quit_ = False
while True:
done = False
while not done:
batch = self._cloud_logger.batch()
items = _get_many(
self._queue,
Expand All @@ -143,20 +142,15 @@ def _thread_main(self):

for item in items:
if item is _WORKER_TERMINATOR:
quit_ = True
# Continue processing items, don't break, try to process
# all items we got back before quitting.
done = True # Continue processing items.
else:
batch.log_struct(**item)

self._safely_commit_batch(batch)

for _ in range(len(items)):
for _ in items:
self._queue.task_done()

if quit_:
break

_LOGGER.debug("Background thread exited gracefully.")

def start(self):
Expand Down