ARROW-12208: [C++] Add the ability to run async tasks without using the CPU thread pool #9892

westonpace · 2021-04-05T14:31:48Z

Added a serial executor which can run async code serially without needing to create threads. In addition, modifies the Scanner::ToTable and FileSystemDataset::Write to execute serially. Scanner::Scan is still non-serial but I think this is OK. The I/O thread pool is still being used however. To truly remove all instances of std::thread we'd need to change the I/O context to use the serial executor but let's see if this makes R happy enough first.

westonpace · 2021-04-05T14:34:27Z

@pitrou I initially wanted to replace the IOContext's executor with the serial executor as well but I wasn't sure what the best way to do that would be. For example, I could create a new IOContext with the same pool & stop token but a different executor and pass that down but I wasn't sure if that would be correct.

I think a simplified version of my question is "does the IOContext come from the calling user (via options) or from the filesystem?"

pitrou · 2021-04-05T14:36:23Z

I think a simplified version of my question is "does the IOContext come from the calling user (via options) or from the filesystem?"

Indeed, I think we should clarify this. I would say: by default, the default IOContext comes from the filesystem but can be overriden by the user. But we may need to modify the IOContext API slightly so that executor = nullptr is taken to mean "take the default".

westonpace · 2021-04-05T14:43:11Z

Ok, that makes sense. I think that RTools is happy enough with the I/O threads (they have always existed anyways) so I'll defer that work for the future.

github-actions · 2021-04-05T17:19:56Z

https://issues.apache.org/jira/browse/ARROW-12208

westonpace · 2021-04-06T03:19:26Z

JNI check seems unrelated. Please review.

lidavidm

I left some tiny nits but this looks good overall.

cpp/src/arrow/util/thread_pool.h

cpp/src/arrow/util/thread_pool_test.cc

pitrou

Nice! I hope we can cut down on the complication by returning Future<T>.

pitrou · 2021-04-06T14:18:27Z

cpp/src/arrow/util/thread_pool.cc

+  return Status::OK();
+}
+
+void SerialExecutor::MarkFinished(bool& finished) {


bool* finished, since it's mutable.

pitrou · 2021-04-06T14:19:39Z

cpp/src/arrow/util/thread_pool.cc

+
+  while (!finished) {
+    while (!state_->task_queue.empty()) {
+      Task& task = state_->task_queue.front();


The task_queue can be mutated from another thread, and I don't think there are any guarantees about reference stability.

True, I thought I was fine since I re-lock before I pop but I suppose the underlying storage could be getting reallocated.

pitrou · 2021-04-06T14:31:48Z

cpp/src/arrow/util/thread_pool.h

+  template <typename T = ::arrow::detail::Empty>
+  using FinishSignal = internal::FnOnce<void(const Result<T>&)>;
+  template <typename T = ::arrow::detail::Empty>
+  using Scheduler = internal::FnOnce<Status(Executor*, FinishSignal<T>)>;


"Scheduler" is confusing. "TopLevelTask" perhaps?

Renamed to TopLevelTask

cpp/src/arrow/util/thread_pool.h

pitrou · 2021-04-06T14:51:24Z

cpp/src/arrow/util/thread_pool_test.cc

+        std::move(finish_signal)(Status::Invalid("XYZ"));
+        return Status::OK();
+      };
+  ASSERT_RAISES(Invalid, SerialExecutor::RunInSerialExecutor(std::move(scheduler)));


Hmm... instead of testing RunInSerialExecutor explicitly, can all these tests call RunSynchronously, so that you can run the same tests for different use_threads?

pitrou · 2021-04-06T14:54:40Z

cpp/src/arrow/dataset/file_csv.cc

@@ -166,14 +174,15 @@ class CsvScanTask : public ScanTask {
        source_(fragment->source()) {}

  Result<RecordBatchIterator> Execute() override {
-    ARROW_ASSIGN_OR_RAISE(auto gen, ExecuteAsync());
+    ARROW_ASSIGN_OR_RAISE(auto gen, ExecuteAsync(internal::GetCpuThreadPool()));


Shouldn't you lookup ScanOptions::use_threads? Or am I missing something?

This is a bit awkward. This method would not be used by either Scanner::ToTable or FileSystemDataset::Write (they will call ExecuteAsync). The only way this could be called is if the user called Scan() directly.

In that case we cannot really use RunInSerialExecutor here because it is returning an iterator.

I suppose that means there is no truly serial way to call Scan() on CSV. It wouldn't be parallel exactly, it would just be juggling the work between the calling thread, the I/O executor, and the CPU executor.

cpp/src/arrow/dataset/scanner.h

pitrou · 2021-04-06T14:57:03Z

cpp/src/arrow/dataset/scanner_internal.h

@@ -29,10 +29,12 @@
 #include "arrow/dataset/partition.h"
 #include "arrow/dataset/scanner.h"
 #include "arrow/util/logging.h"
+#include "arrow/util/thread_pool.h"


Is this required?

pitrou · 2021-04-06T14:57:58Z

cpp/src/arrow/util/thread_pool.h

+  /// always mark a future finished (which can someday be aided by ARROW-12207).
+  template <typename T>
+  static Result<T> RunInSerialExecutor(Scheduler<T> initial_task) {
+    auto serial_executor = std::make_shared<SerialExecutor>();


Nit, but this can probably be SerialExecutor serial_executor;

pitrou · 2021-04-07T08:25:12Z

cpp/src/arrow/util/thread_pool.h

+    return final_result;
+  }
+  void RunLoop(const bool& finished);
+  void MarkFinished(bool& finished);


Also, it seems bool finished may simply be an internal state member of SerialExecutor, e.g.:

void RunLoop(); void FinishRunning();

(though, is the executor meant to be restartable?)

I moved finished into the internal state. The executor is not meant to be restartable at all and should not exist outside of the scope of RunInSerialExecutor. I've moved the constructor for SerialExecutor to private to better reflect this.

westonpace · 2021-04-08T09:13:17Z

I believe I've addressed all comments. The only outstanding question is how we want to handle CsvScanTask::Execute which I've added some comments to.

pitrou · 2021-04-08T12:33:56Z

cpp/src/arrow/dataset/file_base.cc

+      });
+    }
+  }
+  RETURN_NOT_OK(task_group->Finish());


@westonpace Is there a reason why you're not using FinishAsync here?

Hmm, do you mean add FinishAsync to scan_futs and then wait for them all at AllComplete?

cpp/src/arrow/util/thread_pool.h

…y without needing to create threads. In addition, modifies the Scanner::ToTable and FileSystemDataset::Write to execute serially. Scanner::Scan is still non-serial but I think this is OK. The I/O thread pool is still being used however. To truly remove all instances of std::thread we'd need to change the I/O context to use the serial executor but let's see if this makes R happy enough first.

…commit

* Improve tests

…te(futures) I added task_group->FinishAsync to the futures and called AllComplete on that.

westonpace · 2021-04-09T06:31:02Z

I believe all new changes have been addressed.

nealrichardson · 2021-04-09T21:18:46Z

This seems to fix the CI issues at least, so I'm going to merge. We can follow up next week if necessary.

pitrou · 2021-04-13T15:28:57Z

cpp/src/arrow/csv/reader.h

@@ -26,6 +26,7 @@
 #include "arrow/type.h"
 #include "arrow/type_fwd.h"
 #include "arrow/util/future.h"
+#include "arrow/util/thread_pool.h"


arrow/util/type_fwd.h was probably sufficient.

…he CPU thread pool Added a serial executor which can run async code serially without needing to create threads. In addition, modifies the Scanner::ToTable and FileSystemDataset::Write to execute serially. Scanner::Scan is still non-serial but I think this is OK. The I/O thread pool is still being used however. To truly remove all instances of std::thread we'd need to change the I/O context to use the serial executor but let's see if this makes R happy enough first. Closes apache#9892 from westonpace/feature/arrow-12208 Lead-authored-by: Weston Pace <[email protected]> Co-authored-by: Antoine Pitrou <[email protected]> Signed-off-by: Neal Richardson <[email protected]>

github-actions bot added the Component: C++ label Apr 5, 2021

lidavidm approved these changes Apr 6, 2021

View reviewed changes

cpp/src/arrow/util/thread_pool.h Outdated Show resolved Hide resolved

cpp/src/arrow/util/thread_pool_test.cc Outdated Show resolved Hide resolved

pitrou requested changes Apr 6, 2021

View reviewed changes

pitrou reviewed Apr 7, 2021

View reviewed changes

westonpace requested a review from pitrou April 8, 2021 09:11

pitrou reviewed Apr 8, 2021

View reviewed changes

westonpace commented Apr 8, 2021

View reviewed changes

cpp/src/arrow/util/thread_pool.h Show resolved Hide resolved

westonpace and others added 9 commits April 8, 2021 19:40

ARROW-12208: Restoring unit tests accidentally removed with previous …

0d343cf

…commit

ARROW-12208: Restoring unit tests accidentally removed with previous …

d26283a

…commit

ARROW-12208: Forgot to assign the lock guard appropriately.

49a89bd

ARROW-12208: Addressed some of the PR feedback

5fc6f94

ARROW-12208: Forgot to export function

9859392

ARROW-12208: Addressed PR comments

fc2aa0c

* Simplify SerialExecutor a bit

3e8a346

* Improve tests

ARROW-12208: Instead of calling task_group->Finish and then AllComple…

c215a31

…te(futures) I added task_group->FinishAsync to the futures and called AllComplete on that.

westonpace force-pushed the feature/arrow-12208 branch from 32e9c12 to c215a31 Compare April 9, 2021 05:48

westonpace requested a review from pitrou April 9, 2021 06:28

nealrichardson closed this in 868b543 Apr 9, 2021

westonpace mentioned this pull request Apr 12, 2021

ARROW-12288: [C++] Create Scanner interface #9947

Closed

pitrou reviewed Apr 13, 2021

View reviewed changes

westonpace deleted the feature/arrow-12208 branch January 6, 2022 08:17

asfimport mentioned this pull request Apr 14, 2021

[C++] Add the ability to run async tasks without using the CPU thread pool #28026

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ARROW-12208: [C++] Add the ability to run async tasks without using the CPU thread pool #9892

ARROW-12208: [C++] Add the ability to run async tasks without using the CPU thread pool #9892

westonpace commented Apr 5, 2021

westonpace commented Apr 5, 2021

pitrou commented Apr 5, 2021

westonpace commented Apr 5, 2021

github-actions bot commented Apr 5, 2021

westonpace commented Apr 6, 2021

lidavidm left a comment

pitrou left a comment

pitrou Apr 6, 2021

westonpace Apr 6, 2021

pitrou Apr 6, 2021

westonpace Apr 6, 2021

westonpace Apr 6, 2021

pitrou Apr 6, 2021

westonpace Apr 6, 2021

pitrou Apr 6, 2021

westonpace Apr 6, 2021

pitrou Apr 6, 2021

westonpace Apr 7, 2021

pitrou Apr 6, 2021

westonpace Apr 7, 2021

pitrou Apr 6, 2021

westonpace Apr 6, 2021

pitrou Apr 7, 2021

westonpace Apr 7, 2021

westonpace commented Apr 8, 2021

pitrou Apr 8, 2021

westonpace Apr 9, 2021

westonpace Apr 9, 2021

westonpace commented Apr 9, 2021

nealrichardson commented Apr 9, 2021

pitrou Apr 13, 2021

ARROW-12208: [C++] Add the ability to run async tasks without using the CPU thread pool #9892

ARROW-12208: [C++] Add the ability to run async tasks without using the CPU thread pool #9892

Conversation

westonpace commented Apr 5, 2021

westonpace commented Apr 5, 2021

pitrou commented Apr 5, 2021

westonpace commented Apr 5, 2021

github-actions bot commented Apr 5, 2021

westonpace commented Apr 6, 2021

lidavidm left a comment

Choose a reason for hiding this comment

pitrou left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

westonpace commented Apr 8, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

westonpace commented Apr 9, 2021

nealrichardson commented Apr 9, 2021

Choose a reason for hiding this comment