Fix handler memory leak, get rid of mp.Manager

[justheuristic]

This code removes the memory leak from somewhere within handler.py that has something to do with mp SyncManger.

Tested properties:

• testcorrect inference outputs

• test that pushes are processed by the correct handler
  
  

• test against memory leaks with the leak notebook

• overnight memory leak test

• make sure rebalancing is handled correctly (needs @borzunov )

After review:

• async with context manager
• enum or global constants for message codes
• rm todos

[borzunov]

Let's make with in self._iterate_inference_steps()?

[borzunov]

nit: Please do it like this, but with enums

Suggested change

	else:
	assert code == "PUSH", f"unexpected code: {code}"
	maybe_session_queue = self._session_queues.get(session_id)
	if maybe_session_queue is not None:
	maybe_session_queue.put_nowait(payload)
	elif code == "PUSH":
	maybe_session_queue = self._session_queues.get(session_id)
	if maybe_session_queue is not None:
	maybe_session_queue.put_nowait(payload)
	else:
	raise RuntimeError(f"Unexpected code: {code}")

[justheuristic]

applied

[borzunov]

Suggested change

	self._handler_queues = handler_queues
	self._handler_index = handler_index
	self._own_queue = handler_queues[handler_index]
	self._listener_task: Optional[asyncio.Task] = None
	self._session_queues: Dict[str, asyncio.Queue] = dict()
	self._session_handlers: Dict[str, int] = dict()
	self._handler_event_queues = handler_event_queues
	self._handler_index = handler_index
	self._own_event_queue = handler_event_queues[handler_index]
	self._listener_task: Optional[asyncio.Task] = None
	self._session_queues: Dict[str, asyncio.Queue] = dict()
	self._session_handlers: Dict[str, int] = dict()

[justheuristic]

renamed

[borzunov]

Suggested change

	await self._session_queues[session_id].put(request)
	self._session_queues[session_id].put_nowait(request)

Thus, this functon doesn't need to be async - please make it a usual func

[justheuristic]

[borzunov]

Suggested change

	self._session_queues: Dict[str, asyncio.Queue] = dict()
	self._session_handlers: Dict[str, int] = dict()
	self._session_queues: Dict[str, asyncio.Queue] = {}
	self._session_handlers: Dict[str, int] = {}

[justheuristic]

[borzunov]

Suggested change

	await self._session_queues.pop(session_id).put(None) # put None so that the get task will not hang
	self._session_queues.pop(session_id).put_nowait(None) # put None so that the get task will not hang

The contextmanager doesn't have to be async, please make it a usual contextmanager

[justheuristic]

[borzunov]

I've valdated this on local and geo-distributed swarms, seems to work.