Release lock when pushing result

Author: Rafał Hibner

cpp/src/arrow/acero/asof_join_node.cc

@@ -1063,28 +1063,28 @@ class AsofJoinNode : public ExecNode {
     // Process batches while we have data
     for (;;) {
       backpressure_future_.Wait();
+      Result<std::shared_ptr<RecordBatch>> result;
       {
         std::lock_guard<std::mutex> guard(gate_);
         if (!CheckEnded()) {
           return false;
         }
-        Result<std::shared_ptr<RecordBatch>> result = ProcessInner();
-
-        if (result.ok()) {
-          auto out_rb = *result;
-          if (!out_rb) break;
-          ExecBatch out_b(*out_rb);
-          out_b.index = batches_produced_++;
-          DEBUG_SYNC(this, "produce batch ", out_b.index, ":", DEBUG_MANIP(std::endl),
-                     out_rb->ToString(), DEBUG_MANIP(std::endl));
-          Status st = output_->InputReceived(this, std::move(out_b));
-          if (!st.ok()) {
-            EndFromProcessThread(std::move(st));
-          }
-        } else {
-          EndFromProcessThread(result.status());
-          return false;
+        result = ProcessInner();
+      }
+      if (result.ok()) {
+        auto out_rb = *result;
+        if (!out_rb) break;
+        ExecBatch out_b(*out_rb);
+        out_b.index = batches_produced_++;
+        DEBUG_SYNC(this, "produce batch ", out_b.index, ":", DEBUG_MANIP(std::endl),
+                   out_rb->ToString(), DEBUG_MANIP(std::endl));
+        Status st = output_->InputReceived(this, std::move(out_b));
+        if (!st.ok()) {
+          EndFromProcessThread(std::move(st));
         }
+      } else {
+        EndFromProcessThread(result.status());
+        return false;
       }
     }