libgolang: Adjust and require runtimes to provide semaphores with timeout

Previously libgolang was specifying its runtime, among other primitives,
to provide semaphore implementation with acquire and release methods.
The release should be non-blocking operation, and the acquire should be
blocking until the semaphore is acquired.

However for efficient implementation of timers, we will need to have
semaphore acquire that can also be instructed to time out.

-> Adjust thread and gevent runtimes to provide that and adjust runtime
interface specification to require that.

This is generally backward incompatible change, but given that there is
just a few libgolang runtimes, it, hopefully, should not do any
real breakage. So I think it is ok to do it this way.

For the reference - contrary to runtimes - the public user API of
libgolang and pygolang - that most of the pygolang users actually use -
is not changed at all. In other words there is no backward-compatibility
issue for regular pygolang/libgolang users because for them pygolang
stays 100% backward compatible.

/proposed-for-review-on nexedi/pygolang!26

golang/libgolang.h

 #ifndef _NXD_LIBGOLANG_H
 #define _NXD_LIBGOLANG_H
 
-// Copyright (C) 2018-2023  Nexedi SA and Contributors.
+// Copyright (C) 2018-2024  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -346,8 +346,13 @@ typedef struct _libgolang_runtime_ops {
     // previously successfully allocated via sema_alloc.
     void             (*sema_free)   (_libgolang_sema*);
 
-    // sema_acquire/sema_release should acquire/release live semaphore allocated via sema_alloc.
-    void             (*sema_acquire)(_libgolang_sema*);
+    // sema_acquire should try to acquire live semaphore allocated via sema_alloc during given time.
+    // it returns whether acquisition succeeded or timed out.
+    // the timeout is specified in nanoseconds.
+    // UINT64_MAX means no timeout.
+    bool             (*sema_acquire)(_libgolang_sema*, uint64_t timeout_ns);
+
+    // sema_release should release live semaphore allocated via sema_alloc.
     void             (*sema_release)(_libgolang_sema*);
 
     // nanosleep should pause current goroutine for at least dt nanoseconds.

golang/runtime/_libgolang.pxd

 # cython: language_level=2
-# Copyright (C) 2019-2022  Nexedi SA and Contributors.
+# Copyright (C) 2019-2024  Nexedi SA and Contributors.
 #                          Kirill Smelkov <kirr@nexedi.com>
 #
 # This program is free software: you can Use, Study, Modify and Redistribute
@@ -36,7 +36,7 @@ cdef extern from "golang/libgolang.h" namespace "golang" nogil:
 
         _libgolang_sema* (*sema_alloc)  ()
         void             (*sema_free)   (_libgolang_sema*)
-        void             (*sema_acquire)(_libgolang_sema*)
+        bint             (*sema_acquire)(_libgolang_sema*, uint64_t timeout_ns)
         void             (*sema_release)(_libgolang_sema*)
 
         void        (*nanosleep)(uint64_t)

golang/runtime/_runtime_gevent.pyx

 # cython: language_level=2
-# Copyright (C) 2019-2023  Nexedi SA and Contributors.
+# Copyright (C) 2019-2024  Nexedi SA and Contributors.
 #                          Kirill Smelkov <kirr@nexedi.com>
 #
 # This program is free software: you can Use, Study, Modify and Redistribute
@@ -40,7 +40,10 @@ ELSE:
 
 from gevent import sleep as pygsleep
 
-from libc.stdint cimport uint8_t, uint64_t
+from libc.stdint cimport uint8_t, uint64_t, UINT64_MAX
+cdef extern from *:
+    ctypedef bint cbool "bool"
+
 from cpython cimport PyObject, Py_INCREF, Py_DECREF
 from cython cimport final
 
@@ -95,9 +98,12 @@ cdef:
         Py_DECREF(pygsema)
         return True
 
-    bint _sema_acquire(_libgolang_sema *gsema):
+    bint _sema_acquire(_libgolang_sema *gsema, uint64_t timeout_ns, cbool* pacq):
         pygsema = <PYGSema>gsema
-        pygsema.acquire()
+        timeout = None
+        if timeout_ns != UINT64_MAX:
+            timeout = float(timeout_ns) * 1e-9
+        pacq[0] = pygsema.acquire(timeout=timeout)
         return True
 
     bint _sema_release(_libgolang_sema *gsema):
@@ -142,14 +148,16 @@ cdef nogil:
         if not ok:
             panic("pyxgo: gevent: sema: free: failed")
 
-    void sema_acquire(_libgolang_sema *gsema):
+    cbool sema_acquire(_libgolang_sema *gsema, uint64_t timeout_ns):
         cdef PyExc exc
+        cdef cbool acq
         with gil:
             pyexc_fetch(&exc)
