高级农民
- 积分
- 3673
- 大米
- 颗
- 鳄梨
- 个
- 水井
- 尺
- 蓝莓
- 颗
- 萝卜
- 根
- 小米
- 粒
- 学分
- 个
- 注册时间
- 2014-9-5
- 最后登录
- 1970-1-1
|
感谢分享(Scale.AI 的面经是真的少。。) ~ 自己写了个 python 简陋版的,sample input 2 的 output 跑出来和 lz 给的例子一样
- from collections import deque
- class Task:
- def __init__(self, name, duration = 1):
- self.name = name
- self.duration = duration
- self.stage = 0
-
- self.prev_workers = set()
-
- class Worker:
- def __init__(self, name):
- self.name = name
-
- self.task = None
- self.task_start_time = None
- self.task_end_time = None
-
- def assign_task(self, task, task_start_time):
- self.task = task
- self.task_start_time = task_start_time
- self.task_end_time = task_start_time + task.duration
-
- def finish_task(self):
- print(f"Worker {self.name} finished task {self.task.name} for stage L{self.task.stage}")
- self.task.stage += 1
- finished_task = self.task
-
- self.task_start_time = None
- self.task_end_time = None
- self.task = None
-
- return finished_task
-
- def task_finished(self, current_time):
- return current_time >= self.task_end_time
-
- class Scheduler:
- def __init__(self):
- self.time = -1
- self.task_queue = deque()
-
- self.idle_worker_queue = deque()
- self.busy_worker_queue = deque()
-
- def add_task(self, task, stage):
- self.task_queue.append(task)
-
- def add_worker(self, worker):
- self.idle_worker_queue.append(worker)
-
- def assign_worker(self, task, worker):
- worker.assign_task(task, self.time)
-
- def all_work_completed(self):
- if len(self.task_queue) > 0:
- return False
- if len(self.busy_worker_queue) > 0:
- return False
-
- return True
-
-
- def time_lapse(self):
- self.time += 1
- print(f"Time: {self.time}")
- # Check completed tasks and refresh idle worker queue
- for _ in range(len(self.busy_worker_queue)):
- worker = self.busy_worker_queue.popleft()
- if worker.task_finished(self.time):
- finished_task = worker.finish_task()
-
- self.idle_worker_queue.append(worker)
- if finished_task.stage <= 2:
- self.task_queue.append(finished_task)
- else:
- self.busy_worker_queue.append(worker)
- # Try to find open task and assign to idle worker
- if len(self.idle_worker_queue) == 0:
- return
- else:
- for _ in range(len(self.idle_worker_queue)):
- worker = self.idle_worker_queue.popleft()
- found_job = False
- for _ in range(len(self.task_queue)):
- task = self.task_queue.popleft()
- if worker not in task.prev_workers:
- found_job = True
- self.assign_worker(task, worker)
- self.busy_worker_queue.append(worker)
- print(f"Assigning {worker.name} to Task {task.name} for L{task.stage}")
- break
- else:
- self.task_queue.append(task)
- if not found_job:
- self.idle_worker_queue.append(worker)
-
- def test():
- scheduler = Scheduler()
- tasks = [Task('A'), Task('B')]
- workers = [Worker('X'), Worker('Y'), Worker('Z')]
-
- for task in tasks:
- scheduler.add_task(task, 0)
- for worker in workers:
- scheduler.add_worker(worker)
-
- while not scheduler.all_work_completed():
- scheduler.time_lapse()
-
- test()
- # *Sample input 2: *
- # tasks = [Task('A'), Task('B')]
- # workers = [Worker('X'), Worker('Y'), Worker('Z')]
- # *Sample output 2 (one possible correct answer): *
- # 0
- # Assigning X to Task A for L0
- # Assigning Y to Task B for L0
- # 1
- # Worker X finished Task A for L0
- # Worker Y finished Task B for L0
- # Assigning Z to Task A for L1
- # Assigning X to Task B for L1
- # 2
- # Worker Z finished Task A for L1
- # Worker X finished Task B for L1
- # Assigning Y to Task A for L2
- # Assigning Z to Task B for L2
- # 3
- # Worker Y finished Task A for L2
- # Worker Z finished Task B for L2
- # Total time taken: 3 min
-
复制代码 |
|