(*args, **kwargs)
| 270 | |
| 271 | @wraps(func) |
| 272 | def inner(*args, **kwargs): |
| 273 | batchmeta = _find_batchmeta(*args, **kwargs) |
| 274 | if batchmeta is None: |
| 275 | return func(*args, **kwargs) |
| 276 | else: |
| 277 | logger.info( |
| 278 | f"Task {func.__name__} (pid={pid}) is getting len_samples={batchmeta.size}, " |
| 279 | f"global_idx={batchmeta.global_indexes}" |
| 280 | ) |
| 281 | args = [_batchmeta_to_dataproto(arg) if isinstance(arg, BatchMeta) else arg for arg in args] |
| 282 | kwargs = {k: _batchmeta_to_dataproto(v) if isinstance(v, BatchMeta) else v for k, v in kwargs.items()} |
| 283 | output = func(*args, **kwargs) |
| 284 | need_collect = _compute_need_collect(dispatch_mode, args) |
| 285 | if put_data and need_collect: |
| 286 | updated_batch_meta = _update_batchmeta_with_output(output, batchmeta, func.__name__) |
| 287 | return updated_batch_meta |
| 288 | return _postprocess_common(output, put_data, need_collect) |
| 289 | |
| 290 | @wraps(func) |
| 291 | async def async_inner(*args, **kwargs): |
nothing calls this directly
no test coverage detected