Add functions to interrupt processing in a specific thread

- Add Interrupt::requestForCurrentThread (C API: GEOS_interruptThread) to
  request interruption of the current thread only.

- Add Interrupt::registerThreadCallback (C API:
  GEOS_interruptRegisterThreadCallback) to register an interruption callback
  for the current thread only.

capi/geos_c.cpp

@@ -116,12 +116,24 @@ extern "C" {
  return geos::util::Interrupt::registerCallback(cb);
  }
 
+ GEOSInterruptThreadCallback*
+ GEOS_interruptRegisterThreadCallback(GEOSInterruptThreadCallback* cb, void* data)
+ {
+ return geos::util::Interrupt::registerThreadCallback(cb, data);
+ }
+
  void
  GEOS_interruptRequest()
  {
  geos::util::Interrupt::request();
  }
 
+ void
+ GEOS_interruptThread()
+ {
+ geos::util::Interrupt::requestForCurrentThread();
+ }
+
  void
  GEOS_interruptCancel()
  {

capi/geos_c.h.in

@@ -305,8 +305,10 @@ typedef int (*GEOSTransformXYCallback)(
 */
 typedef void (GEOSInterruptCallback)(void);
 
+typedef void (GEOSInterruptThreadCallback)(void*);
+
 /**
-* Register a function to be called when processing is interrupted.
+* Register a function to be called when processing is interrupted on any thread.
 * \param cb Callback function to invoke
 * \return the previously configured callback
 * \see GEOSInterruptCallback
@@ -315,10 +317,29 @@ extern GEOSInterruptCallback GEOS_DLL *GEOS_interruptRegisterCallback(
  GEOSInterruptCallback* cb);
 
 /**
-* Request safe interruption of operations
+* Register a function to be called when processing is interrupted on the current thread.
+*
+* \param cb Callback function to invoke
+* \param context pointer to a context object that will be passed to `cb`
+* \return the previously configured callback
+* \see GEOSInterruptCallback
+*/
+extern GEOSInterruptThreadCallback GEOS_DLL *GEOS_interruptRegisterThreadCallback(
+ GEOSInterruptThreadCallback* cb, void* context);
+
+/**
+* Request safe interruption of operations. The next thread to check for an
+* interrupt will be interrupted. To request interruption of a specific thread,
+* instead call `GEOS_interruptThread` from a callback executed by that thread.
 */
 extern void GEOS_DLL GEOS_interruptRequest(void);
 
+/**
+* Request safe interruption of operations in the current thread. This function
+* should be called from a callback registered by `GEOS_interruptRegisterThreadCallback`.
+*/
+extern void GEOS_DLL GEOS_interruptThread(void);
+
 /**
 * Cancel a pending interruption request
 */

include/geos/util/Interrupt.h

@@ -27,33 +27,55 @@ class GEOS_DLL Interrupt {
 public:
 
  typedef void (Callback)(void);
+ typedef void (ThreadCallback)(void*);
 
  /**
  * Request interruption of operations
  *
  * Operations will be terminated by a GEOSInterrupt
- * exception at first occasion.
+ * exception at first occasion, by the first thread
+ * to check for an interrupt request.
  */
  static void request();
 
+ /**
+ * Request interruption of operations in the current thread
+ *
+ * Operations in the current thread will be terminated by
+ * a GEOSInterrupt at first occasion.
+ */
+ static void requestForCurrentThread();
+
  /** Cancel a pending interruption request */
  static void cancel();
 
  /** Check if an interruption request is pending */
  static bool check();
 
  /** \brief
- * Register a callback that will be invoked
+ * Register a callback that will be invoked by all threads
  * before checking for interruption requests.
  *
  * NOTE that interruption request checking may happen
- * frequently so any callback would better be quick.
+ * frequently so the callback should execute quickly.
  *
  * The callback can be used to call Interrupt::request()
- *
+ * or Interrupt::requestForCurrentThread().
  */
  static Callback* registerCallback(Callback* cb);
 
+ /** \brief
+ * Register a callback that will be invoked the current thread
+ * before checking for interruption requests.
+ *
+ * NOTE that interruption request checking may happen
+ * frequently so the callback shoudl execute quickly.
+ *
+ * The callback can be used to call Interrupt::request()
+ * or Interrupt::requestForCurrentThread().
+ */
+ static ThreadCallback* registerThreadCallback(ThreadCallback* cb, void* data);
+
  /**
  * Invoke the callback, if any. Process pending interruption, if any.
  *

src/util/Interrupt.cpp

@@ -18,8 +18,12 @@
 namespace {
 /* Could these be portably stored in thread-specific space ? */
 bool requested = false;
+thread_local bool requested_for_thread = false;
 
 geos::util::Interrupt::Callback* callback = nullptr;
+thread_local geos::util::Interrupt::ThreadCallback* callback_thread = nullptr;
+thread_local void* callback_thread_data = nullptr;
+
 }
 
 namespace geos {
@@ -37,16 +41,23 @@ Interrupt::request()
  requested = true;
 }
 
+void
+Interrupt::requestForCurrentThread()
+{
+ requested_for_thread = true;
+}
+
 void
 Interrupt::cancel()
 {
  requested = false;
+ requested_for_thread = false;
 }
 
 bool
 Interrupt::check()
 {
- return requested;
+ return requested || requested_for_thread;
 }
 
 Interrupt::Callback*
@@ -57,14 +68,25 @@ Interrupt::registerCallback(Interrupt::Callback* cb)
  return prev;
 }
 
+Interrupt::ThreadCallback*
+Interrupt::registerThreadCallback(ThreadCallback* cb, void* data)
+{
+ ThreadCallback* prev = callback_thread;
+ callback_thread = cb;
+ callback_thread_data = data;
+ return prev;
+}
+
 void
 Interrupt::process()
 {
  if(callback) {
  (*callback)();
  }
- if(requested) {
- requested = false;
+ if(callback_thread) {
+ (*callback_thread)(callback_thread_data);
+ }
+ if(check()) {
  interrupt();
  }
 }
@@ -74,6 +96,7 @@ void
 Interrupt::interrupt()
 {
  requested = false;
+ requested_for_thread = false;
  throw InterruptedException();
 }
 

tests/unit/capi/GEOSInterruptTest.cpp

@@ -9,6 +9,7 @@
 #include <cstdio>
 #include <cstdlib>
 #include <memory>
+#include <thread>
 
 namespace tut {
 //
@@ -18,6 +19,7 @@ namespace tut {
 // Common data used in test cases.
 struct test_capiinterrupt_data {
  static int numcalls;
+ static int maxcalls;
  static GEOSInterruptCallback* nextcb;
 
  static void
@@ -56,9 +58,18 @@ struct test_capiinterrupt_data {
  }
  }
 
+ static void
+ interruptAfterMaxCalls(void* data)
+ {
+ if (++*static_cast<int*>(data) >= maxcalls) {
+ GEOS_interruptThread();
+ }
+ }
+
 };
 
 int test_capiinterrupt_data::numcalls = 0;
+int test_capiinterrupt_data::maxcalls = 0;
 GEOSInterruptCallback* test_capiinterrupt_data::nextcb = nullptr;
 
 typedef test_group<test_capiinterrupt_data> group;
@@ -221,5 +232,40 @@ void object::test<5>
 }
 
 
+// Test callback is thread-local
+template<>
+template<>
+void object::test<6>
+()
+{
+ maxcalls = 3;
+ int calls_1 = 0;
+ int calls_2 = 0;
+
+ initGEOS(notice, notice);
+
+ auto buffer = [](GEOSInterruptThreadCallback* cb, void* data) {
+ GEOSGeometry* geom1 = GEOSGeomFromWKT("LINESTRING (0 0, 1 0)");
+
+ GEOS_interruptRegisterThreadCallback(cb, data);
+
+ GEOSGeometry* geom2 = GEOSBuffer(geom1, 1, 8);
+ GEOSGeom_destroy(geom2);
+ GEOSGeom_destroy(geom1);
+ };
+
+ std::thread t1(buffer, interruptAfterMaxCalls, &calls_1);
+ std::thread t2(buffer, interruptAfterMaxCalls, &calls_2);
+
+ t1.join();
+ t2.join();
+
+ ensure_equals(calls_1, maxcalls);
+ ensure_equals(calls_2, maxcalls);
+
+ finishGEOS();
+}
+
+
 } // namespace tut