| 620 | } |
| 621 | |
| 622 | Status CopyFiles(const std::vector<FileLocator>& sources, |
| 623 | const std::vector<FileLocator>& destinations, |
| 624 | const io::IOContext& io_context, int64_t chunk_size, bool use_threads) { |
| 625 | if (sources.size() != destinations.size()) { |
| 626 | return Status::Invalid("Trying to copy ", sources.size(), " files into ", |
| 627 | destinations.size(), " paths."); |
| 628 | } |
| 629 | |
| 630 | auto copy_one_file = [&](size_t i, |
| 631 | const FileLocator& source_file_locator) -> Result<Future<>> { |
| 632 | if (source_file_locator.filesystem->Equals(destinations[i].filesystem)) { |
| 633 | RETURN_NOT_OK(source_file_locator.filesystem->CopyFile(source_file_locator.path, |
| 634 | destinations[i].path)); |
| 635 | return Future<>::MakeFinished(); |
| 636 | } |
| 637 | |
| 638 | ARROW_ASSIGN_OR_RAISE(auto source, |
| 639 | sources[i].filesystem->OpenInputStream(sources[i].path)); |
| 640 | ARROW_ASSIGN_OR_RAISE(const auto metadata, source->ReadMetadata()); |
| 641 | |
| 642 | ARROW_ASSIGN_OR_RAISE(auto destination, destinations[i].filesystem->OpenOutputStream( |
| 643 | destinations[i].path, metadata)); |
| 644 | RETURN_NOT_OK(internal::CopyStream(source, destination, chunk_size, io_context)); |
| 645 | // Using the blocking Close() here can cause reduced performance and deadlocks because |
| 646 | // FileSystem implementations that implement background_writes need to queue and wait |
| 647 | // for other IO thread(s). There is a risk that most or all the threads in the IO |
| 648 | // thread pool are blocking on a call Close(), leaving no IO threads left to actually |
| 649 | // fulfil the background writes. |
| 650 | return destination->CloseAsync(); |
| 651 | }; |
| 652 | |
| 653 | // Spawn copy_one_file less urgently than default, so that background_writes are done |
| 654 | // with higher priority. Otherwise copy_one_file will keep buffering more data in memory |
| 655 | // without giving the background_writes any chance to upload the data and drop it from |
| 656 | // memory. Therefore, without this large copies would cause OOMs. |
| 657 | TaskHints hints{10}; |
| 658 | auto future = ::arrow::internal::OptionalParallelForAsync( |
| 659 | use_threads, sources, std::move(copy_one_file), io_context.executor(), hints); |
| 660 | |
| 661 | // Wait for all the copy_one_file instances to complete. |
| 662 | ARROW_ASSIGN_OR_RAISE(auto copy_close_async_future, future.result()); |
| 663 | |
| 664 | // Wait for all the futures returned by copy_one_file to complete. When the destination |
| 665 | // filesystem uses background_writes this is when most of the upload happens. |
| 666 | for (const auto& result : copy_close_async_future) { |
| 667 | result.Wait(); |
| 668 | } |
| 669 | return Status::OK(); |
| 670 | } |
| 671 | |
| 672 | Status CopyFiles(const std::shared_ptr<FileSystem>& source_fs, |
| 673 | const FileSelector& source_sel, |