Query Session Pool redesign #476
Conversation
|
🌋 Here are results of SLO test for Python SDK over Table Service: |
|
🌋 Here are results of SLO test for Python SDK over Query Service: |
vgvoleg
force-pushed
the
new_session_pool
branch
from
ff838aa
to
4a5c58e
Compare
vgvoleg
force-pushed
the
new_session_pool
branch
from
3526fc2
to
c087ffb
Compare
|
| pool._size = target_size | ||
| ids = set() | ||
|
|
||
| for i in range(1, target_size + 1): |
There was a problem hiding this comment.
why not
for i in range (target_size)
?
There was a problem hiding this comment.
just for better readability here
assert pool._current_size == i
I think it does not matter where to drop a +1 so it was a coin flip
| ids.add(session._state.session_id) | ||
|
|
||
| with pytest.raises(asyncio.TimeoutError): | ||
| await asyncio.wait_for(pool.acquire(), timeout=0.5) |
| await pool.release(session) | ||
|
|
||
| session = await pool.acquire() | ||
| assert pool._current_size == target_size |
There was a problem hiding this comment.
what about save session id before release and compare aquired session id with exactly saved id instead of one of?
| docker_project.stop() | ||
| try: | ||
| await asyncio.wait_for(pool.acquire(), timeout=0.5) | ||
| except ydb.Error: |
There was a problem hiding this comment.
why ydb.Error instead of TimeoutError?
There was a problem hiding this comment.
Because we will have an error from driver after ydb scale down
ydb/aio/query/pool.py
Outdated
| done, _ = await asyncio.wait((queue_get, task_stop), return_when=asyncio.FIRST_COMPLETED) | ||
| if task_stop in done: | ||
| queue_get.cancel() | ||
| return await self._create_new_session() # TODO: not sure why |
|
|
||
| await asyncio.gather(*tasks) | ||
|
|
||
| logger.debug("All session were deleted.") |
There was a problem hiding this comment.
what about add __del__ for stop the pool? may be hard stop: without wait anything and with a log message?
fot prevent leaks
ydb/aio/query/pool.py
Outdated
| @@ -96,13 +158,17 @@ async def __aexit__(self, exc_type, exc_val, exc_tb): | |||
|
|
|||
|
|
|||
| class SimpleQuerySessionCheckoutAsync: | |||
| def __init__(self, pool: QuerySessionPoolAsync): | |||
| def __init__(self, pool: QuerySessionPoolAsync, timeout: Optional[float] = None): | |||
There was a problem hiding this comment.
why do you need timeout in async?
ydb/query/pool.py
Outdated
| start = time.monotonic() | ||
| if session is None and self._current_size == self._size: | ||
| try: | ||
| _, session = self._queue.get(block=True, timeout=timeout) |
There was a problem hiding this comment.
it is long operation under mutex lock.
How you will enforce smaller timeout on parallel request?
vgvoleg
force-pushed
the
new_session_pool
branch
from
9489f7a
to
1d26890
Compare
ydb/aio/query/pool.py
Outdated
| done, _ = await asyncio.wait((queue_get, task_stop), return_when=asyncio.FIRST_COMPLETED) | ||
| if task_stop in done: | ||
| queue_get.cancel() | ||
| return await self._create_new_session() |
There was a problem hiding this comment.
why it create a session on stop pool?
may be runtime error - similar to acquire on stopped pool?
ydb/aio/query/pool.py
Outdated
| if self._should_stop.is_set() or self._loop.is_closed(): | ||
| return | ||
|
|
||
| self._loop.call_soon(functools.partial(self.stop)) |
vgvoleg
force-pushed
the
new_session_pool
branch
from
3f425bf
to
ef7e50e
Compare
vgvoleg
force-pushed
the
new_session_pool
branch
from
ef7e50e
to
4bb2566
Compare
Pull request type
Please check the type of change your PR introduces:
What is the current behavior?
Issue Number: N/A
What is the new behavior?
Other information