summaryrefslogtreecommitdiff
path: root/static_pool.go
diff options
context:
space:
mode:
authorValery Piashchynski <[email protected]>2020-11-09 14:51:34 +0300
committerValery Piashchynski <[email protected]>2020-11-09 14:51:34 +0300
commit83c14cbad2d7d403b08efbb3cf900df9b52b4938 (patch)
treeb084a2ca99eb7523232f477678f8aa2a82cd5812 /static_pool.go
parentb7b533dbe13d2c1a8e78c0e33a4a388c56884440 (diff)
Add spiral errors
Diffstat (limited to 'static_pool.go')
-rwxr-xr-xstatic_pool.go26
1 files changed, 13 insertions, 13 deletions
diff --git a/static_pool.go b/static_pool.go
index ee81fd39..17ec605e 100755
--- a/static_pool.go
+++ b/static_pool.go
@@ -189,10 +189,8 @@ func (sp *StaticPool) execDebug(p Payload) (Payload, error) {
func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs Payload) (Payload, error) {
const op = errors.Op("Exec")
w, err := sp.ww.GetFreeWorker(context.Background())
- if err != nil && errors.Is(errors.ErrWatcherStopped, err) {
+ if err != nil {
return EmptyPayload, errors.E(op, err)
- } else if err != nil {
- return EmptyPayload, err
}
sw := w.(SyncWorker)
@@ -204,19 +202,19 @@ func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs Payload) (Payload
if sp.cfg.MaxJobs != 0 && w.State().NumExecs() >= sp.cfg.MaxJobs {
err = sp.ww.AllocateNew(bCtx)
if err != nil {
- sp.events.Push(PoolEvent{Event: EventPoolError, Payload: err})
+ sp.events.Push(PoolEvent{Event: EventPoolError, Payload: errors.E(op, err)})
}
w.State().Set(StateInvalid)
err = w.Stop(bCtx)
if err != nil {
- sp.events.Push(WorkerEvent{Event: EventWorkerError, Worker: w, Payload: err})
+ sp.events.Push(WorkerEvent{Event: EventWorkerError, Worker: w, Payload: errors.E(op, err)})
}
} else {
sp.ww.PushWorker(w)
}
- return EmptyPayload, err
+ return EmptyPayload, errors.E(op, err)
}
sw.State().Set(StateInvalid)
@@ -224,10 +222,10 @@ func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs Payload) (Payload
errS := w.Stop(bCtx)
if errS != nil {
- return EmptyPayload, fmt.Errorf("%v, %v", err, errS)
+ return EmptyPayload, errors.E(op, errors.Errorf("%v, %v", err, errS))
}
- return EmptyPayload, err
+ return EmptyPayload, errors.E(op, err)
}
// worker want's to be terminated
@@ -235,7 +233,7 @@ func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs Payload) (Payload
w.State().Set(StateInvalid)
err = w.Stop(bCtx)
if err != nil {
- sp.events.Push(WorkerEvent{Event: EventWorkerError, Worker: w, Payload: err})
+ sp.events.Push(WorkerEvent{Event: EventWorkerError, Worker: w, Payload: errors.E(op, err)})
}
return sp.Exec(rqs)
@@ -244,7 +242,7 @@ func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs Payload) (Payload
if sp.cfg.MaxJobs != 0 && w.State().NumExecs() >= sp.cfg.MaxJobs {
err = sp.ww.AllocateNew(bCtx)
if err != nil {
- return EmptyPayload, err
+ return EmptyPayload, errors.E(op, err)
}
} else {
sp.ww.PushWorker(w)
@@ -259,6 +257,7 @@ func (sp *StaticPool) Destroy(ctx context.Context) {
// allocate required number of stack
func (sp *StaticPool) allocateWorkers(ctx context.Context, numWorkers int64) ([]WorkerBase, error) {
+ const op = errors.Op("allocate workers")
var workers []WorkerBase
// constant number of stack simplify logic
@@ -267,20 +266,21 @@ func (sp *StaticPool) allocateWorkers(ctx context.Context, numWorkers int64) ([]
w, err := sp.factory.SpawnWorkerWithContext(ctx, sp.cmd())
if err != nil {
cancel()
- return nil, err
+ return nil, errors.E(op, err)
}
- cancel()
workers = append(workers, w)
+ cancel()
}
return workers, nil
}
func (sp *StaticPool) checkMaxJobs(ctx context.Context, w WorkerBase) error {
+ const op = errors.Op("check max jobs")
if sp.cfg.MaxJobs != 0 && w.State().NumExecs() >= sp.cfg.MaxJobs {
err := sp.ww.AllocateNew(ctx)
if err != nil {
sp.events.Push(PoolEvent{Event: EventPoolError, Payload: err})
- return err
+ return errors.E(op, err)
}
}
return nil