Bi!ddlZddlZddlmZddlmZddlmZmZddl m Z ddl m Z ddl mZddlmZdd lmZmZmZd Zd ZGd d eeZGddZ ddeedededededeeefdzddfdZGddZy)N) defaultdict)Enum)Queue get_context) BaseContext) BaseProcess) Synchronized)Empty)AnyIterableTypeiXceZdZdZdZdZy) QueueSignalsstopconfirmerrorN)__name__ __module__ __qualname__rrrb/var/www/pod-logistic/pod-ai/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.rworkerr3processed_itemes ` r_workerrC!s~ LL?9+VBIIK=AB!<###-f- hsm %nn-A-CD -N   ^ , - !  "  ( ( * *  $ $ ) $ *  ~i[ :;) -!++,,-" * * !  "  ( ( * *  $ $ ) $ * * *  ~i[ :;sUADE$E E4E E$EE$E!$AG:5G G:G !G:c eZdZdefdedeededzdefdZde ddfd Z d e e d e de de e fd Z d e e d e de de e fd Z d e e d e de de e fdZddZddedzddfdZddZddZy)ParallelWorkerPoolN num_workersr@ start_methodmax_internal_batch_sizec||_||_d|_d|_t ||_g|_|j|z|_d|_d|_ y)NF) r*rFr+r,rctx processes queue_sizeemergency_shutdownr-)r'rFr@rGrHs r__init__zParallelWorkerPool.__init__XsZ#&)-*. +L 9,.**-DD"'48rrrc |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)rJrrLr+r,ValuerF isinstance BaseValuer-rangehasattrrQrCr*copyr$rKappend)r'r ctx_valuer.r(s rr$zParallelWorkerPool.startis88>>$//: HHNN4??;HHNN3(8(89 )Y///"+q$"2"23 +I488Y/ //hh&&%%$$%%++KKM ' G MMO NN ! !' * +rstreamrc/>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 terminatedr4zInput queue is NonezOutput queue is Noner)r$r+r,check_worker_healthrL get_nowaitr r1processing_timeoutjoin_or_terminaterr RuntimeErrorr9rVrFrjoinr<rMcancel_join_threadr=) r'r[rrpushedreadr2out_itemrB_s r unordered_mapz ParallelWorkerPool.unordered_mapsA4 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++, 8  $$\%6%67 8-,,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 )rj 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)Nrr4)rintrmpop)r'r[rrbuffer next_expectedidxr2s r ordered_mapzParallelWorkerPool.ordered_mapssS! ...vGGG #ICF3K6)jj//"  6) #s7AAAc|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)rKis_aliveexitcoderMrbrcpidr'r(s rr_z&ParallelWorkerPool.check_worker_healthsn~~ G##%'*:*:a*?*.'&&(""7;;-/RSZScScRde  rr^cd|_|jD]5}|j||js&|j 7|jj y)zM Emergency shutdown @param timeout: @return: Tr]N)rMrKrdrv terminateclear)r'r^r(s rrbz$ParallelWorkerPool.join_or_terminatesW #'~~ $G LLL )!!!# $ rcz|jD]}|j|jjyr )rKrdr|rys rrdzParallelWorkerPool.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)rKrvr{rys r__del__zParallelWorkerPool.__del__s/~~ $G!!!# $r)rN)r4)rrrMAX_INTERNAL_BATCH_SIZEror rstrrNr r$r rjrmrtr_rbrdrrrrrErEWs $('> 99V 9Dj 9 "% 9"+c+d+050HSM50#5050QYZ]Q^50nFx}FSFCFT\]`TaF#(3-##s#xX[}#  t D  $rrEr )r5r7 collectionsrenumrmultiprocessingrrmultiprocessing.contextrmultiprocessing.processrmultiprocessing.sharedctypesr rUqueuer typingr r r rarrrrrodictrCrErrrrs #.//B&&3 $$%) 3<v,3<3<3<" 3<  3< cNT ! 3< 3