U H$xe6> @svddlZddlZddlZddlZddlZddlZddlmZmZddl m Z ddl m Z ddl mZmZmZmZmZmZmZmZddlZddlmZmZmZmZddlmZddlmZdd l m!Z!m"Z"e#Z$e%Z&erej'Z(ne&j'Z(Gd d d e(Z)Gd d d e(Z*GdddZ+GdddeZ,Gddde,Z-Gddde,Z.deeee/ee0e!dfee/edddZ1dS)N)ABCabstractmethod) ExitStack) Connection)AnyListOptionalTupleSequenceIterator TYPE_CHECKINGUnion) InputFiles OutputFiles InputPathsxopen_rb_raise_limit)Pipeline) Statistics)Progress DummyProgresscsfeZdZdZeeeeje dfdd Z ddZ e e edfdd d Zdd d ZddZZS) ReaderProcessaC Read chunks of FASTA or FASTQ data (single-end or paired) and send them to a worker. The reader repeatedly - reads a chunk from the file(s) - reads a worker index from the Queue - sends the chunk to connections[index] and finally sends the stop token -1 ("poison pills") to all connections. )paths connectionsqueue buffer_sizecsLtt|dkrtd|s*td||_||_||_||_||_dS)a Args: paths: path to input files connections: a list of Connection objects, one for each worker. queue: a Queue of worker indices. A worker writes its own index into this queue to notify the reader that it is ready to receive more data. buffer_size: stdin_fd: Note: This expects the paths to the input files as strings because these can be pickled while file-like objects such as BufferedReader cannot. When using multiprocessing with the "spawn" method, which is the default method on macOS, function arguments must be picklable. z8Reading from more than two files currently not supportedzMust provide at least one fileN) super__init__len ValueError_pathsrrrstdin_fd)selfrrrr!r __class__/lib/python3.8/site-packages/cutadapt/runners.pyr+s  zReaderProcess.__init__c s|jdkr"tjt|jt_zZtBfdd|jD}t|j |D]\}}|j |f|qNW5QRX| WnJt k r}z,|j D] }|d||tfqW5d}~XYnXdS)Ncsg|]}t|qSr%) enter_contextr).0pathstackr%r& Ssz%ReaderProcess.run..)r!sysstdincloseosfdopenrr enumerate _read_chunkssend_to_workershutdown Exceptionrsend traceback format_exc)r"filesindexchunkse connectionr%r+r&runMs      zReaderProcess.run.returncgsht|dkr.t|d|jD] }|fVqn6t|dkr`t|d|d|jD] }|VqRntdS)Nrr)rdnaioZ read_chunksrZread_paired_chunksNotImplementedError)r"r<chunkr>r%r%r&r5`s    zReaderProcess._read_chunksNcCs>|j}|j|}|||||dk r:||dSN)rgetrr9 send_bytes)r" chunk_indexZchunk1Zchunk2 worker_indexr@r%r%r&r6ls     zReaderProcess.send_to_workercCs2tt|jD]}|j}|j|dqdSNr')rangerrrrIr9)r"_rLr%r%r&r7ts zReaderProcess.shutdown)N)__name__ __module__ __qualname____doc__strr rmultiprocessingQueueintrrAr r memoryviewr5r6r7 __classcell__r%r%r#r&rs" rc sPeZdZdZeeeeeeee j dfdd Z ddZ eeeddd Z ZS) WorkerProcessa7 The worker repeatedly reads chunks of data from the read_pipe, runs the pipeline on it and sends the processed chunks to the write_pipe. To notify the reader process that it wants data, it puts its own identifier into the need_work_queue before attempting to read data from the read_pipe. )id_pipeline n_input_filesinterleaved_input orig_outfiles read_pipe write_pipeneed_work_queuec sBt||_||_||_||_||_||_||_| |_ dSrH) rr_id _pipeline_n_input_files_interleaved_input _read_pipe _write_pipe_need_work_queue as_bytesio_original_outfiles) r"r[r\r]r^r_r`rarbr#r%r&rs zWorkerProcess.__init__c srz&t}jjj}|dkr.qn&|dkrTj\}}td||fddtj D}t |dj i}j }j||\}} } jt|| | gjj} || 7}|||jq jj} tddjjrdnd| g} || 7}jdj|WnDtk rl}z$jdj|tfW5d}~XYnXdS)Nr'r.%scsg|]}tjqSr%)ioBytesIOrg recv_bytesr)rOr"r%r&r-sz%WorkerProcess.run.. interleavedr)rriZputrcrgrecvloggererrorrNrerrfrkrjrd process_readsflushcollect_steps_send_outfilesr1 _modifiersZpairedrhr9r8r:r;)r"statsrKr?tb_strr<infilesoutfilesnZbp1Zbp2 cur_statsmZmodifier_statsr%rqr&rAsF        zWorkerProcess.run)rrKn_readscCsR|j||j||D]0}|t|tjs8t|}|j|qdSrH) rhr9rw isinstancermrnAssertionErrorgetvaluerJ)r"rrKrfZprocessed_chunkr%r%r&rzs  zWorkerProcess._send_outfiles)rPrQrRrSrWrboolrrrUrVrrArzrYr%r%r#r&rZ{s (rZc@s(eZdZdZddZddZddZdS) OrderedChunkWriterz We may receive chunks of processed data from worker processes in any order. This class writes them to an output file in the correct order. cCst|_d|_||_dS)Nr)dict_chunks_current_index_outfile)r"Zoutfiler%r%r&rszOrderedChunkWriter.__init__cCsH||j|<|j|jkrD|j|j|j|j|j=|jd7_q dS) rDN)rrrwrite)r"datar=r%r%r&rs    zOrderedChunkWriter.writecCs|j SrH)rrqr%r%r&wrote_everythingsz#OrderedChunkWriter.wrote_everythingN)rPrQrRrSrrrr%r%r%r&rsrc@sNeZdZdZeedddZeedddZ edd Z d d Z d d Z dS)PipelineRunnerz$ A read processing pipeline )r\progresscCs||_||_dSrH)rd _progress)r"r\rr%r%r&rszPipelineRunner.__init__rBcCsdSrHr%rqr%r%r&rAszPipelineRunner.runcCsdSrHr%rqr%r%r&r1szPipelineRunner.closecCs|SrHr%rqr%r%r& __enter__szPipelineRunner.__enter__cGs |dSrH)r1)r"argsr%r%r&__exit__szPipelineRunner.__exit__N) rPrQrRrSrrrrrrAr1rrr%r%r%r&rs rcseZdZdZdeeeeee edfdd Z dde e ddd d Z eeeeefd d d Zed ddZeddZdd ddZZS)ParallelPipelineRunnera Run a Pipeline in parallel - When connect_io() is called, a reader process is spawned. - When run() is called, as many worker processes as requested are spawned. - In the main process, results are written to the output files in the correct order, and statistics are aggregated. If a worker needs work, it puts its own index into a Queue() (_need_work_queue). The reader process listens on this queue and sends the raw data to the worker that has requested work. For sending the data from reader to worker, a Connection() is used. There is one such connection for each worker (self._pipes). For sending the processed data from the worker to the main process, there is a second set of connections, again one for each worker. When the reader is finished, it sends 'poison pills' to all workers. When a worker receives this, it sends a poison pill to the main process, followed by a Statistics object that contains statistics about all the reads processed by that worker. N)r\inpathsrr n_workersrcsNt||||_t|_|dkr*dn||_||_|j|j d|j idS)Ni@rr) rr _n_workersmpctxrVri _buffer_size _outfiles _assign_inputrrr)r"r\rrrrrr#r%r&rs  zParallelPipelineRunner.__init__F)rr)rrrrCcGst||_||_ddt|jD}t|\|_}ztj }Wnt j k rZd}YnXt |||j |j|d|_d|j_|jdS)NcSsg|]}tjddqS)FZduplex)rPiperpr%r%r&r-.sz8ParallelPipelineRunner._assign_input..r')rrrr!T)rrerfrNrzip _connectionsr/r0filenormUnsupportedOperationrrir_reader_processdaemonstart)r"rrrrZconnwrr%r%r&r&s"   z$ParallelPipelineRunner._assign_inputrBc Csxg}g}t|jD]\}tjdd\}}||t||j|j|j|j |j |||j }d|_ | ||q||fS)NFrT)rNrrrappendrZrdrerfrrrirr)r"workersrr=Zconn_rZconn_wZworkerr%r%r&_start_workers@s&   z%ParallelPipelineRunner._start_workerscCs|\}}g}|jD]}|t|qt}|rtj|}|D]h}||}|dkrx||} || 7}| |qD||} |j | |D]} | } | | |qqDq0|D]} | stq|D] } | q|j|j |SrM)rrrrrrUr@wait _try_receiveremoverupdaterorrrjoinrr1)r"rrZwritersrr|Zready_connectionsr@rKrZnumber_of_readswriterrwr%r%r&rAUs4           zParallelPipelineRunner.runcCsF|}|dkrB|\}}td|tD] }|q0||S)z{ Try to receive data over `self.connection` and return it. If an exception was received, raise it. r.rl)rsrtdebugrUZactive_childrenZ terminate)r@resultr?r}Zchildr%r%r&rss    z#ParallelPipelineRunner._try_receivecCs|jdSrH)rr1rqr%r%r&r1szParallelPipelineRunner.close)N)rPrQrRrSrrrrrWrrrTrrr rrZrrrrA staticmethodrr1rYr%r%r#r&rs*  rcsHeZdZdZeeeedfdd Ze dddZ ddd d Z Z S) SerialPipelineRunnerz) Run a Pipeline on a single core )r\r~rrcst||||_||_dSrH)rr_infilesr)r"r\r~rrr#r%r&rszSerialPipelineRunner.__init__rBcCsd|jj|j|j|jd\}}}|jdk r2|jt|jdd}|dk sLtt |||||jj S)N)rr{) rdrvrrrr1getattrrrrxry)r"rZ total1_bpZ total2_bpZ modifiersr%r%r&rAs     zSerialPipelineRunner.runNcCs|jdSrH)rdr1rqr%r%r&r1szSerialPipelineRunner.close) rPrQrRrSrrrrrrrAr1rYr%r%r#r&rs  r)r\rrcoresrrrCc Csr|dks|dkrt}n|dkr&t}|dkrDt||||||d}nt||||}||}W5QRX|S)u Run a pipeline. This uses a SerialPipelineRunner if cores is 1 and a ParallelPipelineRunner otherwise. Args: inpaths: outfiles: cores: number of cores to run the pipeline on (this is actually the number of worker processes, there will be one extra process for reading the input file(s)) progress: Set to False for no progress bar, True for Cutadapt’s default progress bar, or use an object that supports .update() and .close() (e.g. a tqdm instance) buffer_size: Forwarded to `ParallelPipelineRunner()`. Ignored if cores is 1. Returns: A Statistics object NFTrD)rr)rrrropenrA)r\rrrrrZrunnerZ statisticsr%r%r& run_pipelines" r)NN)2rmZloggingrUr2r/r:abcrr contextlibrZmultiprocessing.connectionrtypingrrrr r r r r rEZcutadapt.filesrrrrZcutadapt.pipelinerZcutadapt.reportrZcutadapt.utilsrrZ getLoggerrtZ get_contextrZProcessZ mpctx_ProcessrrZrrrrrWrrr%r%r%r&sJ  (  ]S &