sync: New package that mirrors Go's sync

Add sync.Once and sync.WaitGroup.

README.rst

@@ -182,6 +182,9 @@ handle concurrency in structured ways.
 - `golang.context` provides contexts to propagate cancellation and task-scoped
   values among spawned goroutines.
 
+- `golang.sync` provides low-level primitives - for example
+  `sync.Once` and `sync.WaitGroup` - that are sometimes useful too.
+
 See `Go Concurrency Patterns: Context`__ for overview of contexts.
 
   __ https://blog.golang.org/context

golang/sync.py

+# -*- coding: utf-8 -*-
+# Copyright (C) 2019  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 sync mirrors Go package sync.
+
+See the following link about Go sync package:
+
+    https://golang.org/pkg/sync
+"""
+
+import threading
+from golang import panic
+
+# Once allows to execute an action only once.
+#
+# For example:
+#
+#   once = Once()
+#   ...
+#   once.do(doSomething)
+class Once(object):
+    def __init__(once):
+        once._mu    = threading.Lock()
+        once._done  = False
+
+    def do(once, f):
+        with once._mu:
+            if not once._done:
+                once._done = True
+                f()
+
+
+# WaitGroup allows to wait for collection of tasks to finish.
+class WaitGroup(object):
+    def __init__(wg):
+        wg._mu      = threading.Lock()
+        wg._count   = 0
+        wg._event   = threading.Event()
+
+    def done(wg):
+        wg.add(-1)
+
+    def add(wg, delta):
+        if delta == 0:
+            return
+        with wg._mu:
+            wg._count += delta
+            if wg._count < 0:
+                panic("sync: negative WaitGroup counter")
+            if wg._count == 0:
+                wg._event.set()
+                wg._event = threading.Event()
+
+    def wait(wg):
+        with wg._mu:
+            if wg._count == 0:
+                return
+            event = wg._event
+        event.wait()

golang/sync_test.py

+# -*- coding: utf-8 -*-
+# Copyright (C) 2019  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.
+
+from golang import go, chan, _PanicError
+from golang import sync
+import time
+from pytest import raises
+
+def test_once():
+    once = sync.Once()
+    l = []
+    def _():
+        l.append(1)
+
+    once.do(_)
+    assert l == [1]
+    once.do(_)
+    assert l == [1]
+    once.do(_)
+    assert l == [1]
+
+    once = sync.Once()
+    l = []
+    def _():
+        l.append(2)
+        raise RuntimeError()
+
+    with raises(RuntimeError):
+        once.do(_)
+    assert l == [2]
+    once.do(_)  # no longer raises
+    assert l == [2]
+    once.do(_)  # no longer raises
+    assert l == [2]
+
+
+def test_waitgroup():
+    wg = sync.WaitGroup()
+    wg.add(2)
+
+    ch = chan(3)
+    def _():
+        wg.wait()
+        ch.send('a')
+    for i in range(3):
+        go(_)
+
+    wg.done()
+    assert len(ch) == 0
+    time.sleep(0.1)
+    assert len(ch) == 0
+    wg.done()
+
+    for i in range(3):
+        assert ch.recv() == 'a'
+
+    wg.add(1)
+    go(_)
+    time.sleep(0.1)
+    assert len(ch) == 0
+    wg.done()
+    assert ch.recv() == 'a'
+
+    with raises(_PanicError):
+        wg.done()