This question evaluates understanding of concurrency and synchronization primitives, testing the ability to design a thread-safe type-based mutex that permits concurrent execution of tasks with the same type while preventing concurrent execution across different types.
You are given a multithreaded task runner. Each task has:
task_id: int
task_type: "gen" | "score"
duration: int
(seconds)
Each task runs in its own thread and calls a shared lock before and after doing work.
from dataclasses import dataclass
from typing import Literal
@dataclass
class SampleTask:
task_id: int
task_type: Literal["gen", "score"]
duration: int
class TaskLock:
def acquire(self, task: SampleTask) -> None:
raise NotImplementedError
def release(self, task: SampleTask) -> None:
raise NotImplementedError
The task runner does:
lock.acquire(task)
task.duration
lock.release(task)
Implement a MutexTaskLock(TaskLock) such that:
task_type
may run
concurrently
.
task_type
must
not
run at the same time (i.e., if any
gen
task is running, no
score
task may run, and vice versa).
You may use Python threading primitives such as threading.Lock, threading.Condition, or semaphores.
wait
).