PYCBC-1810: Release the GIL with an RAII guard

Changes
--------

* Add pycbc::gil_release_guard in src/gil_guard.hxx, a non-copyable,
  non-movable wrapper over PyEval_SaveThread/PyEval_RestoreThread that
  restores the GIL when a C++ exception unwinds out of the region
* Replace all 17 Py_BEGIN/END_ALLOW_THREADS regions inside try blocks
  in connection.cxx and connection.hxx with the guard.  The macro pair
  expands to a bare block open and close, so a throw between them
  skipped Py_END_ALLOW_THREADS entirely and the catch block, its
  PyErr_* calls and the return path all ran with the GIL released
* Covers connect, close, open_bucket, close_bucket, diagnostics and
  ping in connection.cxx, and execute_op, dispatch_sync,
  execute_multi_op, execute_streaming_op and execute_mgmt_op in
  connection.hxx
* Scope the guard to the dispatch and wait region only, so the barrier
  result is read back with the GIL held rather than relying on
  Py_END_ALLOW_THREADS landing before the return statement
* Restore the intended brace layout at the sites where the macros had
  forced clang-format into a hanging if (barrier) indent

Change-Id: I8483e939444a0cc15971c991131b687f9c6140af
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/249733
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Sergey Avseyev <sergey.avseyev@gmail.com>

Author: thejcfactor

src/connection.cxx

@@ -177,23 +177,23 @@ Connection::connect(PyObject* kwargs)
     update_cluster_options_from_py(connstr.options, pyObj_options, pyObj_auth);
     couchbase::core::origin origin(creds, connstr);
 
-    Py_BEGIN_ALLOW_THREADS
     {
+      gil_release_guard no_gil;
       cluster_.open(
         origin,
         [callback = pyObj_callback, errback = pyObj_errback, barrier, this](std::error_code ec) {
           handle_connection_operation_callback(
             ec, "connect", callback, errback, barrier, connection_state_action::set_connected);
         });
     }
-    Py_END_ALLOW_THREADS
 
