py-generator-threads.py

#!/usr/bin/env python3
# coding: utf-8

import sys, time, uuid, random, threading, multiprocessing

def process(thread, gen, res, ch=[ "foo", "bar" ], dly=0.1):
	print("[GO] Thread:", thread)
	while gen:
		try:
			res[str(next(gen))] = random.choice(ch)
		except:
			break
		if dly > 0.0:
			time.sleep(dly)
			pass
		continue
	print("[OK] Thread:", thread)
	return thread

def stats(res):
	mapping = dict()
	for k, v in res.items():
		s = str(v)
		if s not in mapping:
			mapping[s] = 0
			pass
		mapping[s] += 1
		continue
	return mapping

if __name__ == "__main__":
	pass
else:
	sys.exit(0)

if len(sys.argv) != 2:
	sys.exit("Thread argument(1) required!")

epoch = time.time()
time.sleep(1.0)
print("...GENERATING:", (time.time() - epoch))
tx_count     = 10_000
thread_count = int(sys.argv[1])
transactions = [ uuid.uuid4() for i in range(tx_count) ]
generator    = iter(transactions) # iterator for parallel-processing
results      = dict() # empty dict
pool         = [
	threading.Thread(
		target=process,
		args=(1+i, generator, results)
	)
	for i in range(thread_count)
]
print("...OKAY      :", (time.time() - epoch))
print("...PROCESSING:", (time.time() - epoch))
start = time.time()
[ t.start() for t in pool ] # start all

while running := [ t for t in pool if t.is_alive() ]:
	print("...Threads:", len(running),
		(100.0 * len(results) / len(transactions)),
		(time.time() - epoch)
	)
	time.sleep(0.3)
	continue

[ t.join()  for t in pool ] # wait  all
end = time.time()
print("...DONE:      ", (time.time() - epoch))

print("DeltaT: ", (end - start))
print("Stats:  ", stats(results))
print("Totals: ", len(results.keys()), len(transactions))
print("Missing:", [ tx for tx in transactions if str(tx) not in results ])
#print("True:   ", len([ tid for tid, result in results.items() if     result  ]))
#print("False:  ", len([ tid for tid, result in results.items() if not result  ]))