fix dead lock
This commit is contained in:
2
makefile
2
makefile
@@ -1,5 +1,5 @@
|
|||||||
debug:
|
debug:
|
||||||
go run -tags "debug" ./...
|
go run -tags "debug" -race ./...
|
||||||
|
|
||||||
all: clean build
|
all: clean build
|
||||||
|
|
||||||
|
|||||||
@@ -51,6 +51,7 @@ func run(ctx context.Context, debug bool) int {
|
|||||||
|
|
||||||
if err := docker.StartWatcher(ctx, config.DaemonConfiguration.PullInterval); err != nil {
|
if err := docker.StartWatcher(ctx, config.DaemonConfiguration.PullInterval); err != nil {
|
||||||
slog.Error("unable to start the docker watcher", "thread", "main", "err", err)
|
slog.Error("unable to start the docker watcher", "thread", "main", "err", err)
|
||||||
|
return failure
|
||||||
}
|
}
|
||||||
|
|
||||||
el, err := runtime.NewEventLoop(ctx, docker)
|
el, err := runtime.NewEventLoop(ctx, docker)
|
||||||
@@ -61,6 +62,7 @@ func run(ctx context.Context, debug bool) int {
|
|||||||
defer el.Close()
|
defer el.Close()
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
|
interruptNum := 0
|
||||||
c := make(chan os.Signal, 10)
|
c := make(chan os.Signal, 10)
|
||||||
signal.Notify(c)
|
signal.Notify(c)
|
||||||
defer close(c)
|
defer close(c)
|
||||||
@@ -69,16 +71,21 @@ func run(ctx context.Context, debug bool) int {
|
|||||||
switch sig {
|
switch sig {
|
||||||
case os.Interrupt:
|
case os.Interrupt:
|
||||||
{
|
{
|
||||||
|
interruptNum++
|
||||||
slog.Info("the process received an interruption signal", "thread", "signal_watcher", "signal", sig.String())
|
slog.Info("the process received an interruption signal", "thread", "signal_watcher", "signal", sig.String())
|
||||||
el.Close()
|
el.Close()
|
||||||
}
|
}
|
||||||
case os.Kill:
|
case os.Kill:
|
||||||
{
|
{
|
||||||
|
interruptNum++
|
||||||
slog.Info("the process received a kill signal", "thread", "signal_watcher", "signal", sig.String())
|
slog.Info("the process received a kill signal", "thread", "signal_watcher", "signal", sig.String())
|
||||||
el.Close()
|
el.Close()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if interruptNum == 3 {
|
||||||
|
slog.Info("received 3 times a stop signal, forcing the program to shut down", "thread", "signal_watch")
|
||||||
|
os.Exit(failure)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
|||||||
@@ -21,18 +21,22 @@ type (
|
|||||||
executors map[string]cron.EntryID
|
executors map[string]cron.EntryID
|
||||||
|
|
||||||
ctx context.Context
|
ctx context.Context
|
||||||
eventMu sync.Mutex
|
cancelFunc context.CancelFunc
|
||||||
|
eventMu sync.RWMutex
|
||||||
eventBuffer []helper.ContainerEvent
|
eventBuffer []helper.ContainerEvent
|
||||||
eventTrigger chan struct{}
|
eventTrigger chan struct{}
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
|
||||||
func NewEventLoop(ctx context.Context, docker *helper.DockerHelper) (*EventLoop, error) {
|
func NewEventLoop(ctx context.Context, docker *helper.DockerHelper) (*EventLoop, error) {
|
||||||
|
loopCtx, cancelFunc := context.WithCancel(contextutil.WithThreadName(context.Background(), "event_loop"))
|
||||||
loop := &EventLoop{
|
loop := &EventLoop{
|
||||||
docker: docker,
|
docker: docker,
|
||||||
cr: cron.New(),
|
cr: cron.New(),
|
||||||
ctx: contextutil.WithThreadName(context.Background(), "event_loop"),
|
ctx: loopCtx,
|
||||||
|
cancelFunc: cancelFunc,
|
||||||
executors: make(map[string]cron.EntryID),
|
executors: make(map[string]cron.EntryID),
|
||||||
|
eventTrigger: make(chan struct{}),
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := loop.firstRun(ctx); err != nil {
|
if err := loop.firstRun(ctx); err != nil {
|
||||||
@@ -46,6 +50,7 @@ func NewEventLoop(ctx context.Context, docker *helper.DockerHelper) (*EventLoop,
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (el *EventLoop) Close() error {
|
func (el *EventLoop) Close() error {
|
||||||
|
el.cancelFunc()
|
||||||
ctx := el.cr.Stop()
|
ctx := el.cr.Stop()
|
||||||
<-ctx.Done()
|
<-ctx.Done()
|
||||||
return nil
|
return nil
|
||||||
@@ -146,15 +151,33 @@ func (el *EventLoop) unregister(container helper.Container) {
|
|||||||
|
|
||||||
func (el *EventLoop) startEventLoop() {
|
func (el *EventLoop) startEventLoop() {
|
||||||
for {
|
for {
|
||||||
if len(el.eventBuffer) == 0 {
|
el.eventMu.RLock()
|
||||||
<-el.eventTrigger
|
eventBufferLen := len(el.eventBuffer)
|
||||||
|
el.eventMu.RUnlock()
|
||||||
|
|
||||||
|
if eventBufferLen == 0 {
|
||||||
|
select {
|
||||||
|
case <-el.eventTrigger:
|
||||||
|
{
|
||||||
|
slog.Debug("event loop waken up", "thread", contextutil.ThreadName(el.ctx))
|
||||||
|
}
|
||||||
|
case <-el.ctx.Done():
|
||||||
|
{
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var ev helper.ContainerEvent
|
||||||
el.eventMu.Lock()
|
el.eventMu.Lock()
|
||||||
ev := el.eventBuffer[0]
|
if len(el.eventBuffer) == 0 {
|
||||||
el.eventBuffer = append([]helper.ContainerEvent{}, el.eventBuffer[1:]...)
|
el.eventMu.Unlock()
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
ev, el.eventBuffer = el.eventBuffer[0], el.eventBuffer[1:]
|
||||||
el.eventMu.Unlock()
|
el.eventMu.Unlock()
|
||||||
|
|
||||||
|
slog.Debug("processing docker event", "thread", contextutil.ThreadName(el.ctx))
|
||||||
el.process(ev)
|
el.process(ev)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user