Use arrow mutex implementation

Author: Rafał Hibner

cpp/src/arrow/acero/pipe_node.cc

@@ -275,7 +275,7 @@ Pipe::Pipe(ExecPlan* plan, std::string pipe_name,
 const Ordering& Pipe::ordering() const { return ordering_; }
 
 void Pipe::Pause(PipeSource* output, int counter) {
-  std::lock_guard<std::mutex> lg(mutex_);
+  auto lock = mutex_.Lock();
   auto& state = state_[output];
   if (state.backpressure_counter < counter) {
     if (!state.paused && !state.stopped) {
@@ -295,7 +295,7 @@ void Pipe::Pause(PipeSource* output, int counter) {
 }
 
 void Pipe::Resume(PipeSource* output, int counter) {
-  std::lock_guard<std::mutex> lg(mutex_);
+  auto lock = mutex_.Lock();
   auto& state = state_[output];
   if (state.backpressure_counter < counter) {
     state.backpressure_counter = counter;
@@ -320,7 +320,7 @@ void Pipe::DoResume(SourceState& state) {
 }
 
 Status Pipe::StopProducing(PipeSource* output) {
-  std::lock_guard<std::mutex> lg(mutex_);
+  auto lock = mutex_.Lock();
   auto& state = state_[output];
   DCHECK(!state.stopped);
   DoResume(state);
@@ -341,7 +341,7 @@ Status Pipe::StopProducing(PipeSource* output) {
 Status Pipe::InputReceived(ExecBatch batch) {
   for (auto& source_node : async_nodes_) {
     {
-      std::lock_guard<std::mutex> lg(mutex_);
+      auto lock = mutex_.Lock();
       if (state_[source_node].stopped) continue;
     }
     plan_->query_context()->ScheduleTask(

cpp/src/arrow/acero/pipe_node.h

@@ -16,11 +16,11 @@
 // under the License.
 
 #pragma once
-#include <mutex>
 #include <string>
 #include "arrow/acero/exec_plan.h"
 #include "arrow/acero/options.h"
 #include "arrow/acero/visibility.h"
+#include "arrow/util/mutex.h"
 
 namespace arrow {
 
@@ -114,10 +114,10 @@ class ARROW_ACERO_EXPORT Pipe {
   Ordering ordering_;
   std::string pipe_name_;
   std::vector<PipeSource*> async_nodes_;
-  PipeSource* sync_node_{nullptr};
+  PipeSource* sync_node_{NULLPTR};
   // backpressure
   std::unordered_map<PipeSource*, SourceState> state_;
-  std::mutex mutex_;
+  util::Mutex mutex_;
   std::atomic_size_t paused_count_{0};
   std::unique_ptr<BackpressureControl> ctrl_;
   // stopProducing