Improve queue and logger context

modules/graceful/context.go

@@ -5,51 +5,8 @@ package graceful
 
 import (
 	"context"
-	"time"
 )
 
-// ChannelContext is a context that wraps a channel and error as a context
-type ChannelContext struct {
-	done <-chan struct{}
-	err  error
-}
-
-// NewChannelContext creates a ChannelContext from a channel and error
-func NewChannelContext(done <-chan struct{}, err error) *ChannelContext {
-	return &ChannelContext{
-		done: done,
-		err:  err,
-	}
-}
-
-// Deadline returns the time when work done on behalf of this context
-// should be canceled. There is no Deadline for a ChannelContext
-func (ctx *ChannelContext) Deadline() (deadline time.Time, ok bool) {
-	return deadline, ok
-}
-
-// Done returns the channel provided at the creation of this context.
-// When closed, work done on behalf of this context should be canceled.
-func (ctx *ChannelContext) Done() <-chan struct{} {
-	return ctx.done
-}
-
-// Err returns nil, if Done is not closed. If Done is closed,
-// Err returns the error provided at the creation of this context
-func (ctx *ChannelContext) Err() error {
-	select {
-	case <-ctx.done:
-		return ctx.err
-	default:
-		return nil
-	}
-}
-
-// Value returns nil for all calls as no values are or can be associated with this context
-func (ctx *ChannelContext) Value(key interface{}) interface{} {
-	return nil
-}
-
 // ShutdownContext returns a context.Context that is Done at shutdown
 // Callers using this context should ensure that they are registered as a running server
 // in order that they are waited for.

modules/graceful/manager.go

@@ -23,6 +23,11 @@ const (
 	stateTerminate
 )
 
+type RunCanceler interface {
+	Run()
+	Cancel()
+}
+
 // There are some places that could inherit sockets:
 //
 // * HTTP or HTTPS main listener
@@ -55,46 +60,19 @@ func InitManager(ctx context.Context) {
 	})
 }
 
-// WithCallback is a runnable to call when the caller has finished
-type WithCallback func(callback func())
-
-// RunnableWithShutdownFns is a runnable with functions to run at shutdown and terminate
-// After the callback to atShutdown is called and is complete, the main function must return.
-// Similarly the callback function provided to atTerminate must return once termination is complete.
-// Please note that use of the atShutdown and atTerminate callbacks will create go-routines that will wait till their respective signals
-// - users must therefore be careful to only call these as necessary.
-type RunnableWithShutdownFns func(atShutdown, atTerminate func(func()))
-
-// RunWithShutdownFns takes a function that has both atShutdown and atTerminate callbacks
-// After the callback to atShutdown is called and is complete, the main function must return.
-// Similarly the callback function provided to atTerminate must return once termination is complete.
-// Please note that use of the atShutdown and atTerminate callbacks will create go-routines that will wait till their respective signals
-// - users must therefore be careful to only call these as necessary.
-func (g *Manager) RunWithShutdownFns(run RunnableWithShutdownFns) {
+// RunWithCancel helps to run a function with a custom context, the Cancel function will be called at shutdown
+// The Cancel function should stop the Run function in predictable time.
+func (g *Manager) RunWithCancel(rc RunCanceler) {
+	g.RunAtShutdown(context.Background(), rc.Cancel)
 	g.runningServerWaitGroup.Add(1)
 	defer g.runningServerWaitGroup.Done()
 	defer func() {
 		if err := recover(); err != nil {
-			log.Critical("PANIC during RunWithShutdownFns: %v\nStacktrace: %s", err, log.Stack(2))
+			log.Critical("PANIC during RunWithCancel: %v\nStacktrace: %s", err, log.Stack(2))
 			g.doShutdown()
 		}
 	}()
-	run(func(atShutdown func()) {
-		g.lock.Lock()
-		defer g.lock.Unlock()
-		g.toRunAtShutdown = append(g.toRunAtShutdown,
-			func() {
-				defer func() {
-					if err := recover(); err != nil {
-						log.Critical("PANIC during RunWithShutdownFns: %v\nStacktrace: %s", err, log.Stack(2))
-						g.doShutdown()
-					}
-				}()
-				atShutdown()
-			})
-	}, func(atTerminate func()) {
-		g.RunAtTerminate(atTerminate)
-	})
+	rc.Run()
 }
 
 // RunWithShutdownContext takes a function that has a context to watch for shutdown.
@@ -151,21 +129,6 @@ func (g *Manager) RunAtShutdown(ctx context.Context, shutdown func()) {
 		})
 }
 
