[mutianf]

Add the shared Cloud Bigtable accelerator system-test harness (_harness.py, _base_async.py, _base_autogen.py, conftest.py, pytest.ini) and the concurrent/multi-process workload system tests (test_concurrency_async.py, test_concurrency_autogen.py).

Part of Stack #18691 (stacked on #18782).

[gemini-code-assist]

Code Review

This pull request introduces integration and concurrency tests for the Google Cloud Bigtable accelerator, including pytest configurations, base test classes, a state-tracking test harness, and multi-process/multi-threaded concurrency tests. The review feedback suggests several key improvements: cleaning up empty row dictionaries from the state model in _harness.py to prevent memory retention, switching the multiprocessing start method from fork to spawn in test_concurrency_async.py to avoid gRPC-related deadlocks, and adopting a shared monotonic deadline pattern when joining background processes during cleanup to prevent excessive blocking.

[gemini-code-assist]

When a column qualifier is completely deleted and the column dictionary becomes empty, it is popped from the row. However, if the row itself becomes empty (i.e., has no more column families/qualifiers), the row key is still left in self._rows as an empty dictionary. This causes keys() to return empty rows and leads to unnecessary memory retention. We should clean up the row key if the row dictionary becomes empty.

Suggested change

	if not col:
	self._rows[row_key].pop((family, qualifier), None)
	if not col:
	self._rows[row_key].pop((family, qualifier), None)
	if not self._rows[row_key]:
	self._rows.pop(row_key, None)

[gemini-code-assist]

Similarly to delete_range_from_column, if all families/qualifiers are deleted from a row during delete_from_family, the row key should be completely removed from self._rows to prevent empty dictionaries from lingering in the state.

Suggested change

	def delete_from_family(self, row_key: bytes, family: str) -> DeleteAllFromFamily:
	row = self._rows.get(row_key)
	if row is not None:
	for key in [k for k in row if k[0] == family]:
	del row[key]
	return DeleteAllFromFamily(family)
	def delete_from_family(self, row_key: bytes, family: str) -> DeleteAllFromFamily:
	row = self._rows.get(row_key)
	if row is not None:
	for key in [k for k in row if k[0] == family]:
	del row[key]
	if not row:
	self._rows.pop(row_key, None)
	return DeleteAllFromFamily(family)

[gemini-code-assist]

Using the fork start method in Python programs that use gRPC or other multi-threaded C-extensions can lead to deadlocks, crashes, or undefined behavior in child processes because background threads are not inherited. Since the child process workload is fully self-contained and all arguments are pickleable, using spawn is much safer and more robust across different platforms (including macOS and Windows).

Suggested change

	ctx = multiprocessing.get_context("fork")
	ctx = multiprocessing.get_context("spawn")

[gemini-code-assist]

When joining multiple background processes during cleanup, using a fixed timeout per process can cause the cleanup block to block for an excessively long time if multiple processes hang (e.g., up to 120 seconds for 4 processes). Using a shared timeout/deadline pattern based on a monotonic clock ensures that the entire cleanup block respects the overall timeout limit.

Suggested change

	for proc in procs:
	proc.join(timeout=30)
	if proc.is_alive():
	proc.kill()
	proc.join()
	import time
	deadline = time.monotonic() + 30.0
	for proc in procs:
	timeout = max(0.0, deadline - time.monotonic())
	proc.join(timeout=timeout)
	if proc.is_alive():
	proc.kill()
	proc.join()

References

1. When joining multiple background threads during cleanup, use a shared timeout/deadline pattern (e.g., calculating remaining time based on a monotonic clock) instead of a fixed timeout per thread, to prevent the cleanup block from blocking for an excessively long time if multiple threads hang.