o rhD@sddlZddlZddlZddlZddlmZddlZddlZddlm Z ddl Z ddl Z ddl Z ddlZddlmZGdddZGdddZd ejejddd d d ddd ddd fddZdS)N) ThreadPool)Queue) get_invlistc@s~eZdZdZ  dddZddZdd 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_float16faissis_similarity_metric metric_type ResultHeaplenrht_accutime t_displayt0)selfr r krr rr\/var/www/html/alpaca_bot/venv/lib/python3.10/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_accu3s zBigBatchSearcher.stop_t_accucCs.|tf|_|jdkrt|ddddSdS)Nr Tendflush)rr rprint)rnamerrrtic6s zBigBatchSearcher.ticcCs:|j\}}t|}|jdkrt|d|dd|S)Nrz: .3fz s)r rrr&)rr'rdtrrrtoc;s   zBigBatchSearcher.toccCs|jdks|jdkr|dkrt|jdkrdSt|j}td|dd|d|jjd |jd d d |jdd d |jdd d|jdd d|jdd d|jdd dtj ||jj|d|ddt |jdkrudndddt|_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 get_mem_usage_kb)rltrrrreportBs6         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)rbsnqrEi0i1q_dis_i q_assign_irrrcoarse_quantizationWs   z$BigBatchSearcher.coarse_quantizationcCsz|d|j}|d7}tj|j|jjddd|_|j|_|j dkr-t d|jd|jdd|_|`| dS)Nz bucket sortr,)nbucketntrz number of -1s:) r(rEr matrix_bucket_sort_inplacer r5 bucket_limsravel query_idsrr&r+)rrErrrreorder_assignes    zBigBatchSearcher.reorder_assignc Cst}|j}|j||j|d}}|j||}|j|}|jr+||j|}t}t|j |\} } |j durA| } n| | } |j rS| d} | d}t} |jd||7<|jd| |7<||| | fS)z4 prepare the queries and database items for bucket lr,Nfloat16r)rr rQrSr by_residualrC reconstructrinvlists decode_funcrRr astyper) rr9rr rHrIq_subsetxq_lt1list_idsxb_lt2rrrprepare_bucketrs&       zBigBatchSearcher.prepare_bucketcCsX|durdSt}|dur|}n||}|j||||jdt|7<dS)z,add the bucket results to the heap structureNr0)rradd_result_subsetr)rr[Dr^Irrrradd_results_to_heapsz$BigBatchSearcher.add_results_to_heapcCs|jj|jj|jjfSr)r shaper r@r5rrrrsizes_in_checkpointsz$BigBatchSearcher.sizes_in_checkpointcCsh|d}t|d}t|||jj|jjfd|dWdn1s'wYt||dS)Nz.tmpwb)sizes completedr) openpickledumprgrrcrdosreplace)rfnamerjtmpnamefrrrwrite_checkpoints z!BigBatchSearcher.write_checkpointcCst|d }t|}Wdn1swY|d|ks$J|dd|jjdd<|dd|jjdd<|dS)Nrbrirrr,rj)rlrmloadrgrrcrd)rrqrsckprrrread_checkpoints  z BigBatchSearcher.read_checkpointN)rF)__name__ __module__ __qualname____doc__rrr!r(r+r;rLrTrarergrtrxrrrrrs"     rc@s,eZdZdZdejejfddZddZdS) BlockComputerz computation within one bucket knn_functioncCs||_|jtjkrt|j|j}dd}d}nL|jtjkr;t|j|j j |j j |j}|j |_ |j j }d|_ |j}n)|jtjkr[t|j|jj|j}|j|_|jj }d|_ |j}n td|jd||_|dkrmdn||_||_||_||_||_dS)NcSs |dS)Nfloat32)view)xrrrs z(BlockComputer.__init__..FTz index type z not supportedr )r __class__r IndexIVFFlat IndexFlatdr IndexIVFPQIndexPQpqMnbitsdecode is_trainedrVIndexIVFScalarQuantizerIndexScalarQuantizersqqtype RuntimeError index_helprYmethodpairwise_distancesknn)rr rrrrrYrVrrrrs8    zBlockComputer.__init__c Ks|jj}|jdks|jdkrd}}||fS|jdkr6t||jjt||j_ |j ||\}}||fS|jdkrI|j |||d}d}||fS|jdkr]|j |||fd|i|\}}||fS)Nrr r)metricr~r) r rsizerr copy_array_to_vectorrcodesrntotalrDrr) rr\r_r^r extra_argsrrcrdrrr block_searchs     zBlockComputer.block_searchN) ryrzr{r|r rrrrrrrrr}s #r}r~Fr,i rkc1 sTj}|dvs J|j}t||tdj}t|tdjtdj}|||}|dkrLtd|d|d|d |d |d d d t|||dt |||dj _ j _ | durn n| _ |dur|j}t}| dur||fdjfksJtj| rtd| | }tdt|ntd|dkrt||D]3}||\}}}}t}|||\}}jdt|7<||||qn$|dkrUfdd}|} d}!td}"t||D]>}||"||!|df}#| \}}}}|||\}}d||||f}!|# } dq j|!|"!nddfdd}$d(fdd }%fdd }&t"d}'t"d}(|$|&dt|(|'})|$|%| ||||'d}*t}+ td"|( },|,sno|,\}-}.}/}0}}}}|-|krddjd|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 rr~r<int64rrzmemory: queries z assign z result z total z = i@r)z GiB)rr )rrrNzrecovering checkpoint: z already completed: z$no checkpoint: starting from scratchr-r,cs*|dur j||jkr|SdS)z` perform the addition for the previous bucket and prefetch the next (if applicable) N)rer5ra)to_addr9)bbsr rradd_results_and_prefetchas    z2big_batch_search..add_results_and_prefetchr1c sz=t|)fddt||D}|D]}|qWdn1s1wYdWdStt )Ncs(g|]}|vrj|fdqS))args) apply_async).0irj input_queue output_queuepooltaskrr szAbig_batch_search..task_manager_thread..) rrAgetclosejoinput traceback print_exc_threadinterrupt_main) r pool_size start_taskend_taskrjrrresrrrrtask_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_managercsfz&td||\}}}}||||||ftd|WdStt)NzPrepare start: z Prepare end: )logginginforarrrrr)task_idrrr[r\r^r_)rrr prepare_tasksz&big_batch_search..prepare_taskc s:ztd|d} t}td||}t|}|dur,|dnZ|\}}} } } td|d|t}dkrSj| | | |d\} } n | | | \} } t|}td |d|t}||||||| | | ft|}q td |WdStt ) NzCompute start: rTzCompute input: task zCompute work: task z , centroid r,) thread_idzCompute output: task z Compute end: ) rrrrrrrrrr)rrr t_wait_outr input_value t_wait_incentroidr[r\r^r_rcrd t_compute)compcomputation_threadsrrr compute_tasks@      z&big_batch_search..compute_taskTzWaiting for resultr2zAdding to heap start: centroid zAdding to heap end: centroid zwriting checkpointz finalize heapr),r@nbytesrr>r=itemsizerrrr}rYrVrLrErTr5setropathexistsrxrAr;rarrrrerrrr!rrraddrtrr(rfinalizer+rcrd)1r r rrrrrthreadedr prefetch_threadsrrE checkpointcheckpoint_freq start_listend_listcrash_atr@ mem_queries mem_assignmem_resmem_totrjr9r[r\r^r_t0ircrdrprefetched_bucketrrprefetched_bucket_arrrprepare_to_compute_queuecompute_to_main_queuecompute_task_managerprepare_task_manager t_checkpointvaluerrrrr)rrrr rrrbig_batch_searchs4                      #           r)rrmrormultiprocessing.poolrrrqueuerrr6numpyr>r faiss.contrib.inspect_toolsrrr}rrrrrrrs>   :