-            ok = _sema_acquire(gsema)
+            ok = _sema_acquire(gsema, timeout_ns, &acq)
             pyexc_restore(exc)
         if not ok:
             panic("pyxgo: gevent: sema: acquire: failed")
+        return acq
 
     void sema_release(_libgolang_sema *gsema):
         cdef PyExc exc

golang/runtime/_runtime_thread.pyx

 # cython: language_level=2
-# Copyright (C) 2019-2022  Nexedi SA and Contributors.
+# Copyright (C) 2019-2024  Nexedi SA and Contributors.
 #                          Kirill Smelkov <kirr@nexedi.com>
 #
 # This program is free software: you can Use, Study, Modify and Redistribute
@@ -35,7 +35,12 @@ from __future__ import print_function, absolute_import
 #
 # NOTE Cython declares PyThread_acquire_lock/PyThread_release_lock as nogil
 from cpython.pythread cimport PyThread_acquire_lock, PyThread_release_lock, \
-        PyThread_type_lock, WAIT_LOCK
+        PyThread_type_lock, WAIT_LOCK, NOWAIT_LOCK, PyLockStatus, PY_LOCK_ACQUIRED, PY_LOCK_FAILURE
+
+cdef extern from * nogil:
+    ctypedef int PY_TIMEOUT_T   # long long there
+    PyLockStatus PyThread_acquire_lock_timed(PyThread_type_lock, PY_TIMEOUT_T timeout_us, int intr_flag)
+
 
 # NOTE On Darwin, even though this is considered as POSIX, Python uses
 # mutex+condition variable to implement its lock, and, as of 20190828, Py2.7
@@ -98,6 +103,9 @@ from libc.errno  cimport errno, EINTR, EBADF
 from posix.fcntl cimport mode_t
 from posix.stat cimport struct_stat
 from posix.strings cimport bzero
+cdef extern from *:
+    ctypedef bint cbool "bool"
+
 IF POSIX:
     from posix.time cimport clock_gettime, nanosleep as posix_nanosleep, timespec, CLOCK_REALTIME
 ELSE:
@@ -138,11 +146,46 @@ cdef nogil:
         pysema = <PyThread_type_lock>gsema
         PyThread_free_lock(pysema)
 
-    void sema_acquire(_libgolang_sema *gsema):
+    cbool sema_acquire(_libgolang_sema *gsema, uint64_t timeout_ns):
         pysema = <PyThread_type_lock>gsema
+        IF PY3:
+            cdef PY_TIMEOUT_T timeout_us
+        ELSE:
+            cdef uint64_t tprev, t, tsleep
+        if timeout_ns == UINT64_MAX:
             ok = PyThread_acquire_lock(pysema, WAIT_LOCK)
             if ok == 0:
                 panic("pyxgo: thread: sema_acquire: PyThread_acquire_lock failed")
+            return 1
+        else:
+            IF PY3:
+                timeout_us = timeout_ns // 1000
+                lkok = PyThread_acquire_lock_timed(pysema, timeout_us, 0)
+                if lkok == PY_LOCK_FAILURE:
+                    return 0
+                elif lkok == PY_LOCK_ACQUIRED:
+                    return 1
+                else:
+                    panic("pyxgo: thread: sema_acquire: PyThread_acquire_lock_timed failed")
+            ELSE:
+                # py2 misses PyThread_acquire_lock_timed - provide fallback ourselves
+                tprev = nanotime()
+                while 1:
+                     ok = PyThread_acquire_lock(pysema, NOWAIT_LOCK)
+                     if ok:
+                         return 1
+                     tsleep = min(timeout_ns, 50*1000)    # poll every 50 μs = 20 Hz
+                     if tsleep == 0:
+                         break
+                     nanosleep(tsleep)
+                     t = nanotime()
+                     if t < tprev:
+                         break  # clock skew
+                     if t - tprev >= timeout_ns:
+                         break
+                     timeout_ns -= t - tprev
+                     tprev = t
+                return 0
 
     void sema_release(_libgolang_sema *gsema):
         pysema = <PyThread_type_lock>gsema

golang/runtime/libgolang.cpp

-// Copyright (C) 2018-2023  Nexedi SA and Contributors.
+// Copyright (C) 2018-2024  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -166,7 +166,15 @@ void _semafree(_sema *sema) {
 }
 
 void _semaacquire(_sema *sema) {
-    _runtime->sema_acquire((_libgolang_sema *)sema);
+    bool ok;
+    ok = _runtime->sema_acquire((_libgolang_sema *)sema, UINT64_MAX);
+    if (!ok)
+        panic("semaacquire: failed");
+}
+
+// NOTE not currently exposed in public API
+bool _semaacquire_timed(_sema *sema, uint64_t timeout_ns) {
+    return _runtime->sema_acquire((_libgolang_sema *)sema, timeout_ns);
 }
 
 void _semarelease(_sema *sema) {