o `۷i# @sddlZddlmZddlmZmZmZmZmZm Z m Z m Z m Z m Z ddlZddlmZddlmZerNddlZddlZddlZddlmZddlmZmZe dd d Ze ed e d fZe d eeee ed fe ed fe eeffZdede fddZ!dede fddZ"dede fddZ#dede fddZ$dede fddZ%edede fddZ&e d e d eed ffZ'eGdddeeZ(eGddde(d Z)eGd!d"d"e(eeej*fZ+eGd#d$d$e(d%Z,eGd&d'd'e)Z-dS)(N)ThreadPoolExecutor) TYPE_CHECKINGAnyDictGenericListMappingOptionalTupleTypeVarUnion) env_integer) DeveloperAPI) DataBatch) CollatedDataTorchDeviceType DataBatchTyper)bound torch.Tensor)r..batchreturncCsddl}t||jS)z*Check if a batch is a single torch.Tensor.rN)torch isinstanceTensor)rrrI/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/data/collate_fn.py _is_tensor3s rcC t|ttfotdd|DS)aDCheck if a batch is a sequence of torch.Tensors. >>> import torch >>> _is_tensor_sequence(torch.ones(1)) False >>> _is_tensor_sequence([torch.ones(1), torch.ones(1)]) True >>> _is_tensor_sequence((torch.ones(1), torch.ones(1))) True >>> _is_tensor_sequence([torch.ones(1), 1]) False cs|]}t|VqdSNr.0trrr Gz&_is_tensor_sequence..rlisttupleallrrrr_is_tensor_sequence:s r+cCr)a=Check if a batch is a sequence of sequences of torch.Tensors. Stops at one level of nesting. >>> import torch >>> _is_nested_tensor_sequence([torch.ones(1), torch.ones(1)]) False >>> _is_nested_tensor_sequence( ... ([torch.ones(1), torch.ones(1)], [torch.ones(1)]) ... ) True csrrr+r!rrrr$W z-_is_nested_tensor_sequence..r&r*rrr_is_nested_tensor_sequenceJs r.cC t|totdd|DS)aCheck if a batch is a mapping of keys to torch.Tensors. >>> import torch >>> _is_tensor_mapping({"a": torch.ones(1), "b": torch.ones(1)}) True >>> _is_tensor_mapping({"a": torch.ones(1), "b": [torch.ones(1), torch.ones(1)]}) False csrrr r"vrrrr$er%z%_is_tensor_mapping..rrr)valuesr*rrr_is_tensor_mapping\s r4cCr/)aECheck if a batch is a mapping of keys to sequences of torch.Tensors. >>> import torch >>> _is_tensor_sequence_mapping({"a": torch.ones(1), "b": torch.ones(1)}) False >>> _is_tensor_sequence_mapping( ... {"a": (torch.ones(1), torch.ones(1)), "b": [torch.ones(1), torch.ones(1)]} ... ) True csrrr,r0rrrr$sr-z._is_tensor_sequence_mapping..r2r*rrr_is_tensor_sequence_mappinghs r5cCs(t|pt|pt|pt|pt|S)a"Check if a batch matches any of the TensorBatchType variants. This function checks if the input batch is one of the following types: 1. A single torch.Tensor 2. A sequence of torch.Tensors 3. A sequence of sequences of torch.Tensors 4. A mapping (e.g., dict) of keys to torch.Tensors 5. A mapping (e.g., dict) of keys to sequences of torch.Tensors Args: batch: The input batch to check. Can be any type. Returns: bool: True if the batch matches any TensorBatchType variant, False otherwise. )rr+r.r4r5r*rrris_tensor_batch_typexsr6c@s(eZdZdZejdeddfddZdS) CollateFnzAbstract interface for collate_fn for `iter_torch_batches`. See doc-string of `collate_fn` in `iter_torch_batches` API for more details. rrrcCdS)zConvert a batch of data to collated format. Args: batch: The input batch to collate. Returns: The collated data in the format expected by the model. Nrselfrrrr__call__s zCollateFn.__call__N)__name__ __module__ __qualname____doc__abcabstractmethodrr;rrrrr7sr7c@eZdZdZd ddZdS) ArrowBatchCollateFnaCollate function that takes pyarrow.Table as the input batch type. Arrow tables with chunked arrays can be efficiently transferred to GPUs without combining the chunks with the `arrow_batch_to_tensors` utility function. See `DefaultCollateFn` for example. r pyarrow.TablerrcCr8)zConvert a batch of pyarrow.Table to collated format. Args: batch: The input pyarrow.Table batch to collate. Returns: The collated data in the format expected by the model. Nrr9rrrr; zArrowBatchCollateFn.__call__N)rrDrrr<r=r>r?r;rrrrrCsrCrDc@s,eZdZdZdeeejfddfddZdS)NumpyBatchCollateFnzQCollate function that takes a dictionary of numpy arrays as the input batch type.rrrcCr8)zConvert a batch of numpy arrays to collated format. Args: batch: The input dictionary of numpy arrays batch to collate. Returns: The collated data in the format expected by the model. Nrr9rrrr;rEzNumpyBatchCollateFn.__call__N) r<r=r>r?rstrnpndarrayr;rrrrrGs rGc@rB) PandasBatchCollateFnzGCollate function that takes a pandas.DataFrame as the input batch type.rpandas.DataFramerrcCr8)zConvert a batch of pandas.DataFrame to collated format. Args: batch: The input pandas.DataFrame batch to collate. Returns: The collated data in the format expected by the model. Nrr9rrrr;rEzPandasBatchCollateFn.__call__N)rrLrrrFrrrrrKsrKrLc seZdZdZeddZdddefdeedee dffded d e d e ffd d Z ddZ dddeee dfee edfffddZZS)DefaultCollateFnzIDefault collate function for converting Arrow batches to PyTorch tensors.2RAY_DATA_DEFAULT_COLLATE_FN_THREADPOOL_MAX_WORKERSNFdtypesz torch.dtypedevicer pin_memory num_workerscsPddl}t||_t|ttfr|||_n||_||_||_ d|_ dS)aHInitialize the collate function. Args: dtypes: The torch dtype(s) for the created tensor(s); if None, the dtype will be inferred from the tensor data. device: The device on which the tensor should be placed. Can be a string (e.g. "cpu", "cuda:0") or a torch.device object. pin_memory: Whether to pin the memory of the created tensors. num_workers: Number of worker threads for parallel tensor conversion. Defaults to `RAY_DATA_DEFAULT_COLLATE_FN_THREADPOOL_MAX_WORKERS`. rN) rsuper__init__rPrrHintrQrRrS _threadpool)r:rPrQrRrSr __class__rrrUs  zDefaultCollateFn.__init__cCs"t|ddr|jjdddSdS)z#Clean up threadpool on destruction.rWNF)wait)getattrrWshutdown)r:rrr__del__s zDefaultCollateFn.__del__rrDrrcCs\ddlm}|jdkr|jdurt|jd|_|jduo!|jjdk}|||j||j|jdS)zConvert an Arrow batch to PyTorch tensors. Args: batch: PyArrow Table to convert Returns: Dictionary mapping column names to lists of tensors r)arrow_batch_to_tensorsN) max_workerscpu)rPcombine_chunksrR threadpool) ray.data.util.torch_utilsr^rSrWrrQtyperPrR)r:rr^rarrrr; s zDefaultCollateFn.__call__)r<r=r>r?r _DEFAULT_NUM_WORKERSr r rrHboolrVrUr]rr; __classcell__rrrXrrMs2rM).r@concurrent.futuresrtypingrrrrrrr r r r numpyrIray._private.ray_constantsr ray.util.annotationsrpandaspyarrowrray.data.blockrray.data.datasetrrrTensorSequenceTyperHTensorBatchTyperfrr+r.r4r5r6TensorBatchReturnTyper7rCrJrGrKrMrrrrsd 0