test_priority_pool.py 2.8 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. import multiprocessing as mp
  2. import time
  3. import pytest
  4. import torch
  5. from src.server.runtime import Runtime
  6. from src.server.task_pool import PrioritizedTaskPool
  7. @pytest.mark.forked
  8. def test_priority_pools():
  9. outputs_queue = mp.SimpleQueue()
  10. results_valid = mp.Event()
  11. def dummy_pool_func(x):
  12. time.sleep(0.1)
  13. y = x**2
  14. outputs_queue.put((x, y))
  15. return (y,)
  16. class DummyBackend:
  17. def __init__(self, pools):
  18. self.pools = pools
  19. def get_pools(self):
  20. return self.pools
  21. pools = (
  22. PrioritizedTaskPool(dummy_pool_func, name="A", max_batch_size=1),
  23. PrioritizedTaskPool(dummy_pool_func, name="B", max_batch_size=1),
  24. )
  25. runtime = Runtime({str(i): DummyBackend([pool]) for i, pool in enumerate(pools)}, prefetch_batches=0)
  26. runtime.start()
  27. def process_tasks():
  28. futures = []
  29. futures.append(pools[0].submit_task(torch.tensor([0]), priority=1))
  30. futures.append(pools[0].submit_task(torch.tensor([1]), priority=1))
  31. time.sleep(0.01)
  32. futures.append(pools[1].submit_task(torch.tensor([2]), priority=1))
  33. futures.append(pools[0].submit_task(torch.tensor([3]), priority=2))
  34. futures.append(pools[0].submit_task(torch.tensor([4]), priority=10))
  35. futures.append(pools[0].submit_task(torch.tensor([5]), priority=0))
  36. futures.append(pools[0].submit_task(torch.tensor([6]), priority=1))
  37. futures.append(pools[1].submit_task(torch.tensor([7]), priority=11))
  38. futures.append(pools[1].submit_task(torch.tensor([8]), priority=1))
  39. for i, f in enumerate(futures):
  40. assert f.result()[0].item() == i**2
  41. results_valid.set()
  42. proc = mp.Process(target=process_tasks)
  43. proc.start()
  44. proc.join()
  45. assert results_valid.is_set()
  46. ordered_outputs = []
  47. while not outputs_queue.empty():
  48. ordered_outputs.append(outputs_queue.get()[0].item())
  49. assert ordered_outputs == [0, 5, 1, 2, 6, 8, 3, 4, 7]
  50. # 0 - first batch is loaded immediately, before everything else
  51. # 5 - highest priority task overall
  52. # 1 - first of several tasks with equal lowest priority (1)
  53. # 2 - second earliest task with priority 1, fetched from pool B
  54. # 6 - third earliest task with priority 1, fetched from pool A again
  55. # 8 - last priority-1 task, pool B
  56. # 3 - task with priority 2 from pool A
  57. # 4 - task with priority 10 from pool A
  58. # 7 - task with priority 11 from pool B