jM3dZddlmZddlZddlZddlZddlmZddlm Z ddl m Z Gdd Z Gd d Z ejdd ZGddZdZ ddZ ddZdS)aCDeterministic cancellation testing for the asyncio dialects. An asyncio task may be cancelled at any ``await``, and reproducing a specific one of those points by racing ``task.cancel()`` against real network traffic is unreliable. Here the choice is made exact. :func:`.greenlet_spawn` drives every await performed by an asyncio-driven SQLAlchemy operation, ``connect()`` included, so instrumenting the greenlet it switches into gives a stable enumeration of a scenario's IO points, and cancelling at point ``N`` is reproducible run to run. The cancellation is delivered to the task the application is running, which is not necessarily the task performing the await; SQLAlchemy runs some operations, such as the ``close()`` performed by an async context manager on exit, in a task of its own. Two things are asserted afterwards: * Every DBAPI connection the pool creates is eventually closed through ``Pool._close_connection()``. A connection that is never closed is leaked; for an asyncio driver the garbage collector cannot close it either, since closing requires the event loop. * Once the application's task has finished and garbage has been collected, no pool slot is still checked out. A slot still checked out at that point is lost: either nothing remains that could return it, or it is being returned by a task SQLAlchemy left running in the background, which the application has no way to wait for. In the latter case, an application that disposes of its engine at that point has the connection returned into the disposed pool afterwards, where nothing closes it. ) annotationsN)mock)exc) concurrencyc"eZdZdZdZdZdZdS)_InstrumentedAwaitablez6An awaitable that invokes ``hook`` when it is awaited. awaitablehookc"||_||_dSNr )selfr r s /home/lighthouse/badge-speaker-benchmark/.benchmark-venv-community/lib64/python3.11/site-packages/sqlalchemy/testing/cancellation.py__init__z_InstrumentedAwaitable.__init__:s" ch||j|jSr)r r __await__rs rrz _InstrumentedAwaitable.__await__>s* $.!!!~'')))rN)__name__ __module__ __qualname____doc__ __slots__rrrrr r 5s=@@%I*****rr cReZdZdZd dZedZedZdZdZ dS) AwaitPointsaqCounts await points, and cancels the running task at each of the ones given. ``cancel_at`` is a tuple of await point numbers; each one is a separate ``task.cancel()`` call. asyncio delivers a given cancellation exactly once, so a second number models a second, independent canceller reaching a task that is still cleaning up after the first. rc>d|_||_g|_d|_dSNr)_count _cancel_at _cancelled_at_task)r cancel_ats rrzAwaitPoints.__init__Ns# # rc|jS)z*The number of await points counted so far.)r rs rcountzAwaitPoints.countTs {rc*t|jS)zMDescriptions of the awaitables at which a cancellation was delivered.)tupler"rs r cancelled_atzAwaitPoints.cancelled_atZs T'(((rcBtj||_|jS)zZRun ``coro`` in a new task, which is the task that cancellations are delivered to.)asyncio create_taskr#)rcoros rr,zAwaitPoints.create_taskas(.. zrc 4|xjdz c_|j|jvrdS|jt |dt ||j|jntj}| Jd| dS)NrzQcancellation can only be delivered to an operation running within an asyncio task) r r!r"appendgetattrreprr#r+ current_taskcancel)rr tasks r__call__zAwaitPoints.__call__hs q ;do - - F !! I~tI ? ?   "Z3tzz9M9O9O %  rNr) rrrrrpropertyr&r)r,r6rrrrrCs X ))X) rrrc#~Kt|tjj}|j|jfdfd}fd}t j|d|5t j|d|5Vdddn #1swxYwYddddS#1swxYwYdS)zlCount the await points performed within the block, cancelling the running task at each of ``cancel_at``.c6|jr|St|Sr)deadr )greenletresultpointss r instrumentz await_points..instruments$ = M%ff555rc0||g|Ri|Srr)rargkwr? real_switchs rswitchzawait_points..switchs/z$ D =3 = = =" = =>>>rc*||g|RSrr)rrAr? real_throws rthrowzawait_points..throws'z$ 4 6# 6 6 6777rrDrGN) rr_concurrency_shim_AsyncIoGreenletrDrGrpatchobject)r$ greenlet_clsrDrGr?r>rCrFs @@@@r await_pointsrMs  # #F0AL%K#J66666??????888888 ,&99 ,77 s6'"B2 B B2B B2!B "B22B69B6cleZdZdZdZejdZdZdddZ e dZ d Z d Z d S) ConnectionAccountingaRecords which DBAPI connections a pool created and which it managed to close, and how many of the pool's slots were lost. Creation is observed at ``Pool._invoke_creator`` rather than through the ``connect`` pool event, because the ``connect`` event is itself where ``dialect.initialize()`` runs; a listener added there does not fire for precisely the connections that a failure in that event strands. c||_|j|_|j|_d|_t |_t |_dSr) _pool_invoke_creator_real_invoke_creator_close_connection_real_close_connection _lost_slotsset_created_closed)rpools rrzConnectionAccounting.__init__sA $($8!&*&<# uu rc# Ktj|jd|j5tj|jd|j5|Vdddn #1swxYwYddddS#1swxYwYdS)z4Account for the pool's connections within the block.rRrTN)rrJrKrQrRrTrs rstartzConnectionAccounting.starts J   -t/C     J   /1G     JJJ                                 s5,BA+ B+A/ /B2A/ 3BB Bcd||}|j||Sr)rSrXadd)rrecdbapi_connections rrRz$ConnectionAccounting._invoke_creators344S99 *+++rF terminatec |||nA#tj$r/|r+|jjjr|j|wxYw|j|dS)Nra)rUr+CancelledErrorrQ_dialect has_terminaterYr^)rr`rbs rrTz&ConnectionAccounting._close_connections   ' '(8I ' N N N N%    3TZ0> 3   !1222    )*****s >Ac |j|jz S)zK|s |r Jd|}t|5}|r-||d{V|r|jdt|5} |||d{Vn#t $r |jsYnwxYwdddn #1swxYwY| |d{Vn#tj $rYnwxYwtj tj hz }|rtj|dd{Vtjdddn #1swxYwY|d|pdfS)aRun ``scenario`` against a fresh engine, optionally cancelling it. ``engine_factory`` is a zero-argument callable returning a new :class:`_asyncio.AsyncEngine`; ``scenario`` is a coroutine function taking that engine. If ``warm`` is True the scenario is run once un-cancelled first, so that the await points counted are those of checkout, execution and return-to-pool rather than those of ``connect()``. If ``recycle`` is also True, the warmed pool is then invalidated, so that the checkout counted closes the old connection before opening its replacement. Returns the :class:`.AwaitPoints` for the run, and a description of what the cancellation cost the pool, or ``None`` if it cost nothing. zrecycle requires 'warm'N)timeoutz; )rwrZ _invalidaterMr, BaseExceptionr)rmdisposerSQLAlchemyErrorr+ all_tasksr3waitrjrkjoinrr) engine_factoryscenarior$warmrecyclerv accountingr>otherss r run_scenariors$ 9w99 9999 ^  F v & &'*  .(6"" " " " " " " " . ''--- ) $ $  ((&)9)9::::::::::   *                " $$&&& .."" " " " " " " " ""    D "$$(<(>(>'??  2,vq111 1 1 1 1 1 1 1 O'''''''''''''''R 499Z<<>>??G4 GGs~?E))B4+$BB4B%"B4$B%%B4( E)4B8 8E);B8 <E)C/.E)/D>E)DAE))E-0E-r/cTKgfddd{VS)aCancel ``scenario`` at each of its await points in turn. With ``cancels=2``, each run that was cancelled at point ``N`` is then repeated with a second cancellation at each await point after ``N``, which is to say at each point of the cleanup the first cancellation set off. Returns a list of strings describing what went wrong, empty if every cancellation point was handled cleanly. The strings are meant to be read in an assertion failure:: eq_(await sweep(engine_factory, scenario), []) c Kt | d{V\}}|j}|r |ddznd}t||dzD]}t ||fz d{V\}}|^dd||fzD}d|j} d|d|d|d |t |dz kr ||fzd{VdS) N)r$rrr/z, c34K|]}t|VdSr)str).0is r z(sweep.._sweep..es(#I#IqCFF#I#I#I#I#I#Irzcancel at await point(s) roz (z): )rr&rangerr)r0rp)r$r>_totalr\indexfailureindexes awaitables_sweepcancelsrrreportedrrs rrzsweep.._sweepRs&             %.5 " !!A5%!),, 3 3E$0#uh. %%%OFG"))#I#II4H#I#I#III!YYv':;; 111U11"11'.119~~!G++fY%1222222222! 3 3rrNr)rrrrrrrs`````@@rsweepr>sr$H333333333338 &** Orr7)rFF)FFr/)r __future__rr+rsrjunittestrrutilrr rrtrMrOrwrrrrrrsrB#"""""  * * * * * * * *;;;;;;;;| :RRRRRRRRj555AF>H>H>H>HDBC111111r