Curious how sqlite3 would handle different race conditions in a multithreaded environment, I created a simple module called deterministic
for serializing the execution of multiple threads that lets the programmer explicitly program in the spots where thread-switching happens. Hopefully this example usage will explain the idea better:
import deterministic
def main():
deterministic.run(consumer0, consumer1, consumer2)
def consumer0():
print('consumer 0 yields to consumer 2')
yield 2
print('consumer 0 exits')
def consumer1():
print('consumer 1 yields to next consumer')
yield
print('consumer 1 exits')
def consumer2():
print('consumer 2 yields to consumer 1')
yield 1
print('consumer 2 yields to next consumer')
yield
print('consumer 2 exits')
if __name__ == '__main__':
main()
Which outputs:
consumer 0 yields to consumer 2
consumer 2 yields to consumer 1
consumer 1 yields to next consumer
consumer 2 yields to next consumer
consumer 0 exits
consumer 1 exits
consumer 2 exits
Again, the idea is to experiment with how real multi-threaded code behaves, but in a highly controller environment.
The actual code follows below. Some specific feedback I'm hoping for:
- Is there a synchronization primitive that better models what I'm trying to do than simple locking?
- Even if not, is there a better way to structure the code to make it easier to reason about?
- Finally, general coding and style advice is more than welcome
import threading, time, types
def run(*consumers):
sync = Syncrhonizer(len(consumers))
_run_threads(
threading.Thread(target=DeterministicRunner(sync, i, c).run)
for i, c in enumerate(consumers)
)
class Syncrhonizer:
def __init__(self, num_instances):
self._instances = list(range(num_instances))
self._lock = threading.Lock()
self._current = 0
self._want_abort = False
def lock(self, idx):
while True:
self._lock.acquire()
if self._want_abort:
self._lock.release()
raise SyncrhonizerAbort()
if self._current == idx:
return
self._lock.release()
time.sleep(0.001)
def die(self):
self._want_abort = True
if self._lock.locked():
self._lock.release()
def yield_to(self, idx=None):
if idx is None:
idx = self._get_next_instance()
if idx not in self._instances:
self.die()
raise Exception('Can not yield to non-existent instance "%s"' % idx)
self._current = idx
self._lock.release()
def _get_next_instance(self):
try:
return next(n for n in self._instances if n > self._current)
except StopIteration:
return self._instances[0]
def remove(self, idx):
self._instances.remove(idx)
if self._lock.locked() and self._instances:
self.yield_to()
class SyncrhonizerAbort(Exception):
"""Abort execution because of an issue in another thread"""
class DeterministicRunner:
def __init__(self, syncer, idx, method):
self._syncer = syncer
self._idx = idx
self._g = method()
if not isinstance(self._g, types.GeneratorType):
raise Exception('Not a generator')
def run(self):
try:
while True:
self._syncer.lock(self._idx)
n = next(self._g)
self._syncer.yield_to(n)
except StopIteration:
pass
finally:
self._syncer.remove(self._idx)
def _run_threads(threads):
threads = list(threads)
for t in threads:
t.start()
for t in threads:
t.join()