separate remove and cancel functions

Signed-off-by: Andrew Thornton <[email protected]>

modules/cron/tasks.go

@@ -81,8 +81,8 @@ func (t *Task) RunWithUser(doer *models.User, config Config) {
 	}()
 	graceful.GetManager().RunWithShutdownContext(func(baseCtx context.Context) {
 		pm := process.GetManager()
-		ctx, cancel := pm.AddContext(baseCtx, config.FormatMessage(t.Name, "process", doer))
-		defer cancel()
+		ctx, _, remove := pm.AddContext(baseCtx, config.FormatMessage(t.Name, "process", doer))
+		defer remove()
 
 		if err := t.fun(ctx, doer, config); err != nil {
 			if models.IsErrCancelled(err) {

modules/git/blame.go

@@ -29,6 +29,7 @@ type BlameReader struct {
 	reader  *bufio.Reader
 	lastSha *string
 	cancel  context.CancelFunc
+	remove  context.CancelFunc
 }
 
 var shaLineRegex = regexp.MustCompile("^([a-z0-9]{40})")
@@ -99,7 +100,8 @@ func (r *BlameReader) NextPart() (*BlamePart, error) {
 
 // Close BlameReader - don't run NextPart after invoking that
 func (r *BlameReader) Close() error {
-	defer r.cancel()
+	defer r.remove()
+	r.cancel()
 
 	_ = r.output.Close()
 
@@ -123,20 +125,21 @@ func CreateBlameReader(ctx context.Context, repoPath, commitID, file string) (*B
 
 func createBlameReader(ctx context.Context, dir string, command ...string) (*BlameReader, error) {
 	// Here we use the provided context - this should be tied to the request performing the blame so that it does not hang around.
-	ctx, cancel := process.GetManager().AddContext(ctx, fmt.Sprintf("GetBlame [repo_path: %s]", dir))
+	ctx, cancel, remove := process.GetManager().AddContext(ctx, fmt.Sprintf("GetBlame [repo_path: %s]", dir))
 
 	cmd := exec.CommandContext(ctx, command[0], command[1:]...)
 	cmd.Dir = dir
 	cmd.Stderr = os.Stderr
 
 	stdout, err := cmd.StdoutPipe()
 	if err != nil {
-		defer cancel()
+		defer remove()
 		return nil, fmt.Errorf("StdoutPipe: %v", err)
 	}
 
 	if err = cmd.Start(); err != nil {
-		defer cancel()
+		defer remove()
+		_ = stdout.Close()
 		return nil, fmt.Errorf("Start: %v", err)
 	}
 
@@ -148,5 +151,6 @@ func createBlameReader(ctx context.Context, dir string, command ...string) (*Bla
 		reader,
 		nil,
 		cancel,
+		remove,
 	}, nil
 }

modules/git/command.go

@@ -148,8 +148,8 @@ func (c *Command) RunWithContext(rc *RunContext) error {
 		desc = fmt.Sprintf("%s %s [repo_path: %s]", c.name, strings.Join(c.args, " "), rc.Dir)
 	}
 
-	ctx, cancel := process.GetManager().AddContextTimeout(c.parentContext, rc.Timeout, desc)
-	defer cancel()
+	ctx, cancel, remove := process.GetManager().AddContextTimeout(c.parentContext, rc.Timeout, desc)
+	defer remove()
 
 	cmd := exec.CommandContext(ctx, c.name, c.args...)
 	if rc.Env == nil {

modules/git/diff.go

@@ -56,8 +56,8 @@ func GetRepoRawDiffForFile(repo *Repository, startCommit, endCommit string, diff
 		fileArgs = append(fileArgs, "--", file)
 	}
 	// FIXME: graceful: These commands should have a timeout
-	ctx, cancel := process.GetManager().AddContext(DefaultContext, fmt.Sprintf("GetRawDiffForFile: [repo_path: %s]", repo.Path))
-	defer cancel()
+	ctx, _, remove := process.GetManager().AddContext(DefaultContext, fmt.Sprintf("GetRawDiffForFile: [repo_path: %s]", repo.Path))
+	defer remove()
 
 	var cmd *exec.Cmd
 	switch diffType {

modules/git/repo_language_stats_nogogit.go

@@ -83,7 +83,7 @@ func (repo *Repository) GetLanguageStats(commitID string) (map[string]int64, err
 					IndexFile:  indexFilename,
 					WorkTree:   tmpWorkTree,
 				}
-				ctx, cancel := context.WithCancel(DefaultContext)
+				ctx, cancel := context.WithCancel(repo.Ctx)
 				if err := checker.Init(ctx); err != nil {
 					log.Error("Unable to open checker for %s. Error: %v", commitID, err)
 				} else {
@@ -104,6 +104,12 @@ func (repo *Repository) GetLanguageStats(commitID string) (map[string]int64, err
 	var content []byte
 	sizes := make(map[string]int64)
 	for _, f := range entries {
+		select {
+		case <-repo.Ctx.Done():
+			return sizes, repo.Ctx.Err()
+		default:
+		}
+
 		contentBuf.Reset()
 		content = contentBuf.Bytes()
 

modules/indexer/stats/db.go

@@ -5,9 +5,13 @@
 package stats
 
 import (
+	"fmt"
+
 	"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/process"
 )
 
 // DBIndexer implements Indexer interface to use database's like search
@@ -16,6 +20,9 @@ type DBIndexer struct {
 
 // Index repository status function
 func (db *DBIndexer) Index(id int64) error {
+	ctx, _, remove := process.GetManager().AddContext(graceful.GetManager().ShutdownContext(), fmt.Sprintf("Stats.DB Index Repo[%d]", id))
+	defer remove()
+
 	repo, err := models.GetRepositoryByID(id)
 	if err != nil {
 		return err
@@ -29,7 +36,7 @@ func (db *DBIndexer) Index(id int64) error {
 		return err
 	}
 
-	gitRepo, err := git.OpenRepository(repo.RepoPath())
+	gitRepo, err := git.OpenRepositoryCtx(ctx, repo.RepoPath())
 	if err != nil {
 		return err
 	}

modules/markup/external/external.go

@@ -106,8 +106,8 @@ func (p *Renderer) Render(ctx *markup.RenderContext, input io.Reader, output io.
 		ctx.Ctx = graceful.GetManager().ShutdownContext()
 	}
 
-	processCtx, cancel := process.GetManager().AddContext(ctx.Ctx, fmt.Sprintf("Render [%s] for %s", commands[0], ctx.URLPrefix))
-	defer cancel()
+	processCtx, _, remove := process.GetManager().AddContext(ctx.Ctx, fmt.Sprintf("Render [%s] for %s", commands[0], ctx.URLPrefix))
+	defer remove()
 
 	cmd := exec.CommandContext(processCtx, commands[0], args...)
 	cmd.Env = append(

modules/process/manager.go

@@ -78,31 +78,31 @@ func GetManager() *Manager {
 }
 
 // AddContext create a new context and add it as a process. The CancelFunc must always be called even if the context is Done()
-func (pm *Manager) AddContext(parent context.Context, description string) (context.Context, context.CancelFunc) {
+func (pm *Manager) AddContext(parent context.Context, description string) (ctx context.Context, cancel, remove context.CancelFunc) {
 	parentPID := GetParentPID(parent)
 
-	ctx, cancel := context.WithCancel(parent)
+	ctx, cancel = context.WithCancel(parent)
 
-	pid, cancel := pm.Add(parentPID, description, cancel)
+	pid, remove := pm.Add(parentPID, description, cancel)
 
 	return &Context{
 		Context: ctx,
 		pid:     pid,
-	}, cancel
+	}, cancel, remove
 }
 
 // AddContextTimeout create a new context and add it as a process
-func (pm *Manager) AddContextTimeout(parent context.Context, timeout time.Duration, description string) (context.Context, context.CancelFunc) {
+func (pm *Manager) AddContextTimeout(parent context.Context, timeout time.Duration, description string) (ctx context.Context, cancel, remove context.CancelFunc) {
 	parentPID := GetParentPID(parent)
 
-	ctx, cancel := context.WithTimeout(parent, timeout)
+	ctx, cancel = context.WithTimeout(parent, timeout)
 
-	pid, cancel := pm.Add(parentPID, description, cancel)
+	pid, remove := pm.Add(parentPID, description, cancel)
 
 	return &Context{
 		Context: ctx,
 		pid:     pid,
-	}, cancel
+	}, cancel, remove
 }
 
 // Add create a new process
@@ -120,9 +120,10 @@ func (pm *Manager) Add(parentPID int64, description string, cancel context.Cance
 		ParentPID:   parentPID,
 		Description: description,
 		Start:       time.Now(),
+		Cancel:      cancel,
 	}
 
-	process.Cancel = func() {
+	remove := func() {
 		cancel()
 		pm.remove(process)
 	}
@@ -133,7 +134,7 @@ func (pm *Manager) Add(parentPID int64, description string, cancel context.Cance
 	pm.processes[pid] = process
 	pm.mutex.Unlock()
 
-	return pid, process.Cancel
+	return pid, remove
 }
 
 // nextPID will return the next available PID. pm.mutex should already be locked.
@@ -264,8 +265,8 @@ func (pm *Manager) ExecDirEnvStdIn(timeout time.Duration, dir, desc string, env
 	stdOut := new(bytes.Buffer)
 	stdErr := new(bytes.Buffer)
 
-	ctx, cancel := pm.AddContextTimeout(DefaultContext, timeout, desc)
-	defer cancel()
+	ctx, _, remove := pm.AddContextTimeout(DefaultContext, timeout, desc)
+	defer remove()
 
 	cmd := exec.CommandContext(ctx, cmdName, args...)
 	cmd.Dir = dir

modules/process/manager_test.go

@@ -27,12 +27,12 @@ func TestManager_AddContext(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	p1Ctx, cancel := pm.AddContext(ctx, "foo")
-	defer cancel()
+	p1Ctx, _, remove := pm.AddContext(ctx, "foo")
+	defer remove()
 	assert.Equal(t, int64(1), GetContext(p1Ctx).GetPID(), "expected to get pid 1 got %d", GetContext(p1Ctx).GetPID())
 
-	p2Ctx, cancel := pm.AddContext(p1Ctx, "bar")
-	defer cancel()
+	p2Ctx, _, remove := pm.AddContext(p1Ctx, "bar")
+	defer remove()
 
 	assert.Equal(t, int64(2), GetContext(p2Ctx).GetPID(), "expected to get pid 2 got %d", GetContext(p2Ctx).GetPID())
 	assert.Equal(t, int64(1), GetContext(p2Ctx).GetParent().GetPID(), "expected to get pid 1 got %d", GetContext(p2Ctx).GetParent().GetPID())
@@ -41,8 +41,8 @@ func TestManager_AddContext(t *testing.T) {
 func TestManager_Cancel(t *testing.T) {
 	pm := Manager{processes: make(map[int64]*Process), next: 1, low: 1}
 
-	ctx, cancel := pm.AddContext(context.Background(), "foo")
-	defer cancel()
+	ctx, _, remove := pm.AddContext(context.Background(), "foo")
+	defer remove()
 
 	pm.Cancel(GetPID(ctx))
 
@@ -51,6 +51,19 @@ func TestManager_Cancel(t *testing.T) {
 	default:
 		assert.Fail(t, "Cancel should cancel the provided context")
 	}
+	remove()
+
+	ctx, cancel, remove := pm.AddContext(context.Background(), "foo")
+	defer remove()
+
+	cancel()
+
+	select {
+	case <-ctx.Done():
+	default:
+		assert.Fail(t, "Cancel should cancel the provided context")
+	}
+	remove()
 }
 
 func TestManager_Remove(t *testing.T) {
@@ -59,12 +72,12 @@ func TestManager_Remove(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	p1Ctx, cancel := pm.AddContext(ctx, "foo")
-	defer cancel()
+	p1Ctx, _, remove := pm.AddContext(ctx, "foo")
+	defer remove()
 	assert.Equal(t, int64(1), GetContext(p1Ctx).GetPID(), "expected to get pid 1 got %d", GetContext(p1Ctx).GetPID())
 
-	p2Ctx, cancel := pm.AddContext(p1Ctx, "bar")
-	defer cancel()
+	p2Ctx, _, remove := pm.AddContext(p1Ctx, "bar")
+	defer remove()
 
 	assert.Equal(t, int64(2), GetContext(p2Ctx).GetPID(), "expected to get pid 2 got %d", GetContext(p2Ctx).GetPID())
 

modules/task/migrate.go

@@ -94,8 +94,8 @@ func runMigrateTask(t *models.Task) (err error) {
 	opts.MigrateToRepoID = t.RepoID
 
 	pm := process.GetManager()
-	ctx, cancel := pm.AddContext(graceful.GetManager().ShutdownContext(), fmt.Sprintf("MigrateTask: %s/%s", t.Owner.Name, opts.RepoName))
-	defer cancel()
+	ctx, _, remove := pm.AddContext(graceful.GetManager().ShutdownContext(), fmt.Sprintf("MigrateTask: %s/%s", t.Owner.Name, opts.RepoName))
+	defer remove()
 
 	t.StartTime = timeutil.TimeStampNow()
 	t.Status = structs.TaskStatusRunning

routers/common/middleware.go

@@ -23,8 +23,8 @@ func Middlewares() []func(http.Handler) http.Handler {
 	var handlers = []func(http.Handler) http.Handler{
 		func(next http.Handler) http.Handler {
 			return http.HandlerFunc(func(resp http.ResponseWriter, req *http.Request) {
-				ctx, cancel := process.GetManager().AddContext(req.Context(), fmt.Sprintf("%s: %s", req.Method, req.RequestURI))
-				defer cancel()
+				ctx, _, remove := process.GetManager().AddContext(req.Context(), fmt.Sprintf("%s: %s", req.Method, req.RequestURI))
+				defer remove()
 				next.ServeHTTP(context.NewResponse(resp), req.WithContext(ctx))
 			})
 		},

routers/web/repo/http.go

@@ -482,8 +482,8 @@ func serviceRPC(h serviceHandler, service string) {
 	}
 
 	// ctx, cancel := gocontext.WithCancel(git.DefaultContext)
-	ctx, cancel := process.GetManager().AddContext(git.DefaultContext, fmt.Sprintf("%s %s %s [repo_path: %s]", git.GitExecutable, service, "--stateless-rpc", h.dir))
-	defer cancel()
+	ctx, _, remove := process.GetManager().AddContext(git.DefaultContext, fmt.Sprintf("%s %s %s [repo_path: %s]", git.GitExecutable, service, "--stateless-rpc", h.dir))
+	defer remove()
 
 	var stderr bytes.Buffer
 	cmd := exec.CommandContext(ctx, git.GitExecutable, service, "--stateless-rpc", h.dir)

services/gitdiff/gitdiff.go

@@ -1226,8 +1226,8 @@ func GetDiffRangeWithWhitespaceBehavior(gitRepo *git.Repository, beforeCommitID,
 	}
 
 	timeout := time.Duration(setting.Git.Timeout.Default) * time.Second
-	ctx, cancel := process.GetManager().AddContextTimeout(git.DefaultContext, timeout, fmt.Sprintf("GetDiffRange [repo_path: %s]", repoPath))
-	defer cancel()
+	ctx, _, remove := process.GetManager().AddContextTimeout(git.DefaultContext, timeout, fmt.Sprintf("GetDiffRange [repo_path: %s]", repoPath))
+	defer remove()
 
 	var cmd *exec.Cmd
 	if (len(beforeCommitID) == 0 || beforeCommitID == git.EmptySHA) && commit.ParentCount() == 0 {

services/mailer/mailer.go

@@ -248,8 +248,8 @@ func (s *sendmailSender) Send(from string, to []string, msg io.WriterTo) error {
 
 	desc := fmt.Sprintf("SendMail: %s %v", setting.MailService.SendmailPath, args)
 
-	ctx, cancel := process.GetManager().AddContextTimeout(graceful.GetManager().HammerContext(), setting.MailService.SendmailTimeout, desc)
-	defer cancel()
+	ctx, cancel, remove := process.GetManager().AddContextTimeout(graceful.GetManager().HammerContext(), setting.MailService.SendmailTimeout, desc)
+	defer remove()
 
 	cmd := exec.CommandContext(ctx, setting.MailService.SendmailPath, args...)
 	pipe, err := cmd.StdinPipe()
@@ -259,6 +259,8 @@ func (s *sendmailSender) Send(from string, to []string, msg io.WriterTo) error {
 	}
 
 	if err = cmd.Start(); err != nil {
+		_ = pipe.Close()
+		cancel()
 		return err
 	}