Make Requests Processes and create process hierarchy. Associate OpenRepository with context.

modules/context/repo.go

@@ -526,7 +526,7 @@ func RepoAssignment(ctx *Context) (cancel context.CancelFunc) {
 		return
 	}
 
-	gitRepo, err := git.OpenRepository(models.RepoPath(userName, repoName))
+	gitRepo, err := git.OpenRepositoryCtx(ctx, models.RepoPath(userName, repoName))
 	if err != nil {
 		ctx.ServerError("RepoAssignment Invalid repo "+models.RepoPath(userName, repoName), err)
 		return
@@ -742,7 +742,7 @@ func RepoRefByType(refType RepoRefType, ignoreNotExistErr ...bool) func(*Context
 
 		if ctx.Repo.GitRepo == nil {
 			repoPath := models.RepoPath(ctx.Repo.Owner.Name, ctx.Repo.Repository.Name)
-			ctx.Repo.GitRepo, err = git.OpenRepository(repoPath)
+			ctx.Repo.GitRepo, err = git.OpenRepositoryCtx(ctx, repoPath)
 			if err != nil {
 				ctx.ServerError("RepoRef Invalid repo "+repoPath, err)
 				return

modules/cron/tasks.go

@@ -80,11 +80,10 @@ func (t *Task) RunWithUser(doer *models.User, config Config) {
 		}
 	}()
 	graceful.GetManager().RunWithShutdownContext(func(baseCtx context.Context) {
-		ctx, cancel := context.WithCancel(baseCtx)
-		defer cancel()
 		pm := process.GetManager()
-		pid := pm.Add(config.FormatMessage(t.Name, "process", doer), cancel)
-		defer pm.Remove(pid)
+		ctx, cancel := pm.AddContext(baseCtx, config.FormatMessage(t.Name, "process", doer))
+		defer cancel()
+
 		if err := t.fun(ctx, doer, config); err != nil {
 			if models.IsErrCancelled(err) {
 				message := err.(models.ErrCancelled).Message

modules/git/batch_reader.go

@@ -28,10 +28,10 @@ type WriteCloserError interface {
 }
 
 // CatFileBatchCheck opens git cat-file --batch-check in the provided repo and returns a stdin pipe, a stdout reader and cancel function
-func CatFileBatchCheck(repoPath string) (WriteCloserError, *bufio.Reader, func()) {
+func CatFileBatchCheck(ctx context.Context, repoPath string) (WriteCloserError, *bufio.Reader, func()) {
 	batchStdinReader, batchStdinWriter := io.Pipe()
 	batchStdoutReader, batchStdoutWriter := io.Pipe()
-	ctx, ctxCancel := context.WithCancel(DefaultContext)
+	ctx, ctxCancel := context.WithCancel(ctx)
 	closed := make(chan struct{})
 	cancel := func() {
 		_ = batchStdinReader.Close()
@@ -67,12 +67,12 @@ func CatFileBatchCheck(repoPath string) (WriteCloserError, *bufio.Reader, func()
 }
 
 // CatFileBatch opens git cat-file --batch in the provided repo and returns a stdin pipe, a stdout reader and cancel function
-func CatFileBatch(repoPath string) (WriteCloserError, *bufio.Reader, func()) {
+func CatFileBatch(ctx context.Context, repoPath string) (WriteCloserError, *bufio.Reader, func()) {
 	// We often want to feed the commits in order into cat-file --batch, followed by their trees and sub trees as necessary.
 	// so let's create a batch stdin and stdout
 	batchStdinReader, batchStdinWriter := io.Pipe()
 	batchStdoutReader, batchStdoutWriter := nio.Pipe(buffer.New(32 * 1024))
-	ctx, ctxCancel := context.WithCancel(DefaultContext)
+	ctx, ctxCancel := context.WithCancel(ctx)
 	closed := make(chan struct{})
 	cancel := func() {
 		_ = batchStdinReader.Close()

modules/git/blame.go

@@ -25,7 +25,6 @@ type BlamePart struct {
 // BlameReader returns part of file blame one by one
 type BlameReader struct {
 	cmd     *exec.Cmd
-	pid     int64
 	output  io.ReadCloser
 	reader  *bufio.Reader
 	lastSha *string
@@ -100,8 +99,7 @@ func (r *BlameReader) NextPart() (*BlamePart, error) {
 
 // Close BlameReader - don't run NextPart after invoking that
 func (r *BlameReader) Close() error {
-	defer process.GetManager().Remove(r.pid)
-	r.cancel()
+	defer r.cancel()
 
 	_ = r.output.Close()
 
@@ -125,7 +123,8 @@ 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 := context.WithCancel(ctx)
+	ctx, cancel := 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
@@ -141,13 +140,10 @@ func createBlameReader(ctx context.Context, dir string, command ...string) (*Bla
 		return nil, fmt.Errorf("Start: %v", err)
 	}
 
-	pid := process.GetManager().Add(fmt.Sprintf("GetBlame [repo_path: %s]", dir), cancel)
-
 	reader := bufio.NewReader(stdout)
 
 	return &BlameReader{
 		cmd,
-		pid,
 		stdout,
 		reader,
 		nil,

modules/git/blob_nogogit.go

@@ -29,7 +29,7 @@ type Blob struct {
 // DataAsync gets a ReadCloser for the contents of a blob without reading it all.
 // Calling the Close function on the result will discard all unread output.
 func (b *Blob) DataAsync() (io.ReadCloser, error) {
-	wr, rd, cancel := b.repo.CatFileBatch()
+	wr, rd, cancel := b.repo.CatFileBatch(b.repo.Ctx)
 
 	_, err := wr.Write([]byte(b.ID.String() + "\n"))
 	if err != nil {
@@ -67,7 +67,7 @@ func (b *Blob) Size() int64 {
 		return b.size
 	}
 
-	wr, rd, cancel := b.repo.CatFileBatchCheck()
+	wr, rd, cancel := b.repo.CatFileBatchCheck(b.repo.Ctx)
 	defer cancel()
 	_, err := wr.Write([]byte(b.ID.String() + "\n"))
 	if err != nil {

modules/git/command.go

@@ -143,7 +143,12 @@ func (c *Command) RunWithContext(rc *RunContext) error {
 		log.Debug("%s: %v", rc.Dir, c)
 	}
 
-	ctx, cancel := context.WithTimeout(c.parentContext, rc.Timeout)
+	desc := c.desc
+	if desc == "" {
+		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()
 
 	cmd := exec.CommandContext(ctx, c.name, c.args...)
@@ -172,13 +177,6 @@ func (c *Command) RunWithContext(rc *RunContext) error {
 		return err
 	}
 
-	desc := c.desc
-	if desc == "" {
-		desc = fmt.Sprintf("%s %s %s [repo_path: %s]", GitExecutable, c.name, strings.Join(c.args, " "), rc.Dir)
-	}
-	pid := process.GetManager().Add(desc, cancel)
-	defer process.GetManager().Remove(pid)
-
 	if rc.PipelineFunc != nil {
 		err := rc.PipelineFunc(ctx, cancel)
 		if err != nil {

modules/git/commit_info_nogogit.go

@@ -99,7 +99,7 @@ func (tes Entries) GetCommitsInfo(ctx context.Context, commit *Commit, treePath
 }
 
 func getLastCommitForPathsByCache(ctx context.Context, commitID, treePath string, paths []string, cache *LastCommitCache) (map[string]*Commit, []string, error) {
-	wr, rd, cancel := cache.repo.CatFileBatch()
+	wr, rd, cancel := cache.repo.CatFileBatch(ctx)
 	defer cancel()
 
 	var unHitEntryPaths []string
@@ -128,7 +128,7 @@ func GetLastCommitForPaths(ctx context.Context, commit *Commit, treePath string,
 		return nil, err
 	}
 
-	batchStdinWriter, batchReader, cancel := commit.repo.CatFileBatch()
+	batchStdinWriter, batchReader, cancel := commit.repo.CatFileBatch(ctx)
 	defer cancel()
 
 	commitsMap := map[string]*Commit{}

modules/git/diff.go

@@ -56,7 +56,7 @@ func GetRepoRawDiffForFile(repo *Repository, startCommit, endCommit string, diff
 		fileArgs = append(fileArgs, "--", file)
 	}
 	// FIXME: graceful: These commands should have a timeout
-	ctx, cancel := context.WithCancel(DefaultContext)
+	ctx, cancel := process.GetManager().AddContext(DefaultContext, fmt.Sprintf("GetRawDiffForFile: [repo_path: %s]", repo.Path))
 	defer cancel()
 
 	var cmd *exec.Cmd
@@ -90,8 +90,6 @@ func GetRepoRawDiffForFile(repo *Repository, startCommit, endCommit string, diff
 	cmd.Dir = repo.Path
 	cmd.Stdout = writer
 	cmd.Stderr = stderr
-	pid := process.GetManager().Add(fmt.Sprintf("GetRawDiffForFile: [repo_path: %s]", repo.Path), cancel)
-	defer process.GetManager().Remove(pid)
 
 	if err = cmd.Run(); err != nil {
 		return fmt.Errorf("Run: %v - %s", err, stderr)

modules/git/pipeline/lfs_nogogit.go

@@ -63,7 +63,7 @@ func FindLFSFile(repo *git.Repository, hash git.SHA1) ([]*LFSResult, error) {
 
 	// Next feed the commits in order into cat-file --batch, followed by their trees and sub trees as necessary.
 	// so let's create a batch stdin and stdout
-	batchStdinWriter, batchReader, cancel := repo.CatFileBatch()
+	batchStdinWriter, batchReader, cancel := repo.CatFileBatch(repo.Ctx)
 	defer cancel()
 
 	// We'll use a scanner for the revList because it's simpler than a bufio.Reader

modules/git/repo_base_gogit.go

@@ -9,6 +9,7 @@
 package git
 
 import (
+	"context"
 	"errors"
 	"path/filepath"
 
@@ -30,10 +31,17 @@ type Repository struct {
 	gogitRepo    *gogit.Repository
 	gogitStorage *filesystem.Storage
 	gpgSettings  *GPGSettings
+
+	ctx context.Context
 }
 
 // OpenRepository opens the repository at the given path.
 func OpenRepository(repoPath string) (*Repository, error) {
+	return OpenRepositoryCtx(DefaultContext, repoPath)
+}
+
+// OpenRepositoryCtx opens the repository at the given path within the context.Context
+func OpenRepositoryCtx(ctx context.Context, repoPath string) (*Repository, error) {
 	repoPath, err := filepath.Abs(repoPath)
 	if err != nil {
 		return nil, err
@@ -60,6 +68,7 @@ func OpenRepository(repoPath string) (*Repository, error) {
 		gogitRepo:    gogitRepo,
 		gogitStorage: storage,
 		tagCache:     newObjectCache(),
+		ctx:          ctx,
 	}, nil
 }
 

modules/git/repo_base_nogogit.go

@@ -32,10 +32,17 @@ type Repository struct {
 	checkCancel context.CancelFunc
 	checkReader *bufio.Reader
 	checkWriter WriteCloserError
+
+	Ctx context.Context
 }
 
 // OpenRepository opens the repository at the given path.
 func OpenRepository(repoPath string) (*Repository, error) {
+	return OpenRepositoryCtx(DefaultContext, repoPath)
+}
+
+// OpenRepositoryCtx opens the repository at the given path with the provided context.
+func OpenRepositoryCtx(ctx context.Context, repoPath string) (*Repository, error) {
 	repoPath, err := filepath.Abs(repoPath)
 	if err != nil {
 		return nil, err
@@ -46,28 +53,29 @@ func OpenRepository(repoPath string) (*Repository, error) {
 	repo := &Repository{
 		Path:     repoPath,
 		tagCache: newObjectCache(),
+		Ctx:      ctx,
 	}
 
-	repo.batchWriter, repo.batchReader, repo.batchCancel = CatFileBatch(repoPath)
-	repo.checkWriter, repo.checkReader, repo.checkCancel = CatFileBatchCheck(repo.Path)
+	repo.batchWriter, repo.batchReader, repo.batchCancel = CatFileBatch(ctx, repoPath)
+	repo.checkWriter, repo.checkReader, repo.checkCancel = CatFileBatchCheck(ctx, repo.Path)
 
 	return repo, nil
 }
 
 // CatFileBatch obtains a CatFileBatch for this repository
-func (repo *Repository) CatFileBatch() (WriteCloserError, *bufio.Reader, func()) {
+func (repo *Repository) CatFileBatch(ctx context.Context) (WriteCloserError, *bufio.Reader, func()) {
 	if repo.batchCancel == nil || repo.batchReader.Buffered() > 0 {
 		log.Debug("Opening temporary cat file batch for: %s", repo.Path)
-		return CatFileBatch(repo.Path)
+		return CatFileBatch(ctx, repo.Path)
 	}
 	return repo.batchWriter, repo.batchReader, func() {}
 }
 
 // CatFileBatchCheck obtains a CatFileBatchCheck for this repository
-func (repo *Repository) CatFileBatchCheck() (WriteCloserError, *bufio.Reader, func()) {
+func (repo *Repository) CatFileBatchCheck(ctx context.Context) (WriteCloserError, *bufio.Reader, func()) {
 	if repo.checkCancel == nil || repo.checkReader.Buffered() > 0 {
 		log.Debug("Opening temporary cat file batch-check: %s", repo.Path)
-		return CatFileBatchCheck(repo.Path)
+		return CatFileBatchCheck(ctx, repo.Path)
 	}
 	return repo.checkWriter, repo.checkReader, func() {}
 }

modules/git/repo_branch_nogogit.go

@@ -23,7 +23,7 @@ func (repo *Repository) IsObjectExist(name string) bool {
 		return false
 	}
 
-	wr, rd, cancel := repo.CatFileBatchCheck()
+	wr, rd, cancel := repo.CatFileBatchCheck(repo.Ctx)
 	defer cancel()
 	_, err := wr.Write([]byte(name + "\n"))
 	if err != nil {
@@ -40,7 +40,7 @@ func (repo *Repository) IsReferenceExist(name string) bool {
 		return false
 	}
 
-	wr, rd, cancel := repo.CatFileBatchCheck()
+	wr, rd, cancel := repo.CatFileBatchCheck(repo.Ctx)
 	defer cancel()
 	_, err := wr.Write([]byte(name + "\n"))
 	if err != nil {

modules/git/repo_commit_nogogit.go

@@ -35,7 +35,7 @@ func (repo *Repository) ResolveReference(name string) (string, error) {
 
 // GetRefCommitID returns the last commit ID string of given reference (branch or tag).
 func (repo *Repository) GetRefCommitID(name string) (string, error) {
-	wr, rd, cancel := repo.CatFileBatchCheck()
+	wr, rd, cancel := repo.CatFileBatchCheck(repo.Ctx)
 	defer cancel()
 	_, _ = wr.Write([]byte(name + "\n"))
 	shaBs, _, _, err := ReadBatchLine(rd)
@@ -53,7 +53,7 @@ func (repo *Repository) IsCommitExist(name string) bool {
 }
 
 func (repo *Repository) getCommit(id SHA1) (*Commit, error) {
-	wr, rd, cancel := repo.CatFileBatch()
+	wr, rd, cancel := repo.CatFileBatch(repo.Ctx)
 	defer cancel()
 
 	_, _ = wr.Write([]byte(id.String() + "\n"))
@@ -132,7 +132,7 @@ func (repo *Repository) ConvertToSHA1(commitID string) (SHA1, error) {
 		}
 	}
 
-	wr, rd, cancel := repo.CatFileBatchCheck()
+	wr, rd, cancel := repo.CatFileBatchCheck(repo.Ctx)
 	defer cancel()
 	_, err := wr.Write([]byte(commitID + "\n"))
 	if err != nil {

modules/git/repo_language_stats_nogogit.go

@@ -26,7 +26,7 @@ import (
 func (repo *Repository) GetLanguageStats(commitID string) (map[string]int64, error) {
 	// We will feed the commit IDs in order into cat-file --batch, followed by blobs as necessary.
 	// so let's create a batch stdin and stdout
-	batchStdinWriter, batchReader, cancel := repo.CatFileBatch()
+	batchStdinWriter, batchReader, cancel := repo.CatFileBatch(repo.Ctx)
 	defer cancel()
 
 	writeID := func(id string) error {

modules/git/repo_tree_nogogit.go

@@ -12,7 +12,7 @@ import (
 )
 
 func (repo *Repository) getTree(id SHA1) (*Tree, error) {
-	wr, rd, cancel := repo.CatFileBatch()
+	wr, rd, cancel := repo.CatFileBatch(repo.Ctx)
 	defer cancel()
 
 	_, _ = wr.Write([]byte(id.String() + "\n"))

modules/git/tree_nogogit.go

@@ -36,7 +36,7 @@ func (t *Tree) ListEntries() (Entries, error) {
 	}
 
 	if t.repo != nil {
-		wr, rd, cancel := t.repo.CatFileBatch()
+		wr, rd, cancel := t.repo.CatFileBatch(t.repo.Ctx)
 		defer cancel()
 
 		_, _ = wr.Write([]byte(t.ID.String() + "\n"))

modules/indexer/code/bleve.go

@@ -271,7 +271,7 @@ func (b *BleveIndexer) Index(repo *models.Repository, sha string, changes *repoC
 	batch := gitea_bleve.NewFlushingBatch(b.indexer, maxBatchSize)
 	if len(changes.Updates) > 0 {
 
-		batchWriter, batchReader, cancel := git.CatFileBatch(repo.RepoPath())
+		batchWriter, batchReader, cancel := git.CatFileBatch(git.DefaultContext, repo.RepoPath())
 		defer cancel()
 
 		for _, update := range changes.Updates {

modules/indexer/code/elastic_search.go

@@ -245,7 +245,7 @@ func (b *ElasticSearchIndexer) Index(repo *models.Repository, sha string, change
 	reqs := make([]elastic.BulkableRequest, 0)
 	if len(changes.Updates) > 0 {
 
-		batchWriter, batchReader, cancel := git.CatFileBatch(repo.RepoPath())
+		batchWriter, batchReader, cancel := git.CatFileBatch(git.DefaultContext, repo.RepoPath())
 		defer cancel()
 
 		for _, update := range changes.Updates {

modules/markup/external/external.go

@@ -5,7 +5,6 @@
 package external
 
 import (
-	"context"
 	"fmt"
 	"io"
 	"os"
@@ -107,12 +106,9 @@ func (p *Renderer) Render(ctx *markup.RenderContext, input io.Reader, output io.
 		ctx.Ctx = graceful.GetManager().ShutdownContext()
 	}
 
-	processCtx, cancel := context.WithCancel(ctx.Ctx)
+	processCtx, cancel := process.GetManager().AddContext(ctx.Ctx, fmt.Sprintf("Render [%s] for %s", commands[0], ctx.URLPrefix))
 	defer cancel()
 
-	pid := process.GetManager().Add(fmt.Sprintf("Render [%s] for %s", commands[0], ctx.URLPrefix), cancel)
-	defer process.GetManager().Remove(pid)
-
 	cmd := exec.CommandContext(processCtx, commands[0], args...)
 	cmd.Env = append(
 		os.Environ(),