o ̳i@sddlZddlZddlZddlmZmZmZmZddlZddl Zddl m Z ddl m Z mZddlmZmZddlmZddlmZmZde d ejd ed ed ejd ejdeeefdeefddZGddde eZdS)N)AnyDictOptionalUnion) pin_memory)BaseNodeT)ExceptionWrapperStartupExceptionWrapper)_SingleThreadedMapper)MonotonicIndex SnapshotStoresourceqsnapshot_storesnapshot_frequency semaphore stop_event device_iddevicec st  ddtdtttttftfffdd }zOt dtj d|d kr4tj |n|d kr?tj |n|tjkrStttj} | |t|tr\|d kscJd |j|d Wntytd|d} j| d YdSwd } |s|jdddsqz&t|} t| |} | d7} d} |d kr| |d kr|} || d| dWn2ty} z| } || ddWYd} ~ dSd} ~ wtytd|d} || ddYdSw|rdSdS)zThis is fork of from torch.utils.data._utils.pin_memory import _pin_memory_loop to remove the index tuples. This setting is thread local, and prevents the copy in pin_memory from consuming all CPU cores. TNblocksnapshotcs0}|r j||dj||f|ddS)N)rversionr)getappendput)itemrr_idxidxrrN/home/ubuntu/.local/lib/python3.10/site-packages/torchdata/nodes/pin_memory.py_put+sz_pin_memory_loop.._put pt_data_pincudaxpurz5snapshot_frequency must be non-negative integer! Got )rz'in _pin_memory_loop startup for device )whereg?)blockingtimeoutF)rrrzin _pin_memory_loop for device )TN)r boolrrrstrrr torchset_num_threadsmultiprocessing_set_thread_namer& set_devicer'_C_get_privateuse1_backend_namegetattr isinstanceintappend_initial_snapshot state_dict Exceptionis_setacquirenextr StopIterationr )rrrrrrrrr#custom_device_modeyieldedrrr!rr"_pin_memory_loopsh          rAcsveZdZdZ  ddeededeffdd Zdd e e ee fffd d Z d dZ de ee ffddZZS) PinMemoryauPins the data of the underlying node to a device. This is backed by torch.utils.data._utils.pin_memory._pin_memory_loop. Args: source (BaseNode[T]): The source node to pin the data from. pin_memory_device (str): The device to pin the data to. Default is "". snapshot_frequency (int): The frequency at which to snapshot the state of the source node. Default is 1, which means that the state of the source node will be snapshotted after every item. If set to a higher value, the state of the source node will be snapshotted after every snapshot_frequency items. r$rpin_memory_devicercst||_||_t|dkrd|_n||_|jdkr$tj|_ n|jtj kr:t ttj }||_ ntj |_ d|_dS)Nrr')super__init__rrlen_pin_memory_devicer-r'current_device_current_devicer2r3r4r&_it)selfrrDrr> __class__r!r"rFms      zPinMemory.__init__N initial_statecsPt||jdur|j|`t|jdtjt|j |j d|j |d|_dS)Nr$)rr)rprefetch_factorworkerrrO) rEresetrK _shutdownr r functoolspartialrArJrHr)rLrOrMr!r"rRs    zPinMemory.resetcCs t|jSN)r<rKrLr!r!r"r< zPinMemory.nextreturncCs |jSrV)rK get_staterWr!r!r"rZrXzPinMemory.get_state)rCr$rV)__name__ __module__ __qualname____doc__rrr,r6rFrrrrRr<rZ __classcell__r!r!rMr"rBas rB)rTqueue threadingtypingrrrrr-torch.multiprocessing"torch.utils.data._utils.pin_memoryrtorchdata.nodes.base_noderr!torchdata.nodes.exception_wrapperr r torchdata.nodes.mapr torchdata.nodes.snapshot_storer r Queuer6BoundedSemaphoreEventr,rArBr!r!r!r"s:    I