summaryrefslogtreecommitdiffstats
path: root/modules/indexer/code/queue.go
blob: 82cd8ded53937dcef5bdbdd31fbda6a006232d15 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
// Copyright 2019 The Gitea Authors. All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.

package code

import (
	"os"

	"code.gitea.io/gitea/models"
	"code.gitea.io/gitea/modules/graceful"
	"code.gitea.io/gitea/modules/log"
	"code.gitea.io/gitea/modules/setting"
)

type repoIndexerOperation struct {
	repoID   int64
	deleted  bool
	watchers []chan<- error
}

var repoIndexerOperationQueue chan repoIndexerOperation

func processRepoIndexerOperationQueue(indexer Indexer) {
	repoIndexerOperationQueue = make(chan repoIndexerOperation, setting.Indexer.UpdateQueueLength)
	for {
		select {
		case op := <-repoIndexerOperationQueue:
			var err error
			if op.deleted {
				if err = indexer.Delete(op.repoID); err != nil {
					log.Error("indexer.Delete: %v", err)
				}
			} else {
				if err = indexer.Index(op.repoID); err != nil {
					log.Error("indexer.Index: %v", err)
				}
			}
			for _, watcher := range op.watchers {
				watcher <- err
			}
		case <-graceful.GetManager().IsShutdown():
			log.Info("PID: %d Repository indexer queue processing stopped", os.Getpid())
			return
		}
	}
}

// DeleteRepoFromIndexer remove all of a repository's entries from the indexer
func DeleteRepoFromIndexer(repo *models.Repository, watchers ...chan<- error) {
	addOperationToQueue(repoIndexerOperation{repoID: repo.ID, deleted: true, watchers: watchers})
}

// UpdateRepoIndexer update a repository's entries in the indexer
func UpdateRepoIndexer(repo *models.Repository, watchers ...chan<- error) {
	addOperationToQueue(repoIndexerOperation{repoID: repo.ID, deleted: false, watchers: watchers})
}

func addOperationToQueue(op repoIndexerOperation) {
	if !setting.Indexer.RepoIndexerEnabled {
		return
	}
	select {
	case repoIndexerOperationQueue <- op:
		break
	default:
		go func() {
			repoIndexerOperationQueue <- op
		}()
	}
}

// populateRepoIndexer populate the repo indexer with pre-existing data. This
// should only be run when the indexer is created for the first time.
func populateRepoIndexer() {
	log.Info("Populating the repo indexer with existing repositories")

	isShutdown := graceful.GetManager().IsShutdown()

	exist, err := models.IsTableNotEmpty("repository")
	if err != nil {
		log.Fatal("System error: %v", err)
	} else if !exist {
		return
	}

	// if there is any existing repo indexer metadata in the DB, delete it
	// since we are starting afresh. Also, xorm requires deletes to have a
	// condition, and we want to delete everything, thus 1=1.
	if err := models.DeleteAllRecords("repo_indexer_status"); err != nil {
		log.Fatal("System error: %v", err)
	}

	var maxRepoID int64
	if maxRepoID, err = models.GetMaxID("repository"); err != nil {
		log.Fatal("System error: %v", err)
	}

	// start with the maximum existing repo ID and work backwards, so that we
	// don't include repos that are created after gitea starts; such repos will
	// already be added to the indexer, and we don't need to add them again.
	for maxRepoID > 0 {
		select {
		case <-isShutdown:
			log.Info("Repository Indexer population shutdown before completion")
			return
		default:
		}
		ids, err := models.GetUnindexedRepos(maxRepoID, 0, 50)
		if err != nil {
			log.Error("populateRepoIndexer: %v", err)
			return
		} else if len(ids) == 0 {
			break
		}
		for _, id := range ids {
			select {
			case <-isShutdown:
				log.Info("Repository Indexer population shutdown before completion")
				return
			default:
			}
			repoIndexerOperationQueue <- repoIndexerOperation{
				repoID:  id,
				deleted: false,
			}
			maxRepoID = id - 1
		}
	}
	log.Info("Done (re)populating the repo indexer with existing repositories")
}