1
0
Fork 0
mirror of https://codeberg.org/forgejo/forgejo.git synced 2025-01-09 15:28:22 -05:00

As per @guillep2k add mutex locks on shutdown/terminate

This commit is contained in:
Andrew Thornton 2020-01-04 14:59:39 +00:00
parent 9941ae2966
commit e852cb620f
No known key found for this signature in database
GPG key ID: 3CDE74631F13A748
2 changed files with 29 additions and 9 deletions

View file

@ -9,6 +9,7 @@ import (
"encoding/json"
"fmt"
"reflect"
"sync"
"time"
"code.gitea.io/gitea/modules/log"
@ -38,6 +39,7 @@ type LevelQueue struct {
queue *levelqueue.Queue
closed chan struct{}
terminated chan struct{}
lock sync.Mutex
exemplar interface{}
workers int
name string
@ -173,6 +175,8 @@ func (l *LevelQueue) Push(data Data) error {
// Shutdown this queue and stop processing
func (l *LevelQueue) Shutdown() {
l.lock.Lock()
defer l.lock.Unlock()
log.Trace("LevelQueue: %s Shutdown", l.name)
select {
case <-l.closed:
@ -185,10 +189,13 @@ func (l *LevelQueue) Shutdown() {
func (l *LevelQueue) Terminate() {
log.Trace("LevelQueue: %s Terminating", l.name)
l.Shutdown()
l.lock.Lock()
select {
case <-l.terminated:
l.lock.Unlock()
default:
close(l.terminated)
l.lock.Unlock()
if err := l.queue.Close(); err != nil && err.Error() != "leveldb: closed" {
log.Error("Error whilst closing internal queue in %s: %v", l.name, err)
}

View file

@ -11,6 +11,7 @@ import (
"fmt"
"reflect"
"strings"
"sync"
"time"
"code.gitea.io/gitea/modules/log"
@ -30,13 +31,15 @@ type redisClient interface {
// RedisQueue redis queue
type RedisQueue struct {
pool *WorkerPool
client redisClient
queueName string
closed chan struct{}
exemplar interface{}
workers int
name string
pool *WorkerPool
client redisClient
queueName string
closed chan struct{}
terminated chan struct{}
exemplar interface{}
workers int
name string
lock sync.Mutex
}
// RedisQueueConfiguration is the configuration for the redis queue
@ -195,19 +198,29 @@ func (r *RedisQueue) Push(data Data) error {
// Shutdown processing from this queue
func (r *RedisQueue) Shutdown() {
log.Trace("Shutdown: %s", r.name)
r.lock.Lock()
select {
case <-r.closed:
default:
close(r.closed)
}
r.lock.Unlock()
}
// Terminate this queue and close the queue
func (r *RedisQueue) Terminate() {
log.Trace("Terminating: %s", r.name)
r.Shutdown()
if err := r.client.Close(); err != nil {
log.Error("Error whilst closing internal redis client in %s: %v", r.name, err)
r.lock.Lock()
select {
case <-r.terminated:
r.lock.Unlock()
default:
close(r.terminated)
r.lock.Unlock()
if err := r.client.Close(); err != nil {
log.Error("Error whilst closing internal redis client in %s: %v", r.name, err)
}
}
}