add syncs lib

2019-02-02 17:51:44 -06:00
parent a731856bb1
commit 7ddf1fc096
7 changed files with 352 additions and 0 deletions
@@ -165,6 +165,14 @@
  revision = "e7d08d0194d613b8854de2e487bf7732500fa153"
  version = "v1.2.0"

+[[projects]]
+  digest = "1:98626887dd476c5187317d8235da0c5489ed4bc9b3a86dd32b5e158f46b4a8ae"
+  name = "github.com/go-pkgz/syncs"
+  packages = ["."]
+  pruneopts = "UT"
+  revision = "ac098f93e9edc6a9c213ada2e74074b12acf9dbe"
+  version = "v1.0.0"
+
 [[projects]]
  digest = "1:ffc060c551980d37ee9e428ef528ee2813137249ccebb0bfc412ef83071cac91"
  name = "github.com/golang/protobuf"
@@ -411,6 +419,7 @@
    "github.com/go-pkgz/rest",
    "github.com/go-pkgz/rest/cache",
    "github.com/go-pkgz/rest/logger",
+    "github.com/go-pkgz/syncs",
    "github.com/google/uuid",
    "github.com/gorilla/feeds",
    "github.com/hashicorp/go-multierror",
@@ -0,0 +1,16 @@
+language: go
+
+go:
+  - "1.11.x"
+
+go_import_path: github.com/go-pkgz/syncs
+
+before_install:
+  - go get github.com/mattn/goveralls
+  - go get gopkg.in/alecthomas/gometalinter.v2
+  - $GOPATH/bin/gometalinter.v2 --install
+
+script:
+  - go test ./...
+  - $GOPATH/bin/gometalinter.v2 --exclude=test --exclude=mock --exclude=vendor --exclude=maligned ./...
+  - $GOPATH/bin/goveralls -service=travis-ci
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2018 Umputun
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
@@ -0,0 +1,66 @@
+# Syncs - additional synchronization primitives 
+
+[![Build Status](https://travis-ci.org/go-pkgz/syncs.svg?branch=master)](https://travis-ci.org/go-pkgz/syncs) [![Go Report Card](https://goreportcard.com/badge/github.com/go-pkgz/syncs)](https://goreportcard.com/report/github.com/go-pkgz/syncs) [![Coverage Status](https://coveralls.io/repos/github/go-pkgz/syncs/badge.svg?branch=master)](https://coveralls.io/github/go-pkgz/syncs?branch=master)
+
+Package syncs provides additional synchronization primitives.
+
+## Install and update
+
+`go get -u github.com/go-pkgz/syncs`
+
+## Details
+
+### Semaphore
+
+Implements `sync.Locker` interface but for given capacity, thread safe. Lock increases count and Unlock - decreases. Unlock on 0 count will be blocked.
+
+```go
+    sema := syncs.NewSemaphore(10) // make semaphore with 10 initial capacity
+    for i :=0; i<10; i++ {
+        sema.Lock() // all 10 locks will pass, i.w. won't lock
+    }
+    sema.Lock() // this is 11 - will lock for real
+
+    // in some other place/goroutine
+    sema.Unlock() // decrease semaphore counter
+```
+
+### SizedGroup
+
+Mix semaphore and WaitGroup to provide sized waiting group. The result is a wait group allowing limited number of goroutine to run in parallel.
+
+The locking happens inside of goroutine, i.e. **every call will be non-blocked**, but some goroutines may wait if semaphore locked. It means - technically it doesn't limit number of goroutines, but rather number of running (active) goroutines. 
+
+```go
+    swg := syncs.NewSizedGroup(5) // wait group with max size=5
+     for i :=0; i<10; i++ {
+        swg.Go(fn func(){
+            doThings() // only 5 of these will run in parallel
+        })
+    }
+    swg.Wait()
+```
+
+### ErrSizedGroup
+
+Sized error group is a SizedGroup with error control. 
+Works the same as errgrp.Group, i.e. returns first error.
+Can work as regular errgrp.Group or with early termination.
+Thread safe.
+
+Supports both in-goroutine-wait via `NewErrSizedGroup` as well as outside of goroutine wait with `Preemptive()` option. Another options are  `TermOnErr` which will skip (won't start) all other goroutines if any error returned, and `Context`.
+
+Important! With `Preemptive` Go call **can block**. In case if maximum size reached the call will wait till number of running goroutines 
+dropped under max. This way we not only limiting number of running goroutines but also number of waiting goroutines.
+
+
+```go
+    ewg := syncs.NewErrSizedGroup(5, syncs.Preemptive()) // error wait group with max size=5, don't try to start more if any error happened
+     for i :=0; i<10; i++ {
+        ewg.Go(fn func() error { // Go here could be blocked if trying to run >5 at the same time 
+           err := doThings()     // only 5 of these will run in parallel
+           return err
+        })
+    }
+    err := ewg.Wait()
+```
@@ -0,0 +1,172 @@
+package syncs
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"strings"
+	"sync"
+)
+
+// ErrSizedGroup is a SizedGroup with error control. Works the same as errgrp.Group, i.e. returns first error.
+// Can work as regular errgrp.Group or with early termination. Thread safe.
+// ErrSizedGroup interface enforces constructor usage and doesn't allow direct creation of errSizedGroup
+type ErrSizedGroup interface {
+	Go(fn func() error)
+	Wait() error
+}
+
+type errSizedGroup struct {
+	wg          sync.WaitGroup
+	sema        sync.Locker
+	ctx         context.Context
+	cancel      func()
+	termOnError bool
+	preLock     bool
+
+	err     *multierror
+	errLock sync.RWMutex
+	errOnce sync.Once
+}
+
+// NewErrSizedGroup makes wait group with limited size alive goroutines.
+// By default all goroutines will be started but will wait inside. For limited number of goroutines use Preemptive() options.
+// TermOnErr will skip (won't start) all other goroutines if any error returned.
+func NewErrSizedGroup(size int, options ...ESGOption) ErrSizedGroup {
+	res := errSizedGroup{
+		sema: NewSemaphore(size),
+		err:  new(multierror),
+	}
+
+	for _, opt := range options {
+		if err := opt(&res); err != nil {
+			log.Printf("[WARN] failed to set cache option, %v", err)
+		}
+	}
+
+	return &res
+}
+
+// Go calls the given function in a new goroutine.
+// The first call to return a non-nil error cancels the group if termOnError; its error will be
+// returned by Wait. If no termOnError all errors will be collected in multierror.
+func (g *errSizedGroup) Go(f func() error) {
+
+	g.wg.Add(1)
+
+	if g.preLock {
+		g.sema.Lock()
+	}
+
+	go func() {
+		defer g.wg.Done()
+
+		// terminated will be true if any error happened before and g.termOnError
+		terminated := func() bool {
+			if !g.termOnError {
+				return false
+			}
+			g.errLock.RLock()
+			defer g.errLock.RUnlock()
+			return g.err.errorOrNil() != nil
+		}
+
+		if terminated() {
+			return // terminated due prev error, don't run anything in this group anymore
+		}
+
+		if !g.preLock {
+			g.sema.Lock()
+		}
+
+		if err := f(); err != nil {
+
+			g.errLock.Lock()
+			g.err = g.err.append(err)
+			g.errLock.Unlock()
+
+			g.errOnce.Do(func() { // call context cancel once
+				if g.cancel != nil {
+					g.cancel()
+				}
+			})
+		}
+		g.sema.Unlock()
+	}()
+}
+
+// Wait blocks until all function calls from the Go method have returned, then
+// returns the first all errors (if any) wrapped with multierror from them.
+func (g *errSizedGroup) Wait() error {
+	g.wg.Wait()
+	if g.cancel != nil {
+		g.cancel()
+	}
+	return g.err.errorOrNil()
+}
+
+// ESGOption functional option type
+type ESGOption func(esg *errSizedGroup) error
+
+// Context passes ctx and makes it cancelable
+func Context(ctx context.Context) ESGOption {
+	return func(esg *errSizedGroup) error {
+		ctxWithCancel, cancel := context.WithCancel(ctx)
+		esg.cancel = cancel
+		esg.ctx = ctxWithCancel
+		return nil
+	}
+}
+
+// Preemptive sets locking mode preventing spawning waiting goroutine. May cause Go call to block!
+func Preemptive() ESGOption {
+	return func(esg *errSizedGroup) error {
+		esg.preLock = true
+		return nil
+	}
+}
+
+// TermOnErr prevents new goroutines to start after first error
+func TermOnErr() ESGOption {
+	return func(esg *errSizedGroup) error {
+		esg.termOnError = true
+		return nil
+	}
+}
+
+type multierror struct {
+	errors []error
+	lock   sync.Mutex
+}
+
+func (m *multierror) append(err error) *multierror {
+	m.lock.Lock()
+	m.errors = append(m.errors, err)
+	m.lock.Unlock()
+	return m
+}
+
+func (m *multierror) errorOrNil() *multierror {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	if len(m.errors) == 0 {
+		return nil
+	}
+	return m
+}
+
+// Error returns multierror string
+func (m *multierror) Error() string {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	if len(m.errors) == 0 {
+		return ""
+	}
+
+	errs := []string{}
+
+	for n, e := range m.errors {
+		errs = append(errs, fmt.Sprintf("[%d] {%s}", n, e.Error()))
+	}
+	return fmt.Sprintf("%d error(s) occurred: %s", len(m.errors), strings.Join(errs, ", "))
+}
@@ -0,0 +1,27 @@
+package syncs
+
+import "sync"
+
+// Semaphore implementation, counted lock only. Implements sync.Locker interface, thread safe.
+type semaphore struct {
+	sync.Locker
+	ch chan struct{}
+}
+
+// NewSemaphore makes Semaphore with given capacity
+func NewSemaphore(capacity int) sync.Locker {
+	if capacity <= 0 {
+		capacity = 1
+	}
+	return &semaphore{ch: make(chan struct{}, capacity)}
+}
+
+// Lock acquires semaphore, can block if out of capacity.
+func (s *semaphore) Lock() {
+	s.ch <- struct{}{}
+}
+
+// Unlock releases semaphore, can block if nothing acquired before.
+func (s *semaphore) Unlock() {
+	<-s.ch
+}
@@ -0,0 +1,41 @@
+package syncs
+
+import "sync"
+
+// SizedGroup has the same role as WaitingGroup but adds a limit of the amount of goroutines started concurrently.
+// Uses similar Go() scheduling as errgrp.Group, thread safe.
+// SizedGroup interface enforces constructor usage and doesn't allow direct creation of sizedGroup
+type SizedGroup interface {
+	Go(fn func())
+	Wait()
+}
+
+type sizedGroup struct {
+	wg   sync.WaitGroup
+	sema sync.Locker
+}
+
+// NewSizedGroup makes wait group with limited size alive goroutines
+func NewSizedGroup(size int) SizedGroup {
+	return &sizedGroup{sema: NewSemaphore(size)}
+}
+
+// Go calls the given function in a new goroutine.
+// Every call will be unblocked, but some goroutines may wait if semaphore locked.
+func (g *sizedGroup) Go(fn func()) {
+	g.wg.Add(1)
+
+	go func() {
+		defer g.wg.Done()
+
+		g.sema.Lock()
+		fn()
+		g.sema.Unlock()
+	}()
+}
+
+// Wait blocks until the SizedGroup counter is zero.
+// See sync.WaitGroup documentation for more information.
+func (g *sizedGroup) Wait() {
+	g.wg.Wait()
+}