Concurrency Model — threads, async/await, actors, CSP: একই সমস্যার চার উত্তর

লক্ষ্য: asyncio-র কঙ্কাল নিজের হাতে — আর দেখানো যে cooperative scheduling-এর সবচেয়ে লুকানো উপহারটা determinism: wall-clock বাদ দিলে একই seed মানে একই execution history, যা OS-thread জগতে অসম্ভব।

# sched.py — cooperative event loop: virtual clock + Channel
import collections
import heapq
import random


class Channel:
    """Buffered channel — CSP-র rendezvous primitive"""

    def __init__(self, cap=1):
        self.cap = cap
        self.buf = collections.deque()
        self.recv_q = collections.deque()   # অপেক্ষমাণ receiver (Task)
        self.send_q = collections.deque()   # parked sender (Task, value)


class Task:
    _next = 0

    def __init__(self, gen):
        Task._next += 1
        self.id = Task._next
        self.gen = gen
        self.value = None                   # পরের resume-এ পাঠানো মান


class Loop:
    def __init__(self, seed=7):
        self.ready = collections.deque()
        self.timers = []                    # (wake_tick, seq, Task)
        self.clock = 0                      # virtual time — কখনো পিছায় না
        self.seq = 0
        self.rng = random.Random(seed)      # seeded → deterministic
        self.alive = 0

    def spawn(self, gen):
        self.alive += 1
        self.ready.append(Task(gen))

    def wake(self, task, value):
        task.value = value
        self.ready.append(task)

    def step(self, task):
        """task-কে পরের yield পর্যন্ত চালাও, request অনুযায়ী সাজাও"""
        try:
            req = task.gen.send(task.value)
        except StopIteration:
            self.alive -= 1
            return
        task.value = None
        if isinstance(req, int):                      # sleep n tick
            self.seq += 1
            heapq.heappush(self.timers, (self.clock + req, self.seq, task))
        elif req[0] == 'send':
            _, ch, item = req
            if ch.recv_q:                             # rendezvous hand-off
                self.wake(ch.recv_q.popleft(), item)
                self.ready.append(task)
            elif len(ch.buf) < ch.cap:
                ch.buf.append(item)
                self.ready.append(task)
            else:
                ch.send_q.append((task, item))        # parked — backpressure
        elif req[0] == 'recv':
            _, ch = req
            if ch.buf:
                item = ch.buf.popleft()
                if ch.send_q:                         # জায়গা ফাঁকা — sender ঢোকাও
                    s, v = ch.send_q.popleft()
                    ch.buf.append(v)
                    self.ready.append(s)
                self.wake(task, item)
            else:
                ch.recv_q.append(task)
        else:
            raise ValueError(f"unknown request: {req!r}")

    def run(self):
        while self.alive:
            if not self.ready:
                if not self.timers:
                    raise RuntimeError("deadlock: সব task blocked")
                self.clock = max(self.clock, self.timers[0][0])
                while self.timers and self.timers[0][0] <= self.clock:
                    self.ready.append(heapq.heappop(self.timers)[2])
            self.step(self.ready.popleft())


# ── pipeline: ৩ producer → bounded channel → ২ consumer ──

state = {'done': 0}

def producer(loop, ch, name, n):
    for i in range(1, n + 1):
        yield loop.rng.randint(1, 3)              # "কাজ" — seeded সময়
        yield ('send', ch, f"{name}#{i}")
    state['done'] += 1

def closer(loop, ch, n_consumers):
    while state['done'] < 3:                      # সব producer শেষ হোক
        yield 1
    for _ in range(n_consumers):
        yield ('send', ch, None)                  # poison pill

def consumer(loop, ch):
    while True:
        item = yield ('recv', ch)
        if item is None:
            return                                # poison — শেষ
        print(f"    tick {loop.clock:>3}: consumed {item}")


if __name__ == '__main__':
    loop = Loop(seed=7)
    ch = Channel(cap=2)                           # bounded → backpressure
    for name in ('P1', 'P2', 'P3'):
        loop.spawn(producer(loop, ch, name, 3))
    loop.spawn(consumer(loop, ch))
    loop.spawn(consumer(loop, ch))
    loop.spawn(closer(loop, ch, n_consumers=2))
    loop.run()
    print(f"done at tick {loop.clock}")

এক run-এর output — এই কোডে কোনো wall-clock নেই, rng seeded; নিজে দুইবার চালিয়ে diff করুন: byte-for-byte identical আসার কথা।

    tick   1: consumed P2#1
    tick   2: consumed P1#1
    tick   2: consumed P3#1
    tick   3: consumed P1#2
    tick   3: consumed P3#2
    tick   4: consumed P2#2
    tick   4: consumed P3#3
    tick   6: consumed P1#3
    tick   6: consumed P2#3
done at tick 7

যা ঘটল: নয়টা item, দুই consumer, সব interleaving ঠিক একই — কারণ scheduling-এর প্রতিটা সিদ্ধান্ত (ready deque-র FIFO ক্রম, timer-এর seq tie-breaker, seeded rng) deterministic। OS-thread জগতে এই প্রশ্নই করা যায় না — race condition debug করার অসম্ভবতার মূল কারণ। আর cap=2 bounded channel-টাই খেয়াল করুন: producer দ্রুত হলে send_q-তে park হয় — backpressure, কোনো unbounded memory নেই।

নিজে বাড়ান:

1. seed বদলে দেখুন interleaving বদলায় কিন্তু ফলাফল invariant থাকে (৯টা item, সব consumed) — এটাই সঠিক concurrent program-এর সংজ্ঞা।
2. Consumer একটা মুছে দিন — closer এখন ২ poison পাঠায় কিন্তু receiver একজন: দ্বিতীয় poison কোথায় আটকায়? Deadlock-এর message-টা দেখুন।
3. ('join', other_task) request যোগ করুন — task completion-এর অপেক্ষা, WaitGroup-এর হাতে-লেখা রূপ।
4. একটা task-এ infinite yield 1 দিন — loop-এর deadlock detection কি ধরতে পারে? (পারে না — সে alive; starvation detection কেন কঠিন, লিখুন।)