Ngh=ddlZddlZddlZddlmZddlmZddlm Z ddl m Z m Z m Z mZ eGddZGdd ZGd d ejZGd d ejZdS)N) dataclass)Queue)Thread)CallableIOOptionalUnionc(eZdZUdZeed<eed<dS)PathDataa Manage the IO job queue and polling thread for a single path. This is done to ensure that write calls to the same path are serialized so they are written in the same order as they were called. On each `f.write` call where `f` is of type `NonBlockingIO`, we send the job to the manager where it is enqueued to the Queue. The polling Thread picks up on the job, executes it, waits for it to finish, and then continues to poll. queuethreadN)__name__ __module__ __qualname____doc__r__annotations__rY/var/www/html/ai-engine/env/lib/python3.11/site-packages/iopath/common/non_blocking_io.pyr r s.   LLL NNNNNrr c ,eZdZdZ ddeedeejjddfdZ dd e d e e e e e fd eedgdfd eedejf d ZdeegdfddfdZdd ee defdZdefdZdS)NonBlockingIOManagera5 All `opena` calls pass through this class so that it can keep track of the threads for proper cleanup at the end of the script. Each path that is opened with `opena` is assigned a single queue and polling thread that is kept open until it is cleaned up by `PathManager.async_join()`. FNbufferedexecutorreturnci|_||_|jrtnt|_|pt j|_dS)av Args: buffered (bool): IO instances will be `NonBlockingBufferedIO` or `NonBlockingIO` based on this value. This bool is set manually for each `PathHandler` in `_opena`. executor: User can optionally attach a custom executor to perform async operations through `PathHandler.__init__`. N) _path_to_data _bufferedNonBlockingBufferedIO NonBlockingIO_IO concurrentfuturesThreadPoolExecutor_pool)selfrrs r__init__zNonBlockingIOManager.__init__3sF !,0NM(( H!3!F!F!H!H rpathio_objcallback_after_file_close bufferingc>js|dkrtd|djvrQt}t j|f}|t||j<jsind|i}jdfd||d|S) a Called by `PathHandler._opena` with the path and returns a `NonBlockingIO` instance. Args: path (str): A path str to operate on. This path should be simplified to ensure that each absolute path has only a single path str that maps onto it. For example, in `NativePathHandler`, we can use `os.path.normpath`. io_obj (IO): a reference to the IO object returned by the `PathHandler._open` function. callback_after_file_close (Callable): An optional argument that can be passed to perform operations that depend on the asynchronous writes being completed. The file is first written to the local disk and then the callback is executed. buffering (int): An optional argument to set the buffer size for buffered asynchronous writing. r'z`NonBlockingIO is not using a buffered writer but `buffering` arg is set to non-default value of z != -1.)targetargsr+cNjj|SN)rr put) io_callabler(r%s rz:NonBlockingIOManager.get_non_blocking_io..rs$"4(.22;??r)notify_managerr)r*r) r ValueErrorrrr _poll_jobsstartr r )r%r(r)r*r+r tkwargss`` rget_non_blocking_ioz(NonBlockingIOManager.get_non_blocking_ioGs2~ )r//I6?III  t) ) )GGEdoUH===A GGIII'/q'9'9D t $>G Y/Gtx &?       rr c |}|dS|j|E)a A single thread runs this loop. It waits for an IO callable to be placed in a specific path's `Queue` where the queue contains callable functions. It then waits for the IO job to be completed before looping to ensure write order. TN)getr$submitresult)r%r funcs rr6zNonBlockingIOManager._poll_jobszsG -99;;D| J  d # # * * , , , -rc|r||jvrt|d|d|r|gn%t|j}d}|D]} |j|}|jd|jQ#t$r7tj t}| d|dd}YwxYw|S)a Waits for write jobs for a specific path or waits for all write jobs for the path handler if no path is provided. Args: path (str): Pass in a file path and will wait for the asynchronous jobs to be completed for that file path. If no path is passed in, then all threads operating on all file paths will be joined. z6 has no async IO associated with it. Make sure `opena(z)` is called first.TNz`NonBlockingIO` thread for z failed to join.F)rr5listkeyspopr r1r join Exceptionlogging getLoggerr exception)r%r(paths_to_closesuccess_path path_dataloggers r_joinzNonBlockingIOManager._joins+  D 222>>$(>>>  $(L$T$2D2I2I2K2K-L-L#  E  .22599 ##D))) %%''''    *844  !Vu!V!V!VWWW sA B!!>C"!C"c |jn?#t$r2tjt }|dYdSwxYwdS)z( Closes the ThreadPool. z,`NonBlockingIO` thread pool failed to close.FT)r$shutdownrErFrGrrH)r%rMs r_close_thread_poolz'NonBlockingIOManager._close_thread_poolsl  J   ! ! ! !   &x00F   K L L L55 ts8AA)FNNr'r0)rrrrrboolr!r"Executorr&strr rbytesrintioIOBaser:r6rNrQrrrrr*sd$):>II4.I:-67I  IIII0GK#% 1 1 1 bgr%y()1 $,HdVT\,B#C 1 C= 1  1 1 1 1 f-"d(); <-----&(3-4> D      rrc eZdZ ddeegdfgdfdeeeeefdeedgdfddffd Z de fdZ de fdZ de fd Z d eeefddfd Zdd ededefdZdefdZddedefdZddZxZS)rNr4r)r*rct||_||_||_d|_dS)a Returned to the user on an `opena` call. Uses a Queue to manage the IO jobs that need to be run to ensure order preservation and a polling Thread that checks the Queue. Implementation for these are lifted to `NonBlockingIOManager` since `NonBlockingIO` closes upon leaving the context block. NOTE: Writes to the same path are serialized so they are written in the same order as they were called but writes to distinct paths can happen concurrently. Args: notify_manager (Callable): a callback function passed in from the `NonBlockingIOManager` so that all IO jobs can be stored in the manager. It takes in a single argument, namely another callable function. Example usage: ``` notify_manager(lambda: file.write(data)) notify_manager(lambda: file.close()) ``` Here, we tell `NonBlockingIOManager` to add a write callable to the path's Queue, and then to add a close callable to the path's Queue. The path's polling Thread then executes the write callable, waits for it to finish, and then executes the close callable. Using `lambda` allows us to pass callables to the manager. io_obj (IO): a reference to the IO object returned by the `PathHandler._open` function. callback_after_file_close (Callable): An optional argument that can be passed to perform operations that depend on the asynchronous writes being completed. The file is first written to the local disk and then the callback is executed. FN)superr&_notify_manager_io_callback_after_file_close _close_called)r%r4r)r* __class__s rr&zNonBlockingIO.__init__sAP -*C'"rcdSNFrr%s rreadablezNonBlockingIO.readableurcdSNTrrds rwritablezNonBlockingIO.writabletrcdSrhrrds rseekablezNonBlockingIO.seekablerjrbc<fddS)Q Called on `f.write()`. Gives the manager the write job to call. c8jSr0r^write)rmr%srr3z%NonBlockingIO.write..sTX^^A%6%6rNr])r%rms``rrrzNonBlockingIO.writes- 6666677777rroffsetwhencec@fddS)z' Called on `f.seek()`. c:jSr0)r^seek)rtr%rusrr3z$NonBlockingIO.seek..sTX]]66%B%BrNrs)r%rtrus```rrxzNonBlockingIO.seeks1 BBBBBBCCCCCrc td)z' Called on `f.tell()`. z2ioPath async writes does not support `tell` calls.)r5rds rtellzNonBlockingIO.tellsMNNNrsizec<fddS)z+ Called on `f.truncate()`. c8jSr0)r^truncater%r{srr3z(NonBlockingIO.truncate..sTX%6%6t%<%<rNrsrs``rr~zNonBlockingIO.truncate s- <<<<<=====rcfdjs!jrjd_dS) Called on `f.close()` or automatically by the context manager. We add the `close` call to the file's queue to make sure that the file is not closed before all of the write jobs are complete. c6jSr0r^closerdsrr3z%NonBlockingIO.close..TX^^%5%5rTN)r]r`r_rds`rrzNonBlockingIO.closesa 5555666! Bd&E B  !@ A A A!rr0)rrN)rrrrr rrUrVrr&rSrerirl bytearrayrrrWrxrzr~r __classcell__ras@rrrs GK -#-# (2t8"4!5t!;<-#bgr%y()-#$,HdVT\,B#C -#  -#-#-#-#-#-#^$$$8uUI-.848888DD3DDCDDDDOcOOOO>>S>C>>>>""""""""rrc eZdZdZ ddeegdfgdfdeeeeefde edgdfde ddf fd Z de fd Z de fd Zde fd Zd eeefddfdZddZddZxZS)riNr'r4r)r*r+rct||_||_||_t jg|_|dkr|n|j|_ d|_ dS)aW Buffered version of `NonBlockingIO`. All write data is stored in an IO buffer until the buffer is full, or `flush` or `close` is called. Args: Same as `NonBlockingIO` args. buffering (int): An optional argument to set the buffer size for buffered asynchronous writing. rFN) r\r&r]r^r_rXBytesIO_buffersMAX_BUFFER_BYTES _buffer_sizer`)r%r4r)r*r+ras rr&zNonBlockingBufferedIO.__init__+sf -*C' )2QIID.asT]2%6%<%<%>%>rc6jSr0rrdsrr3z-NonBlockingBufferedIO.close..crrTN)rr]r`r_rds`rrzNonBlockingBufferedIO.closeYs  >>>>??? 5555666! Bd&E B  !@ A A A!rcjddkrdSd}dtj}}||kr<|||jz}|ffd |jz }||k<fdjtj dS)aR Called on `f.write()` if the buffer is filled (or overfilled). Can also be explicitly called by user. NOTE: Buffering is used in a strict manner. Any buffer that exceeds `self._buffer_size` will be broken into multiple write jobs where each has a write call with `self._buffer_size` size. r'rNc8j|Sr0rq)itemr%s rr3z-NonBlockingBufferedIO.flush..~s48>>$3G3Grc,Sr0)r)rsrr3z-NonBlockingBufferedIO.flush..sV\\^^r) rrzrxrXSEEK_END getbufferrr]appendr)r%pos total_sizerrrs` @rrzNonBlockingBufferedIO.flushjsr" ;;==A   F[[BK00 !!JcD$5556D  T!G!G!G!G!G H H H 4$ $C J 3333444 RZ\\*****rrRr)rrrrrr rrUrVrrWr&rSrerirlrrrrrrrs@rrr's_' GK ## (2t8"4!5t!;<#bgr%y()#$,HdVT\,B#C #  #  ######4$$$ uUI-. 4    """""++++++++rr)concurrent.futuresr!rXrF dataclassesrr r threadingrtypingrrrr r rrYrrrrrrs^ !!!!!!000000000000  $LLLLLLLL`h"h"h"h"h"BIh"h"h"Z\+\+\+\+\+BI\+\+\+\+\+r