set some pprof labels for different goroutines

Author: cce

agreement/service.go

@@ -19,6 +19,7 @@ package agreement
 //go:generate dbgen -i agree.sql -p agreement -n agree -o agreeInstall.go -h ../scripts/LICENSE_HEADER
 import (
 	"context"
+	"runtime/pprof"
 	"time"
 
 	"github.com/algorand/go-algorand/config"
@@ -143,8 +144,12 @@ func (s *Service) Start() {
 	input := make(chan externalEvent)
 	output := make(chan []action)
 	ready := make(chan externalDemuxSignals)
-	go s.demuxLoop(ctx, input, output, ready)
-	go s.mainLoop(input, output, ready)
+	pprof.Do(context.Background(), pprof.Labels("worker", "agreement.demux"), func(_ context.Context) {
+		go s.demuxLoop(ctx, input, output, ready)
+	})
+	pprof.Do(context.Background(), pprof.Labels("worker", "agreement.main"), func(_ context.Context) {
+		go s.mainLoop(input, output, ready)
+	})
 }
 
 // Shutdown the execution of the protocol.

data/txHandler.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"runtime/pprof"
 	"sync"
 
 	"github.com/algorand/go-algorand/crypto"
@@ -100,7 +101,9 @@ func (handler *TxHandler) Start() {
 		{Tag: protocol.TxnTag, MessageHandler: network.HandlerFunc(handler.processIncomingTxn)},
 	})
 	handler.backlogWg.Add(1)
-	go handler.backlogWorker()
+	pprof.Do(context.Background(), pprof.Labels("worker", "TxHandler.backlogWorker"), func(_ context.Context) {
+		go handler.backlogWorker()
+	})
 }
 
 // Stop suspends the processing of incoming messages at the transaction handler

ledger/notifier.go

@@ -19,6 +19,7 @@ package ledger
 import (
 	"context"
 	"database/sql"
+	"runtime/pprof"
 	"sync"
 
 	"github.com/algorand/go-deadlock"
@@ -92,7 +93,9 @@ func (bn *blockNotifier) loadFromDisk(l ledgerForTracker, _ basics.Round) error
 	bn.running = true
 	bn.pendingBlocks = nil
 	bn.closing.Add(1)
-	go bn.worker()
+	pprof.Do(context.Background(), pprof.Labels("worker", "blockNotifier"), func(_ context.Context) {
+		go bn.worker()
+	})
 	return nil
 }
 

network/wsNetwork.go

