Concurrency

def worker(name: str) -> None:
    print(f"{name} starting")
    time.sleep(1)
    print(f"{name} done")


threads = [threading.Thread(target=worker, args=(f"worker-{i}",)) for i in range(5)]
for t in threads:
    t.start()

for t in threads:
    t.start()
print("all done")

start = time.perf_counter()
threads = [threading.Thread(target=worker, args=(f"worker-{i}",)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"all done after {time.perf_counter() - start:.2f} s")

counter = 0


def increment(times: int) -> None:
    global counter
    for _ in range(times):
        value = counter
        time.sleep(0)
        counter = value + 1

for _run in range(3):
    counter = 0
    threads = [threading.Thread(target=increment, args=(10_000,)) for _ in range(2)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(counter)

def increment_plain(times: int) -> None:
    global counter
    for _ in range(times):
        counter += 1


for _run in range(3):
    counter = 0
    threads = [threading.Thread(target=increment_plain, args=(10_000,)) for _ in range(2)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(counter)

# Only lucky : counter += 1 is still three steps, and a thread switch between two of them
# loses an update. Python switches so rarely inside this loop that the total usually comes out right.

lock = threading.Lock()


def increment_locked(times: int) -> None:
    global counter
    for _ in range(times):
        with lock:
            value = counter
            time.sleep(0)
            counter = value + 1


for target in (increment, increment_locked):
    for _run in range(3):
        counter = 0
        start = time.perf_counter()
        threads = [threading.Thread(target=target, args=(10_000,)) for _ in range(2)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        print(f"{target.__name__:17} total {counter:6}  {time.perf_counter() - start:.2f} s")

start = time.perf_counter()
for _ in range(4):
    count_primes(200_000)
print(f"one thread   : {time.perf_counter() - start:.2f} s")

start = time.perf_counter()
threads = [threading.Thread(target=count_primes, args=(200_000,)) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"four threads : {time.perf_counter() - start:.2f} s")

# The GIL lets only one thread run Python bytecode at a time, so four threads share one core's
# worth of interpreter. Nothing was waiting, so there was never a moment to hand over.

start = time.perf_counter()
for _ in range(4):
    count_primes(200_000)
alone = time.perf_counter() - start

start = time.perf_counter()
with Pool(4) as pool:
    results = pool.map(count_primes, [200_000] * 4)
pooled = time.perf_counter() - start
print(f"one process {alone:.3f} s, pool of four {pooled:.3f} s")

print("cores on this machine :", os.cpu_count())
print(f"speed-up : {alone / pooled:.1f}")
# The pool has four workers, so the speed-up stays below four, however many cores there are.
# Starting the workers and pickling the arguments take part of it away.

for limit in (200_000, 1_000):
    start = time.perf_counter()
    for _ in range(4):
        count_primes(limit)
    alone = time.perf_counter() - start

    start = time.perf_counter()
    with Pool(4) as pool:
        results = pool.map(count_primes, [limit] * 4)
    pooled = time.perf_counter() - start
    print(f"limit {limit:>7} : one process {alone:.3f} s, pool of four {pooled:.3f} s")

with Pool(4) as pool:
    handles = [pool.apply_async(count_primes, (limit,)) for limit in (50_000, 100_000, 150_000)]
    print("apply_async :", [h.get() for h in handles])

async def fetch(name: str, seconds: float) -> str:
    print(f"{name} started")
    await asyncio.sleep(seconds)
    return f"{name} took {seconds}s"


start = time.perf_counter()
for name, seconds in (("a", 1), ("b", 2), ("c", 1)):
    print(await fetch(name, seconds))
print(f"one after another : {time.perf_counter() - start:.2f} s")

start = time.perf_counter()
print(await asyncio.gather(fetch("a", 1), fetch("b", 2), fetch("c", 1)))
print(f"gather            : {time.perf_counter() - start:.2f} s")

async def fetch(name: str, seconds: float) -> str:
    print(f"{name} started")
    await asyncio.sleep(seconds)
    if seconds < 0:
        raise ValueError(f"{name} failed")
    return f"{name} took {seconds}s"


results = await asyncio.gather(fetch("a", 1), fetch("bad", -1), return_exceptions=True)
print("gather keeps the others :", results)

try:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fetch("a", 1))
        tg.create_task(fetch("bad", -1))
except* ValueError as group:
    print("TaskGroup cancels the rest and raises :", group.exceptions)

# TaskGroup : when one call fails, the other two results are usually of no use, and the group stops them at once.
# gather with return_exceptions=True suits calls that do not depend on each other and where a partial result is worth having.

async def polite(name: str) -> None:
    print(f"{name} on {threading.current_thread().name}")
    await asyncio.sleep(1)


start = time.perf_counter()
await asyncio.gather(polite("a"), polite("b"), polite("c"))
print(f"asyncio.sleep : {time.perf_counter() - start:.2f} s")

async def blocking(name: str) -> None:
    print(f"{name} on {threading.current_thread().name}")
    time.sleep(1)


start = time.perf_counter()
await asyncio.gather(blocking("a"), blocking("b"), blocking("c"))
print(f"time.sleep    : {time.perf_counter() - start:.2f} s")

# time.sleep does not hand control back to the event loop, so the other coroutines cannot start.
# All three share one thread, and they run one after another. One blocking call anywhere on the
# path, or one library without an async version, and the concurrency is gone without an error.