[i|#ddlZddlZddlmZddlmZddlmZddlm Z m Z ddl m Z ddl mZddlmZdd lmZdd lmZmZmZmZd Zd ZGd deeZGddZ ddeede de dededeeeefddfdZ GddZ!y)N) defaultdict)deepcopy)Enum)Queue get_context) BaseContext) BaseProcess) Synchronized)Empty)AnyIterableOptionalTypeiXceZdZdZdZdZy) QueueSignalsstopconfirmerrorN)__name__ __module__ __qualname__rrrX/mnt/e/genesis-system/.venv/lib/python3.12/site-packages/fastembed/parallel_processor.pyrrs DG ErrcZeZdZedededdfdZdeeeefdeeeeffdZ y)WorkerargskwargsreturnctNNotImplementedError)clsrrs rstartz Worker.starts !##ritemsctr"r#)selfr's rprocesszWorker.processs !##rN) rrr classmethodr r&r tupleintr*rrrrrsV$#$$$$$XeCHo6$8E#s(O;T$rr worker_class input_queue output_queuenum_active_workers worker_idrr c  |i}tjd|dtjd|j dd |j d i|}dt tffd }|j|D]}|j| j|jj|j|j!5|xj"d zc_dddtjd |d y#t$r>} tj| |jtjYd} ~ d} ~ wwxYw#1swYmxYw#j|jj|j|j!5|xj"d zc_dddn #1swYnxYwtjd |d wxYw) 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: z Device: device_idCPUr c3`K j}|tjk(ry|)wr")getrr)itemr/s rinput_queue_iterablez%_worker..input_queue_iterable;s1"(<,,, s+.zReader worker z finishedr)logginginfoosgetpidr7r&r r r*put Exception exceptionrrclose join_threadget_lockvalue) r.r/r0r1r2rworkerr9processed_itemes ` r_workerrI#s~ LL )F299;-yKY^A_@`a!<###-f- hsm %nn-A-CD -N   ^ , - !  "  ( ( * *  $ $ ) $ *  ~i[ :;) -!++,,-" * * !  "  ( ( * *  $ $ ) $ * * *  ~i[ :;sVAD!E7)E+! E(*4E#E7#E((E7+E47AH G' H 'G0 ,!H c eZdZ ddedeedeedeeede f dZ de d dfd Z d e e d e de d e e fd Zd e e d e de d e eee ffdZddZddeed dfdZddZddZy)ParallelWorkerPoolN num_workersrF start_method device_idscudac||_||_d|_d|_t ||_g|_|jtz|_d|_ ||_ ||_ d|_ y)NF) r.rLr/r0rctx processesmax_internal_batch_size queue_sizeemergency_shutdownrNrOr1)r)rLrFrMrNrOs r__init__zParallelWorkerPool.__init__\sh#&,0-1 +L 9,.**-DD"'$ 7;rrr c $|jj|j|_|jj|j|_|jj d|j }t|tsJ||_ td|j D]}t|}|jr9|j|t|jz}||d<|j|d<t|jdsJ|jj!t"|j$|j|j|j||f}|j'|j(j+|y)Nirr4rOProcess)targetr)rQrrTr/r0ValuerL isinstance BaseValuer1rangerrNlenrOhasattrrYrIr.r&rRappend)r)r ctx_valuer2 worker_kwargsr4r*s rr&zParallelWorkerPool.startpsC88>>$//: HHNN4??;HHNN3(8(89 )Y///"+q$"2"23 +I$V,M OOIDOO8L,LM -6 k*(, f%488Y/ //hh&&%%$$%%++! ' G MMO NN ! !' *) +rstreamrc/Ktt}d}|j|g|i|D],\}}|||<||vs|j||dz }||vr.yw)Nrr:)rr semi_ordered_mappop)r)rdrrbuffer next_expectedidxr8s r ordered_mapzParallelWorkerPool.ordered_mapss(3C(8 ...vGGG #ICF3K6)jj//"  6) #s7AAAc/zK |jd i||jJd|jJdd}d}t|D]\}}|j ||z |j kr |jj }n! |jjt}|7|tjk(r|jtd||dz }|jj||f|dz }t|j D]+} |jjtj"-||krm|j |jjt}|tjk(r|jtd||dz }||krm|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 terminatedr:zInput queue is NonezOutput queue is Noner)r&r/r0 enumeratecheck_worker_healthrT get_nowaitr r7processing_timeoutjoin_or_terminaterr RuntimeErrorr?r^rLrjoinrBrUcancel_join_threadrC) r)rdrrpushedreadrjr8out_itemrH_s rrfz#ParallelWorkerPool.semi_ordered_mapsZ6 0 DJJ  ##/ R1R R/$$0 T2T T0FD&v.  T((*D=4??2(#'#4#4#?#?#A #'#4#4#8#8AS#8#T '<#5#55..0*+KLL"NAID  $$c4[1! - 04++, 8  $$\%6%67 8-((*,,009K0L|111**,&'GHH -##/ F1F F/$$0 H2H H0 IIK    " " $    # # %&&  335!!446  ,,.!!--/S!(#'( ! ..0 4##/ F1F F/$$0 H2H H0 IIK    " " $    # # %&&  335!!446  ,,.!!--/shN;A/K4J"K J40DKC N;" J1-K0J11K4 K=KKKC!N88N;c|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)rRis_aliveexitcoderUrsrtpidr)r*s rrpz&ParallelWorkerPool.check_worker_healthsn~~ G##%'*:*:a*?*.'&&(""7;;-/RSZScScRde  rrnc|jD]5}|j||js&|j7|jj y)zM Emergency shutdown @param timeout: @return: rmN)rRrur| terminateclear)r)rnr*s rrsz$ParallelWorkerPool.join_or_terminatesO ~~ $G LLL )!!!# $ rcz|jD]}|j|jjyr")rRrurrs rruzParallelWorkerPool.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)rRr|rrs r__del__zParallelWorkerPool.__del__s/~~ $G!!!# $r)NNF)r N)r:)rrrr-rrrstrlistboolrVr r&r rkr,rfrprsrurrrrrKrK[s '+*. <<V <sm < T#Y' <  <(+c+d+<#(3-##s#xX[}#90sm90,/90;>90 %S/ "90v  # t  $rrKr")"r;r= collectionsrcopyrenumrmultiprocessingrrmultiprocessing.contextrmultiprocessing.processr multiprocessing.sharedctypesr r]queuer typingr r rrrrrSrrrr-dictrIrKrrrrs #.//B003 $$(, 5<v,5<5<5<" 5<  5< T#s(^ $ 5< 5