Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- # try this with Python 3.15.0b1 and 3.15.0b1t
- # the freethreaded version looses results
- # if time.sleep(0) is removed, it's even worse
- # the non-freethreaded version does not have this issue
- import time
- from threading import Thread, synchronized_iterator
- from itertools import batched
- @synchronized_iterator
- def numbers(n):
- for x in range(n):
- yield x
- def consume(iterable, results):
- numbers_append = results.append
- for item in iterable:
- numbers_append(item)
- time.sleep(0)
- def is_ok(numbers):
- for batch in batched(numbers, 2):
- if len(batch) == 2:
- last, current = batch
- else:
- # handling odd length of numbers
- last, current = current, batch[0]
- if last + 1 != current:
- return False
- return True
- THREADS = 2
- success = 0
- errors = 0
- all_results = []
- for try_number in range(100):
- print(f"Try number: {try_number + 1}")
- results = []
- all_results.append(results)
- gen = numbers(1_001)
- threads = [Thread(target=consume, args=(gen, results)) for _ in range(THREADS)]
- for t in threads:
- t.start()
- for t in threads:
- t.join()
- if is_ok(results):
- success += 1
- else:
- errors += 1
- print(f"{success} times success and {errors} errors")
Advertisement