From a9a1acc8806768bd6c6524e881697e25818183c3 Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Tue, 16 Feb 2021 15:17:37 +0300
Subject: [PATCH] [#378] shard: Control the completion of all handlers of the
previous event
Group handlers of the particular event to a WaitGroup and wait for it before
the next event handling. This will ensure that all handlers complete and
prevent potential conflicts between past and present jobs.
Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
---
pkg/local_object_storage/shard/gc.go | 9 +++++++++
1 file changed, 9 insertions(+)
diff --git a/pkg/local_object_storage/shard/gc.go b/pkg/local_object_storage/shard/gc.go
index 504748473..3af5db40c 100644
--- a/pkg/local_object_storage/shard/gc.go
+++ b/pkg/local_object_storage/shard/gc.go
@@ -2,6 +2,7 @@ package shard
import (
"context"
+ "sync"
"time"
"github.com/nspcc-dev/neofs-api-go/pkg/object"
@@ -25,6 +26,8 @@ const (
type eventHandler func(context.Context, Event)
type eventHandlers struct {
+ prevGroup sync.WaitGroup
+
cancelFunc context.CancelFunc
handlers []eventHandler
@@ -97,18 +100,24 @@ func (gc *gc) listenEvents() {
}
v.cancelFunc()
+ v.prevGroup.Wait()
var ctx context.Context
ctx, v.cancelFunc = context.WithCancel(context.Background())
+ v.prevGroup.Add(len(v.handlers))
+
for _, h := range v.handlers {
err := gc.workerPool.Submit(func() {
h(ctx, event)
+ v.prevGroup.Done()
})
if err != nil {
gc.log.Warn("could not submit GC job to worker pool",
zap.String("error", err.Error()),
)
+
+ v.prevGroup.Done()
}
}
}