'Xh " ddlZddlZddlmZddlmZddlmZmZddl m Z ddl m Z ddl mZddlmZdd lmZmZmZmZd Zd ZGd d eeZGddZ ddeedededededeeeefddfdZGddZy)N) defaultdict)Enum)Queue get_context) BaseContext) BaseProcess) Synchronized)Empty)AnyIterableOptionalTypeiXceZdZdZdZdZy) QueueSignalsstopconfirmerrorN)__name__ __module__ __qualname__rrrJ/RAG/venv/lib/python3.12/site-packages/qdrant_client/parallel_processor.pyrrs DG ErrcFeZdZedededdfdZdeedeefdZy)WorkerargskwargsreturnctNNotImplementedError)clsrrs rstartz Worker.starts !##ritemsctr!r")selfr&s rprocesszWorker.processs !##rN)rrr classmethodr r%r r)rrrrrsD$#$$$$$Xc]$x}$rr worker_class input_queue output_queuenum_active_workers worker_idrrc|i}tjd|dtj |jd i|}dt t ffd }|j|D]}|j| j|jj|j|j5|xj dzc_dddtjd|dy#t$r>} tj| |jtjYd} ~ d} ~ wwxYw#1swYmxYw#j|jj|j|j5|xj dzc_dddn #1swYnxYwtjd|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 j}|tjk(ry|)wr!)getrr)itemr,s rinput_queue_iterablez%_worker..input_queue_iterable7s1"(<,,, s+.zReader worker z finishedr)logginginfoosgetpidr%r r r)put Exception exceptionrrclose join_threadget_lockvalue) r+r,r-r.r/rworkerr4processed_itemes ` r_workerrD!s~ LL?9+VBIIK=AB!<###-f- hsm %nn-A-CDN   ^ ,E !  "  ( ( *  $ $ ) $+  ~i[ :;) -!++,,-"+ * !  "  ( ( *  $ $ ) $+ * *  ~i[ :;sUADE$E E4E E$EE$E!$AG:5G G:G !G:c eZdZdefdedeedeedefdZ de ddfd Z d e e d e de de e fd Z d e e d e de de e fd Zd e e d e de de e fdZddZddeeddfdZddZddZy)ParallelWorkerPoolN num_workersrA start_methodmax_internal_batch_sizec||_||_d|_d|_t ||_g|_|j|z|_d|_d|_ y)NF) r+rGr,r-rctx processes queue_sizeemergency_shutdownr.)r(rGrArHrIs r__init__zParallelWorkerPool.__init__XsZ#&,0-1 +L 9,.**-DD"'7;rrrc |jj|j|_|jj|j|_|jj d|j }t|tsJ||_ td|j D]}t|jdsJ|jjt|j|j|j|j||jf}|j!|j"j%|y)NirProcess)targetr)rKrrMr,r-ValuerG isinstance BaseValuer.rangehasattrrRrDr+copyr%rLappend)r(r ctx_valuer/r)s rr%zParallelWorkerPool.startis88>>$//: HHNN4??;HHNN3(8(89 )Y///"+q$"2"23I488Y/ //hh&&%%$$%%++KKM ' G MMO NN ! !' *4rstreamrc/>K |jd i||jJd|jJdd}d}|D]}|j||z |jkr |jj }n! |jjt}|7|tjk(r|jtd||dz }|jj||dz }t|jD]+} |jjtj -||kr]|jjt}|tjk(r|jtd||dz }||kr]|jJd|jJd|j#|jj%|jj%|j&r5|jj)|jj)y|jj+|jj+y#t $rd}YwxYw#t $r}|j|d}~wwxYw#|jJd|jJd|j#|jj%|jj%|j&r5|jj)|jj)w|jj+|jj+wxYww) NzInput queue was not initializedz Output queue was not initializedrtimeoutzThread unexpectedly terminatedr5zInput queue is NonezOutput queue is Noner)r%r,r-check_worker_healthrM get_nowaitr r2processing_timeoutjoin_or_terminaterr RuntimeErrorr:rWrGrjoinr=rNcancel_join_threadr>) r(r\rrpushedreadr3out_itemrC_s r unordered_mapz ParallelWorkerPool.unordered_maps74 0 DJJ  ##/ R1R R/$$0 T2T T0FD((*D=4??2(#'#4#4#?#?#A #'#4#4#8#8AS#8#T '<#5#55..0*+KLL"NAID  $$T*! +.4++,  $$\%6%67--,,009K0L|111**,&'GHH  -##/ F1F F/$$0 H2H H0 IIK    " " $    # # %&&  335!!446  ,,.!!--/O!(#'( ! ..0 0##/ F1F F/$$0 H2H H0 IIK    " " $    # # %&&  335!!446  ,,.!!--/shNA#J9(JJ9 J$C?J9$C N JJ9JJ9 J6J11J66J99C!NNc@|jt|g|i|Sr!)rk enumerate)r(r\rrs rsemi_ordered_mapz#ParallelWorkerPool.semi_ordered_maps$!t!!)F"3EdEfEErc/Ktt}d}|j|g|i|D],\}}|||<||vs|j||dz }||vr.yw)Nrr5)rintrnpop)r(r\rrbuffer next_expectedidxr3s r ordered_mapzParallelWorkerPool.ordered_mapsoS! ...vGGGICF3K6)jj//"  6)Hs7AAAc|jD]^}|jr|jdk7s$d|_|j t d|j d|jy)zJ Checks if any worker process has terminated unexpectedly rTz Worker PID: z# terminated unexpectedly with code N)rLis_aliveexitcoderNrcrdpidr(r)s rr`z&ParallelWorkerPool.check_worker_healthsi~~G##%'*:*:a*?*.'&&(""7;;-/RSZScScRde &rr_cd|_|jD]5}|j||js&|j 7|jj y)zM Emergency shutdown @param timeout: @return: Tr^N)rNrLrerw terminateclear)r(r_r)s rrcz$ParallelWorkerPool.join_or_terminatesR #'~~G LLL )!!!#& rcz|jD]}|j|jjyr!)rLrer}rzs rrezParallelWorkerPool.joins)~~G LLN& rch|jD]#}|js|j%y)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)rLrwr|rzs r__del__zParallelWorkerPool.__del__s*~~G!!!#&r)rN)r5)rrrMAX_INTERNAL_BATCH_SIZErprrr strrOr r%r rkrnrur`rcrerrrrrFrFWs '+'> <<V <sm < "% <"+c+d+050HSM50#5050QYZ]Q^50nFx}FSFCFT\]`TaF#(3-##s#xX[}#  # t  $rrFr!) r6r8 collectionsrenumrmultiprocessingrrmultiprocessing.contextrmultiprocessing.processrmultiprocessing.sharedctypesr rVqueuer typingr r r rrbrrrrrpdictrDrFrrrrs #.//B003 $$(, 3<v,3<3<3<" 3<  3< T#s(^ $ 3< 3