Ngz,ddlZddlZddlmZddlmZmZddlmZddl m Z ddl m Z ddlmZddlmZmZmZmZmZmZd Zd ZGd d eeZGd dZ ddeededede dedeeeefddfdZGddZdS)N)Enum)Queue get_context) BaseContext) BaseProcess) Synchronized)Empty)AnyDictIterableListOptionalTypeiXceZdZdZdZdZdS) QueueSignalsstopconfirmerrorN)__name__ __module__ __qualname__rrr\/var/www/html/ai-engine/env/lib/python3.11/site-packages/qdrant_client/parallel_processor.pyrrs DG EEErrcVeZdZededdfdZdeedeefdZdS)Workerkwargsreturnc tNNotImplementedError)clsrs rstartz Worker.starts!###ritemsctr!r")selfr&s rprocesszWorker.processs!###rN)rrr classmethodr r%r r)rrrrrsj$S$X$$$[$$Xc]$x}$$$$$$rr worker_class input_queue output_queuenum_active_workers worker_idrrc|i}tjd|dtj |jd i|}dt t ffd }||D]}||nJ#t$r=} tj | |tj Yd} ~ nd} ~ wwxYw| ||5|xjdzc_dddn #1swxYwYtjd|ddS#| ||5|xjdzc_dddn #1swxYwYtjd|dwxYw) z A worker that pulls data pints off the input queue, and places the execution result on the output queue. When there are no data pints left on the input queue, it decrements num_active_workers to signal completion. NzReader worker: z PID: rc3`K }|tjkrdS|V+r!)getrr)itemr,s rinput_queue_iterablez%_worker..input_queue_iterable6s< "((<,,,E  rzReader worker z finishedr)logginginfoosgetpidr%r r r)put Exception exceptionrrclose join_threadget_lockvalue) r+r,r-r.r/rworkerr4processed_itemes ` r_workerrD s~ LA9AABIKKAABBB<##--f-- hsm      %nn-A-A-C-CDD - -N   ^ , , , , - ---!+,,,,,,,,-   """  ( ( * * * *  $ $ ) $ $ * * * * * * * * * * * * * * *  :i:::;;;;;   """  ( ( * * * *  $ $ ) $ $ * * * * * * * * * * * * * * *  :i:::;;;;sfAB E C3C ECED00D47D4=GF/# G/F3 3G6F3 7Gc eZdZddedeedeefdZde ddfdZ d e e d e de de e fd Z dd eeddfdZ ddZddZdS)ParallelWorkerPoolN num_workersrA start_methodc||_||_d|_d|_t ||_g|_|jtz|_d|_ dSr!) r+rGr,r-rctx processesmax_internal_batch_size queue_sizer.)r(rGrArHs r__init__zParallelWorkerPool.__init__UsV"&,0-1 +L 9 9,.*-DD7;rrrc f|j|j|_|j|j|_|jd|j}t|tsJ||_ td|jD]}t|jdsJ|j t|j|j|j|j ||f}||j|dS)NirProcess)targetargs)rJrrMr,r-ValuerG isinstance BaseValuer.rangehasattrrQrDr+copyr%rKappend)r(r ctx_valuer/r)s rr%zParallelWorkerPool.start`s8>>$/:: HNN4?;;HNN3(899 )Y/////"+q$"233 + +I48Y// / / /h&&%$%+KKMM '  G MMOOO N ! !' * * * * + +rstreamrSc/K |jd i||j Jd|j Jdd}d}|D]}||z |jkr- |j}n\#t $rd}YnNwxYw |jt}n(#t $r}||d}~wwxYw|<|tj kr#|td|V|dz }|j ||dz }t|jD]&} |j tj'||krb|jt}|tj kr#|td|V|dz }||kb|j Jd|j Jd|j|jdS#|j Jd|j Jd|j|jwxYw) NzInput queue was not initializedz Output queue was not initializedrtimeoutzThread unexpectedly terminatedr5zInput queue is NonezOutput queue is Noner)r%r,r-rM get_nowaitr r2processing_timeoutjoin_or_terminaterr RuntimeErrorr:rWrGrr=) r(r\rSrpushedreadr3out_itemrC_s r unordered_mapz ParallelWorkerPool.unordered_mapxs- & DJ  #//1R///$002T000FD  D=4?22(#'#4#?#?#A#A (((#'( #'#4#8#8AS#8#T#T    ..000 '<#555..000*+KLLL"NNNAID $$T***! 4+,, 8 8 $$\%67777--,009K0LL|111**,,,&'GHHH  --#//1F///$002H000   " " $ $ $   # # % % % % %#//1F///$002H000   " " $ $ $   # # % % % %sVAH A$#H$ A30H2A33H7 BH B="B88B==DHAI-r5r_c|jD]@}|||r|A|jdS)zM Emergency shutdown @param timeout: @return: r^N)rKjoinis_alive terminateclear)r(r_r)s rrbz$ParallelWorkerPool.join_or_terminatesj ~ $ $G LLL ) ) )!! $!!### rct|jD]}||jdSr!)rKrjrmr(r)s rrjzParallelWorkerPool.joins<~  G LLNNNN rcB|jD]}|dS)a  Terminate processes if the user hasn't joined. This is necessary as leaving stray processes running can corrupt shared state. In brief, we've observed shared memory counters being reused (when the memory was free from the perspective of the parent process) while the stray workers still held a reference to them. For a discussion of using destructors in Python in this manner, see https://eli.thegreenplace.net/2009/06/12/safely-using-destructors-in-python/. N)rKrlros r__del__zParallelWorkerPool.__del__s4~  G         rr!)r5)rN)rrrintrrrstrrNr r%r rhrbrjrqrrrrFrFTs < r{s ........////////////BBBBBB<<<<<<<<<<<<<<<<3 $$$$$$$$(, 1<1<v,1<1<1<" 1<  1< T#s(^ $ 1< 1<1<1<1