a ^gD@sddlZddlZddlZddlZddlmZddlZddlZddlm Z ddl Z ddl Z ddl Z ddlZddlmZGdddZGdddZd ejejddd d d ddd ddd fddZdS)N) ThreadPool)Queue) get_invlistc@szeZdZdZdddZddZdd Zd d Zd d ZddZ ddZ ddZ ddZ ddZ ddZddZddZdS) BigBatchSearcherz Object that manages all the data related to the computation except the actual within-bucket matching and the organization of the computation (parallel or not) rFcCs`||_g|_||_||_||_t|j}tjt |||d|_ dgd|_ t |_ |_dS)N)keep_maxr)verbosetictocxqindex use_float16faissZis_similarity_metric metric_typeZ ResultHeaplenrht_accutime t_displayt0)selfr r krr rrl/var/www/html/cobodadashboardai.evdpl.com/venv/lib/python3.9/site-packages/faiss/contrib/big_batch_search.py__init__s  zBigBatchSearcher.__init__cCst|_dSN)r t_accu_t0rrrr start_t_accu0szBigBatchSearcher.start_t_accucCs |j|t|j7<dSr)rrr)rnrrr stop_t_accu3szBigBatchSearcher.stop_t_accucCs*|tf|_|jdkr&t|ddddS)Nr Tendflush)rr rprint)rnamerrrtic6s zBigBatchSearcher.ticcCs:|j\}}t|}|jdkr6t|d|dd|S)Nrz: .3fz s)r rrr$)rr%rdtrrrtoc;s    zBigBatchSearcher.toccCs|jdks.|jdkr2|dkr2t|jdkr2dSt|j}td|dd|d|jjd |jd d d |jdd d |jdd d|jdd d|jdd d|jdd dtj ||jj|d|ddt |jdkrdndddt|_dS)Nig?[z.1fz s] list /z times prep q rr'z prep b z comp z res z wait in z wait out z eta )secondsz mem r  Tr!) rrrrr$r nlistrdatetime timedeltar Zget_mem_usage_kb)rltrrrreportBs>        zBigBatchSearcher.reportcCs|dd}t|j}tj||jjfdd}td||D]@}t|||}|jj |j|||jj\}}||||<q:| ||_ dS)Nzcoarse quantizationiint32)dtyper) r&rr npemptyr nproberangemin quantizersearchr)q_assign)rbsZnqrBi0i1Zq_dis_iZ q_assign_irrrcoarse_quantizationWs  z$BigBatchSearcher.coarse_quantizationcCsz|d|j}|d7}tj|j|jjddd|_|j|_|j dkrZt d|jd|jdd|_|`| dS)Nz bucket sortr*)Znbucketntrz number of -1s:) r&rBr Zmatrix_bucket_sort_inplacer r3 bucket_limsravel query_idsrr$r))rrBrrrreorder_assignes   zBigBatchSearcher.reorder_assignc Cst}|j}|j||j|d}}|j||}|j|}|jrV||j|}t}t|j |\} } |j dur| } n | | } |j r| d} | d}t} |jd||7<|jd| |7<||| | fS)z4 prepare the queries and database items for bucket lr*NZfloat16r)rr rIrKr by_residualr@Z reconstructrZinvlists decode_funcrJr Zastyper) rr6rr rDrEq_subsetxq_lt1list_idsxb_lt2rrrprepare_bucketrs&      zBigBatchSearcher.prepare_bucketcCsX|dur dSt}|dur"|}n||}|j||||jdt|7<dS)z,add the bucket results to the heap structureNr.)rrZadd_result_subsetr)rrODrRIrrrradd_results_to_heapsz$BigBatchSearcher.add_results_to_heapcCs|jj|jj|jjfSr)r shaper r=r3rrrrsizes_in_checkpointsz$BigBatchSearcher.sizes_in_checkpointcCsh|d}t|d6}t|||jj|jjfd|dWdn1sN0Yt||dS)Nz.tmpwb)sizes completedr) openpickledumprZrrVrWosreplace)rfnamer]Ztmpnamefrrrwrite_checkpoints "z!BigBatchSearcher.write_checkpointcCst|d}t|}Wdn1s*0Y|d|ksHJ|dd|jjdd<|dd|jjdd<|dS)Nrbr\rrr*r])r_r`loadrZrrVrW)rrdreZckprrrread_checkpoints  (z BigBatchSearcher.read_checkpointN)rF)__name__ __module__ __qualname____doc__rrrr&r)r8rFrLrUrXrZrfrirrrrrs      rc@s,eZdZdZdejejfddZddZdS) BlockComputerz computation within one bucket knn_functioncCs||_|jtjkr0t|j|j}dd}d}n|jtjkrvt|j|j j |j j |j}|j |_ |j j }d|_ |j}nR|jtjkrt|j|jj|j}|j|_|jj }d|_ |j}ntd|jd||_|dkrdn||_||_||_||_||_dS)NcSs |dS)Nfloat32)view)xrrrz(BlockComputer.__init__..FTz index type z not supportedr )r __class__r Z IndexIVFFlatZ IndexFlatdrZ IndexIVFPQZIndexPQZpqMnbitsdecodeZ is_trainedrMZIndexIVFScalarQuantizerZIndexScalarQuantizersqZqtype RuntimeError index_helprNmethodpairwise_distancesknn)rr r}r~rr|rNrMrrrrs8   zBlockComputer.__init__c Ks|jj}|jdks|jdkr&d}}n|jdkr`t||jjt||j_ |j ||\}}nH|jdkr|j |||d}d}n(|jdkr|j |||fd|i|\}}||fS)Nrr r~)metricror) r rsizer}r Zcopy_array_to_vectorr|codesrZntotalrAr~r) rrPrSrRr extra_argsrrVrWrrr block_searchs     zBlockComputer.block_searchN) rjrkrlrmr r~rrrrrrrrns  #rnroFr*i r^c1 s`j}|dvsJ|j}t||tdj}t|tdjtdj}|||}|dkrtd|d|d|d |d |d d d t|||dt |||dj _ j _ | dur܈ n| _ |durj}t}| durh||fdjfks Jtj| r^td| | }tdt|n td|dkrt||D]h}||\}}}}t}|||\}}jdt|7<||||q|nJ|dkrfdd}|} d}!td}"t||D]|}||"||!|df}#| \}}}}|||\}}d||||f}!|# } dq"j|!|"!n~ddfdd}$d'fdd }%fdd }&t"d}'t"d}(|$|&dt|(|'})|$|%| ||||'d}*t}+td!|( },|,sFq$|,\}-}.}/}0}}}}|-|krlddjd|07<jd|.7<jd"|/7<td#|-||||td$|-|#|-|-| dur*t|+| kr*td%$| |t}+q*|*%|)%&d&j'()j'j*j'j+fS)(a Search queries xq in the IVF index, with a search function that collects batches of query vectors per inverted list. This can be faster than the regular search indexes. Supports IVFFlat, IVFPQ and IVFScalarQuantizer. Supports three computation methods: method = "index": build a flat index and populate it separately for each index method = "pairwise_distances": decompress codes and compute all pairwise distances for the queries and index and add result to heap method = "knn_function": decompress codes and compute knn results for the queries threaded=0: sequential execution threaded=1: prefetch next bucket while computing the current one threaded=2: prefetch prefetch_threads buckets at a time. compute_threads>1: the knn function will get an additional thread_no that tells which worker should handle this. In threaded mode, the computation is tiled with the bucket perparation and the writeback of results (useful to maximize GPU utilization). use_float16: convert all matrices to float16 (faster for GPU gemm) q_assign: override coarse assignment, should be a matrix of size nq * nprobe checkpointing (only for threaded > 1): checkpoint: file where the checkpoints are stored checkpoint_freq: when to perform checkpoinging. Should be a multiple of threaded start_list, end_list: process only a subset of invlists )r r~ror9Zint64rprzmemory: queries z assign z result z total z = i@r'z GiB)rr )r}r~rNzrecovering checkpoint: z already completed: z$no checkpoint: starting from scratchr+r*cs*|durj||jkr&|SdS)z` perform the addition for the previous bucket and prefetch the next (if applicable) N)rXr3rU)to_addr6)bbsr rradd_results_and_prefetchas  z2big_batch_search..add_results_and_prefetchr/c szxt|Rfddt||D}|D] }|q0Wdn1sb0YdWntt Yn0dS)Ncs(g|] }|vrj|fdqS))args) apply_async).0ir] input_queue output_queuepooltaskrr s zAbig_batch_search..task_manager_thread..) rr>getclosejoinput traceback print_exc_threadinterrupt_main) rZ pool_size start_taskZend_taskr]rrresrrrrtask_manager_thread~s   &z-big_batch_search..task_manager_threadcs tj|d}d|_||S)N)targetrT) threadingThreaddaemonstart)r task_manager)rrrrsz&big_batch_search..task_managercsnzJtd||\}}}}||||||ftd|WnttYn0dS)NzPrepare start: z Prepare end: )logginginforUrrrrr)task_idrrrOrPrRrS)rrr prepare_tasksz&big_batch_search..prepare_taskc sDztd|d}t}td||}t|}|durZ|dq|\}}} } } td|d|t}dkrj| | | |d\} } n| | | \} } t|}td|d|t}||||||| | | ft|}qtd |Wntt Yn0dS) NzCompute start: rzCompute input: task zCompute work: task z , centroid r*) thread_idzCompute output: task z Compute end: ) rrrrrrrrrr)rrr t_wait_outrZ input_value t_wait_incentroidrOrPrRrSrVrW t_compute)compcomputation_threadsrrr compute_tasks<    z&big_batch_search..compute_taskzWaiting for resultr0zAdding to heap start: centroid zAdding to heap end: centroid zwriting checkpointz finalize heap)N),r=nbytesrr;r:itemsizerrrrnrNrMrFrBrLr3setrbpathexistsrir>r8rUrrrrXrrrrrrraddrfrr&rfinalizer)rVrW)1r r rr}r~rrZthreadedr Zprefetch_threadsrrB checkpointZcheckpoint_freq start_listZend_listZcrash_atr=Z mem_queriesZ mem_assignZmem_resZmem_totr]r6rOrPrRrSZt0irVrWrZprefetched_bucketrrZprefetched_bucket_arrrZprepare_to_compute_queueZcompute_to_main_queueZcompute_task_managerZprepare_task_managerZ t_checkpointvaluerrrrr)rrrr rrrbig_batch_searchs4                      #            r)rr`rbrZmultiprocessing.poolrrrqueuerrr4numpyr;r Zfaiss.contrib.inspect_toolsrrrnr~rrrrrrs<   :