Move TestPullRequests over to use UniqueQueue

Author: zeripath

modules/setting/queue.go

@@ -149,6 +149,17 @@ func NewQueueService() {
 	if _, ok := sectionMap["LENGTH"]; !ok {
 		_, _ = section.NewKey("LENGTH", fmt.Sprintf("%d", Cfg.Section("mailer").Key("SEND_BUFFER_LEN").MustInt(100)))
 	}
+
+	// Handle the old test pull requests configuration
+	// Please note this will be a unique queue
+	section = Cfg.Section("queue.test_pull_requests")
+	sectionMap = map[string]bool{}
+	for _, key := range section.Keys() {
+		sectionMap[key.Name()] = true
+	}
+	if _, ok := sectionMap["LENGTH"]; !ok {
+		_, _ = section.NewKey("LENGTH", fmt.Sprintf("%d", Repository.PullRequestQueueLength))
+	}
 }
 
 // ParseQueueConnStr parses a queue connection string

routers/init.go

@@ -113,7 +113,9 @@ func GlobalInit(ctx context.Context) {
 		code_indexer.Init()
 		mirror_service.InitSyncMirrors()
 		webhook.InitDeliverHooks()
-		pull_service.Init()
+		if err := pull_service.Init(); err != nil {
+			log.Fatal("Failed to initialize test pull requests queue: %v", err)
+		}
 		if err := task.Init(); err != nil {
 			log.Fatal("Failed to initialize task scheduler: %v", err)
 		}

services/pull/check.go

@@ -10,31 +10,40 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"strconv"
 	"strings"
 
 	"code.gitea.io/gitea/models"
 	"code.gitea.io/gitea/modules/git"
 	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/notification"
-	"code.gitea.io/gitea/modules/setting"
-	"code.gitea.io/gitea/modules/sync"
+	"code.gitea.io/gitea/modules/queue"
 	"code.gitea.io/gitea/modules/timeutil"
 
 	"github.com/unknwon/com"
 )
 
