# 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")
Comments
0 B
|👍
/👎
0 B
|👍
/👎
0 B
|👍
/👎
0 B
|👍
/👎
0 B
|👍
/👎