diff --git a/backend/Gopkg.lock b/backend/Gopkg.lock index acc5ddc4..1c83a71d 100644 --- a/backend/Gopkg.lock +++ b/backend/Gopkg.lock @@ -174,12 +174,12 @@ version = "v1.2.0" [[projects]] - digest = "1:98626887dd476c5187317d8235da0c5489ed4bc9b3a86dd32b5e158f46b4a8ae" + digest = "1:92b44856ee15e8a98b91d751a60b512017e0ba227d1ed9d2c02ad13d67062ff8" name = "github.com/go-pkgz/syncs" packages = ["."] pruneopts = "UT" - revision = "ac098f93e9edc6a9c213ada2e74074b12acf9dbe" - version = "v1.0.0" + revision = "72b3cd427a3495479f72a3a06c8949e39485dfc0" + version = "v1.1.0" [[projects]] digest = "1:ffc060c551980d37ee9e428ef528ee2813137249ccebb0bfc412ef83071cac91" diff --git a/backend/app/migrator/native.go b/backend/app/migrator/native.go index bf60d834..99cf9d34 100644 --- a/backend/app/migrator/native.go +++ b/backend/app/migrator/native.go @@ -2,6 +2,7 @@ package migrator import ( "bytes" + "context" "encoding/json" "io" "sync/atomic" @@ -109,7 +110,7 @@ func (n *Native) Import(reader io.Reader, siteID string) (size int, err error) { if n.Concurrent > 0 { concurrent = n.Concurrent } - grp := syncs.NewErrSizedGroup(concurrent, syncs.Preemptive()) + grp := syncs.NewSizedGroup(concurrent, syncs.Preemptive) for { comment := store.Comment{} @@ -127,22 +128,21 @@ func (n *Native) Import(reader io.Reader, siteID string) (size int, err error) { } // write comments in parallel - grp.Go(func() error { + grp.Go(func(context.Context) { if _, e := n.DataStore.Create(comment); e != nil { atomic.AddInt64(&failed, 1) log.Printf("[WARN] can't write %+v to store, %s", comment, e) - return nil + return } n := atomic.AddInt64(&comments, 1) if n%1000 == 0 { log.Printf("[DEBUG] imported %d comments", n) } - return nil }) } - _ = grp.Wait() + grp.Wait() if failed > 0 { return int(comments), errors.Errorf("failed to save %d comments", failed) diff --git a/backend/vendor/github.com/go-pkgz/syncs/LICENSE b/backend/vendor/github.com/go-pkgz/syncs/LICENSE index ca125214..ac540250 100644 --- a/backend/vendor/github.com/go-pkgz/syncs/LICENSE +++ b/backend/vendor/github.com/go-pkgz/syncs/LICENSE @@ -1,6 +1,6 @@ MIT License -Copyright (c) 2018 Umputun +Copyright (c) 2019 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 diff --git a/backend/vendor/github.com/go-pkgz/syncs/README.md b/backend/vendor/github.com/go-pkgz/syncs/README.md index 3acf3384..9599cb95 100644 --- a/backend/vendor/github.com/go-pkgz/syncs/README.md +++ b/backend/vendor/github.com/go-pkgz/syncs/README.md @@ -29,13 +29,14 @@ Implements `sync.Locker` interface but for given capacity, thread safe. Lock inc 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. +By default 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. +In order to block goroutines from even starting use `Preemptive` option (see below). ```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.Go(fn func(ctx context.Context){ + doThings(ctx) // only 5 of these will run in parallel }) } swg.Wait() @@ -48,17 +49,17 @@ 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`. +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` for early termination/timeouts. 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 + 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 + ewg.Go(fn func(ctx context.Context) error { // Go here could be blocked if trying to run >5 at the same time + err := doThings(ctx) // only 5 of these will run in parallel return err }) } diff --git a/backend/vendor/github.com/go-pkgz/syncs/errsizedgroup.go b/backend/vendor/github.com/go-pkgz/syncs/errsizedgroup.go index bbe02ea0..356b0503 100644 --- a/backend/vendor/github.com/go-pkgz/syncs/errsizedgroup.go +++ b/backend/vendor/github.com/go-pkgz/syncs/errsizedgroup.go @@ -1,9 +1,7 @@ package syncs import ( - "context" "fmt" - "log" "strings" "sync" ) @@ -11,18 +9,10 @@ import ( // 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 +type ErrSizedGroup struct { + options + wg sync.WaitGroup + sema sync.Locker err *multierror errLock sync.RWMutex @@ -32,16 +22,15 @@ type errSizedGroup struct { // 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{ +func NewErrSizedGroup(size int, options ...GroupOption) *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) - } + opt(&res.options) } return &res @@ -50,7 +39,7 @@ func NewErrSizedGroup(size int, options ...ESGOption) ErrSizedGroup { // 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) { +func (g *ErrSizedGroup) Go(f func() error) { g.wg.Add(1) @@ -97,7 +86,7 @@ func (g *errSizedGroup) Go(f func() error) { // 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 { +func (g *ErrSizedGroup) Wait() error { g.wg.Wait() if g.cancel != nil { g.cancel() @@ -105,35 +94,6 @@ func (g *errSizedGroup) Wait() error { 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 diff --git a/backend/vendor/github.com/go-pkgz/syncs/go.mod b/backend/vendor/github.com/go-pkgz/syncs/go.mod new file mode 100644 index 00000000..20a3a4d6 --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/syncs/go.mod @@ -0,0 +1,3 @@ +module github.com/go-pkgz/syncs + +require github.com/stretchr/testify v1.3.0 diff --git a/backend/vendor/github.com/go-pkgz/syncs/go.sum b/backend/vendor/github.com/go-pkgz/syncs/go.sum new file mode 100644 index 00000000..4347755a --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/syncs/go.sum @@ -0,0 +1,7 @@ +github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= diff --git a/backend/vendor/github.com/go-pkgz/syncs/group_options.go b/backend/vendor/github.com/go-pkgz/syncs/group_options.go new file mode 100644 index 00000000..8f27643f --- /dev/null +++ b/backend/vendor/github.com/go-pkgz/syncs/group_options.go @@ -0,0 +1,30 @@ +package syncs + +import "context" + +type options struct { + ctx context.Context + cancel context.CancelFunc + preLock bool + termOnError bool +} + +// GroupOption functional option type +type GroupOption func(o *options) + +// Context passes ctx and makes it cancelable +func Context(ctx context.Context) GroupOption { + return func(o *options) { + o.ctx, o.cancel = context.WithCancel(ctx) + } +} + +// Preemptive sets locking mode preventing spawning waiting goroutine. May cause Go call to block! +func Preemptive(o *options) { + o.preLock = true +} + +// TermOnErr prevents new goroutines to start after first error +func TermOnErr(o *options) { + o.termOnError = true +} diff --git a/backend/vendor/github.com/go-pkgz/syncs/sizedgroup.go b/backend/vendor/github.com/go-pkgz/syncs/sizedgroup.go index 27e57c91..66b09be7 100644 --- a/backend/vendor/github.com/go-pkgz/syncs/sizedgroup.go +++ b/backend/vendor/github.com/go-pkgz/syncs/sizedgroup.go @@ -1,41 +1,70 @@ package syncs -import "sync" +import ( + "context" + "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 { +type SizedGroup struct { + options 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)} +func NewSizedGroup(size int, opts ...GroupOption) *SizedGroup { + res := SizedGroup{sema: NewSemaphore(size)} + res.options.ctx = context.Background() + for _, opt := range opts { + opt(&res.options) + } + return &res } // 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()) { +func (g *SizedGroup) Go(fn func(ctx context.Context)) { + + canceled := func() bool { + select { + case <-g.ctx.Done(): + return true + default: + return false + } + } + + if canceled() { + return + } + g.wg.Add(1) + if g.preLock { + g.sema.Lock() + } + go func() { defer g.wg.Done() - g.sema.Lock() - fn() + if canceled() { + return + } + + if !g.preLock { + g.sema.Lock() + } + + fn(g.ctx) g.sema.Unlock() }() } // Wait blocks until the SizedGroup counter is zero. // See sync.WaitGroup documentation for more information. -func (g *sizedGroup) Wait() { +func (g *SizedGroup) Wait() { g.wg.Wait() }