-// pullRequestQueue represents a queue to handle update pull request tests
-var pullRequestQueue = sync.NewUniqueQueue(setting.Repository.PullRequestQueueLength)
+// prQueue represents a queue to handle update pull request tests
+var prQueue queue.UniqueQueue
 
 // AddToTaskQueue adds itself to pull request test task queue.
 func AddToTaskQueue(pr *models.PullRequest) {
-	go pullRequestQueue.AddFunc(pr.ID, func() {
-		pr.Status = models.PullRequestStatusChecking
-		if err := pr.UpdateCols("status"); err != nil {
-			log.Error("AddToTaskQueue.UpdateCols[%d].(add to queue): %v", pr.ID, err)
+	go func() {
+		err := prQueue.PushFunc(strconv.FormatInt(pr.ID, 10), func() error {
+			pr.Status = models.PullRequestStatusChecking
+			err := pr.UpdateCols("status")
+			if err != nil {
+				log.Error("AddToTaskQueue.UpdateCols[%d].(add to queue): %v", pr.ID, err)
+			} else {
+				log.Trace("Adding PR ID: %d to the test pull requests queue", pr.ID)
+			}
+			return err
+		})
+		if err != nil && err != queue.ErrAlreadyInQueue {
+			log.Error("Error adding prID %d to the test pull requests queue: %v", pr.ID, err)
 		}
-	})
+	}()
 }
 
 // checkAndUpdateStatus checks if pull request is possible to leaving checking status,
@@ -46,7 +55,12 @@ func checkAndUpdateStatus(pr *models.PullRequest) {
 	}
 
 	// Make sure there is no waiting test to process before leaving the checking status.
-	if !pullRequestQueue.Exist(pr.ID) {
+	has, err := prQueue.Has(strconv.FormatInt(pr.ID, 10))
+	if err != nil {
+		log.Error("Unable to check if the queue is waiting to reprocess pr.ID %d. Error: %v", pr.ID, err)
+	}
+
+	if !has {
 		if err := pr.UpdateCols("status, conflicted_files"); err != nil {
 			log.Error("Update[%d]: %v", pr.ID, err)
 		}
@@ -155,61 +169,65 @@ func manuallyMerged(pr *models.PullRequest) bool {
 	return false
 }
 
-// TestPullRequests checks and tests untested patches of pull requests.
-// TODO: test more pull requests at same time.
-func TestPullRequests(ctx context.Context) {
-
-	go func() {
-		prs, err := models.GetPullRequestIDsByCheckStatus(models.PullRequestStatusChecking)
-		if err != nil {
-			log.Error("Find Checking PRs: %v", err)
+// InitializePullRequests checks and tests untested patches of pull requests.
+func InitializePullRequests(ctx context.Context) {
+	prs, err := models.GetPullRequestIDsByCheckStatus(models.PullRequestStatusChecking)
+	if err != nil {
+		log.Error("Find Checking PRs: %v", err)
+		return
+	}
+	for _, prID := range prs {
+		select {
+		case <-ctx.Done():
 			return
-		}
-		for _, prID := range prs {
-			select {
-			case <-ctx.Done():
-				return
-			default:
-				pullRequestQueue.Add(prID)
+		default:
+			if err := prQueue.PushFunc(strconv.FormatInt(prID, 10), func() error {
+				log.Trace("Adding PR ID: %d to the test pull requests queue", prID)
+				return nil
+			}); err != nil {
+				log.Error("Error adding prID: %s to the test pull requests queue %v", prID, err)
 			}
 		}
-	}()
+	}
+}
 
-	// Start listening on new test requests.
-	for {
-		select {
-		case prID := <-pullRequestQueue.Queue():
-			log.Trace("TestPullRequests[%v]: processing test task", prID)
-			pullRequestQueue.Remove(prID)
+// handle passed PR IDs and test the PRs
+func handle(data ...queue.Data) {
+	for _, datum := range data {
+		prID := datum.(string)
+		id := com.StrTo(prID).MustInt64()
 
-			id := com.StrTo(prID).MustInt64()
+		log.Trace("Testing PR ID %d from the test pull requests queue", id)
 
-			pr, err := models.GetPullRequestByID(id)
-			if err != nil {
-				log.Error("GetPullRequestByID[%s]: %v", prID, err)
-				continue
-			} else if pr.Status != models.PullRequestStatusChecking {
-				continue
-			} else if manuallyMerged(pr) {
-				continue
-			} else if err = TestPatch(pr); err != nil {
-				log.Error("testPatch[%d]: %v", pr.ID, err)
-				pr.Status = models.PullRequestStatusError
-				if err := pr.UpdateCols("status"); err != nil {
-					log.Error("update pr [%d] status to PullRequestStatusError failed: %v", pr.ID, err)
-				}
-				continue
+		pr, err := models.GetPullRequestByID(id)
+		if err != nil {
+			log.Error("GetPullRequestByID[%s]: %v", prID, err)
+			continue
+		} else if pr.Status != models.PullRequestStatusChecking {
+			continue
+		} else if manuallyMerged(pr) {
+			continue
+		} else if err = TestPatch(pr); err != nil {
+			log.Error("testPatch[%d]: %v", pr.ID, err)
+			pr.Status = models.PullRequestStatusError
+			if err := pr.UpdateCols("status"); err != nil {
+				log.Error("update pr [%d] status to PullRequestStatusError failed: %v", pr.ID, err)
 			}
-			checkAndUpdateStatus(pr)
-		case <-ctx.Done():
-			pullRequestQueue.Close()
-			log.Info("PID: %d Pull Request testing shutdown", os.Getpid())
-			return
+			continue
 		}
+		checkAndUpdateStatus(pr)
 	}
 }
 
 // Init runs the task queue to test all the checking status pull requests
-func Init() {
-	go graceful.GetManager().RunWithShutdownContext(TestPullRequests)
+func Init() error {
+	prQueue = queue.CreateUniqueQueue("test_pull_requests", handle, "").(queue.UniqueQueue)
+
+	if prQueue == nil {
+		return fmt.Errorf("Unable to create test_pull_requests Queue")
+	}
+
+	go graceful.GetManager().RunWithShutdownFns(prQueue.Run)
+	go graceful.GetManager().RunWithShutdownContext(InitializePullRequests)
+	return nil
 }

services/pull/check_test.go

@@ -6,29 +6,82 @@
 package pull
 
 import (
+	"context"
 	"strconv"
 	"testing"
 	"time"
 
 	"code.gitea.io/gitea/models"
+	"code.gitea.io/gitea/modules/queue"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/unknwon/com"
 )
 
 func TestPullRequest_AddToTaskQueue(t *testing.T) {
 	assert.NoError(t, models.PrepareTestDatabase())
 
+	idChan := make(chan int64, 10)
+
+	q, err := queue.NewChannelUniqueQueue(func(data ...queue.Data) {
+		for _, datum := range data {
+			prID := datum.(string)
+			id := com.StrTo(prID).MustInt64()
+			idChan <- id
+		}
+	}, queue.ChannelUniqueQueueConfiguration{
+		WorkerPoolConfiguration: queue.WorkerPoolConfiguration{
+			QueueLength: 10,
+			BatchLength: 1,
+		},
+		Workers: 1,
+		Name:    "temporary-queue",
+	}, "")
+	assert.NoError(t, err)
+
+	queueShutdown := []func(){}
+	queueTerminate := []func(){}
+
+	prQueue = q.(queue.UniqueQueue)
+
 	pr := models.AssertExistsAndLoadBean(t, &models.PullRequest{ID: 1}).(*models.PullRequest)
 	AddToTaskQueue(pr)
 
+	assert.Eventually(t, func() bool {
+		pr = models.AssertExistsAndLoadBean(t, &models.PullRequest{ID: 1}).(*models.PullRequest)
+		return pr.Status == models.PullRequestStatusChecking
+	}, 1*time.Second, 100*time.Millisecond)
+
+	has, err := prQueue.Has(strconv.FormatInt(pr.ID, 10))
+	assert.True(t, has)
+	assert.NoError(t, err)
+
+	prQueue.Run(func(_ context.Context, shutdown func()) {
+		queueShutdown = append(queueShutdown, shutdown)
+	}, func(_ context.Context, terminate func()) {
+		queueTerminate = append(queueTerminate, terminate)
+	})
+
 	select {
-	case id := <-pullRequestQueue.Queue():
-		assert.EqualValues(t, strconv.FormatInt(pr.ID, 10), id)
+	case id := <-idChan:
+		assert.EqualValues(t, pr.ID, id)
 	case <-time.After(time.Second):
 		assert.Fail(t, "Timeout: nothing was added to pullRequestQueue")
 	}
 
-	assert.True(t, pullRequestQueue.Exist(pr.ID))
+	has, err = prQueue.Has(strconv.FormatInt(pr.ID, 10))
+	assert.False(t, has)
+	assert.NoError(t, err)
+
 	pr = models.AssertExistsAndLoadBean(t, &models.PullRequest{ID: 1}).(*models.PullRequest)
 	assert.Equal(t, models.PullRequestStatusChecking, pr.Status)
+
+	for _, callback := range queueShutdown {
+		callback()
+	}
+	for _, callback := range queueTerminate {
+		callback()
+	}
+
+	prQueue = nil
 }