Ki4dZddlmZddlZddlZddlmZmZmZddlm Z m Z m Z ddl m Z mZdd lmZmZmZmZmZej*d Ze r dd lmZdd lmZd ZdZGddeZGddedZy)z Concurrency limiting dependency.) annotationsN)datetime timedeltatimezone) TYPE_CHECKINGAnyoverload)CANCEL_MSG_CLEANUP cancel_task)AdmissionBlocked Dependencycurrent_docketcurrent_executioncurrent_workerzdocket.dependencies)Redis) Executionc$eZdZdZdfd ZxZS)ConcurrencyBlockedz:Raised when a task cannot start due to concurrency limits.cT||_||_d|d|}t| ||y)Nzconcurrency limit (z max) on )reason)concurrency_keymax_concurrentsuper__init__)self executionrrr __class__s l/home/jay/workspace/scripts/.codegraph-venv/lib/python3.12/site-packages/docket/dependencies/_concurrency.pyrzConcurrencyBlocked.__init__(s9.,&~&6i?PQ 62)rrrstrrint)__name__ __module__ __qualname____doc__r __classcell__)r s@r!rr%sD33r"rc eZdZUdZdZded<edd ddZe d dd Zeddd  dd Z d dd Zdd ZddZ ddZ ddZ ddZ ddZ eddZy) ConcurrencyLimitaConfigures concurrency limits for task execution. Can limit concurrency globally for a task, or per specific argument value. Works both as a default parameter and as ``Annotated`` metadata:: # Default-parameter style async def process_customer( customer_id: int, concurrency: ConcurrencyLimit = ConcurrencyLimit("customer_id", 1), ) -> None: ... # Annotated style (parameter name auto-inferred) async def process_customer( customer_id: Annotated[int, ConcurrencyLimit(1)], ) -> None: ... # Per-task (no argument grouping) async def expensive( concurrency: ConcurrencyLimit = ConcurrencyLimit(max_concurrent=3), ) -> None: ... TboolsingleN)scopecy)z9Annotated style: ``Annotated[int, ConcurrencyLimit(1)]``.Nrrr.s r!rzConcurrencyLimit.__init__Ir"r cy)z/Default-param style with per-argument grouping.Nr0r argument_namerr.s r!rzConcurrencyLimit.__init__Sr2r"rr.cy)z,Per-task concurrency (no argument grouping).Nr0r1s r!rzConcurrencyLimit.__init__\r2r"ct|trd|_||_n||_||_||_d|_d|_d|_d|_y)NF) isinstancer$r5rr._concurrency_key _initialized _task_key _renewal_taskr4s r!rzConcurrencyLimit.__init__esT mS )-1D '4D !.D "0D  ,0"'%)8<r"cz|j |jn|}t||j|jS)zGBind to an ``Annotated`` parameter, inferring argument_name if not set.r6)r5r+rr.)rnamevaluer5s r!bind_to_parameterz"ConcurrencyLimit.bind_to_parameterws;.2.@.@.L**RV  ..**  r"c Kddlm}tj}t j}t j}|j xs |j}|j1 |j|j}|d|jd|}n|d|j}t!|j|j"|j } || _d| _|j(| _|j-4d{} | j/| |j0|j2d{} | st5|||j"dddd{t7j8| j;|j2|jd|j( | _|j>j} | jA| jB| jAtD| j<tF| S#t$rB}td|jdt|jj|d}~wwxYw7U7)7#1d{7swYxYww) Nr )_DependszConcurrencyLimit argument 'z*' not found in task arguments. Available: z :concurrency::Tz - concurrency lease:)r?)$ _functionalrCrgetrrr.r?r5 get_argumentKeyError ValueErrorlistkwargskeys function_namer+rr:r;keyr<redis _acquire_slot redeliveredredelivery_timeoutrasyncio create_task_renew_lease_loopr=stackpush_async_callback _release_slotr r ) rrCrdocketworkerr.argument_valueerlimitrOacquiredrVs r! __aenter__zConcurrencyLimit.__aenter__sM)%))+ ##%##% )fkk    ) !*!7!78J8J!K't'9'9&:!N;KL  "'}Y5L5L4MNO !!3!3T5H5H$**U!0!#--<<>  U"00y,,f.G.GH(0C0C   &11  # #F$=$= >KK= 5imm_E ""$ !!%"5"56 !!+u/B/BDVW W  1$2D2D1EF226y7G7G7L7L7N2O1PR *     sA)I?,HBI? I  I?+I):I#;I) I?#I&$B.I? I=III?#I)&I?)I</I2 0I<7I?c Kyw)Nr0)rexc_type exc_value tracebacks r! __aexit__zConcurrencyLimit.__aexit__s  sc K|jd}tjtjj }||j z }ttt|j tz}||jg|j|j||rdnd||gd{}t|S7w)aAtomically acquire a concurrency slot. Uses a Redis sorted set to track concurrency slots per task. Each entry is keyed by task_key with the timestamp as the score. When XAUTOCLAIM reclaims a message (because the original worker stopped renewing its lease), is_redelivery=True signals that slot takeover is safe. If the message is NOT a redelivery and a slot already exists, we block to prevent duplicate execution. Slots are refreshed during lease renewal every redelivery_timeout/4. If all slots are full, we scavenge any slot older than redelivery_timeout (meaning it hasn't been refreshed and the worker must be dead). a9 local key = KEYS[1] local max_concurrent = tonumber(ARGV[1]) local task_key = ARGV[2] local current_time = tonumber(ARGV[3]) local is_redelivery = tonumber(ARGV[4]) local stale_threshold = tonumber(ARGV[5]) local key_ttl = tonumber(ARGV[6]) -- Check if this task already has a slot (from a previous delivery attempt) local slot_time = redis.call('ZSCORE', key, task_key) if slot_time then slot_time = tonumber(slot_time) if is_redelivery == 1 and slot_time <= stale_threshold then -- Redelivery AND slot is stale: original worker stopped renewing, -- safe to take over the slot. redis.call('ZADD', key, current_time, task_key) redis.call('EXPIRE', key, key_ttl) return 1 else -- Either not a redelivery, or slot is still fresh (original worker -- is just slow, not dead). Don't take over. return 0 end end -- No existing slot for this task - check if we can acquire a new one if redis.call('ZCARD', key) < max_concurrent then redis.call('ZADD', key, current_time, task_key) redis.call('EXPIRE', key, key_ttl) return 1 end -- All slots are full. Scavenge any stale slot (not refreshed recently). -- Slots are refreshed every redelivery_timeout/4, so anything older than -- redelivery_timeout hasn't been refreshed and the worker must be dead. local stale_slots = redis.call('ZRANGEBYSCORE', key, 0, stale_threshold, 'LIMIT', 0, 1) if #stale_slots > 0 then redis.call('ZREM', key, stale_slots[1]) redis.call('ZADD', key, current_time, task_key) redis.call('EXPIRE', key, key_ttl) return 1 end return 0 r rrLargsN)register_scriptrnowrutc timestamp total_secondsmaxMINIMUM_TTL_SECONDSr$LEASE_RENEWAL_FACTORr:rr<r,) rrO is_redeliveryrRacquire_script current_timestale_thresholdkey_ttlresults r!rPzConcurrencyLimit._acquire_slots*..- / b ||HLL1;;= &);)I)I)KK  "0025II J  &''(##"    F|  sB:C<C =CcbK|jr |jsJtj}|j 4d{}|j d}||jg|jgd{dddd{y7N77 #1d{7swYyxYww)z/Release a concurrency slot when task completes.Nz redis.call('ZREM', KEYS[1], ARGV[1]) if redis.call('ZCARD', KEYS[1]) == 0 then redis.call('DEL', KEYS[1]) end rf)r:r<rrFrOrh)rrYrOrelease_scripts r!rXzConcurrencyLimit._release_slot s$$77##%<<> V VU#22N!t'<'<&=T^^DTU U U V V V V V V V VsZAB/BB/ 5B>B?B B/BB/BB/B, B# !B,(B/cKtj}|jtz }t t t |jtz} tj|d{ |j4d{}tjtjj}|j|j |j"|id{|j%|j |d{dddd{777=77 #1d{7swYxYw#t&$r%t(j+d|j dYMwxYww)z:Periodically refresh slot timestamp to prevent expiration.TNz'Concurrency lease renewal failed for %s)exc_info)rrFrlrormrnr$rSsleeprOrrirrjrkzaddr:r<expire Exceptionloggerwarning)rrRrYrenewal_intervalrtrOrrs r!rUz"ConcurrencyLimit._renew_lease_loop3sX##%-;;=@TT  "0025II J  -- 01 1 1 !<<>GGU#+<< #=#G#G#IL**--6 ,,t'<'>> >  >>  ;; ;  ;;+/ ='== =  =$ 8t  ,  (  .    YY+/YENY YvV&8%%r"r+)r( __future__rrSloggingrrrtypingrrr _cancellationr r _baserrrrr getLoggerr~ redis.asynciorrrrornrr+r0r"r!rsy&"22//;   0 1#% 3)3i%z"45i%r"