-      if (barrier)
-    {
+    if (barrier) {
       PyObject* result = nullptr;
-      Py_BEGIN_ALLOW_THREADS result = fut.get();
-      Py_END_ALLOW_THREADS if (result == nullptr)
       {
+        gil_release_guard no_gil;
+        result = fut.get();
+      }
+      if (result == nullptr) {
         // The IO thread's result construction failed; that failure was already
         // reported via PyErr_WriteUnraisable there, so nothing is pending on this
         // thread.
@@ -231,8 +231,8 @@ Connection::close(PyObject* kwargs)
   }
 
   try {
-    Py_BEGIN_ALLOW_THREADS
     {
+      gil_release_guard no_gil;
       cluster_.close([callback = pyObj_callback, errback = pyObj_errback, barrier, this]() {
         handle_connection_operation_callback(std::error_code{},
                                              "close",
@@ -242,14 +242,14 @@ Connection::close(PyObject* kwargs)
                                              connection_state_action::set_disconnected);
       });
     }
-    Py_END_ALLOW_THREADS
 
-      if (barrier)
-    {
+    if (barrier) {
       PyObject* result = nullptr;
-      Py_BEGIN_ALLOW_THREADS result = fut.get();
-      Py_END_ALLOW_THREADS if (result == nullptr)
       {
+        gil_release_guard no_gil;
+        result = fut.get();
+      }
+      if (result == nullptr) {
         // The IO thread's result construction failed; that failure was already
         // reported via PyErr_WriteUnraisable there, so nothing is pending on this
         // thread.
@@ -293,22 +293,22 @@ Connection::open_bucket(PyObject* kwargs)
       throw std::invalid_argument("bucket_name must be a valid UTF-8 string");
     }
 
-    Py_BEGIN_ALLOW_THREADS
     {
+      gil_release_guard no_gil;
       cluster_.open_bucket(
         bucket_name,
         [callback = pyObj_callback, errback = pyObj_errback, barrier, this](std::error_code ec) {
           handle_connection_operation_callback(ec, "open_bucket", callback, errback, barrier);
         });
     }
-    Py_END_ALLOW_THREADS
 
-      if (barrier)
-    {
+    if (barrier) {
       PyObject* result = nullptr;
-      Py_BEGIN_ALLOW_THREADS result = fut.get();
-      Py_END_ALLOW_THREADS if (result == nullptr)
       {
+        gil_release_guard no_gil;
+        result = fut.get();
+      }
+      if (result == nullptr) {
         // The IO thread's result construction failed; that failure was already
         // reported via PyErr_WriteUnraisable there, so nothing is pending on this
         // thread.
@@ -352,22 +352,22 @@ Connection::close_bucket(PyObject* kwargs)
       throw std::invalid_argument("bucket_name must be a valid UTF-8 string");
     }
 
-    Py_BEGIN_ALLOW_THREADS
     {
+      gil_release_guard no_gil;
       cluster_.close_bucket(
         bucket_name,
         [callback = pyObj_callback, errback = pyObj_errback, barrier, this](std::error_code ec) {
           handle_connection_operation_callback(ec, "close_bucket", callback, errback, barrier);
         });
     }
-    Py_END_ALLOW_THREADS
 
-      if (barrier)
-    {
+    if (barrier) {
       PyObject* result = nullptr;
-      Py_BEGIN_ALLOW_THREADS result = fut.get();
-      Py_END_ALLOW_THREADS if (result == nullptr)
       {
+        gil_release_guard no_gil;
+        result = fut.get();
+      }
+      if (result == nullptr) {
         // The IO thread's result construction failed; that failure was already
         // reported via PyErr_WriteUnraisable there, so nothing is pending on this
         // thread.
@@ -445,23 +445,23 @@ Connection::diagnostics(PyObject* kwargs)
   }
 
   try {
-    Py_BEGIN_ALLOW_THREADS
     {
+      gil_release_guard no_gil;
       cluster_.diagnostics(report_id,
                            [pyObj_callback, pyObj_errback, barrier, this](
                              couchbase::core::diag::diagnostics_result resp) {
                              handle_cluster_operation_callback(
                                resp, pyObj_callback, pyObj_errback, barrier);
                            });
     }
-    Py_END_ALLOW_THREADS
 
-      if (barrier)
-    {
+    if (barrier) {
       PyObject* result = nullptr;
-      Py_BEGIN_ALLOW_THREADS result = fut.get();
-      Py_END_ALLOW_THREADS if (result == nullptr)
       {
+        gil_release_guard no_gil;
+        result = fut.get();
+      }
+      if (result == nullptr) {
         // The IO thread's result construction failed; that failure was already
         // reported via PyErr_WriteUnraisable there, so nothing is pending on this
         // thread.
@@ -508,8 +508,8 @@ Connection::ping(PyObject* kwargs)
   }
 
   try {
-    Py_BEGIN_ALLOW_THREADS
     {
+      gil_release_guard no_gil;
       cluster_.ping(
         report_id,
         bucket_name,
@@ -519,14 +519,14 @@ Connection::ping(PyObject* kwargs)
           handle_cluster_operation_callback(resp, pyObj_callback, pyObj_errback, barrier);
         });
     }
-    Py_END_ALLOW_THREADS
 
-      if (barrier)
-    {
+    if (barrier) {
       PyObject* result = nullptr;
-      Py_BEGIN_ALLOW_THREADS result = fut.get();
-      Py_END_ALLOW_THREADS if (result == nullptr)
       {
+        gil_release_guard no_gil;
+        result = fut.get();
+      }
+      if (result == nullptr) {
         // The IO thread's result construction failed; that failure was already
         // reported via PyErr_WriteUnraisable there, so nothing is pending on this
         // thread.

src/connection.hxx

@@ -20,6 +20,7 @@
 #include "Python.h"
 #include "error_contexts.hxx"
 #include "exceptions.hxx"
+#include "gil_guard.hxx"
 #include "operations_autogen.hxx"
 #include "pycbc_kv_request.hxx"
 #include "pytocbpp_defs.hxx"
@@ -234,8 +235,8 @@ private:
     std::optional<std::chrono::system_clock::time_point> start_time = std::nullopt)
   {
     using response_type = typename Request::response_type;
-    Py_BEGIN_ALLOW_THREADS
     {
+      gil_release_guard no_gil;
       cluster_.execute(
         req,
         [pyObj_callback, pyObj_errback, barrier, wrapper_span, start_time, this](
@@ -276,7 +277,6 @@ private:
           PyGILState_Release(state);
         });
     }
-    Py_END_ALLOW_THREADS
   }
 
   template<typename PyType>
@@ -366,11 +366,14 @@ Connection::dispatch_sync(Request&& req)
   using Response = typename Request::response_type;
   auto barrier = std::make_shared<std::promise<Response>>();
   auto fut = barrier->get_future();
-  Py_BEGIN_ALLOW_THREADS cluster_.execute(std::forward<Request>(req), [barrier](Response resp) {
-    barrier->set_value(std::move(resp));
-  });
-  fut.wait();
-  Py_END_ALLOW_THREADS return fut.get();
+  {
+    gil_release_guard no_gil;
+    cluster_.execute(std::forward<Request>(req), [barrier](Response resp) {
+      barrier->set_value(std::move(resp));
+    });
+    fut.wait();
+  }
+  return fut.get();
 }
 
 template<typename Request, typename Response>
@@ -523,20 +526,21 @@ Connection::execute_multi_op(PyObject* arg)
                         std::move(fut) });
   }
 
-  Py_BEGIN_ALLOW_THREADS for (auto& s : staging)
   {
-    auto barrier = s.barrier;
-    cluster_.execute(s.req, [barrier](Response resp) {
-      barrier->set_value(std::move(resp));
-    });
-  }
+    gil_release_guard no_gil;
+    for (auto& s : staging) {
+      auto barrier = s.barrier;
+      cluster_.execute(s.req, [barrier](Response resp) {
+        barrier->set_value(std::move(resp));
+      });
+    }
 
-  for (auto& s : staging) {
-    s.fut.wait();
+    for (auto& s : staging) {
+      s.fut.wait();
+    }
   }
-  Py_END_ALLOW_THREADS
 
-    bool all_okay = true;
+  bool all_okay = true;
   for (auto& s : staging) {
     PyObject* res =
       finalize_kv_result<Request>(s.fut.get(), std::move(s.wrapper_span), std::move(s.start_time));
@@ -622,8 +626,8 @@ Connection::execute_streaming_op(PyObject* kwargs)
     // };
 
     using response_type = typename Request::response_type;
-    Py_BEGIN_ALLOW_THREADS
     {
+      gil_release_guard no_gil;
       cluster_.execute(
         req,
         // we pass the shared_ptr rows separately b/c we need t o allow the rows queue to survive
@@ -708,9 +712,8 @@ Connection::execute_streaming_op(PyObject* kwargs)
           PyGILState_Release(state);
         });
     }
-    Py_END_ALLOW_THREADS
 
-      return reinterpret_cast<PyObject*>(streamed_res);
+    return reinterpret_cast<PyObject*>(streamed_res);
 
   } catch (const std::exception& e) {
     Py_XDECREF(pyObj_callback);
@@ -764,9 +767,11 @@ Connection::execute_mgmt_op(PyObject* kwargs)
     execute_op(req, pyObj_callback, pyObj_errback, barrier, wrapper_span);
     if (barrier) {
       PyObject* result = nullptr;
-      Py_BEGIN_ALLOW_THREADS result = fut.get();
-      Py_END_ALLOW_THREADS if (result == nullptr)
       {
+        gil_release_guard no_gil;
+        result = fut.get();
+      }
+      if (result == nullptr) {
         // The IO thread's conversion failed; that failure was already reported via
         // PyErr_WriteUnraisable there, so nothing is pending on this thread.
         set_runtime_error_if_unset("Failed to process operation result.");

src/gil_guard.hxx

@@ -0,0 +1,49 @@
+/*
+ *   Copyright 2016-2026. Couchbase, Inc.
+ *   All Rights Reserved.
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+#pragma once
+
+#include "Python.h"
+
+namespace pycbc
+{
+
+// Releases the GIL for the guard's lifetime. Unlike Py_BEGIN/END_ALLOW_THREADS, the GIL is
+// restored when a C++ exception unwinds out of the region, so catch blocks run with it held.
+class gil_release_guard
+{
+public:
+  gil_release_guard()
+    : state_(PyEval_SaveThread())
+  {
+  }
+
+  ~gil_release_guard()
+  {
+    PyEval_RestoreThread(state_);
+  }
+
+  gil_release_guard(const gil_release_guard&) = delete;
+  gil_release_guard& operator=(const gil_release_guard&) = delete;
+  gil_release_guard(gil_release_guard&&) = delete;
+  gil_release_guard& operator=(gil_release_guard&&) = delete;
+
+private:
+  PyThreadState* state_;
+};
+
+} // namespace pycbc