Feat: daemon.BackgroundWorker has to be named

packages/daemon/daemon.go

@@ -5,29 +5,53 @@ import (
 )
 
 var (
-	running           bool
-	wg                sync.WaitGroup
-	ShutdownSignal    = make(chan int, 1)
-	backgroundWorkers = make([]func(), 0)
-	lock              = sync.Mutex{}
+	running                  bool
+	wg                       sync.WaitGroup
+	ShutdownSignal           = make(chan int, 1)
+	backgroundWorkers        = make([]func(), 0)
+	backgroundWorkerNames    = make([]string, 0)
+	runningBackgroundWorkers = make(map[string]bool)
+	lock                     = sync.Mutex{}
 )
 
-func runBackgroundWorker(backgroundWorker func()) {
+func GetRunningBackgroundWorkers() []string {
+	lock.Lock()
+
+	result := make([]string, 0)
+	for runningBackgroundWorker := range runningBackgroundWorkers {
+		result = append(result, runningBackgroundWorker)
+	}
+
+	lock.Unlock()
+
+	return result
+}
+
+func runBackgroundWorker(name string, backgroundWorker func()) {
 	wg.Add(1)
 
 	go func() {
+		lock.Lock()
+		runningBackgroundWorkers[name] = true
+		lock.Unlock()
+
 		backgroundWorker()
 
+		lock.Lock()
+		delete(runningBackgroundWorkers, name)
+		lock.Unlock()
+
 		wg.Done()
 	}()
 }
 
-func BackgroundWorker(handler func()) {
+func BackgroundWorker(name string, handler func()) {
 	lock.Lock()
 
 	if IsRunning() {
-		runBackgroundWorker(handler)
+		runBackgroundWorker(name, handler)
 	} else {
+		backgroundWorkerNames = append(backgroundWorkerNames, name)
 		backgroundWorkers = append(backgroundWorkers, handler)
 	}
 
@@ -45,8 +69,8 @@ func Run() {
 
 			Events.Run.Trigger()
 
-			for _, backgroundWorker := range backgroundWorkers {
-				runBackgroundWorker(backgroundWorker)
+			for i, backgroundWorker := range backgroundWorkers {
+				runBackgroundWorker(backgroundWorkerNames[i], backgroundWorker)
 			}
 		}
 

packages/node/plugin.go

@@ -55,3 +55,16 @@ func (plugin *Plugin) LogFailure(message string) {
 func (plugin *Plugin) LogDebug(message string) {
 	plugin.Node.LogDebug(plugin.Name, message)
 }
+
+var TestNode = &Node{
+	loggers:       make([]*Logger, 0),
+	wg:            &sync.WaitGroup{},
+	loadedPlugins: make([]*Plugin, 0),
+}
+
+func (plugin *Plugin) InitTest() {
+	plugin.Node = TestNode
+
+	plugin.Events.Configure.Trigger(plugin)
+	plugin.Events.Run.Trigger(plugin)
+}

packages/transactionspammer/transactionspammer.go

@@ -23,7 +23,7 @@ func Start(tps int64) {
 		shutdownSignal = make(chan int, 1)
 
 		func(shutdownSignal chan int) {
-			daemon.BackgroundWorker(func() {
+			daemon.BackgroundWorker("Transaction Spammer", func() {
 				for {
 					start := time.Now()
 					sentCounter := int64(0)

plugins/analysis/client/plugin.go

@@ -21,7 +21,7 @@ import (
 )
 
 func Run(plugin *node.Plugin) {
-	daemon.BackgroundWorker(func() {
+	daemon.BackgroundWorker("Analysis Client", func() {
 		shuttingDown := false
 
 		for !shuttingDown {

plugins/analysis/server/plugin.go

@@ -36,7 +36,7 @@ func Configure(plugin *node.Plugin) {
 }
 
 func Run(plugin *node.Plugin) {
-	daemon.BackgroundWorker(func() {
+	daemon.BackgroundWorker("Analysis Server", func() {
 		plugin.LogInfo("Starting Server (port " + strconv.Itoa(*SERVER_PORT.Value) + ") ...")
 
 		server.Listen(*SERVER_PORT.Value)

plugins/analysis/webinterface/httpserver/plugin.go

@@ -32,7 +32,7 @@ func Configure(plugin *node.Plugin) {
 }
 
 func Run(plugin *node.Plugin) {
-	daemon.BackgroundWorker(func() {
+	daemon.BackgroundWorker("Analysis HTTP Server", func() {
 		httpServer.ListenAndServe()
 	})
 }

plugins/autopeering/instances/neighborhood/instance.go

@@ -35,7 +35,7 @@ func Configure(plugin *node.Plugin) {
 }
 
 func Run(plugin *node.Plugin) {
-	daemon.BackgroundWorker(func() {
+	daemon.BackgroundWorker("Neighborhood Updater", func() {
 		timeutil.Ticker(updateNeighborHood, 1*time.Second)
 	})
 }

plugins/autopeering/protocol/outgoing_request_processor.go

@@ -41,7 +41,11 @@ func createOutgoingRequestProcessor(plugin *node.Plugin) func() {
 
 func sendOutgoingRequests(plugin *node.Plugin) {
 	for _, chosenNeighborCandidate := range chosenneighbors.CANDIDATES.Clone() {
-		time.Sleep(5 * time.Second)
+		select {
+		case <-daemon.ShutdownSignal:
+			return
+		case <-time.After(5 * time.Second):
+		}
 
 		if candidateShouldBeContacted(chosenNeighborCandidate) {
 			if dialed, err := chosenNeighborCandidate.Send(outgoingrequest.INSTANCE.Marshal(), types.PROTOCOL_TYPE_TCP, true); err != nil {

plugins/autopeering/protocol/plugin.go

@@ -20,8 +20,8 @@ func Configure(plugin *node.Plugin) {
 }
 
 func Run(plugin *node.Plugin) {
-	daemon.BackgroundWorker(createChosenNeighborDropper(plugin))
-	daemon.BackgroundWorker(createAcceptedNeighborDropper(plugin))
-	daemon.BackgroundWorker(createOutgoingRequestProcessor(plugin))
-	daemon.BackgroundWorker(createOutgoingPingProcessor(plugin))
+	daemon.BackgroundWorker("Autopeering Chosen Neighbor Dropper", createChosenNeighborDropper(plugin))
+	daemon.BackgroundWorker("Autopeering Accepted Neighbor Dropper", createAcceptedNeighborDropper(plugin))
+	daemon.BackgroundWorker("Autopeering Outgoing Request Processor", createOutgoingRequestProcessor(plugin))
+	daemon.BackgroundWorker("Autopeering Outgoing Ping Processor", createOutgoingPingProcessor(plugin))
 }

plugins/autopeering/saltmanager/saltmanager.go

@@ -68,7 +68,7 @@ func scheduleUpdateForSalt(saltToUpdate *salt.Salt, settingsKey []byte, lifeSpan
 	if saltToUpdate.ExpirationTime.Before(now) {
 		updatePublicSalt(saltToUpdate, settingsKey, lifeSpan, callback)
 	} else {
-		daemon.BackgroundWorker(func() {
+		daemon.BackgroundWorker("Salt Updater", func() {
 			select {
 			case <-time.After(saltToUpdate.ExpirationTime.Sub(now)):
 				updatePublicSalt(saltToUpdate, settingsKey, lifeSpan, callback)

plugins/autopeering/server/tcp/server.go

@@ -37,7 +37,7 @@ func ConfigureServer(plugin *node.Plugin) {
 }
 
 func RunServer(plugin *node.Plugin) {
-	daemon.BackgroundWorker(func() {
+	daemon.BackgroundWorker("Autopeering TCP Server", func() {
 		if *parameters.ADDRESS.Value == "0.0.0.0" {
 			plugin.LogInfo("Starting TCP Server (port " + strconv.Itoa(*parameters.PORT.Value) + ") ...")
 		} else {

plugins/autopeering/server/udp/server.go

@@ -41,7 +41,7 @@ func ConfigureServer(plugin *node.Plugin) {
 }
 
 func RunServer(plugin *node.Plugin) {
-	daemon.BackgroundWorker(func() {
+	daemon.BackgroundWorker("Autopeering UDP Server", func() {
 		if *parameters.ADDRESS.Value == "0.0.0.0" {
 			plugin.LogInfo("Starting UDP Server (port " + strconv.Itoa(*parameters.PORT.Value) + ") ...")
 		} else {

plugins/bundleprocessor/bundleprocessor.go

@@ -28,7 +28,7 @@ func ProcessSolidBundleHead(headTransaction *value_transaction.ValueTransaction)
 			if currentTransaction.IsHead() && currentTransaction != headTransaction {
 				newBundle.SetTransactionHashes(mapTransactionsToTransactionHashes(bundleTransactions))
 
-				Events.InvalidBundleReceived.Trigger(newBundle, bundleTransactions)
+				Events.InvalidBundle.Trigger(newBundle, bundleTransactions)
 
 				return nil, ErrProcessBundleFailed.Derive(errors.New("invalid bundle found"), "missing bundle tail")
 			}
@@ -52,11 +52,7 @@ func ProcessSolidBundleHead(headTransaction *value_transaction.ValueTransaction)
 			if currentTransaction.IsTail() {
 				newBundle.SetTransactionHashes(mapTransactionsToTransactionHashes(bundleTransactions))
 
-				if newBundle.IsValueBundle() {
-					Events.ValueBundleReceived.Trigger(newBundle, bundleTransactions)
-				} else {
-					Events.DataBundleReceived.Trigger(newBundle, bundleTransactions)
-				}
+				Events.BundleSolid.Trigger(newBundle, bundleTransactions)
 
 				return newBundle, nil
 			}

plugins/bundleprocessor/bundleprocessor_test.go

+package bundleprocessor
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/iotaledger/goshimmer/packages/events"
+	"github.com/iotaledger/goshimmer/packages/model/bundle"
+	"github.com/iotaledger/goshimmer/packages/model/value_transaction"
+	"github.com/iotaledger/goshimmer/packages/ternary"
+	"github.com/iotaledger/goshimmer/plugins/tangle"
+	"github.com/magiconair/properties/assert"
+)
+
+func TestProcessSolidBundleHead(t *testing.T) {
+	tangle.PLUGIN.InitTest()
+
+	tx := value_transaction.New()
+	tx.SetTail(true)
+	tx.SetValue(3)
+
+	tx1 := value_transaction.New()
+	tx1.SetTrunkTransactionHash(tx.GetHash())
+	tx1.SetHead(true)
+
+	tangle.StoreTransaction(tx)
+	tangle.StoreTransaction(tx1)
+
+	Events.BundleSolid.Attach(events.NewClosure(func(bundle *bundle.Bundle, transactions []*value_transaction.ValueTransaction) {
+		fmt.Println("IT HAPPENED")
+		fmt.Println(bundle.GetHash())
+	}))
+
+	result, err := ProcessSolidBundleHead(tx1)
+	if err != nil {
+		t.Error(err)
+	} else {
+		assert.Equal(t, result.GetHash(), ternary.Trytes("UFWJYEWKMEQDNSQUCUWBGOFRHVBGHVVYEZCLCGRDTRQSMAFALTIPMJEEYFDPMQCNJWLXUWFMBZGHQRO99"), "invalid bundle hash")
+		assert.Equal(t, result.IsValueBundle(), true, "invalid value bundle status")
+	}
+}

plugins/bundleprocessor/events.go

@@ -7,15 +7,13 @@ import (
 )
 
 var Events = pluginEvents{
-	DataBundleReceived:    events.NewEvent(bundleEventCaller),
-	ValueBundleReceived:   events.NewEvent(bundleEventCaller),
-	InvalidBundleReceived: events.NewEvent(bundleEventCaller),
+	BundleSolid:   events.NewEvent(bundleEventCaller),
+	InvalidBundle: events.NewEvent(bundleEventCaller),
 }
 
 type pluginEvents struct {
-	DataBundleReceived    *events.Event
-	ValueBundleReceived   *events.Event
-	InvalidBundleReceived *events.Event
+	BundleSolid   *events.Event
+	InvalidBundle *events.Event
 }
 
 func bundleEventCaller(handler interface{}, params ...interface{}) {

plugins/bundleprocessor/plugin.go

@@ -34,7 +34,7 @@ func configure(plugin *node.Plugin) {
 func run(plugin *node.Plugin) {
 	plugin.LogInfo("Starting Bundle Processor ...")
 
-	daemon.BackgroundWorker(func() {
+	daemon.BackgroundWorker("Bundle Processor", func() {
 		plugin.LogSuccess("Starting Bundle Processor ... done")
 
 		workerPool.Run()

plugins/gossip/neighbors.go

@@ -47,7 +47,7 @@ func runNeighbors(plugin *node.Plugin) {
 }
 
 func manageConnection(plugin *node.Plugin, neighbor *Neighbor) {
-	daemon.BackgroundWorker(func() {
+	daemon.BackgroundWorker("Connection Manager ("+neighbor.Identity.StringIdentifier+")", func() {
 		failedConnectionAttempts := 0
 
 		for _, exists := GetNeighbor(neighbor.Identity.StringIdentifier); exists && failedConnectionAttempts < CONNECTION_MAX_ATTEMPTS; {

plugins/gossip/send_queue.go

@@ -26,7 +26,7 @@ func configureSendQueue(plugin *node.Plugin) {
 func runSendQueue(plugin *node.Plugin) {
 	plugin.LogInfo("Starting Send Queue Dispatcher ...")
 
-	daemon.BackgroundWorker(func() {
+	daemon.BackgroundWorker("Gossip Send Queue Dispatcher", func() {
 		plugin.LogSuccess("Starting Send Queue Dispatcher ... done")
 
 		for {
@@ -54,7 +54,7 @@ func runSendQueue(plugin *node.Plugin) {
 
 	connectedNeighborsMutex.Lock()
 	for _, neighborQueue := range neighborQueues {
-		startNeighborSendQueue(neighborQueue)
+		startNeighborSendQueue(neighborQueue.protocol.Neighbor, neighborQueue)
 	}
 	connectedNeighborsMutex.Unlock()
 }
@@ -104,13 +104,13 @@ func setupEventHandlers(neighbor *Neighbor) {
 		}))
 
 		if daemon.IsRunning() {
-			startNeighborSendQueue(queue)
+			startNeighborSendQueue(neighbor, queue)
 		}
 	}))
 }
 
-func startNeighborSendQueue(neighborQueue *neighborQueue) {
-	daemon.BackgroundWorker(func() {
+func startNeighborSendQueue(neighbor *Neighbor, neighborQueue *neighborQueue) {
+	daemon.BackgroundWorker("Gossip Send Queue ("+neighbor.Identity.StringIdentifier+")", func() {
 		for {
 			select {
 			case <-daemon.ShutdownSignal:

plugins/gossip/server.go

@@ -73,7 +73,7 @@ func configureServer(plugin *node.Plugin) {
 func runServer(plugin *node.Plugin) {
 	plugin.LogInfo("Starting TCP Server (port " + strconv.Itoa(*PORT.Value) + ") ...")
 
-	daemon.BackgroundWorker(func() {
+	daemon.BackgroundWorker("Gossip TCP Server", func() {
 		plugin.LogSuccess("Starting TCP Server (port " + strconv.Itoa(*PORT.Value) + ") ... done")
 
 		TCPServer.Listen(*PORT.Value)

plugins/gracefulshutdown/plugin.go

@@ -4,6 +4,7 @@ import (
 	"os"
 	"os/signal"
 	"strconv"
+	"strings"
 	"syscall"
 	"time"
 
@@ -31,7 +32,13 @@ var PLUGIN = node.NewPlugin("Graceful Shutdown", func(plugin *node.Plugin) {
 				secondsSinceStart := x.Sub(start).Seconds()
 
 				if secondsSinceStart <= WAIT_TO_KILL_TIME_IN_SECONDS {
-					plugin.LogWarning("Received shutdown request - waiting (max " + strconv.Itoa(WAIT_TO_KILL_TIME_IN_SECONDS-int(secondsSinceStart)) + " seconds) to finish processing ...")
+					processList := ""
+					runningBackgroundWorkers := daemon.GetRunningBackgroundWorkers()
+					if len(runningBackgroundWorkers) >= 1 {
+						processList = "(" + strings.Join(runningBackgroundWorkers, ", ") + ") "
+					}
+
+					plugin.LogWarning("Received shutdown request - waiting (max " + strconv.Itoa(WAIT_TO_KILL_TIME_IN_SECONDS-int(secondsSinceStart)) + " seconds) to finish processing " + processList + "...")
 				} else {
 					plugin.LogFailure("Background processes did not terminate in time! Forcing shutdown ...")