@@ -30,6 +30,7 @@ import (
 	"path"
 	"regexp"
 	"runtime"
+	"runtime/pprof"
 	"strconv"
 	"strings"
 	"sync"
@@ -811,7 +812,9 @@ func (wn *WebsocketNetwork) Start() {
 	}
 	if wn.listener != nil {
 		wn.wg.Add(1)
-		go wn.httpdThread()
+		pprof.Do(context.Background(), pprof.Labels("worker", "httpdThread"), func(ctx context.Context) {
+			go wn.httpdThread()
+		})
 	}
 	wn.wg.Add(1)
 	go wn.meshThread()
@@ -821,11 +824,13 @@ func (wn *WebsocketNetwork) Start() {
 		wn.peersConnectivityCheckTicker.Stop()
 	}
 	wn.peersConnectivityCheckTicker = time.NewTicker(connectionActivityMonitorInterval)
-	for i := 0; i < incomingThreads; i++ {
-		wn.wg.Add(1)
-		// We pass the peersConnectivityCheckTicker.C here so that we don't need to syncronize the access to the ticker's data structure.
-		go wn.messageHandlerThread(wn.peersConnectivityCheckTicker.C)
-	}
+	pprof.Do(context.Background(), pprof.Labels("worker", "messageHandlerThread"), func(_ context.Context) {
+		for i := 0; i < incomingThreads; i++ {
+			wn.wg.Add(1)
+			// We pass the peersConnectivityCheckTicker.C here so that we don't need to syncronize the access to the ticker's data structure.
+			go wn.messageHandlerThread(wn.peersConnectivityCheckTicker.C)
+		}
+	})
 	wn.wg.Add(1)
 	go wn.broadcastThread()
 	if wn.prioScheme != nil {

node/node.go

@@ -24,6 +24,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"runtime/pprof"
 	"strings"
 	"sync"
 	"time"
@@ -208,9 +209,11 @@ func MakeFull(log logging.Logger, rootDir string, cfg config.Local, phonebookAdd
 		return nil, err
 	}
 
-	node.cryptoPool = execpool.MakePool(node)
-	node.lowPriorityCryptoVerificationPool = execpool.MakeBacklog(node.cryptoPool, 2*node.cryptoPool.GetParallelism(), execpool.LowPriority, node)
-	node.highPriorityCryptoVerificationPool = execpool.MakeBacklog(node.cryptoPool, 2*node.cryptoPool.GetParallelism(), execpool.HighPriority, node)
+	node.cryptoPool = execpool.MakePool(node, "worker", "cryptoPool")
+	node.lowPriorityCryptoVerificationPool = execpool.MakeBacklog(node.cryptoPool, 2*node.cryptoPool.GetParallelism(),
+		execpool.LowPriority, node, "worker", "lowPriorityCryptoVerificationPool")
+	node.highPriorityCryptoVerificationPool = execpool.MakeBacklog(node.cryptoPool, 2*node.cryptoPool.GetParallelism(),
+		execpool.HighPriority, node, "worker", "highPriorityCryptoVerificationPool")
 	node.ledger, err = data.LoadLedger(node.log, ledgerPathnamePrefix, false, genesis.Proto, genalloc, node.genesisID, node.genesisHash, []ledger.BlockListener{}, cfg)
 	if err != nil {
 		log.Errorf("Cannot initialize ledger (%s): %v", ledgerPathnamePrefix, err)
@@ -378,8 +381,9 @@ func (node *AlgorandFullNode) startMonitoringRoutines() {
 	node.monitoringRoutinesWaitGroup.Add(2)
 	go node.txPoolGaugeThread(node.ctx.Done())
 	// Delete old participation keys
-	go node.oldKeyDeletionThread(node.ctx.Done())
-
+	pprof.Do(context.Background(), pprof.Labels("worker", "oldKeyDeletionThread"), func(_ context.Context) {
+		go node.oldKeyDeletionThread(node.ctx.Done())
+	})
 	// TODO re-enable with configuration flag post V1
 	//go logging.UsageLogThread(node.ctx, node.log, 100*time.Millisecond, nil)
 }

util/execpool/backlog.go

@@ -18,6 +18,7 @@ package execpool
 
 import (
 	"context"
+	"runtime/pprof"
 	"sync"
 )
 
@@ -46,7 +47,7 @@ type BacklogPool interface {
 }
 
 // MakeBacklog creates a backlog
-func MakeBacklog(execPool ExecutionPool, backlogSize int, priority Priority, owner interface{}) BacklogPool {
+func MakeBacklog(execPool ExecutionPool, backlogSize int, priority Priority, owner interface{}, profLabels ...string) BacklogPool {
 	if backlogSize < 0 {
 		return nil
 	}
@@ -58,7 +59,7 @@ func MakeBacklog(execPool ExecutionPool, backlogSize int, priority Priority, own
 	bl.ctx, bl.ctxCancel = context.WithCancel(context.Background())
 	if bl.pool == nil {
 		// create one internally.
-		bl.pool = MakePool(bl)
+		bl.pool = MakePool(bl, append(profLabels, "execpool", "internal")...)
 	}
 	if backlogSize == 0 {
 		// use the number of cpus in the system.
@@ -67,7 +68,9 @@ func MakeBacklog(execPool ExecutionPool, backlogSize int, priority Priority, own
 	bl.buffer = make(chan backlogItemTask, backlogSize)
 
 	bl.wg.Add(1)
-	go bl.worker()
+	pprof.Do(context.Background(), pprof.Labels(profLabels...), func(_ context.Context) {
+		go bl.worker()
+	})
 	return bl
 }
 

util/execpool/pool.go

@@ -19,6 +19,7 @@ package execpool
 import (
 	"context"
 	"runtime"
+	"runtime/pprof"
 	"sync"
 )
 
@@ -68,7 +69,7 @@ type enqueuedTask struct {
 }
 
 // MakePool creates a pool.
-func MakePool(owner interface{}) ExecutionPool {
+func MakePool(owner interface{}, profLabels ...string) ExecutionPool {
 	p := &pool{
 		inputs:  make([]chan enqueuedTask, numPrios),
 		numCPUs: runtime.NumCPU(),
@@ -81,10 +82,11 @@ func MakePool(owner interface{}) ExecutionPool {
 	}
 
 	p.wg.Add(p.numCPUs)
-	for i := 0; i < p.numCPUs; i++ {
-		go p.worker()
-	}
-
+	pprof.Do(context.Background(), pprof.Labels(profLabels...), func(_ context.Context) {
+		for i := 0; i < p.numCPUs; i++ {
+			go p.worker()
+		}
+	})
 	return p
 }