FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Add a test which provokes abort-during-read during 'run_in_transacti… · landrito/google-cloud-python@4c55a81 · GitHub

/ google-cloud-python Public
forked from googleapis/google-cloud-python

Commit 4c55a81

Browse files
tseaver
authored and
landrito
committed
Add a test which provokes abort-during-read during 'run_in_transaction'. (googleapis#3663)
1 parent 1c5199d commit 4c55a81

2 files changed

Lines changed: 114 additions & 21 deletions

File tree

‎spanner/google/cloud/spanner/session.py‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424

2525
# pylint: disable=ungrouped-imports
2626
from google.cloud.exceptions import NotFound
27+
from google.cloud.exceptions import GrpcRendezvous
2728
from google.cloud.spanner._helpers import _options_with_prefix
2829
from google.cloud.spanner.batch import Batch
2930
from google.cloud.spanner.snapshot import Snapshot
@@ -286,7 +287,7 @@ def run_in_transaction(self, func, *args, **kw):
286287
txn.begin()
287288
try:
288289
return_value = func(txn, *args, **kw)
289-
except GaxError as exc:
290+
except (GaxError, GrpcRendezvous) as exc:
290291
_delay_until_retry(exc, deadline)
291292
del self._transaction
292293
continue
@@ -318,15 +319,20 @@ def _delay_until_retry(exc, deadline):
318319
:type deadline: float
319320
:param deadline: maximum timestamp to continue retrying the transaction.
320321
"""
321-
if exc_to_code(exc.cause) != StatusCode.ABORTED:
322+
if isinstance(exc, GrpcRendezvous): # pragma: NO COVER see #3663
323+
cause = exc
324+
else:
325+
cause = exc.cause
326+
327+
if exc_to_code(cause) != StatusCode.ABORTED:
322328
raise
323329

324330
now = time.time()
325331

326332
if now >= deadline:
327333
raise
328334

329-
delay = _get_retry_delay(exc)
335+
delay = _get_retry_delay(cause)
330336
if delay is not None:
331337

332338
if now + delay > deadline:
@@ -336,7 +342,7 @@ def _delay_until_retry(exc, deadline):
336342
# pylint: enable=misplaced-bare-raise
337343

338344

339-
def _get_retry_delay(exc):
345+
def _get_retry_delay(cause):
340346
"""Helper for :func:`_delay_until_retry`.
341347
342348
:type exc: :class:`google.gax.errors.GaxError`
@@ -345,7 +351,7 @@ def _get_retry_delay(exc):
345351
:rtype: float
346352
:returns: seconds to wait before retrying the transaction.
347353
"""
348-
metadata = dict(exc.cause.trailing_metadata())
354+
metadata = dict(cause.trailing_metadata())
349355
retry_info_pb = metadata.get('google.rpc.retryinfo-bin')
350356
if retry_info_pb is not None:
351357
retry_info = RetryInfo()

‎spanner/tests/system/test_system.py‎

Lines changed: 103 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,8 @@
5757
'google-cloud-python-systest')
5858
DATABASE_ID = 'test_database'
5959
EXISTING_INSTANCES = []
60+
COUNTERS_TABLE = 'counters'
61+
COUNTERS_COLUMNS = ('name', 'value')
6062

6163

6264
class Config(object):
@@ -360,11 +362,6 @@ class TestSessionAPI(unittest.TestCase, _TestData):
360362
'description',
361363
'exactly_hwhen',
362364
)
363-
COUNTERS_TABLE = 'counters'
364-
COUNTERS_COLUMNS = (
365-
'name',
366-
'value',
367-
)
368365
SOME_DATE = datetime.date(2011, 1, 17)
369366
SOME_TIME = datetime.datetime(1989, 1, 17, 17, 59, 12, 345612)
370367
NANO_TIME = TimestampWithNanoseconds(1995, 8, 31, nanosecond=987654321)
@@ -554,9 +551,7 @@ def _transaction_concurrency_helper(self, unit_of_work, pkey):
554551

555552
with session.batch() as batch:
556553
batch.insert_or_update(
557-
self.COUNTERS_TABLE,
558-
self.COUNTERS_COLUMNS,
559-
[[pkey, INITIAL_VALUE]])
554+
COUNTERS_TABLE, COUNTERS_COLUMNS, [[pkey, INITIAL_VALUE]])
560555

561556
# We don't want to run the threads' transactions in the current
562557
# session, which would fail.
@@ -582,21 +577,19 @@ def _transaction_concurrency_helper(self, unit_of_work, pkey):
582577

583578
keyset = KeySet(keys=[(pkey,)])
584579
rows = list(session.read(
585-
self.COUNTERS_TABLE, self.COUNTERS_COLUMNS, keyset))
580+
COUNTERS_TABLE, COUNTERS_COLUMNS, keyset))
586581
self.assertEqual(len(rows), 1)
587582
_, value = rows[0]
588583
self.assertEqual(value, INITIAL_VALUE + len(threads))
589584

590585
def _read_w_concurrent_update(self, transaction, pkey):
591586
keyset = KeySet(keys=[(pkey,)])
592587
rows = list(transaction.read(
593-
self.COUNTERS_TABLE, self.COUNTERS_COLUMNS, keyset))
588+
COUNTERS_TABLE, COUNTERS_COLUMNS, keyset))
594589
self.assertEqual(len(rows), 1)
595590
pkey, value = rows[0]
596591
transaction.update(
597-
self.COUNTERS_TABLE,
598-
self.COUNTERS_COLUMNS,
599-
[[pkey, value + 1]])
592+
COUNTERS_TABLE, COUNTERS_COLUMNS, [[pkey, value + 1]])
600593

601594
def test_transaction_read_w_concurrent_updates(self):
602595
PKEY = 'read_w_concurrent_updates'
@@ -613,15 +606,48 @@ def _query_w_concurrent_update(self, transaction, pkey):
613606
self.assertEqual(len(rows), 1)
614607
pkey, value = rows[0]
615608
transaction.update(
616-
self.COUNTERS_TABLE,
617-
self.COUNTERS_COLUMNS,
618-
[[pkey, value + 1]])
609+
COUNTERS_TABLE, COUNTERS_COLUMNS, [[pkey, value + 1]])
619610

620611
def test_transaction_query_w_concurrent_updates(self):
621612
PKEY = 'query_w_concurrent_updates'
622613
self._transaction_concurrency_helper(
623614
self._query_w_concurrent_update, PKEY)
624615

616+
def test_transaction_read_w_abort(self):
617+
618+
retry = RetryInstanceState(_has_all_ddl)
619+
retry(self._db.reload)()
620+
621+
session = self._db.session()
622+
session.create()
623+
624+
trigger = _ReadAbortTrigger()
625+
626+
with session.batch() as batch:
627+
batch.delete(COUNTERS_TABLE, self.ALL)
628+
batch.insert(
629+
COUNTERS_TABLE,
630+
COUNTERS_COLUMNS,
631+
[[trigger.KEY1, 0], [trigger.KEY2, 0]])
632+
633+
provoker = threading.Thread(
634+
target=trigger.provoke_abort, args=(self._db,))
635+
handler = threading.Thread(
636+
target=trigger.handle_abort, args=(self._db,))
637+
638+
provoker.start()
639+
trigger.provoker_started.wait()
640+
641+
handler.start()
642+
trigger.handler_done.wait()
643+
644+
provoker.join()
645+
handler.join()
646+
647+
rows = list(session.read(COUNTERS_TABLE, COUNTERS_COLUMNS, self.ALL))
648+
self._check_row_data(
649+
rows, expected=[[trigger.KEY1, 1], [trigger.KEY2, 1]])
650+
625651
@staticmethod
626652
def _row_data(max_index):
627653
for index in range(max_index):
@@ -1103,3 +1129,64 @@ def __init__(self, db):
11031129

11041130
def delete(self):
11051131
self._db.drop()
1132+
1133+
1134+
class _ReadAbortTrigger(object):
1135+
"""Helper for tests provoking abort-during-read."""
1136+
1137+
KEY1 = 'key1'
1138+
KEY2 = 'key2'
1139+
1140+
def __init__(self):
1141+
self.provoker_started = threading.Event()
1142+
self.provoker_done = threading.Event()
1143+
self.handler_running = threading.Event()
1144+
self.handler_done = threading.Event()
1145+
1146+
def _provoke_abort_unit_of_work(self, transaction):
1147+
keyset = KeySet(keys=[(self.KEY1,)])
1148+
rows = list(
1149+
transaction.read(COUNTERS_TABLE, COUNTERS_COLUMNS, keyset))
1150+
1151+
assert len(rows) == 1
1152+
row = rows[0]
1153+
value = row[1]
1154+
1155+
self.provoker_started.set()
1156+
1157+
self.handler_running.wait()
1158+
1159+
transaction.update(
1160+
COUNTERS_TABLE, COUNTERS_COLUMNS, [[self.KEY1, value + 1]])
1161+
1162+
def provoke_abort(self, database):
1163+
database.run_in_transaction(self._provoke_abort_unit_of_work)
1164+
self.provoker_done.set()
1165+
1166+
def _handle_abort_unit_of_work(self, transaction):
1167+
keyset_1 = KeySet(keys=[(self.KEY1,)])
1168+
rows_1 = list(
1169+
transaction.read(COUNTERS_TABLE, COUNTERS_COLUMNS, keyset_1))
1170+
1171+
assert len(rows_1) == 1
1172+
row_1 = rows_1[0]
1173+
value_1 = row_1[1]
1174+
1175+
self.handler_running.set()
1176+
1177+
self.provoker_done.wait()
1178+
1179+
keyset_2 = KeySet(keys=[(self.KEY2,)])
1180+
rows_2 = list(
1181+
transaction.read(COUNTERS_TABLE, COUNTERS_COLUMNS, keyset_2))
1182+
1183+
assert len(rows_2) == 1
1184+
row_2 = rows_2[0]
1185+
value_2 = row_2[1]
1186+
1187+
transaction.update(
1188+
COUNTERS_TABLE, COUNTERS_COLUMNS, [[self.KEY2, value_1 + value_2]])
1189+
1190+
def handle_abort(self, database):
1191+
database.run_in_transaction(self._handle_abort_unit_of_work)
1192+
self.handler_done.set()

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL