Graceful: Xorm, RepoIndexer, Cron and Others

models/pull_list.go

@@ -68,11 +68,12 @@ func GetUnmergedPullRequestsByBaseInfo(repoID int64, branch string) ([]*PullRequ
 		Find(&prs)
 }
 
-// GetPullRequestsByCheckStatus returns all pull requests according the special checking status.
-func GetPullRequestsByCheckStatus(status PullRequestStatus) ([]*PullRequest, error) {
-	prs := make([]*PullRequest, 0, 10)
-	return prs, x.
+// GetPullRequestIDsByCheckStatus returns all pull requests according the special checking status.
+func GetPullRequestIDsByCheckStatus(status PullRequestStatus) ([]int64, error) {
+	prs := make([]int64, 0, 10)
+	return prs, x.Table("pull_request").
 		Where("status=?", status).
+		Cols("pull_request.id").
 		Find(&prs)
 }
 

models/repo_indexer.go

@@ -5,6 +5,8 @@
 package models
 
 import (
+	"fmt"
+
 	"xorm.io/builder"
 )
 
@@ -22,10 +24,10 @@ func GetUnindexedRepos(maxRepoID int64, page, pageSize int) ([]int64, error) {
 	cond := builder.Cond(builder.IsNull{
 		"repo_indexer_status.id",
 	})
-	sess := x.Table("repo").Join("LEFT OUTER", "repo_indexer_status", "repo.id = repo_indexer_status.repoID")
+	sess := x.Table("repository").Join("LEFT OUTER", "repo_indexer_status", "repository.id = repo_indexer_status.repo_id")
 	if maxRepoID > 0 {
 		cond = builder.And(cond, builder.Lte{
-			"repo.id": maxRepoID,
+			"repository.id": maxRepoID,
 		})
 	}
 	if page >= 0 && pageSize > 0 {
@@ -36,7 +38,7 @@ func GetUnindexedRepos(maxRepoID int64, page, pageSize int) ([]int64, error) {
 		sess.Limit(pageSize, start)
 	}
 
-	sess.Where(cond).Cols("repo.id")
+	sess.Where(cond).Cols("repository.id")
 	err := sess.Find(&ids)
 	return ids, err
 }
@@ -60,15 +62,18 @@ func (repo *Repository) GetIndexerStatus() error {
 // UpdateIndexerStatus updates indexer status
 func (repo *Repository) UpdateIndexerStatus(sha string) error {
 	if err := repo.GetIndexerStatus(); err != nil {
-		return err
+		return fmt.Errorf("UpdateIndexerStatus: Unable to getIndexerStatus for repo: %s/%s Error: %v", repo.MustOwnerName(), repo.Name, err)
 	}
 	if len(repo.IndexerStatus.CommitSha) == 0 {
 		repo.IndexerStatus.CommitSha = sha
 		_, err := x.Insert(repo.IndexerStatus)
-		return err
+		return fmt.Errorf("UpdateIndexerStatus: Unable to insert repoIndexerStatus for repo: %s/%s Sha: %s Error: %v", repo.MustOwnerName(), repo.Name, sha, err)
 	}
 	repo.IndexerStatus.CommitSha = sha
 	_, err := x.ID(repo.IndexerStatus.ID).Cols("commit_sha").
 		Update(repo.IndexerStatus)
-	return err
+	if err != nil {
+		return fmt.Errorf("UpdateIndexerStatus: Unable to update repoIndexerStatus for repo: %s/%s Sha: %s Error: %v", repo.MustOwnerName(), repo.Name, sha, err)
+	}
+	return nil
 }

modules/indexer/code/bleve.go

@@ -124,33 +124,33 @@ func populateRepoIndexer(maxRepoID int64) {
 func updateRepoIndexer(repoID int64) error {
 	repo, err := models.GetRepositoryByID(repoID)
 	if err != nil {
-		return err
+		return fmt.Errorf("UpdateRepoIndexer: Unable to GetRepositoryByID: %d, Error: %v", repoID, err)
 	}
 
 	sha, err := getDefaultBranchSha(repo)
 	if err != nil {
-		return err
+		return fmt.Errorf("UpdateRepoIndexer: Unable to GetDefaultBranchSha for: %s/%s, Error: %v", repo.MustOwnerName(), repo.Name, err)
 	}
 	changes, err := getRepoChanges(repo, sha)
 	if err != nil {
-		return err
+		return fmt.Errorf("UpdateRepoIndexer: Unable to GetRepoChanges for: %s/%s Sha: %s Error: %v", repo.MustOwnerName(), repo.Name, sha, err)
 	} else if changes == nil {
 		return nil
 	}
 
 	batch := RepoIndexerBatch()
 	for _, update := range changes.Updates {
 		if err := addUpdate(update, repo, batch); err != nil {
-			return err
+			return fmt.Errorf("UpdateRepoIndexer: Unable to addUpdate to: %s/%s Sha: %s, update: %s(%s) Error: %v", repo.MustOwnerName(), repo.Name, sha, update.Filename, update.BlobSha, err)
 		}
 	}
 	for _, filename := range changes.RemovedFilenames {
 		if err := addDelete(filename, repo, batch); err != nil {
-			return err
+			return fmt.Errorf("UpdateRepoIndexer: Unable to addDelete to: %s/%s Sha: %s, filename: %s Error: %v", repo.MustOwnerName(), repo.Name, sha, filename, err)
 		}
 	}
 	if err = batch.Flush(); err != nil {
-		return err
+		return fmt.Errorf("UpdateRepoIndexer: Unable to flush batch to indexer for repo: %s/%s Error: %v", repo.MustOwnerName(), repo.Name, err)
 	}
 	return repo.UpdateIndexerStatus(sha)
 }

modules/sync/unique_queue.go

@@ -6,8 +6,6 @@
 package sync
 
 import (
-	"context"
-
 	"github.com/unknwon/com"
 )
 
@@ -18,8 +16,9 @@ import (
 // This queue is particularly useful for preventing duplicated task
 // of same purpose.
 type UniqueQueue struct {
-	table *StatusTable
-	queue chan string
+	table  *StatusTable
+	queue  chan string
+	closed chan struct{}
 }
 
 // NewUniqueQueue initializes and returns a new UniqueQueue object.
@@ -29,11 +28,43 @@ func NewUniqueQueue(queueLength int) *UniqueQueue {
 	}
 
 	return &UniqueQueue{
-		table: NewStatusTable(),
-		queue: make(chan string, queueLength),
+		table:  NewStatusTable(),
+		queue:  make(chan string, queueLength),
+		closed: make(chan struct{}),
+	}
+}
+
+// Close closes this queue
+func (q *UniqueQueue) Close() {
+	select {
+	case <-q.closed:
+	default:
+		q.table.lock.Lock()
+		select {
+		case <-q.closed:
+		default:
+			close(q.closed)
+		}
+		q.table.lock.Unlock()
 	}
 }
 
+// IsClosed returns a channel that is closed when this Queue is closed
+func (q *UniqueQueue) IsClosed() <-chan struct{} {
+	return q.closed
+}
+
+// IDs returns the current ids in the pool
+func (q *UniqueQueue) IDs() []interface{} {
+	q.table.lock.Lock()
+	defer q.table.lock.Unlock()
+	ids := make([]interface{}, 0, len(q.table.pool))
+	for id := range q.table.pool {
+		ids = append(ids, id)
+	}
+	return ids
+}
+
 // Queue returns channel of queue for retrieving instances.
 func (q *UniqueQueue) Queue() <-chan string {
 	return q.queue
@@ -48,15 +79,8 @@ func (q *UniqueQueue) Exist(id interface{}) bool {
 // AddFunc adds new instance to the queue with a custom runnable function,
 // the queue is blocked until the function exits.
 func (q *UniqueQueue) AddFunc(id interface{}, fn func()) {
-	q.AddCtxFunc(context.Background(), id, fn)
-}
-
-// AddCtxFunc adds new instance to the queue with a custom runnable function,
-// the queue is blocked until the function exits. If the context is done before
-// the id is added to the queue it will not be added and false will be returned.
-func (q *UniqueQueue) AddCtxFunc(ctx context.Context, id interface{}, fn func()) bool {
 	if q.Exist(id) {
-		return true
+		return
 	}
 
 	idStr := com.ToStr(id)
@@ -67,10 +91,10 @@ func (q *UniqueQueue) AddCtxFunc(ctx context.Context, id interface{}, fn func())
 	}
 	q.table.lock.Unlock()
 	select {
-	case <-ctx.Done():
-		return false
+	case <-q.closed:
+		return
 	case q.queue <- idStr:
-		return true
+		return
 	}
 }
 
@@ -79,11 +103,6 @@ func (q *UniqueQueue) Add(id interface{}) {
 	q.AddFunc(id, nil)
 }
 
-// AddCtx adds new instance to the queue with a context - if the context is done before the id is added to the queue it is cancelled
-func (q *UniqueQueue) AddCtx(ctx context.Context, id interface{}) bool {
-	return q.AddCtxFunc(ctx, id, nil)
-}
-
 // Remove removes instance from the queue.
 func (q *UniqueQueue) Remove(id interface{}) {
 	q.table.Stop(com.ToStr(id))

modules/webhook/deliver.go

@@ -177,6 +177,7 @@ func DeliverHooks(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
+			hookQueue.Close()
 			return
 		case repoIDStr := <-hookQueue.Queue():
 			log.Trace("DeliverHooks [repo_id: %v]", repoIDStr)

services/mirror/mirror.go

@@ -308,7 +308,7 @@ func Update(ctx context.Context) {
 		case <-ctx.Done():
 			return fmt.Errorf("Aborted due to shutdown")
 		default:
-			_ = mirrorQueue.AddCtx(ctx, m.RepoID)
+			mirrorQueue.Add(m.RepoID)
 			return nil
 		}
 	}); err != nil {
@@ -323,6 +323,7 @@ func SyncMirrors(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
+			mirrorQueue.Close()
 			return
 		case repoID := <-mirrorQueue.Queue():
 			syncMirror(repoID)

services/pull/check.go

@@ -154,37 +154,22 @@ func manuallyMerged(pr *models.PullRequest) bool {
 // TestPullRequests checks and tests untested patches of pull requests.
 // TODO: test more pull requests at same time.
 func TestPullRequests(ctx context.Context) {
-	prs, err := models.GetPullRequestsByCheckStatus(models.PullRequestStatusChecking)
-	if err != nil {
-		log.Error("Find Checking PRs: %v", err)
-		return
-	}
-
-	var checkedPRs = make(map[int64]struct{})
 
-	// Update pull request status.
-	for _, pr := range prs {
-		checkedPRs[pr.ID] = struct{}{}
-		if err := pr.GetBaseRepo(); err != nil {
-			log.Error("GetBaseRepo: %v", err)
-			continue
-		}
-		if manuallyMerged(pr) {
-			continue
-		}
-		if err := TestPatch(pr); err != nil {
-			log.Error("testPatch: %v", err)
-			continue
-		}
-
-		checkAndUpdateStatus(pr)
-		select {
-		case <-ctx.Done():
-			log.Info("PID: %d Pull Request testing shutdown", os.Getpid())
+	go func() {
+		prs, err := models.GetPullRequestIDsByCheckStatus(models.PullRequestStatusChecking)
+		if err != nil {
+			log.Error("Find Checking PRs: %v", err)
 			return
-		default:
 		}
-	}
+		for _, prID := range prs {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				pullRequestQueue.Add(prID)
+			}
+		}
+	}()
 
 	// Start listening on new test requests.
 	for {
@@ -194,23 +179,20 @@ func TestPullRequests(ctx context.Context) {
 			pullRequestQueue.Remove(prID)
 
 			id := com.StrTo(prID).MustInt64()
-			if _, ok := checkedPRs[id]; ok {
-				continue
-			}
 
 			pr, err := models.GetPullRequestByID(id)
 			if err != nil {
 				log.Error("GetPullRequestByID[%s]: %v", prID, err)
 				continue
 			} else if manuallyMerged(pr) {
 				continue
-			} else if err = pr.TestPatch(); err != nil {
+			} else if err = TestPatch(pr); err != nil {
 				log.Error("testPatch[%d]: %v", pr.ID, err)
 				continue
 			}
-
 			checkAndUpdateStatus(pr)
 		case <-ctx.Done():
+			pullRequestQueue.Close()
 			log.Info("PID: %d Pull Request testing shutdown", os.Getpid())
 			return
 		}