Preemptive Resource

Source and Output

"""Example: preemptive resource where a high-priority process evicts a low-priority one."""

from asimpy import Environment, Interrupt, Preempted, PreemptiveResource, Process
from _util import example

RESOURCE_CAPACITY = 1  # single-slot resource
LOW_PRIORITY = 2  # lower priority (higher number)
HIGH_PRIORITY = 0  # higher priority (lower number)
LOW_WORK = 8  # ticks the low-priority worker needs
HIGH_ARRIVE = 3  # tick at which the high-priority worker arrives
HIGH_WORK = 2  # ticks the high-priority worker needs


class LowPriorityWorker(Process):
    def init(self, resource):
        self._resource = resource
        self._remaining = LOW_WORK

    async def run(self):
        while self._remaining > 0:
            self._env.log("low", f"acquire (remaining={self._remaining})")
            try:
                await self._resource.acquire(priority=LOW_PRIORITY)
                self._env.log("low", "working")
                await self.timeout(self._remaining)
                self._remaining = 0
                self._resource.release()
                self._env.log("low", "done")
            except Interrupt as exc:
                if isinstance(exc.cause, Preempted):
                    used = self.now - exc.cause.usage_since
                    self._remaining -= used
                    self._env.log(
                        "low",
                        f"preempted after {used} ticks, {self._remaining} remaining",
                    )
                else:
                    raise


class HighPriorityWorker(Process):
    def init(self, resource):
        self._resource = resource

    async def run(self):
        await self.timeout(HIGH_ARRIVE)
        self._env.log("high", "acquire (preempt=True)")
        await self._resource.acquire(priority=HIGH_PRIORITY, preempt=True)
        self._env.log("high", "working")
        await self.timeout(HIGH_WORK)
        self._resource.release()
        self._env.log("high", "done")


def main():
    env = Environment()
    resource = PreemptiveResource(env, capacity=RESOURCE_CAPACITY)
    LowPriorityWorker(env, resource)
    HighPriorityWorker(env, resource)
    env.run()
    return env


if __name__ == "__main__":
    example(main)

time	name	event
0	low	acquire (remaining=8)
0	low	working
3	high	acquire (preempt=True)
3	high	working
3	low	preempted after 3 ticks, 5 remaining
3	low	acquire (remaining=5)
5	high	done
5	low	working
10	low	done

Key Points

PreemptiveResource uses priority numbers where a lower number means higher priority (0 is best). When preempt=True and the resource is full, a new request with better priority ejects the worst current user.
The evicted process receives an Interrupt whose cause is a Preempted dataclass with two fields: by (the process that caused the eviction) and usage_since (the simulation time when the evicted process acquired the resource).
Do not call release() when handling a Preempted interrupt. The preemptor has already removed the evicted process from the user list. Calling release() a second time will raise RuntimeError.
The low-priority worker tracks remaining work and re-acquires after the high-priority worker finishes. It resumes at t=5 and completes at t=10 (5 remaining ticks from where it was interrupted at t=3).

Check for Understanding

What would change if HighPriorityWorker called acquire(priority=HIGH_PRIORITY, preempt=False)? Would the low-priority worker be interrupted, and when would the high-priority worker start working?