mirrors/remark42
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

switch native import lock to sized group

Browse Source
This commit is contained in:
Umputun
2019-02-02 22:28:16 -06:00
parent e6f30b0768
commit 3314223120
9 changed files with 109 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Gopkg.lock
Generated
+3 -3
View File
@@ -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"
backend/app/migrator/native.go
+5 -5
View File
@@ -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)
backend/vendor/github.com/go-pkgz/syncs/LICENSE
Generated Vendored
+1 -1
View File
@@ -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
backend/vendor/github.com/go-pkgz/syncs/README.md
Generated Vendored
+8 -7
View File
@@ -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
})
}
backend/vendor/github.com/go-pkgz/syncs/errsizedgroup.go
Generated Vendored
+10 -50
View File
@@ -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
backend/vendor/github.com/go-pkgz/syncs/go.mod
Generated Vendored
+3
View File
@@ -0,0 +1,3 @@
module github.com/go-pkgz/syncs
require github.com/stretchr/testify v1.3.0
backend/vendor/github.com/go-pkgz/syncs/go.sum
Generated Vendored
+7
View File
@@ -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=
backend/vendor/github.com/go-pkgz/syncs/group_options.go
Generated Vendored
+30
View File
@@ -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
}
backend/vendor/github.com/go-pkgz/syncs/sizedgroup.go
Generated Vendored
+42 -13
View File
@@ -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()
}