| 370 | } |
| 371 | |
| 372 | static Future<std::shared_ptr<StreamingReaderImpl>> MakeAsync( |
| 373 | std::shared_ptr<DecodeContext> context, std::shared_ptr<io::InputStream> stream, |
| 374 | io::IOContext io_context, Executor* cpu_executor, const ReadOptions& read_options) { |
| 375 | ARROW_ASSIGN_OR_RAISE( |
| 376 | auto buffer_it, |
| 377 | io::MakeInputStreamIterator(std::move(stream), read_options.block_size)); |
| 378 | ARROW_ASSIGN_OR_RAISE( |
| 379 | auto buffer_gen, |
| 380 | MakeBackgroundGenerator(std::move(buffer_it), io_context.executor())); |
| 381 | |
| 382 | AsyncGenerator<DecodedBlock> decoding_gen; |
| 383 | int max_readahead = 0; |
| 384 | if (read_options.use_threads) { |
| 385 | // Prepare a source generator capable of async-reentrancy and parallel execution |
| 386 | if (!cpu_executor) { |
| 387 | cpu_executor = GetCpuThreadPool(); |
| 388 | } |
| 389 | max_readahead = cpu_executor->GetCapacity(); |
| 390 | |
| 391 | // Since the chunking/decoding steps are heavy we want to schedule them as a |
| 392 | // separate task so as to maximize task distribution across CPU cores |
| 393 | // |
| 394 | // TODO: Add an `always_transfer` parameter to `MakeTransferredGenerator`? |
| 395 | buffer_gen = [source = std::move(buffer_gen), cpu_executor] { |
| 396 | return cpu_executor->TransferAlways(source()); |
| 397 | }; |
| 398 | auto chunking_gen = MakeChunkingGenerator(std::move(buffer_gen), |
| 399 | MakeChunker(context->parse_options())); |
| 400 | |
| 401 | // At this stage, we want to allow the decoding tasks for each chunked block to run |
| 402 | // in parallel on the CPU executor. However: |
| 403 | // - Chunking is inherently serial and not thread-safe |
| 404 | // - The chunking generator is not async-reentrant, won't play well with readahead |
| 405 | // |
| 406 | // Fortunately, `MappingGenerator` queues pending jobs and keeps only one future |
| 407 | // from its source active at a time - which takes care of those concerns. In |
| 408 | // addition, it will start the next job within the continuation of the previous one, |
| 409 | // but before invoking its map function (in our case, `DecodingOperator`). This |
| 410 | // allows for decoding tasks to gradually saturate multiple CPU cores over multiple |
| 411 | // iterations. At a high level, this is how the full pipeline would operate in cases |
| 412 | // where decoding tasks are disproportionately expensive: |
| 413 | // |
| 414 | // -------------------------------------------------------------------------- |
| 415 | // Reading: IoThread(?) --> Chunking: CpuThread(0) ... Decoding: CpuThread(0) |
| 416 | // -------------------------------------------------------------------------- |
| 417 | // Decoding: CpuThread(0) |
| 418 | // Reading: IoThread(?) --> Chunking: CpuThread(1) ... Decoding: CpuThread(1) |
| 419 | // -------------------------------------------------------------------------- |
| 420 | // Decoding: CpuThread(0) |
| 421 | // Decoding: CpuThread(1) |
| 422 | // Reading: IoThread(?) --> Chunking: CpuThread(2) ... Decoding: CpuThread(2) |
| 423 | // -------------------------------------------------------------------------- |
| 424 | // |
| 425 | // Remember that we should already be on the CPU executor following chunking, so the |
| 426 | // decoding task simply continues to use that thread rather than spawning a new one. |
| 427 | decoding_gen = |
| 428 | MakeMappedGenerator(std::move(chunking_gen), DecodingOperator(context)); |
| 429 | } else { |
nothing calls this directly
no test coverage detected