Pause queues (go-gitea#15928)

* Start adding mechanism to return unhandled data

Signed-off-by: Andrew Thornton <art27@cantab.net>

* Create pushback interface

Signed-off-by: Andrew Thornton <art27@cantab.net>

* Add Pausable interface to WorkerPool and Manager

Signed-off-by: Andrew Thornton <art27@cantab.net>

* Implement Pausable and PushBack for the bytefifos

Signed-off-by: Andrew Thornton <art27@cantab.net>

* Implement Pausable and Pushback for ChannelQueues and ChannelUniqueQueues

Signed-off-by: Andrew Thornton <art27@cantab.net>

* Wire in UI for pausing

Signed-off-by: Andrew Thornton <art27@cantab.net>

* add testcases and fix a few issues

Signed-off-by: Andrew Thornton <art27@cantab.net>

* fix build

Signed-off-by: Andrew Thornton <art27@cantab.net>

* prevent "race" in the test

Signed-off-by: Andrew Thornton <art27@cantab.net>

* fix jsoniter mismerge

Signed-off-by: Andrew Thornton <art27@cantab.net>

* fix conflicts

Signed-off-by: Andrew Thornton <art27@cantab.net>

* fix format

Signed-off-by: Andrew Thornton <art27@cantab.net>

* Add warnings for no worker configurations and prevent data-loss with redis/levelqueue

Signed-off-by: Andrew Thornton <art27@cantab.net>

* Use StopTimer

Signed-off-by: Andrew Thornton <art27@cantab.net>

Co-authored-by: Lauris BH <lauris@nix.lv>
Co-authored-by: 6543 <6543@obermui.de>
Co-authored-by: techknowlogick <techknowlogick@gitea.io>
Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>

modules/indexer/code/indexer.go

@@ -133,11 +133,11 @@ func Init() {
  // Create the Queue
  switch setting.Indexer.RepoType {
  case "bleve", "elasticsearch":
- handler := func(data ...queue.Data) {
+ handler := func(data ...queue.Data) []queue.Data {
  idx, err := indexer.get()
  if idx == nil || err != nil {
  log.Error("Codes indexer handler: unable to get indexer!")
- return
+ return data
  }
 
  for _, datum := range data {
@@ -153,6 +153,7 @@ func Init() {
  continue
  }
  }
+ return nil
  }
 
  indexerQueue = queue.CreateUniqueQueue("code_indexer", handler, &IndexerData{})

modules/indexer/issues/indexer.go

@@ -103,11 +103,11 @@ func InitIssueIndexer(syncReindex bool) {
  // Create the Queue
  switch setting.Indexer.IssueType {
  case "bleve", "elasticsearch":
- handler := func(data ...queue.Data) {
+ handler := func(data ...queue.Data) []queue.Data {
  indexer := holder.get()
  if indexer == nil {
  log.Error("Issue indexer handler: unable to get indexer!")
- return
+ return data
  }
 
  iData := make([]*IndexerData, 0, len(data))
@@ -127,6 +127,7 @@ func InitIssueIndexer(syncReindex bool) {
  if err := indexer.Index(iData); err != nil {
  log.Error("Error whilst indexing: %v Error: %v", iData, err)
  }
+ return nil
  }
 
  issueIndexerQueue = queue.CreateQueue("issue_indexer", handler, &IndexerData{})

modules/indexer/stats/queue.go

@@ -17,13 +17,14 @@ import (
 var statsQueue queue.UniqueQueue
 
 // handle passed PR IDs and test the PRs
-func handle(data ...queue.Data) {
+func handle(data ...queue.Data) []queue.Data {
  for _, datum := range data {
  opts := datum.(int64)
  if err := indexer.Index(opts); err != nil {
  log.Error("stats queue indexer.Index(%d) failed: %v", opts, err)
  }
  }
+ return nil
 }
 
 func initStatsQueue() error {

modules/notification/ui/ui.go

@@ -38,13 +38,14 @@ func NewNotifier() base.Notifier {
  return ns
 }
 
-func (ns *notificationService) handle(data ...queue.Data) {
+func (ns *notificationService) handle(data ...queue.Data) []queue.Data {
  for _, datum := range data {
  opts := datum.(issueNotificationOpts)
  if err := models.CreateOrUpdateIssueNotifications(opts.IssueID, opts.CommentID, opts.NotificationAuthorID, opts.ReceiverID); err != nil {
  log.Error("Was unable to create issue notification: %v", err)
  }
  }
+ return nil
 }
 
 func (ns *notificationService) Run() {

modules/queue/bytefifo.go

@@ -16,6 +16,8 @@ type ByteFIFO interface {
  Pop(ctx context.Context) ([]byte, error)
  // Close this fifo
  Close() error
+ // PushBack pushes data back to the top of the fifo
+ PushBack(ctx context.Context, data []byte) error
 }
 
 // UniqueByteFIFO defines a FIFO that Uniques its contents
@@ -50,6 +52,11 @@ func (*DummyByteFIFO) Len(ctx context.Context) int64 {
  return 0
 }
 
+// PushBack pushes data back to the top of the fifo
+func (*DummyByteFIFO) PushBack(ctx context.Context, data []byte) error {
+ return nil
+}
+
 var _ UniqueByteFIFO = &DummyUniqueByteFIFO{}
 
 // DummyUniqueByteFIFO represents a dummy unique fifo

modules/queue/manager.go

@@ -54,6 +54,18 @@ type Flushable interface {
  IsEmpty() bool
 }
 
+// Pausable represents a pool or queue that is Pausable
+type Pausable interface {
+ // IsPaused will return if the pool or queue is paused
+ IsPaused() bool
+ // Pause will pause the pool or queue
+ Pause()
+ // Resume will resume the pool or queue
+ Resume()
+ // IsPausedIsResumed will return a bool indicating if the pool or queue is paused and a channel that will be closed when it is resumed
+ IsPausedIsResumed() (paused, resumed <-chan struct{})
+}
+
 // ManagedPool is a simple interface to get certain details from a worker pool
 type ManagedPool interface {
  // AddWorkers adds a number of worker as group to the pool with the provided timeout. A CancelFunc is provided to cancel the group
@@ -192,6 +204,14 @@ func (m *Manager) FlushAll(baseCtx context.Context, timeout time.Duration) error
  wg.Done()
  continue
  }
+ if pausable, ok := mq.Managed.(Pausable); ok {
+ // no point flushing paused queues
+ if pausable.IsPaused() {
+ wg.Done()
+ continue
+ }
+ }
+
  allEmpty = false
  if flushable, ok := mq.Managed.(Flushable); ok {
  log.Debug("Flushing (flushable) queue: %s", mq.Name)
@@ -215,7 +235,7 @@ func (m *Manager) FlushAll(baseCtx context.Context, timeout time.Duration) error
  log.Debug("All queues are empty")
  break
  }
- // Ensure there are always at least 100ms between loops but not more if we've actually been doing some flushign
+ // Ensure there are always at least 100ms between loops but not more if we've actually been doing some flushing
  // but don't delay cancellation here.
  select {
  case <-ctx.Done():
@@ -298,6 +318,12 @@ func (q *ManagedQueue) AddWorkers(number int, timeout time.Duration) context.Can
  return nil
 }
 
+// Flushable returns true if the queue is flushable
+func (q *ManagedQueue) Flushable() bool {
+ _, ok := q.Managed.(Flushable)
+ return ok
+}
+
 // Flush flushes the queue with a timeout
 func (q *ManagedQueue) Flush(timeout time.Duration) error {
  if flushable, ok := q.Managed.(Flushable); ok {
@@ -315,6 +341,34 @@ func (q *ManagedQueue) IsEmpty() bool {
  return true
 }
 
+// Pausable returns whether the queue is Pausable
+func (q *ManagedQueue) Pausable() bool {
+ _, ok := q.Managed.(Pausable)
+ return ok
+}
+
+// Pause pauses the queue
+func (q *ManagedQueue) Pause() {
+ if pausable, ok := q.Managed.(Pausable); ok {
+ pausable.Pause()
+ }
+}
+
+// IsPaused reveals if the queue is paused
+func (q *ManagedQueue) IsPaused() bool {
+ if pausable, ok := q.Managed.(Pausable); ok {
+ return pausable.IsPaused()
+ }
+ return false
+}
+
+// Resume resumes the queue
+func (q *ManagedQueue) Resume() {
+ if pausable, ok := q.Managed.(Pausable); ok {
+ pausable.Resume()
+ }
+}
+
 // NumberOfWorkers returns the number of workers in the queue
 func (q *ManagedQueue) NumberOfWorkers() int {
  if pool, ok := q.Managed.(ManagedPool); ok {

modules/queue/queue.go

@@ -36,7 +36,7 @@ type Type string
 type Data interface{}
 
 // HandlerFunc is a function that takes a variable amount of data and processes it
-type HandlerFunc func(...Data)
+type HandlerFunc func(...Data) (unhandled []Data)
 
 // NewQueueFunc is a function that creates a queue
 type NewQueueFunc func(handler HandlerFunc, config, exemplar interface{}) (Queue, error)
@@ -61,6 +61,12 @@ type Queue interface {
  Push(Data) error
 }
 
+// PushBackable queues can be pushed back to
+type PushBackable interface {
+ // PushBack pushes data back to the top of the fifo
+ PushBack(Data) error
+}
+
 // DummyQueueType is the type for the dummy queue
 const DummyQueueType Type = "dummy"