(*args, **kwargs)
| 289 | |
| 290 | @wraps(func) |
| 291 | async def async_inner(*args, **kwargs): |
| 292 | batchmeta = _find_batchmeta(*args, **kwargs) |
| 293 | if batchmeta is None: |
| 294 | return await func(*args, **kwargs) |
| 295 | else: |
| 296 | logger.info( |
| 297 | f"Task {func.__name__} (pid={pid}) is getting len_samples={batchmeta.size}, " |
| 298 | f"global_idx={batchmeta.global_indexes}" |
| 299 | ) |
| 300 | args = [await _async_batchmeta_to_dataproto(arg) if isinstance(arg, BatchMeta) else arg for arg in args] |
| 301 | kwargs = { |
| 302 | k: await _async_batchmeta_to_dataproto(v) if isinstance(v, BatchMeta) else v |
| 303 | for k, v in kwargs.items() |
| 304 | } |
| 305 | output = await func(*args, **kwargs) |
| 306 | need_collect = _compute_need_collect(dispatch_mode, args) |
| 307 | if put_data and need_collect: |
| 308 | updated_batchmeta = await _async_update_batchmeta_with_output(output, batchmeta, func.__name__) |
| 309 | return updated_batchmeta |
| 310 | return _postprocess_common(output, put_data, need_collect) |
| 311 | |
| 312 | @wraps(func) |
| 313 | def dummy_inner(*args, **kwargs): |
nothing calls this directly
no test coverage detected