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()
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 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()