blob: 00e69fbe405c4154afe7445f65f52837df8ec214 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
|
"""
One connection per group for the whole run, and a ceiling on how many are being
negotiated at once — the application's `ConnectionPool` rules
(docs/MESHBAY_DESIGN.md §9.11).
The ceiling is on concurrency, never a batch: a group whose node does not
answer holds one place for its own deadline and delays nobody else. Six stays
far below what the hub admits per account (32 pending offers), even with the
person's application and a browser open at the same time.
"""
import asyncio
import contextlib
from .account import Account
from .groups import open_group
from .session import GroupSession
MAX_IN_FLIGHT = 6
class Pool:
def __init__(self, account: Account, max_in_flight: int = MAX_IN_FLIGHT):
self.account = account
self._gate = asyncio.Semaphore(max_in_flight)
self._sessions: dict[str, asyncio.Task] = {}
async def session(self, group: dict) -> GroupSession:
"""The group's session, negotiated once and shared by every caller."""
task = self._sessions.get(group["id"])
if task is None:
task = asyncio.ensure_future(self._open(group))
self._sessions[group["id"]] = task
return await asyncio.shield(task)
async def _open(self, group: dict) -> GroupSession:
async with self._gate:
return await open_group(self.account, group)
async def close(self) -> None:
for task in self._sessions.values():
if task.done() and not task.cancelled() and task.exception() is None:
await task.result().close()
else:
task.cancel()
with contextlib.suppress(BaseException):
await task
self._sessions.clear()
|