internal/atomic: New package

This package provides special kind of atomic that is automatically reset
to zero after fork in child. This kind of atomic will be used in os
package to implement IO that does not deadlock in Close after fork.

/reviewed-on nexedi/pygolang!17

MANIFEST.in

@@ -6,6 +6,8 @@ include golang/pyx/runtime.h
 include golang/pyx/runtime.cpp
 include golang/pyx/testprog/golang_dso_user/dsouser/dso.cpp
 include golang/runtime/internal.h
+include golang/runtime/internal/atomic.h
+include golang/runtime/internal/atomic.cpp
 include golang/runtime/internal/syscall.h
 include golang/runtime/internal/syscall.cpp
 include golang/context.h

golang/pyx/build.py

@@ -174,6 +174,7 @@ def _with_build_defaults(kw):   # -> (pygo, kw')
     dependv.extend(['%s/golang/%s' % (pygo, _) for _ in [
         'libgolang.h',
         'runtime/internal.h',
+        'runtime/internal/atomic.h',
         'runtime/internal/syscall.h',
         'context.h',
         'cxx.h',

golang/runtime/internal/atomic.cpp

+// Copyright (C) 2022  Nexedi SA and Contributors.
+//                     Kirill Smelkov <kirr@nexedi.com>
+//
+// This program is free software: you can Use, Study, Modify and Redistribute
+// it under the terms of the GNU General Public License version 3, or (at your
+// option) any later version, as published by the Free Software Foundation.
+//
+// You can also Link and Combine this program with other software covered by
+// the terms of any of the Free Software licenses or any of the Open Source
+// Initiative approved licenses and Convey the resulting work. Corresponding
+// source of such a combination shall include the source code for all other
+// software used.
+//
+// This program is distributed WITHOUT ANY WARRANTY; without even the implied
+// warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+//
+// See COPYING file for full licensing terms.
+// See https://www.nexedi.com/licensing for rationale and options.
+
+#include "golang/runtime/internal/atomic.h"
+#include "golang/libgolang.h"
+
+#include <pthread.h>
+
+// golang::internal::atomic::
+namespace golang {
+namespace internal {
+namespace atomic {
+
+// _forkEpoch is incremented in child processes on every fork
+//
+// NOTE being plain int32_t is ok, but ThreadSanitizer thinks that after fork
+// _forkEpoch++ is a race, tries to report it and deadlocks in its own runtime:
+// https://github.com/google/sanitizers/issues/1116
+//
+// -> workaround it via pretentding that _forkEpoch is atomic.
+static std::atomic<int32_t> _forkEpoch (0);
+
+static void _forkNewEpoch() {
+    _forkEpoch++;
+}
+
+void _init() {
+    int e = pthread_atfork(/*prepare*/nil, /*inparent*/nil, /*inchild*/_forkNewEpoch);
+    if (e != 0)
+        panic("pthread_atfork failed");
+}
+
+
+int32ForkReset::int32ForkReset() {
+    int32ForkReset& x = *this;
+    x.store(0);
+}
+
+int32ForkReset::int32ForkReset(int32_t value) {
+    int32ForkReset& x = *this;
+    x.store(value);
+}
+
+void int32ForkReset::store(int32_t value) noexcept {
+    int32ForkReset& x = *this;
+    x._state.store( (((uint64_t)(_forkEpoch)) << 32) | ((uint64_t)value) );
+}
+
+int32_t int32ForkReset::load() noexcept {
+    int32ForkReset& x = *this;
+
+    while (1) {
+        uint64_t s  = x._state.load();
+        int32_t  epoch = (int32_t)(s >> 32);
+        int32_t  value = (int32_t)(s & ((1ULL << 32) - 1));
+        if (epoch == _forkEpoch)
+            return value;
+
+        uint64_t s_ = (((uint64_t)(_forkEpoch)) << 32) | 0;  // reset to 0
+        if (x._state.compare_exchange_strong(s, s_))
+            return 0;
+    }
+}
+
+int32_t int32ForkReset::fetch_add(int32_t delta) noexcept {
+    int32ForkReset& x = *this;
+
+    while (1) {
+        int32_t  value  = x.load(); // resets value to 0 on new fork epoch
+        int32_t  value_ = value + delta;
+        uint64_t s  = (((uint64_t)(_forkEpoch)) << 32) | ((uint64_t)value);
+        uint64_t s_ = (((uint64_t)(_forkEpoch)) << 32) | ((uint64_t)value_);
+        if (x._state.compare_exchange_strong(s, s_))
+            return value;
+    }
+}
+
+
+}}} // golang::internal::atomic::

golang/runtime/internal/atomic.h

+#ifndef _NXD_LIBGOLANG_RUNTIME_INTERNAL_ATOMIC_H
+#define _NXD_LIBGOLANG_RUNTIME_INTERNAL_ATOMIC_H
+
+// Copyright (C) 2022  Nexedi SA and Contributors.
+//                     Kirill Smelkov <kirr@nexedi.com>
+//
+// This program is free software: you can Use, Study, Modify and Redistribute
+// it under the terms of the GNU General Public License version 3, or (at your
+// option) any later version, as published by the Free Software Foundation.
+//
+// You can also Link and Combine this program with other software covered by
+// the terms of any of the Free Software licenses or any of the Open Source
+// Initiative approved licenses and Convey the resulting work. Corresponding
+// source of such a combination shall include the source code for all other
+// software used.
+//
+// This program is distributed WITHOUT ANY WARRANTY; without even the implied
+// warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+//
+// See COPYING file for full licensing terms.
+// See https://www.nexedi.com/licensing for rationale and options.
+
+// Package internal/atomic provides specialized low-level atomic types.
+
+#include <atomic>
+#include <stdint.h>
+
+// golang::internal::atomic::
+namespace golang {
+namespace internal {
+namespace atomic {
+
+// int32ForkReset is atomic int32 that is reset to zero in forked child.
+//
+// It helps to organize locks that do not deadlock in forked child.
+// See NOTES in https://man7.org/linux/man-pages/man3/pthread_atfork.3.html for
+// rationale and details.
+struct int32ForkReset {
+private:
+    std::atomic<uint64_t> _state; // [fork_epoch]₃₂[value]₃₂
+
+public:
+    int32ForkReset();
+    int32ForkReset(int32_t value);
+
+    void    store(int32_t value) noexcept;
+    int32_t load() noexcept;
+    int32_t fetch_add(int32_t delta) noexcept;
+};
+
+}}} // golang::internal::atomic::
+
+#endif  // _NXD_LIBGOLANG_RUNTIME_INTERNAL_ATOMIC_H

golang/runtime/libgolang.cpp

@@ -118,10 +118,13 @@ const _libgolang_runtime_ops *_runtime = nil;
 }
 using internal::_runtime;
 
+namespace internal { namespace atomic { extern void _init(); } }
 void _libgolang_init(const _libgolang_runtime_ops *runtime_ops) {
     if (_runtime != nil) // XXX better check atomically
         panic("libgolang: double init");
     _runtime = runtime_ops;
+
+    internal::atomic::_init();
 }
 
 void _taskgo(void (*f)(void *), void *arg) {

setup.py

@@ -197,6 +197,7 @@ setup(
 
     x_dsos      = [DSO('golang.runtime.libgolang',
                         ['golang/runtime/libgolang.cpp',
+                         'golang/runtime/internal/atomic.cpp',
                          'golang/runtime/internal/syscall.cpp',
                          'golang/context.cpp',
                          'golang/errors.cpp',
@@ -208,6 +209,7 @@ setup(
                         depends = [
                             'golang/libgolang.h',
                             'golang/runtime/internal.h',
+                            'golang/runtime/internal/atomic.h',
                             'golang/runtime/internal/syscall.h',
                             'golang/context.h',
                             'golang/cxx.h',