Skip to content
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

Jump to bottom

fix: add lock to flow control #899

Merged
merged 3 commits into from Dec 12, 2023
Merged

fix: add lock to flow control #899

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8a63ce4
add lock to flow control
daniel-sanche Dec 12, 2023
918b5b9
Merge branch 'main' into fix_flow_control_threads2
daniel-sanche Dec 12, 2023
d167267
moved set_flow_control_status out of lock
daniel-sanche Dec 12, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
11 changes: 7 additions & 4 deletions google/cloud/bigtable/batcher.py
View file
Open in desktop
Expand Up @@ -110,6 +110,7 @@ def __init__(
self.inflight_size = 0
self.event = threading.Event()
self.event.set()
self._lock = threading.Lock()

def is_blocked(self):
"""Returns True if:
Expand All @@ -128,8 +129,9 @@ def control_flow(self, batch_info):
Calculate the resources used by this batch
"""

self.inflight_mutations += batch_info.mutations_count
self.inflight_size += batch_info.mutations_size
with self._lock:
self.inflight_mutations += batch_info.mutations_count
self.inflight_size += batch_info.mutations_size
self.set_flow_control_status()
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe we can keep set_flow_control_status out of the lock. Even if it occasionally gets clobbered by later changes, every add and release ends with another call to set_flow_control_status, so it should be eventually correct.


def wait(self):
Expand All @@ -154,8 +156,9 @@ def release(self, batch_info):
Release the resources.
Decrement the row size to allow enqueued mutations to be run.
"""
self.inflight_mutations -= batch_info.mutations_count
self.inflight_size -= batch_info.mutations_size
with self._lock:
self.inflight_mutations -= batch_info.mutations_count
self.inflight_size -= batch_info.mutations_size
self.set_flow_control_status()


Expand Down