-// RunAtHammer creates a go-routine to run the provided function at shutdown
-func (g *Manager) RunAtHammer(hammer func()) {
-	g.lock.Lock()
-	defer g.lock.Unlock()
-	g.toRunAtHammer = append(g.toRunAtHammer,
-		func() {
-			defer func() {
-				if err := recover(); err != nil {
-					log.Critical("PANIC during RunAtHammer: %v\nStacktrace: %s", err, log.Stack(2))
-				}
-			}()
-			hammer()
-		})
-}
-
 func (g *Manager) doShutdown() {
 	if !g.setStateTransition(stateRunning, stateShuttingDown) {
 		g.DoImmediateHammer()
@@ -206,9 +169,6 @@ func (g *Manager) doHammerTime(d time.Duration) {
 		g.hammerCtxCancel()
 		atHammerCtx := pprof.WithLabels(g.terminateCtx, pprof.Labels("graceful-lifecycle", "post-hammer"))
 		pprof.SetGoroutineLabels(atHammerCtx)
-		for _, fn := range g.toRunAtHammer {
-			go fn()
-		}
 	}
 	g.lock.Unlock()
 }

modules/graceful/manager_unix.go

@@ -41,7 +41,6 @@ type Manager struct {
 	terminateWaitGroup     sync.WaitGroup
 
 	toRunAtShutdown  []func()
-	toRunAtHammer    []func()
 	toRunAtTerminate []func()
 }
 

modules/graceful/manager_windows.go

@@ -50,7 +50,6 @@ type Manager struct {
 	shutdownRequested      chan struct{}
 
 	toRunAtShutdown  []func()
-	toRunAtHammer    []func()
 	toRunAtTerminate []func()
 }
 

modules/indexer/code/indexer.go

@@ -166,7 +166,7 @@ func Init() {
 		handler := func(items ...*IndexerData) (unhandled []*IndexerData) {
 			idx, err := indexer.get()
 			if idx == nil || err != nil {
-				log.Error("Codes indexer handler: unable to get indexer!")
+				log.Warn("Codes indexer handler: indexer is not ready, retry later.")
 				return items
 			}
 
@@ -201,7 +201,7 @@ func Init() {
 			return unhandled
 		}
 
-		indexerQueue = queue.CreateUniqueQueue("code_indexer", handler)
+		indexerQueue = queue.CreateUniqueQueue(ctx, "code_indexer", handler)
 		if indexerQueue == nil {
 			log.Fatal("Unable to create codes indexer queue")
 		}
@@ -259,7 +259,7 @@ func Init() {
 		indexer.set(rIndexer)
 
 		// Start processing the queue
-		go graceful.GetManager().RunWithShutdownFns(indexerQueue.Run)
+		go graceful.GetManager().RunWithCancel(indexerQueue)
 
 		if populate {
 			go graceful.GetManager().RunWithShutdownContext(populateRepoIndexer)

modules/indexer/issues/indexer.go

@@ -102,15 +102,15 @@ var (
 func InitIssueIndexer(syncReindex bool) {
 	ctx, _, finished := process.GetManager().AddTypedContext(context.Background(), "Service: IssueIndexer", process.SystemProcessType, false)
 
-	waitChannel := make(chan time.Duration, 1)
+	indexerInitWaitChannel := make(chan time.Duration, 1)
 
 	// Create the Queue
 	switch setting.Indexer.IssueType {
 	case "bleve", "elasticsearch", "meilisearch":
 		handler := func(items ...*IndexerData) (unhandled []*IndexerData) {
 			indexer := holder.get()
 			if indexer == nil {
-				log.Error("Issue indexer handler: unable to get indexer.")
+				log.Warn("Issue indexer handler: indexer is not ready, retry later.")
 				return items
 			}
 			toIndex := make([]*IndexerData, 0, len(items))
@@ -138,15 +138,17 @@ func InitIssueIndexer(syncReindex bool) {
 			return unhandled
 		}
 
-		issueIndexerQueue = queue.CreateSimpleQueue("issue_indexer", handler)
+		issueIndexerQueue = queue.CreateSimpleQueue(ctx, "issue_indexer", handler)
 
 		if issueIndexerQueue == nil {
 			log.Fatal("Unable to create issue indexer queue")
 		}
 	default:
-		issueIndexerQueue = queue.CreateSimpleQueue[*IndexerData]("issue_indexer", nil)
+		issueIndexerQueue = queue.CreateSimpleQueue[*IndexerData](ctx, "issue_indexer", nil)
 	}
 
+	graceful.GetManager().RunAtTerminate(finished)
+
 	// Create the Indexer
 	go func() {
 		pprof.SetGoroutineLabels(ctx)
@@ -178,51 +180,41 @@ func InitIssueIndexer(syncReindex bool) {
 				if issueIndexer != nil {
 					issueIndexer.Close()
 				}
-				finished()
 				log.Info("PID: %d Issue Indexer closed", os.Getpid())
 			})
 			log.Debug("Created Bleve Indexer")
 		case "elasticsearch":
-			graceful.GetManager().RunWithShutdownFns(func(_, atTerminate func(func())) {
-				pprof.SetGoroutineLabels(ctx)
-				issueIndexer, err := NewElasticSearchIndexer(setting.Indexer.IssueConnStr, setting.Indexer.IssueIndexerName)
-				if err != nil {
-					log.Fatal("Unable to initialize Elastic Search Issue Indexer at connection: %s Error: %v", setting.Indexer.IssueConnStr, err)
-				}
-				exist, err := issueIndexer.Init()
-				if err != nil {
-					log.Fatal("Unable to issueIndexer.Init with connection %s Error: %v", setting.Indexer.IssueConnStr, err)
-				}
-				populate = !exist
-				holder.set(issueIndexer)
-				atTerminate(finished)
-			})
+			issueIndexer, err := NewElasticSearchIndexer(setting.Indexer.IssueConnStr, setting.Indexer.IssueIndexerName)
+			if err != nil {
+				log.Fatal("Unable to initialize Elastic Search Issue Indexer at connection: %s Error: %v", setting.Indexer.IssueConnStr, err)
+			}
+			exist, err := issueIndexer.Init()
+			if err != nil {
+				log.Fatal("Unable to issueIndexer.Init with connection %s Error: %v", setting.Indexer.IssueConnStr, err)
+			}
+			populate = !exist
+			holder.set(issueIndexer)
 		case "db":
 			issueIndexer := &DBIndexer{}
 			holder.set(issueIndexer)
-			graceful.GetManager().RunAtTerminate(finished)
 		case "meilisearch":
-			graceful.GetManager().RunWithShutdownFns(func(_, atTerminate func(func())) {
-				pprof.SetGoroutineLabels(ctx)
-				issueIndexer, err := NewMeilisearchIndexer(setting.Indexer.IssueConnStr, setting.Indexer.IssueConnAuth, setting.Indexer.IssueIndexerName)
-				if err != nil {
-					log.Fatal("Unable to initialize Meilisearch Issue Indexer at connection: %s Error: %v", setting.Indexer.IssueConnStr, err)
-				}
-				exist, err := issueIndexer.Init()
-				if err != nil {
-					log.Fatal("Unable to issueIndexer.Init with connection %s Error: %v", setting.Indexer.IssueConnStr, err)
-				}
-				populate = !exist
-				holder.set(issueIndexer)
-				atTerminate(finished)
-			})
+			issueIndexer, err := NewMeilisearchIndexer(setting.Indexer.IssueConnStr, setting.Indexer.IssueConnAuth, setting.Indexer.IssueIndexerName)
+			if err != nil {
+				log.Fatal("Unable to initialize Meilisearch Issue Indexer at connection: %s Error: %v", setting.Indexer.IssueConnStr, err)
+			}
+			exist, err := issueIndexer.Init()
+			if err != nil {
+				log.Fatal("Unable to issueIndexer.Init with connection %s Error: %v", setting.Indexer.IssueConnStr, err)
+			}
+			populate = !exist
+			holder.set(issueIndexer)
 		default:
 			holder.cancel()
 			log.Fatal("Unknown issue indexer type: %s", setting.Indexer.IssueType)
 		}
 
 		// Start processing the queue
-		go graceful.GetManager().RunWithShutdownFns(issueIndexerQueue.Run)
+		go graceful.GetManager().RunWithCancel(issueIndexerQueue)
 
 		// Populate the index
 		if populate {
@@ -232,13 +224,14 @@ func InitIssueIndexer(syncReindex bool) {
 				go graceful.GetManager().RunWithShutdownContext(populateIssueIndexer)
 			}
 		}
-		waitChannel <- time.Since(start)
-		close(waitChannel)
+
+		indexerInitWaitChannel <- time.Since(start)
+		close(indexerInitWaitChannel)
 	}()
 
 	if syncReindex {
 		select {
-		case <-waitChannel:
+		case <-indexerInitWaitChannel:
 		case <-graceful.GetManager().IsShutdown():
 		}
 	} else if setting.Indexer.StartupTimeout > 0 {
@@ -249,7 +242,7 @@ func InitIssueIndexer(syncReindex bool) {
 				timeout += setting.GracefulHammerTime
 			}
 			select {
-			case duration := <-waitChannel:
+			case duration := <-indexerInitWaitChannel:
 				log.Info("Issue Indexer Initialization took %v", duration)
 			case <-graceful.GetManager().IsShutdown():
 				log.Warn("Shutdown occurred before issue index initialisation was complete")

modules/indexer/stats/queue.go

@@ -29,13 +29,11 @@ func handler(items ...int64) []int64 {
 }
 
 func initStatsQueue() error {
-	statsQueue = queue.CreateUniqueQueue("repo_stats_update", handler)
+	statsQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "repo_stats_update", handler)
 	if statsQueue == nil {
-		return fmt.Errorf("Unable to create repo_stats_update Queue")
+		return fmt.Errorf("unable to create repo_stats_update queue")
 	}
-
-	go graceful.GetManager().RunWithShutdownFns(statsQueue.Run)
-
+	go graceful.GetManager().RunWithCancel(statsQueue)
 	return nil
 }
 

modules/log/event_writer_base.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"regexp"
+	"runtime/pprof"
 	"time"
 )
 
@@ -143,9 +144,17 @@ func eventWriterStartGo(ctx context.Context, w EventWriter, shared bool) {
 	}
 	w.Base().shared = shared
 	w.Base().stopped = make(chan struct{})
+
+	ctxDesc := "Logger: EventWriter: " + w.GetWriterName()
+	if shared {
+		ctxDesc = "Logger: EventWriter (shared): " + w.GetWriterName()
+	}
+	writerCtx, writerCancel := newContext(ctx, ctxDesc)
 	go func() {
+		defer writerCancel()
 		defer close(w.Base().stopped)
-		w.Run(ctx)
+		pprof.SetGoroutineLabels(writerCtx)
+		w.Run(writerCtx)
 	}()
 }
 

modules/log/event_writer_conn_test.go

@@ -40,7 +40,7 @@ func TestConnLogger(t *testing.T) {
 	level := INFO
 	flags := LstdFlags | LUTC | Lfuncname
 
-	logger := NewLoggerWithWriters(context.Background(), NewEventWriterConn("test-conn", WriterMode{
+	logger := NewLoggerWithWriters(context.Background(), "test", NewEventWriterConn("test-conn", WriterMode{
 		Level:        level,
 		Prefix:       prefix,
 		Flags:        FlagsFromBits(flags),