Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings
This repository was archived by the owner on Apr 1, 2026. It is now read-only.

fix: add lock to flow control #899

Merged
daniel-sanche 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.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 7 additions & 4 deletions 11 google/cloud/bigtable/batcher.py
View file
Open in desktop
Original file line number Diff line number Diff line change
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()

@daniel-sanche daniel-sanche Dec 12, 2023 •
edited
Loading

Copy link
Copy Markdown
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
Morty Proxy This is a proxified and sanitized view of the page, visit original site.