GH-35176: [C++] Add support for disabling threading for emscripten

ci/scripts/cpp_build.sh

@@ -106,6 +106,7 @@ cmake \
   -DARROW_C_FLAGS_RELWITHDEBINFO="${ARROW_C_FLAGS_RELWITHDEBINFO:-}" \
   -DARROW_DATASET=${ARROW_DATASET:-ON} \
   -DARROW_DEPENDENCY_SOURCE=${ARROW_DEPENDENCY_SOURCE:-AUTO} \
+  -DARROW_ENABLE_THREADING=${ARROW_ENABLE_THREADING:-ON} \
   -DARROW_ENABLE_TIMING_TESTS=${ARROW_ENABLE_TIMING_TESTS:-ON} \
   -DARROW_EXTRA_ERROR_CONTEXT=${ARROW_EXTRA_ERROR_CONTEXT:-OFF} \
   -DARROW_FILESYSTEM=${ARROW_FILESYSTEM:-ON} \

cpp/cmake_modules/DefineOptions.cmake

@@ -196,6 +196,8 @@ takes precedence over ccache if a storage backend is configured" ON)
 
   define_option(ARROW_WITH_MUSL "Whether the system libc is musl or not" OFF)
 
+  define_option(ARROW_ENABLE_THREADING "Enable threading in Arrow core" ON)
+
   #----------------------------------------------------------------------
   set_option_category("Test and benchmark")
 

cpp/src/arrow/acero/CMakeLists.txt

@@ -173,7 +173,14 @@ add_arrow_acero_test(hash_join_node_test SOURCES hash_join_node_test.cc
                      bloom_filter_test.cc)
 add_arrow_acero_test(pivot_longer_node_test SOURCES pivot_longer_node_test.cc
                      test_nodes.cc)
-add_arrow_acero_test(asof_join_node_test SOURCES asof_join_node_test.cc test_nodes.cc)
+
+# asof_join_node uses std::thread internally
+# and doesn't use ThreadPool so it will
+# be broken if threading is turned off
+if(ARROW_ENABLE_THREADING)
+  add_arrow_acero_test(asof_join_node_test SOURCES asof_join_node_test.cc test_nodes.cc)
+endif()
+
 add_arrow_acero_test(tpch_node_test SOURCES tpch_node_test.cc)
 add_arrow_acero_test(union_node_test SOURCES union_node_test.cc)
 add_arrow_acero_test(aggregate_node_test SOURCES aggregate_node_test.cc)
@@ -221,7 +228,9 @@ if(ARROW_BUILD_BENCHMARKS)
   add_arrow_acero_benchmark(project_benchmark SOURCES benchmark_util.cc
                             project_benchmark.cc)
 
-  add_arrow_acero_benchmark(asof_join_benchmark SOURCES asof_join_benchmark.cc)
+  if(ARROW_ENABLE_THREADING)
+    add_arrow_acero_benchmark(asof_join_benchmark SOURCES asof_join_benchmark.cc)
+  endif()
 
   add_arrow_acero_benchmark(tpch_benchmark SOURCES tpch_benchmark.cc)
 
@@ -244,7 +253,9 @@ if(ARROW_BUILD_BENCHMARKS)
     target_link_libraries(arrow-acero-expression-benchmark PUBLIC arrow_acero_static)
     target_link_libraries(arrow-acero-filter-benchmark PUBLIC arrow_acero_static)
     target_link_libraries(arrow-acero-project-benchmark PUBLIC arrow_acero_static)
-    target_link_libraries(arrow-acero-asof-join-benchmark PUBLIC arrow_acero_static)
+    if(ARROW_ENABLE_THREADING)
+      target_link_libraries(arrow-acero-asof-join-benchmark PUBLIC arrow_acero_static)
+    endif()
     target_link_libraries(arrow-acero-tpch-benchmark PUBLIC arrow_acero_static)
     if(ARROW_BUILD_OPENMP_BENCHMARKS)
       target_link_libraries(arrow-acero-hash-join-benchmark PUBLIC arrow_acero_static)
@@ -253,7 +264,9 @@ if(ARROW_BUILD_BENCHMARKS)
     target_link_libraries(arrow-acero-expression-benchmark PUBLIC arrow_acero_shared)
     target_link_libraries(arrow-acero-filter-benchmark PUBLIC arrow_acero_shared)
     target_link_libraries(arrow-acero-project-benchmark PUBLIC arrow_acero_shared)
-    target_link_libraries(arrow-acero-asof-join-benchmark PUBLIC arrow_acero_shared)
+    if(ARROW_ENABLE_THREADING)
+      target_link_libraries(arrow-acero-asof-join-benchmark PUBLIC arrow_acero_shared)
+    endif()
     target_link_libraries(arrow-acero-tpch-benchmark PUBLIC arrow_acero_shared)
     if(ARROW_BUILD_OPENMP_BENCHMARKS)
       target_link_libraries(arrow-acero-hash-join-benchmark PUBLIC arrow_acero_shared)

cpp/src/arrow/acero/asof_join_node.cc

@@ -49,6 +49,7 @@
 #include "arrow/type_traits.h"
 #include "arrow/util/bit_util.h"
 #include "arrow/util/checked_cast.h"
+#include "arrow/util/config.h"
 #include "arrow/util/future.h"
 #include "arrow/util/string.h"
 
@@ -1707,6 +1708,10 @@ class AsofJoinNode : public ExecNode {
   }
 
   Status StartProducing() override {
+#ifndef ARROW_ENABLE_THREADING
+    return Status::NotImplemented("ASOF join requires threading enabled");
+#endif
+
     ARROW_ASSIGN_OR_RAISE(process_task_, plan_->query_context()->BeginExternalTask(
                                              "AsofJoinNode::ProcessThread"));
     if (!process_task_.is_valid()) {

cpp/src/arrow/acero/bloom_filter.cc

@@ -20,6 +20,7 @@
 #include "arrow/acero/util.h"       // PREFETCH
 #include "arrow/util/bit_util.h"    // Log2
 #include "arrow/util/bitmap_ops.h"  // CountSetBits
+#include "arrow/util/config.h"
 
 namespace arrow {
 namespace acero {
@@ -426,6 +427,9 @@ void BloomFilterBuilder_Parallel::CleanUp() {
 
 std::unique_ptr<BloomFilterBuilder> BloomFilterBuilder::Make(
     BloomFilterBuildStrategy strategy) {
+#ifndef ARROW_ENABLE_THREADING
+  strategy = BloomFilterBuildStrategy::SINGLE_THREADED;
+#endif
   switch (strategy) {
     case BloomFilterBuildStrategy::SINGLE_THREADED: {
       std::unique_ptr<BloomFilterBuilder> impl{new BloomFilterBuilder_SingleThreaded()};

cpp/src/arrow/acero/bloom_filter_test.cc

@@ -29,6 +29,8 @@
 #include "arrow/acero/util.h"
 #include "arrow/compute/key_hash.h"
 #include "arrow/util/bitmap_ops.h"
+#include "arrow/util/config.h"
+#include "arrow/util/cpu_info.h"
 
 namespace arrow {
 
@@ -468,7 +470,7 @@ TEST(BloomFilter, Basic) {
 
   std::vector<BloomFilterBuildStrategy> strategies;
   strategies.push_back(BloomFilterBuildStrategy::SINGLE_THREADED);
-#ifndef ARROW_VALGRIND
+#if defined(ARROW_ENABLE_THREADING) && !defined(ARROW_VALGRIND)
   strategies.push_back(BloomFilterBuildStrategy::PARALLEL);
 #endif
 
@@ -501,7 +503,10 @@ TEST(BloomFilter, Scaling) {
   num_build.push_back(4000000);
 
   std::vector<BloomFilterBuildStrategy> strategies;
+#ifdef ARROW_ENABLE_THREADING
   strategies.push_back(BloomFilterBuildStrategy::PARALLEL);
+#endif
+  strategies.push_back(BloomFilterBuildStrategy::SINGLE_THREADED);
 
   for (const auto hardware_flags : HardwareFlagsForTesting()) {
     for (const auto& strategy : strategies) {

cpp/src/arrow/acero/plan_test.cc

@@ -36,6 +36,7 @@
 #include "arrow/testing/matchers.h"
 #include "arrow/testing/random.h"
 #include "arrow/util/async_generator.h"
+#include "arrow/util/config.h"
 #include "arrow/util/logging.h"
 #include "arrow/util/macros.h"
 #include "arrow/util/thread_pool.h"
@@ -1619,6 +1620,9 @@ TEST(ExecPlan, SourceEnforcesBatchLimit) {
 }
 
 TEST(ExecPlanExecution, SegmentedAggregationWithMultiThreading) {
+#ifndef ARROW_ENABLE_THREADING
+  GTEST_SKIP() << "Test requires threading enabled";
+#endif
   BatchesWithSchema data;
   data.batches = {ExecBatchFromJSON({int32()}, "[[1]]")};
   data.schema = schema({field("i32", int32())});

cpp/src/arrow/acero/task_util.cc

@@ -20,6 +20,7 @@
 #include <algorithm>
 #include <mutex>
 
+#include "arrow/util/config.h"
 #include "arrow/util/logging.h"
 
 namespace arrow {
@@ -316,7 +317,11 @@ Status TaskSchedulerImpl::StartScheduling(size_t thread_id, ScheduleImpl schedul
                                           int num_concurrent_tasks,
                                           bool use_sync_execution) {
   schedule_impl_ = std::move(schedule_impl);
+#ifdef ARROW_ENABLE_THREADING
   use_sync_execution_ = use_sync_execution;
+#else
+  use_sync_execution_ = true;
+#endif
   num_concurrent_tasks_ = num_concurrent_tasks;
   num_tasks_to_schedule_.value += num_concurrent_tasks;
   return ScheduleMore(thread_id);

cpp/src/arrow/acero/task_util.h

@@ -24,6 +24,7 @@
 
 #include "arrow/acero/visibility.h"
 #include "arrow/status.h"
+#include "arrow/util/config.h"
 #include "arrow/util/logging.h"
 
 namespace arrow {

cpp/src/arrow/acero/task_util_test.cc

@@ -27,6 +27,7 @@
 
 #include "arrow/acero/util.h"
 #include "arrow/testing/gtest_util.h"
+#include "arrow/util/config.h"
 #include "arrow/util/thread_pool.h"
 
 namespace arrow {
@@ -101,6 +102,9 @@ TaskScheduler::TaskGroupContinuationImpl MakeFinalContinuation(
 // concurrently.  When all groups in that stage finish the next
 // stage is started.
 TEST(TaskScheduler, Stress) {
+#ifndef ARROW_ENABLE_THREADING
+  GTEST_SKIP() << "Test requires threading support";
+#endif
   constexpr int kNumThreads = 8;
   constexpr int kNumGroups = 8;
   constexpr int kGroupsPerStage = 3;
@@ -176,6 +180,9 @@ TEST(TaskScheduler, Stress) {
 // thread starts a task group while another thread is finishing
 // the last of its tasks.
 TEST(TaskScheduler, StressTwo) {
+#ifndef ARROW_ENABLE_THREADING
+  GTEST_SKIP() << "Test requires threading support";
+#endif
   constexpr int kNumThreads = 16;
   constexpr int kNumGroups = 8;
   constexpr int kTasksPerGroup = 1;

cpp/src/arrow/dataset/dataset_writer_test.cc

@@ -31,6 +31,7 @@
 #include "arrow/table.h"
 #include "arrow/testing/future_util.h"
 #include "arrow/testing/gtest_util.h"
+#include "arrow/util/config.h"
 #include "gtest/gtest.h"
 
 using namespace std::string_view_literals;  // NOLINT
@@ -380,6 +381,9 @@ TEST_F(DatasetWriterTestFixture, MinRowGroupBackpressure) {
 }
 
 TEST_F(DatasetWriterTestFixture, ConcurrentWritesSameFile) {
+#ifndef ARROW_ENABLE_THREADING
+  GTEST_SKIP() << "Concurrent writes tests need threads";
+#endif
   // Use a gated filesystem to queue up many writes behind a file open to make sure the
   // file isn't opened multiple times.
   auto gated_fs = UseGatedFs();
@@ -394,6 +398,9 @@ TEST_F(DatasetWriterTestFixture, ConcurrentWritesSameFile) {
 }
 
 TEST_F(DatasetWriterTestFixture, ConcurrentWritesDifferentFiles) {
+#ifndef ARROW_ENABLE_THREADING
+  GTEST_SKIP() << "Concurrent writes tests need threads";
+#endif
   // NBATCHES must be less than I/O executor concurrency to avoid deadlock / test failure
   constexpr int NBATCHES = 6;
   auto gated_fs = UseGatedFs();
@@ -412,6 +419,9 @@ TEST_F(DatasetWriterTestFixture, ConcurrentWritesDifferentFiles) {
 }
 
 TEST_F(DatasetWriterTestFixture, MaxOpenFiles) {
+#ifndef ARROW_ENABLE_THREADING
+  GTEST_SKIP() << "Concurrent writes tests need threads";
+#endif
   auto gated_fs = UseGatedFs();
   std::atomic<bool> paused = false;
   write_options_.max_open_files = 2;

cpp/src/arrow/engine/substrait/serde_test.cc

@@ -71,6 +71,7 @@
 #include "arrow/type_fwd.h"
 #include "arrow/util/async_generator_fwd.h"
 #include "arrow/util/checked_cast.h"
+#include "arrow/util/config.h"
 #include "arrow/util/decimal.h"
 #include "arrow/util/future.h"
 #include "arrow/util/hash_util.h"
@@ -4458,6 +4459,9 @@ TEST(Substrait, SetRelationBasic) {
 }
 
 TEST(Substrait, PlanWithAsOfJoinExtension) {
+#ifndef ARROW_ENABLE_THREADING
+  GTEST_SKIP() << "ASOF join requires threading";
+#endif
   // This demos an extension relation
   std::string substrait_json = R"({
     "extensionUris": [],
@@ -5477,6 +5481,10 @@ TEST(Substrait, MixedSort) {
 }
 
 TEST(Substrait, PlanWithExtension) {
+#ifndef ARROW_ENABLE_THREADING
+  GTEST_SKIP() << "ASOF join requires threading";
+#endif
+
   // This demos an extension relation
   std::string substrait_json = R"({
     "extensionUris": [],
@@ -5665,6 +5673,9 @@ TEST(Substrait, PlanWithExtension) {
 }
 
 TEST(Substrait, AsOfJoinDefaultEmit) {
+#ifndef ARROW_ENABLE_THREADING
+  GTEST_SKIP() << "ASOF join requires threading";
+#endif
   std::string substrait_json = R"({
     "extensionUris": [],
     "extensions": [],

cpp/src/arrow/io/memory_test.cc

@@ -42,6 +42,7 @@
 #include "arrow/testing/util.h"
 #include "arrow/util/bit_util.h"
 #include "arrow/util/checked_cast.h"
+#include "arrow/util/config.h"
 #include "arrow/util/future.h"
 #include "arrow/util/iterator.h"
 #include "arrow/util/logging.h"
@@ -918,6 +919,9 @@ TEST(CacheOptions, Basics) {
 }
 
 TEST(IOThreadPool, Capacity) {
+#ifndef ARROW_ENABLE_THREADING
+  GTEST_SKIP() << "Test requires threading enabled";
+#endif
   // Simple sanity check
   auto pool = internal::GetIOThreadPool();
   int capacity = pool->GetCapacity();