Rewrite queue

cmd/hook.go

@@ -18,7 +18,6 @@ import (
 	"code.gitea.io/gitea/modules/private"
 	repo_module "code.gitea.io/gitea/modules/repository"
 	"code.gitea.io/gitea/modules/setting"
-	"code.gitea.io/gitea/modules/util"
 
 	"github.com/urfave/cli"
 )
@@ -141,7 +140,7 @@ func (d *delayWriter) Close() error {
 	if d == nil {
 		return nil
 	}
-	stopped := util.StopTimer(d.timer)
+	stopped := d.timer.Stop()
 	if stopped || d.buf == nil {
 		return nil
 	}

custom/conf/app.example.ini

@@ -926,12 +926,6 @@ ROUTER = console
 ;; Global limit of repositories per user, applied at creation time. -1 means no limit
 ;MAX_CREATION_LIMIT = -1
 ;;
-;; Mirror sync queue length, increase if mirror syncing starts hanging (DEPRECATED: please use [queue.mirror] LENGTH instead)
-;MIRROR_QUEUE_LENGTH = 1000
-;;
-;; Patch test queue length, increase if pull request patch testing starts hanging (DEPRECATED: please use [queue.pr_patch_checker] LENGTH instead)
-;PULL_REQUEST_QUEUE_LENGTH = 1000
-;;
 ;; Preferred Licenses to place at the top of the List
 ;; The name here must match the filename in options/license or custom/options/license
 ;PREFERRED_LICENSES = Apache License 2.0,MIT License
@@ -1376,22 +1370,6 @@ ROUTER = console
 ;; Set to -1 to disable timeout.
 ;STARTUP_TIMEOUT = 30s
 ;;
-;; Issue indexer queue, currently support: channel, levelqueue or redis, default is levelqueue (deprecated - use [queue.issue_indexer])
-;ISSUE_INDEXER_QUEUE_TYPE = levelqueue; **DEPRECATED** use settings in `[queue.issue_indexer]`.
-;;
-;; When ISSUE_INDEXER_QUEUE_TYPE is levelqueue, this will be the path where the queue will be saved.
-;; This can be overridden by `ISSUE_INDEXER_QUEUE_CONN_STR`.
-;; default is queues/common
-;ISSUE_INDEXER_QUEUE_DIR = queues/common; **DEPRECATED** use settings in `[queue.issue_indexer]`. Relative paths will be made absolute against `%(APP_DATA_PATH)s`.
-;;
-;; When `ISSUE_INDEXER_QUEUE_TYPE` is `redis`, this will store the redis connection string.
-;; When `ISSUE_INDEXER_QUEUE_TYPE` is `levelqueue`, this is a directory or additional options of
-;; the form `leveldb://path/to/db?option=value&....`, and overrides `ISSUE_INDEXER_QUEUE_DIR`.
-;ISSUE_INDEXER_QUEUE_CONN_STR = "addrs=127.0.0.1:6379 db=0"; **DEPRECATED** use settings in `[queue.issue_indexer]`.
-;;
-;; Batch queue number, default is 20
-;ISSUE_INDEXER_QUEUE_BATCH_NUMBER = 20; **DEPRECATED** use settings in `[queue.issue_indexer]`.
-
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Repository Indexer settings
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@@ -1418,8 +1396,6 @@ ROUTER = console
 ;; A comma separated list of glob patterns to exclude from the index; ; default is empty
 ;REPO_INDEXER_EXCLUDE =
 ;;
-;;
-;UPDATE_BUFFER_LEN = 20; **DEPRECATED** use settings in `[queue.issue_indexer]`.
 ;MAX_FILE_SIZE = 1048576
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@@ -1449,37 +1425,16 @@ ROUTER = console
 ;; Connection string for redis queues this will store the redis connection string.
 ;; When `TYPE` is `persistable-channel`, this provides a directory for the underlying leveldb
 ;; or additional options of the form `leveldb://path/to/db?option=value&....`, and will override `DATADIR`.
-;CONN_STR = "addrs=127.0.0.1:6379 db=0"
+;CONN_STR = "redis://127.0.0.1:6379/0"
 ;;
 ;; Provides the suffix of the default redis/disk queue name - specific queues can be overridden within in their [queue.name] sections.
 ;QUEUE_NAME = "_queue"
 ;;
 ;; Provides the suffix of the default redis/disk unique queue set name - specific queues can be overridden within in their [queue.name] sections.
 ;SET_NAME = "_unique"
 ;;
-;; If the queue cannot be created at startup - level queues may need a timeout at startup - wrap the queue:
-;WRAP_IF_NECESSARY = true
-;;
-;; Attempt to create the wrapped queue at max
-;MAX_ATTEMPTS = 10
-;;
-;; Timeout queue creation
-;TIMEOUT = 15m30s
-;;
-;; Create a pool with this many workers
-;WORKERS = 0
-;;
 ;; Dynamically scale the worker pool to at this many workers
 ;MAX_WORKERS = 10
-;;
-;; Add boost workers when the queue blocks for BLOCK_TIMEOUT
-;BLOCK_TIMEOUT = 1s
-;;
-;; Remove the boost workers after BOOST_TIMEOUT
-;BOOST_TIMEOUT = 5m
-;;
-;; During a boost add BOOST_WORKERS
-;BOOST_WORKERS = 1
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

docs/content/doc/administration/config-cheat-sheet.en-us.md

@@ -89,10 +89,6 @@ In addition there is _`StaticRootPath`_ which can be set as a built-in at build
 - `DEFAULT_PUSH_CREATE_PRIVATE`: **true**: Default private when creating a new repository with push-to-create.
 - `MAX_CREATION_LIMIT`: **-1**: Global maximum creation limit of repositories per user,
    `-1` means no limit.
-- `PULL_REQUEST_QUEUE_LENGTH`: **1000**: Length of pull request patch test queue, make it. **DEPRECATED** use `LENGTH` in `[queue.pr_patch_checker]`.
-   as large as possible. Use caution when editing this value.
-- `MIRROR_QUEUE_LENGTH`: **1000**: Patch test queue length, increase if pull request patch
-   testing starts hanging. **DEPRECATED** use `LENGTH` in `[queue.mirror]`.
 - `PREFERRED_LICENSES`: **Apache License 2.0,MIT License**: Preferred Licenses to place at
    the top of the list. Name must match file name in options/license or custom/options/license.
 - `DISABLE_HTTP_GIT`: **false**: Disable the ability to interact with repositories over the
@@ -466,10 +462,6 @@ relation to port exhaustion.
 - `ISSUE_INDEXER_NAME`: **gitea_issues**: Issue indexer name, available when ISSUE_INDEXER_TYPE is elasticsearch
 - `ISSUE_INDEXER_PATH`: **indexers/issues.bleve**: Index file used for issue search; available when ISSUE_INDEXER_TYPE is bleve and elasticsearch. Relative paths will be made absolute against _`AppWorkPath`_.
 - The next 4 configuration values are deprecated and should be set in `queue.issue_indexer` however are kept for backwards compatibility:
-- `ISSUE_INDEXER_QUEUE_TYPE`: **levelqueue**: Issue indexer queue, currently supports:`channel`, `levelqueue`, `redis`. **DEPRECATED** use settings in `[queue.issue_indexer]`.
-- `ISSUE_INDEXER_QUEUE_DIR`: **queues/common**: When `ISSUE_INDEXER_QUEUE_TYPE` is `levelqueue`, this will be the path where the queue will be saved. **DEPRECATED** use settings in `[queue.issue_indexer]`. Relative paths will be made absolute against `%(APP_DATA_PATH)s`.
-- `ISSUE_INDEXER_QUEUE_CONN_STR`: **addrs=127.0.0.1:6379 db=0**: When `ISSUE_INDEXER_QUEUE_TYPE` is `redis`, this will store the redis connection string. When `ISSUE_INDEXER_QUEUE_TYPE` is `levelqueue`, this is a directory or additional options of the form `leveldb://path/to/db?option=value&....`, and overrides `ISSUE_INDEXER_QUEUE_DIR`. **DEPRECATED** use settings in `[queue.issue_indexer]`.
-- `ISSUE_INDEXER_QUEUE_BATCH_NUMBER`: **20**: Batch queue number. **DEPRECATED** use settings in `[queue.issue_indexer]`.
 
 - `REPO_INDEXER_ENABLED`: **false**: Enables code search (uses a lot of disk space, about 6 times more than the repository size).
 - `REPO_INDEXER_TYPE`: **bleve**: Code search engine type, could be `bleve` or `elasticsearch`.
@@ -480,7 +472,6 @@ relation to port exhaustion.
 - `REPO_INDEXER_INCLUDE`: **empty**: A comma separated list of glob patterns (see https://github.com/gobwas/glob) to **include** in the index. Use `**.txt` to match any files with .txt extension. An empty list means include all files.
 - `REPO_INDEXER_EXCLUDE`: **empty**: A comma separated list of glob patterns (see https://github.com/gobwas/glob) to **exclude** from the index. Files that match this list will not be indexed, even if they match in `REPO_INDEXER_INCLUDE`.
 - `REPO_INDEXER_EXCLUDE_VENDORED`: **true**: Exclude vendored files from index.
-- `UPDATE_BUFFER_LEN`: **20**: Buffer length of index request. **DEPRECATED** use settings in `[queue.issue_indexer]`.
 - `MAX_FILE_SIZE`: **1048576**: Maximum size in bytes of files to be indexed.
 - `STARTUP_TIMEOUT`: **30s**: If the indexer takes longer than this timeout to start - fail. (This timeout will be added to the hammer time above for child processes - as bleve will not start until the previous parent is shutdown.) Set to -1 to never timeout.
 
@@ -496,15 +487,7 @@ Configuration at `[queue]` will set defaults for queues with overrides for indiv
 - `QUEUE_NAME`: **_queue**: The suffix for default redis and disk queue name. Individual queues will default to **`name`**`QUEUE_NAME` but can be overridden in the specific `queue.name` section.
 - `SET_NAME`: **_unique**: The suffix that will be added to the default redis and disk queue `set` name for unique queues. Individual queues will default to
  **`name`**`QUEUE_NAME`_`SET_NAME`_ but can be overridden in the specific `queue.name` section.
-- `WRAP_IF_NECESSARY`: **true**: Will wrap queues with a timeoutable queue if the selected queue is not ready to be created - (Only relevant for the level queue.)
-- `MAX_ATTEMPTS`: **10**: Maximum number of attempts to create the wrapped queue
-- `TIMEOUT`: **GRACEFUL_HAMMER_TIME + 30s**: Timeout the creation of the wrapped queue if it takes longer than this to create.
-- Queues by default come with a dynamically scaling worker pool. The following settings configure this:
-- `WORKERS`: **0**: Number of initial workers for the queue.
 - `MAX_WORKERS`: **10**: Maximum number of worker go-routines for the queue.
-- `BLOCK_TIMEOUT`: **1s**: If the queue blocks for this time, boost the number of workers - the `BLOCK_TIMEOUT` will then be doubled before boosting again whilst the boost is ongoing.
-- `BOOST_TIMEOUT`: **5m**: Boost workers will timeout after this long.
-- `BOOST_WORKERS`: **1**: This many workers will be added to the worker pool if there is a boost.
 
 Gitea creates the following non-unique queues:
 

models/unittest/testdb.go

@@ -89,6 +89,7 @@ func MainTest(m *testing.M, testOpts *TestOptions) {
 	}
 
 	if err = CreateTestEngine(opts); err != nil {
+		_, _ = fmt.Fprintln(os.Stderr, `sqlite3 requires: import _ "github.com/mattn/go-sqlite3" or -tags sqlite,sqlite_unlock_notify`)
 		fatalTestError("Error creating test engine: %v\n", err)
 	}
 

modules/indexer/code/bleve.go

@@ -273,10 +273,6 @@ func (b *BleveIndexer) Close() {
 	log.Info("PID: %d Repository Indexer closed", os.Getpid())
 }
 
-// SetAvailabilityChangeCallback does nothing
-func (b *BleveIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
-}
-
 // Ping does nothing
 func (b *BleveIndexer) Ping() bool {
 	return true

modules/indexer/code/elastic_search.go

@@ -42,12 +42,11 @@ var _ Indexer = &ElasticSearchIndexer{}
 
 // ElasticSearchIndexer implements Indexer interface
 type ElasticSearchIndexer struct {
-	client               *elastic.Client
-	indexerAliasName     string
-	available            bool
-	availabilityCallback func(bool)
-	stopTimer            chan struct{}
-	lock                 sync.RWMutex
+	client           *elastic.Client
+	indexerAliasName string
+	available        bool
+	stopTimer        chan struct{}
+	lock             sync.RWMutex
 }
 
 type elasticLogger struct {
@@ -198,13 +197,6 @@ func (b *ElasticSearchIndexer) init() (bool, error) {
 	return exists, nil
 }
 
-// SetAvailabilityChangeCallback sets callback that will be triggered when availability changes
-func (b *ElasticSearchIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
-	b.lock.Lock()
-	defer b.lock.Unlock()
-	b.availabilityCallback = callback
-}
-
 // Ping checks if elastic is available
 func (b *ElasticSearchIndexer) Ping() bool {
 	b.lock.RLock()
@@ -529,8 +521,4 @@ func (b *ElasticSearchIndexer) setAvailability(available bool) {
 	}
 
 	b.available = available
-	if b.availabilityCallback != nil {
-		// Call the callback from within the lock to ensure that the ordering remains correct
-		b.availabilityCallback(b.available)
-	}
 }

modules/indexer/code/indexer.go

@@ -44,7 +44,6 @@ type SearchResultLanguages struct {
 // Indexer defines an interface to index and search code contents
 type Indexer interface {
 	Ping() bool
-	SetAvailabilityChangeCallback(callback func(bool))
 	Index(ctx context.Context, repo *repo_model.Repository, sha string, changes *repoChanges) error
 	Delete(repoID int64) error
 	Search(ctx context.Context, repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error)
@@ -81,7 +80,7 @@ type IndexerData struct {
 	RepoID int64
 }
 
-var indexerQueue queue.UniqueQueue
+var indexerQueue *queue.WorkerPoolQueue[*IndexerData]
 
 func index(ctx context.Context, indexer Indexer, repoID int64) error {
 	repo, err := repo_model.GetRepositoryByID(ctx, repoID)
@@ -137,37 +136,46 @@ func Init() {
 	// Create the Queue
 	switch setting.Indexer.RepoType {
 	case "bleve", "elasticsearch":
-		handler := func(data ...queue.Data) []queue.Data {
+		handler := func(items ...*IndexerData) []*IndexerData {
 			idx, err := indexer.get()
 			if idx == nil || err != nil {
 				log.Error("Codes indexer handler: unable to get indexer!")
-				return data
+				return items
+			}
+			if !idx.Ping() {
+				log.Error("Code indexer handler: indexer is unavailable.")
+				return items
 			}
 
-			unhandled := make([]queue.Data, 0, len(data))
-			for _, datum := range data {
-				indexerData, ok := datum.(*IndexerData)
-				if !ok {
-					log.Error("Unable to process provided datum: %v - not possible to cast to IndexerData", datum)
-					continue
-				}
+			// the old logic did: if indexer.Ping() { return nil }, skip all failed items
+
+			for _, indexerData := range items {
 				log.Trace("IndexerData Process Repo: %d", indexerData.RepoID)
 
+				// FIXME: it seems there is a bug in `CatFileBatch` or `nio.Pipe`, which will cause the process to hang forever in rare cases
+				/*
+					sync.(*Cond).Wait(cond.go:70)
+					github.com/djherbis/nio/v3.(*PipeReader).Read(sync.go:106)
+					bufio.(*Reader).fill(bufio.go:106)
+					bufio.(*Reader).ReadSlice(bufio.go:372)
+					bufio.(*Reader).collectFragments(bufio.go:447)
+					bufio.(*Reader).ReadString(bufio.go:494)
+					code.gitea.io/gitea/modules/git.ReadBatchLine(batch_reader.go:149)
+					code.gitea.io/gitea/modules/indexer/code.(*BleveIndexer).addUpdate(bleve.go:214)
+					code.gitea.io/gitea/modules/indexer/code.(*BleveIndexer).Index(bleve.go:296)
+					code.gitea.io/gitea/modules/indexer/code.(*wrappedIndexer).Index(wrapped.go:74)
+					code.gitea.io/gitea/modules/indexer/code.index(indexer.go:105)
+				*/
 				if err := index(ctx, indexer, indexerData.RepoID); err != nil {
 					if !setting.IsInTesting {
 						log.Error("indexer index error for repo %v: %v", indexerData.RepoID, err)
 					}
-					if indexer.Ping() {
-						continue
-					}
-					// Add back to queue
-					unhandled = append(unhandled, datum)
 				}
 			}
-			return unhandled
+			return nil
 		}
 
-		indexerQueue = queue.CreateUniqueQueue("code_indexer", handler, &IndexerData{})
+		indexerQueue = queue.CreateUniqueQueue("code_indexer", handler)
 		if indexerQueue == nil {
 			log.Fatal("Unable to create codes indexer queue")
 		}
@@ -224,18 +232,6 @@ func Init() {
 
 		indexer.set(rIndexer)
 
-		if queue, ok := indexerQueue.(queue.Pausable); ok {
-			rIndexer.SetAvailabilityChangeCallback(func(available bool) {
-				if !available {
-					log.Info("Code index queue paused")
-					queue.Pause()
-				} else {
-					log.Info("Code index queue resumed")
-					queue.Resume()
-				}
-			})
-		}
-
 		// Start processing the queue
 		go graceful.GetManager().RunWithShutdownFns(indexerQueue.Run)
 

modules/indexer/code/wrapped.go

@@ -56,16 +56,6 @@ func (w *wrappedIndexer) get() (Indexer, error) {
 	return w.internal, nil
 }
 
-// SetAvailabilityChangeCallback sets callback that will be triggered when availability changes
-func (w *wrappedIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
-	indexer, err := w.get()
-	if err != nil {
-		log.Error("Failed to get indexer: %v", err)
-		return
-	}
-	indexer.SetAvailabilityChangeCallback(callback)
-}
-
 // Ping checks if elastic is available
 func (w *wrappedIndexer) Ping() bool {
 	indexer, err := w.get()

modules/indexer/issues/bleve.go

@@ -187,10 +187,6 @@ func (b *BleveIndexer) Init() (bool, error) {
 	return false, err
 }
 
-// SetAvailabilityChangeCallback does nothing
-func (b *BleveIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
-}
-
 // Ping does nothing
 func (b *BleveIndexer) Ping() bool {
 	return true

modules/indexer/issues/db.go

@@ -18,10 +18,6 @@ func (i *DBIndexer) Init() (bool, error) {
 	return false, nil
 }
 
-// SetAvailabilityChangeCallback dummy function
-func (i *DBIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
-}
-
 // Ping checks if database is available
 func (i *DBIndexer) Ping() bool {
 	return db.GetEngine(db.DefaultContext).Ping() != nil

modules/indexer/issues/elastic_search.go

@@ -22,12 +22,11 @@ var _ Indexer = &ElasticSearchIndexer{}
 
 // ElasticSearchIndexer implements Indexer interface
 type ElasticSearchIndexer struct {
-	client               *elastic.Client
-	indexerName          string
-	available            bool
-	availabilityCallback func(bool)
-	stopTimer            chan struct{}
-	lock                 sync.RWMutex
+	client      *elastic.Client
+	indexerName string
+	available   bool
+	stopTimer   chan struct{}
+	lock        sync.RWMutex
 }
 
 type elasticLogger struct {
@@ -138,13 +137,6 @@ func (b *ElasticSearchIndexer) Init() (bool, error) {
 	return true, nil
 }
 
-// SetAvailabilityChangeCallback sets callback that will be triggered when availability changes
-func (b *ElasticSearchIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
-	b.lock.Lock()
-	defer b.lock.Unlock()
-	b.availabilityCallback = callback
-}
-
 // Ping checks if elastic is available
 func (b *ElasticSearchIndexer) Ping() bool {
 	b.lock.RLock()
@@ -305,8 +297,4 @@ func (b *ElasticSearchIndexer) setAvailability(available bool) {
 	}
 
 	b.available = available
-	if b.availabilityCallback != nil {
-		// Call the callback from within the lock to ensure that the ordering remains correct
-		b.availabilityCallback(b.available)
-	}
 }