ciVBfddlZddlZddlZddlmZmZmZGddZdejdefdZ dS)N)AnyCallable Coroutinec eZdZdZdZdZdZdedefdZ dedefd Z ded ed e e e e fffd Z d dd ed e e e e ffdedzde fdZd ed e e e e ffddfdZddefdZdZdejded ed e e e e fffdZdejded ed e e e e fffdZded ed e e e e fffdZdejdedefdZdejdedefdZdejdedefdZdS)BackgroundSchedulerzf Schedules background tasks execution either in separate thread or in the running event loop. cd|_g|_tj|_d|_d|_d|_tj|_ dS)NF) _next_timer _event_loops threadingLock_lock_stopped_health_check_loop_health_check_threadEvent_health_check_loop_readyselfs k/var/www/html/web/mlink/mlink_AI_Server/mlink-backend/venv/lib/python3.11/site-packages/redis/background.py__init__zBackgroundScheduler.__init__ sN^%%  DH=A!(1(9(9%%%c.|dSN)stoprs r__del__zBackgroundScheduler.__del__s rcn|j5|jr ddddSd|_|jr |jd|_|jD]0}|r||j1|jddddS#1swxYwYdS)zB Stop all scheduled tasks and clean up resources. NT) r rr cancelr is_runningcall_soon_threadsaferclear)rloops rrzBackgroundScheduler.stops(Z & &}  & & & & & & & &!DM ( '')))#' ) 9 9??$$9--di888   # # % % % & & & & & & & & & & & & & & & & & &s B*A?B**B.1B.delaycallbackc|j5|jr ddddS dddn #1swxYwYtj}|j5|j|dddn #1swxYwYt jt||j ||g|Rd}| dS)zI Runs callable task once after certain delay in seconds. NTtargetargsdaemon) r rasyncionew_event_loopr appendr Thread_start_event_loop_in_thread _call_laterstart)rr"r#r'r!threads rrun_oncezBackgroundScheduler.run_once.shZ  }                          %'' Z + +   $ $T * * * + + + + + + + + + + + + + + +!.(%ADAA      +//A77A;>A;intervalc|j5|jr ddddS dddn #1swxYwYtj}|j5|j|dddn #1swxYwYt jt||j ||g|Rd}| dS)zN Runs recurring callable task with given interval in seconds. NTr%) r rr)r*r r+r r,r-_call_later_recurringr/)rr3r#r'r!r0s r run_recurringz!BackgroundScheduler.run_recurringCshZ  }                          %'' Z + +   $ $T * * * + + + + + + + + + + + + + + +!.2HhNNN     r2coro.c|j5|jr ddddS dddn #1swxYwY||j5|j}dddn #1swxYwY|j|j|||g|RdS)a Runs recurring coroutine with given interval in seconds in a background thread. Uses a shared event loop to ensure connection pools remain valid across calls. This is useful for sync code that needs to run async health checks. N)r r_ensure_health_check_looprr_call_later_recurring_coro)rr3r7r'r!s rrun_recurring_coroz&BackgroundScheduler.run_recurring_coroXsPZ  }                          &&((( Z + +*D + + + + + + + + + + + + + + + "!  +T8T DH       +//A%%A),A)g$@timeoutr>Nreturnc|j5|jrtd dddn #1swxYwY||j5|j}dddn #1swxYwYt j|||} ||S#t$r| wxYw)ag Runs a coroutine synchronously and returns its result. Uses the shared health check event loop to ensure connection pools created here remain valid for subsequent recurring health checks. This is useful for running the initial health check before starting recurring checks. Args: coro: Coroutine function to execute *args: Arguments to pass to the coroutine timeout: Maximum seconds to wait for the result. None means wait forever. Default is 10 seconds to avoid blocking indefinitely if the event loop is busy with long-running health checks. Returns: The result of the coroutine Raises: TimeoutError: If the coroutine doesn't complete within timeout Any exception raised by the coroutine zScheduler is stoppedNr=) r r RuntimeErrorr9rr)run_coroutine_threadsaferesult TimeoutErrorr)rr7r>r'r!futures r run_coro_syncz!BackgroundScheduler.run_coro_syncpsV:Z ; ;} ;"#9::: ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; &&((( Z + +*D + + + + + + + + + + + + + + +1$$+tDD ===11 1    MMOOO  s*,00A&&A*-A* B Cc^|j5|jr ddddS dddn #1swxYwY||j5|j}dddn #1swxYwYdtjfd}t j|||}||dS)a Schedule a coroutine for execution on the shared health check loop without waiting for the result. Exceptions are logged but not raised. This is useful for HALF_OPEN recovery health checks that need to run on the same event loop where connection pools were created. Args: coro: Coroutine function to execute *args: Arguments to pass to the coroutine NrEc6|r.tjtddS|Btjtd|dSdS)z&Log any exceptions from the coroutine.z#Fire-and-forget coroutine cancelledNz*Fire-and-forget coroutine raised exceptionexc_info) cancelledlogging getLogger__name__debug exception)rEs r on_completezABackgroundScheduler.run_coro_fire_and_forget..on_completes!! !(++112WXXXXX!!##/!(++11@#--//20/r)r rr9rr)FuturerBadd_done_callback)rr7r'r!rQrEs rrun_coro_fire_and_forgetz,BackgroundScheduler.run_coro_fire_and_forgetseZ  }                          &&((( Z + +*D + + + + + + + + + + + + + + +     1$$+tDD  -----r<@c|jrM|j5|j'|jr ddddSdddn #1swxYwY|j5|j'|jr ddddS|jt j|_|j |jtj |j d|_ |j |j|sj|j}d|_||jvr|j| |n#t$$rYnwxYwt'd|d ddddS#1swxYwYdS)z Ensure the shared health check loop and thread are running. Args: timeout: Maximum seconds to wait for the loop to start. Raises: RuntimeError: If the loop fails to start within the timeout. NT)r&r(r=z/Health check event loop failed to start within z seconds)ris_setr rrr r)r*r r+r r,_run_health_check_looprr/waitremoveclose ExceptionrA)rr> failed_loops rr9z-BackgroundScheduler._ensure_health_check_loops  ( / / 1 1   +7/::<<8                        Z' ' '3+66884 ' ' ' ' ' ' ' '   ) / / 1 1 1'.&<&>&>D #   $ $T%< = = =)2(82)))D %  % + + - - - 055g5FF #5 *.'$"333%,,[999%%'''' D"WgWWW 5' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' sN"AA #A ."F9CF93FF9 FF9FF99F=F=c Ntj|j|j|jj |j tj|j}|D]}||j tj |ddin#t$rYnwxYw|j dS#|j wxYw# tj|j}|D]}||j tj |ddin#t$rYnwxYw|j w#|j wxYwxYw)z'Run the shared health check event loop.return_exceptionsTN) r)set_event_loopr call_soonrset run_forever all_tasksrrun_until_completegatherr\r[)rpendingtasks rrXz*BackgroundScheduler._run_health_check_loopst6777 ))$*G*KLLL 0  # / / 1 1 1 0!+D,CDD#""DKKMMMM'::NGDtDD    '--/////'--//// 0!+D,CDD#""DKKMMMM'::NGDtDD    '--////'--////smC>AB54C 5 C?C CC C;>F$AEF E)&F(E))F,F$F!!F$r!c|j5|jr ddddS dddn #1swxYwY|j|||j|||g|RdS)z0Schedule first execution of recurring coroutine.N)r rr._execute_recurring_coro)rr!r3r7r's rr:z.BackgroundScheduler._call_later_recurring_corosZ  }                           (D8$$ QU      +//c j5jr ddddS dddn #1swxYwYdtjffd } tj}||dS#t $rRj5jr dddYdS dddn #1swxYwYjjgRYdSwxYw)z Executes recurring coroutine with given interval in seconds. Schedules next execution only after current one completes to prevent overlap. Nrhct|rnT|@tjtd|j5jr ddddS dddn #1swxYwYjj gRdS).on_complete7sB~~ !!-!(++11;!^^--2   =                        D ,        s2 BBB) r rr)Task ensure_futurerSr\r.rj)rr!r3r7r'rQrhs````` rrjz+BackgroundScheduler._execute_recurring_coro(sZ  }                          gl          6 (t55D  " "; / / / / /     =                         D ,          sH 044,BC B5 C) C5B9 9C<B9 =CCcKj5jr ddddS dddn #1swxYwYtjfdfd_dS)a Runs recurring coroutine with given interval in seconds in the current event loop. To be used only from an async context. No additional threads are created. Prevents overlapping executions by scheduling the next run only after the current one completes. Raises: RuntimeError: If called without a running event loop (programming error) Ncj5jr ddddS dddn #1swxYwY_dS)zBackgroundScheduler.run_recurring_async..schedule_next{s  =                        $x9OPPD   s ,00cj5jr ddddS dddn #1swxYwYdtjffd } tj}||dS#t $r<tjt ddYdSwxYw)z=Execute the coroutine and schedule next run after completion.Nrhc|rnT|@tjtd|dS)rnNz*Recurring async coroutine raised exceptionrI)rKrPrLrMrNrO)rhrvs rrQz\BackgroundScheduler.run_recurring_async..execute_and_reschedule..on_completess>>##^^%%1%h//55D!%!1!16  rz,Failed to schedule recurring async coroutineTrI) r rr)rorprSr\rLrMrNrO)rQrhr'r7rvrs rruzGBackgroundScheduler.run_recurring_async..execute_and_reschedulesQ  =                        ',  ,TT4[99&&{33333   !(++11BT2   s ,00 ,A88AB>=B>)r rr)get_running_looprtr )rr3r7r'rur!rvs````@@@rrun_recurring_asyncz'BackgroundScheduler.run_recurring_asynces-Z  }                         ')) Q Q Q Q Q Q Q Q        < ??85KLLs 488c|j5|jr ddddS dddn #1swxYwY|j||g|R|_dSrrs)rr!r"r#r's rr.zBackgroundScheduler._call_latersZ  }                         +4?5(BTBBBrkc|j5|jr ddddS dddn #1swxYwY|j|||j|||g|RdSr)r rr._execute_recurringrr!r3r#r's rr5z)BackgroundScheduler._call_later_recurringsZ  }                           (D3T8X PT      rkc8|j5|jr ddddS dddn #1swxYwY ||n#t$rYnwxYw|j5|jr ddddS dddn #1swxYwY|j|||j|||g|RdS)zR Executes recurring callable task with given interval in seconds. N)r rr\r.r}r~s rr}z&BackgroundScheduler._execute_recurringsZ  }                           HdOOO    D Z  }                           (D3T8X PT      s- +//= A  A  A77A;>A;)rU)rN __module__ __qualname____doc__rrrfloatrr1r6rrr;rFrTr9rXr)AbstractEventLoopr:rjrzr.r5r}rrrrs : : :&&&(ex*ex*  %-c9S#s]3K.K%L    8!% ...sIc3m445. . ....`$.S)CcM"::;$. $.$.$.$.L::::::x000.  '    sIc3m445     ;';;sIc3m445 ;;;;z;M;M%-c9S#s]3K.K%L;M;M;M;MzC-C6;CGOCCCC  '          '         rr event_loop call_soon_cbc tj||j||g|R | tj|}|D]}||tj|ddin#t$rYnwxYw| dS#| wxYw# tj|}|D]}||tj|ddin#t$rYnwxYw| w#| wxYwxYw)z Starts event loop in a thread and schedule callback as soon as event loop is ready. Used to be able to schedule tasks using loop.call_later. :param event_loop: :return: r_TN) r)r`rarcrdrrerfr\r[)rrr'rgrhs rr-r-s :&&&Jz9D9999    ' 33G    ) )@4@@        D         J       ' 33G    ) )@4@@        D        J      slC AB B2 BB2BB22C E ADE D,)E+D,,E/EEE) r)rLr typingrrrrrr-rrrrs++++++++++K K K K K K K K \)9Ar