diff --git a/backend/Gopkg.lock b/backend/Gopkg.lock index f1b7b5ae..954d6b43 100644 --- a/backend/Gopkg.lock +++ b/backend/Gopkg.lock @@ -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", diff --git a/backend/vendor/github.com/go-pkgz/syncs/.travis.yml b/backend/vendor/github.com/go-pkgz/syncs/.travis.yml new file mode 100644 index 00000000..eb49afa8 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/syncs/.travis.yml @@ -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 diff --git a/backend/vendor/github.com/go-pkgz/syncs/LICENSE b/backend/vendor/github.com/go-pkgz/syncs/LICENSE new file mode 100644 index 00000000..ca125214 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/syncs/LICENSE @@ -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. diff --git a/backend/vendor/github.com/go-pkgz/syncs/README.md b/backend/vendor/github.com/go-pkgz/syncs/README.md new file mode 100644 index 00000000..3acf3384 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/syncs/README.md @@ -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() +``` \ No newline at end of file diff --git a/backend/vendor/github.com/go-pkgz/syncs/errsizedgroup.go b/backend/vendor/github.com/go-pkgz/syncs/errsizedgroup.go new file mode 100644 index 00000000..bbe02ea0 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/syncs/errsizedgroup.go @@ -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, ", ")) +} diff --git a/backend/vendor/github.com/go-pkgz/syncs/semaphore.go b/backend/vendor/github.com/go-pkgz/syncs/semaphore.go new file mode 100644 index 00000000..9eb84599 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/syncs/semaphore.go @@ -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 +} diff --git a/backend/vendor/github.com/go-pkgz/syncs/sizedgroup.go b/backend/vendor/github.com/go-pkgz/syncs/sizedgroup.go new file mode 100644 index 00000000..27e57c91 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/syncs/sizedgroup.go @@ -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() +}