Parallel miner

Co-authored-by: stasatdaglabs <39559713+stasatdaglabs@users.noreply.github.com>

.github/workflows/go-deploy.yml

@@ -37,24 +37,33 @@ jobs:
         # `-extldflags=-static` - means static link everything, `-tags netgo,osusergo` means use pure go replacements for "os/user" and "net"
         # `-s -w` strips the binary to produce smaller size binaries
         run: |
-          binary="kaspad-${{ github.event.release.tag_name }}-linux"
-          echo "binary=${binary}" >> $GITHUB_ENV
-          go build -v -ldflags="-s -w -extldflags=-static" -tags netgo,osusergo -o "${binary}"
+          go build -v -ldflags="-s -w -extldflags=-static" -tags netgo,osusergo -o ./bin/ ./...
+          archive="bin/kaspad-${{ github.event.release.tag_name }}-linux.zip"
+          asset_name="kaspad-${{ github.event.release.tag_name }}-linux.zip"
+          zip -r "${archive}" ./bin/*
+          echo "archive=${archive}" >> $GITHUB_ENV
+          echo "asset_name=${asset_name}" >> $GITHUB_ENV
 
       - name: Build on Windows
         if: runner.os == 'Windows'
         shell: bash
         run: |
-          binary="kaspad-${{ github.event.release.tag_name }}-win64.exe"
-          echo "binary=${binary}" >> $GITHUB_ENV
-          go build -v -ldflags="-s -w" -o "${binary}"
+          go build -v -ldflags="-s -w" -o bin/ ./...
+          archive="bin/kaspad-${{ github.event.release.tag_name }}-win64.zip"
+          asset_name="kaspad-${{ github.event.release.tag_name }}-win64.zip"
+          powershell "Compress-Archive bin/* \"${archive}\""
+          echo "archive=${archive}" >> $GITHUB_ENV
+          echo "asset_name=${asset_name}" >> $GITHUB_ENV
 
       - name: Build on MacOS
         if: runner.os == 'macOS'
         run: |
-          binary="kaspad-${{ github.event.release.tag_name }}-osx"
-          echo "binary=${binary}" >> $GITHUB_ENV
-          go build -v -ldflags="-s -w" -o "${binary}"
+          go build -v -ldflags="-s -w" -o ./bin/ ./...
+          archive="bin/kaspad-${{ github.event.release.tag_name }}-osx.zip"
+          asset_name="kaspad-${{ github.event.release.tag_name }}-osx.zip"
+          zip -r "${archive}" ./bin/*
+          echo "archive=${archive}" >> $GITHUB_ENV
+          echo "asset_name=${asset_name}" >> $GITHUB_ENV
 
 
       - name: Upload Release Asset
@@ -63,6 +72,6 @@ jobs:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         with:
           upload_url: ${{ github.event.release.upload_url }}
-          asset_path: "./${{ env.binary }}"
-          asset_name: "${{ env.binary }}"
+          asset_path: "./${{ env.archive }}"
+          asset_name: "${{ env.asset_name }}"
           asset_content_type: application/zip

app/protocol/flows/blockrelay/handle_relay_invs.go

@@ -1,6 +1,8 @@
 package blockrelay
 
 import (
+	"time"
+
 	"github.com/kaspanet/kaspad/app/appmessage"
 	"github.com/kaspanet/kaspad/app/protocol/common"
 	peerpkg "github.com/kaspanet/kaspad/app/protocol/peer"
@@ -11,6 +13,7 @@ import (
 	"github.com/kaspanet/kaspad/domain/consensus/utils/consensushashing"
 	"github.com/kaspanet/kaspad/infrastructure/config"
 	"github.com/kaspanet/kaspad/infrastructure/network/netadapter/router"
+	"github.com/kaspanet/kaspad/util/mstime"
 	"github.com/pkg/errors"
 )
 
@@ -109,6 +112,8 @@ func (flow *handleRelayInvsFlow) start() error {
 		if err != nil {
 			return err
 		}
+		msTime := mstime.UnixMilliseconds(block.Header.TimeInMilliseconds())
+		time.Sleep(time.Until(msTime.ToNativeTime().Add(config.DelayDuration)))
 
 		log.Debugf("Processing block %s", inv.Hash)
 		missingParents, blockInsertionResult, err := flow.processBlock(block)

app/protocol/protocol.go

@@ -238,7 +238,7 @@ func (m *Manager) registerTransactionRelayFlow(router *routerpkg.Router, isStopp
 	outgoingRoute := router.OutgoingRoute()
 
 	return []*flow{
-		m.registerFlow("HandleRelayedTransactions", router,
+		m.registerFlowWithCapacity("HandleRelayedTransactions", 10_000, router,
 			[]appmessage.MessageCommand{appmessage.CmdInvTransaction, appmessage.CmdTx, appmessage.CmdTransactionNotFound}, isStopping, errChan,
 			func(incomingRoute *routerpkg.Route, peer *peerpkg.Peer) error {
 				return transactionrelay.HandleRelayedTransactions(m.context, incomingRoute, outgoingRoute)
@@ -274,6 +274,24 @@ func (m *Manager) registerFlow(name string, router *routerpkg.Router, messageTyp
 		panic(err)
 	}
 
+	return m.registerFlowForRoute(route, name, isStopping, errChan, initializeFunc)
+}
+
+func (m *Manager) registerFlowWithCapacity(name string, capacity int, router *routerpkg.Router,
+	messageTypes []appmessage.MessageCommand, isStopping *uint32,
+	errChan chan error, initializeFunc flowInitializeFunc) *flow {
+
+	route, err := router.AddIncomingRouteWithCapacity(capacity, messageTypes)
+	if err != nil {
+		panic(err)
+	}
+
+	return m.registerFlowForRoute(route, name, isStopping, errChan, initializeFunc)
+}
+
+func (m *Manager) registerFlowForRoute(route *routerpkg.Route, name string, isStopping *uint32,
+	errChan chan error, initializeFunc flowInitializeFunc) *flow {
+
 	return &flow{
 		name: name,
 		executeFunc: func(peer *peerpkg.Peer) {

changelog.txt

@@ -1,4 +1,12 @@
 
+Kaspad v0.9.2 - 2021-03-31
+===========================
+* Increase the route capacity of InvTransaction messages. (#1603) (#1637)
+
+Kaspad v0.9.1 - 2021-03-14
+===========================
+* Testnet network reset
+
 Kaspad v0.9.0 - 2021-03-04
 ===========================
 

cmd/kaspaminer/mineloop.go

@@ -2,7 +2,9 @@ package main
 
 import (
 	nativeerrors "errors"
+	"fmt"
 	"math/rand"
+	"runtime"
 	"sync/atomic"
 	"time"
 
@@ -42,39 +44,43 @@ func mineLoop(client *minerClient, numberOfBlocks uint64, targetBlocksPerSecond
 		templatesLoop(client, miningAddr, errChan)
 	})
 
-	spawn("blocksLoop", func() {
-		const windowSize = 10
-		var expectedDurationForWindow time.Duration
-		var windowExpectedEndTime time.Time
-		hasBlockRateTarget := targetBlocksPerSecond != 0
-		if hasBlockRateTarget {
-			expectedDurationForWindow = time.Duration(float64(windowSize)/targetBlocksPerSecond) * time.Second
-			windowExpectedEndTime = time.Now().Add(expectedDurationForWindow)
-		}
-		blockInWindowIndex := 0
-
-		sleepTime := 0 * time.Second
-
-		for {
-			foundBlockChan <- mineNextBlock(mineWhenNotSynced)
-
+	for c := 0; c < (runtime.NumCPU()/2)+1; c++ {
+		c := c
+		spawn(fmt.Sprintf("blocksLoop %d", c), func() {
+			const windowSize = 10
+			var expectedDurationForWindow time.Duration
+			var windowExpectedEndTime time.Time
+			hasBlockRateTarget := targetBlocksPerSecond != 0
 			if hasBlockRateTarget {
-				blockInWindowIndex++
-				if blockInWindowIndex == windowSize-1 {
-					deviation := windowExpectedEndTime.Sub(time.Now())
-					if deviation > 0 {
-						sleepTime = deviation / windowSize
-						log.Infof("Finished to mine %d blocks %s earlier than expected. Setting the miner "+
-							"to sleep %s between blocks to compensate",
-							windowSize, deviation, sleepTime)
-					}
-					blockInWindowIndex = 0
-					windowExpectedEndTime = time.Now().Add(expectedDurationForWindow)
-				}
-				time.Sleep(sleepTime)
+				expectedDurationForWindow = time.Duration(float64(windowSize)/targetBlocksPerSecond) * time.Second
+				windowExpectedEndTime = time.Now().Add(expectedDurationForWindow)
 			}
-		}
-	})
+			blockInWindowIndex := 0
+
+			sleepTime := 0 * time.Second
+
+			for {
+				foundBlockChan <- mineNextBlock(mineWhenNotSynced)
+
+				if hasBlockRateTarget {
+					blockInWindowIndex++
+					if blockInWindowIndex == windowSize-1 {
+						deviation := windowExpectedEndTime.Sub(time.Now())
+						if deviation > 0 {
+							sleepTime = deviation / windowSize
+							log.Infof("cpu: %d Finished to mine %d blocks %s earlier than expected. Setting the miner "+
+								"to sleep %s between blocks to compensate",
+								c, windowSize, deviation, sleepTime)
+						}
+						blockInWindowIndex = 0
+						windowExpectedEndTime = time.Now().Add(expectedDurationForWindow)
+					}
+					time.Sleep(sleepTime)
+				}
+			}
+		})
+
+	}
 
 	spawn("handleFoundBlock", func() {
 		for i := uint64(0); numberOfBlocks == 0 || i < numberOfBlocks; i++ {

domain/consensus/processes/dagtraversalmanager/window_test.go

@@ -130,37 +130,37 @@ func TestBlockWindow(t *testing.T) {
 			{
 				parents:                          []string{"H", "F"},
 				id:                               "I",
-				expectedWindowWithGenesisPadding: []string{"F", "H", "C", "D", "B", "G", "A", "A", "A", "A"},
+				expectedWindowWithGenesisPadding: []string{"F", "C", "H", "D", "G", "B", "A", "A", "A", "A"},
 			},
 			{
 				parents:                          []string{"I"},
 				id:                               "J",
-				expectedWindowWithGenesisPadding: []string{"I", "F", "H", "C", "D", "B", "G", "A", "A", "A"},
+				expectedWindowWithGenesisPadding: []string{"I", "F", "C", "H", "D", "G", "B", "A", "A", "A"},
 			},
 			{
 				parents:                          []string{"J"},
 				id:                               "K",
-				expectedWindowWithGenesisPadding: []string{"J", "I", "F", "H", "C", "D", "B", "G", "A", "A"},
+				expectedWindowWithGenesisPadding: []string{"J", "I", "F", "C", "H", "D", "G", "B", "A", "A"},
 			},
 			{
 				parents:                          []string{"K"},
 				id:                               "L",
-				expectedWindowWithGenesisPadding: []string{"K", "J", "I", "F", "H", "C", "D", "B", "G", "A"},
+				expectedWindowWithGenesisPadding: []string{"K", "J", "I", "F", "C", "H", "D", "G", "B", "A"},
 			},
 			{
 				parents:                          []string{"L"},
 				id:                               "M",
-				expectedWindowWithGenesisPadding: []string{"L", "K", "J", "I", "F", "H", "C", "D", "B", "G"},
+				expectedWindowWithGenesisPadding: []string{"L", "K", "J", "I", "F", "C", "H", "D", "G", "B"},
 			},
 			{
 				parents:                          []string{"M"},
 				id:                               "N",
-				expectedWindowWithGenesisPadding: []string{"M", "L", "K", "J", "I", "F", "H", "C", "D", "B"},
+				expectedWindowWithGenesisPadding: []string{"M", "L", "K", "J", "I", "F", "C", "H", "D", "G"},
 			},
 			{
 				parents:                          []string{"N"},
 				id:                               "O",
-				expectedWindowWithGenesisPadding: []string{"N", "M", "L", "K", "J", "I", "F", "H", "C", "D"},
+				expectedWindowWithGenesisPadding: []string{"N", "M", "L", "K", "J", "I", "F", "C", "H", "D"},
 			},
 		},
 		dagconfig.DevnetParams.Name: {

domain/dagconfig/genesis.go

@@ -182,10 +182,10 @@ var testnetGenesisCoinbaseTx = transactionhelper.NewSubnetworkTransaction(0,
 // testnetGenesisHash is the hash of the first block in the block DAG for the test
 // network (genesis block).
 var testnetGenesisHash = externalapi.NewDomainHashFromByteArray(&[externalapi.DomainHashSize]byte{
-	0x5d, 0xb7, 0x49, 0xc1, 0x6e, 0xfb, 0x4e, 0x7a,
-	0x0c, 0x9f, 0xd1, 0x80, 0x74, 0x91, 0x60, 0xd0,
-	0x1b, 0x84, 0xc7, 0x92, 0xa8, 0x5f, 0xcf, 0x9b,
-	0x1d, 0x8c, 0x8c, 0x34, 0xa9, 0x41, 0x5f, 0xa5,
+	0x49, 0xa6, 0x19, 0xe4, 0x25, 0xa0, 0x8d, 0xae,
+	0xd9, 0x67, 0x76, 0x82, 0xd8, 0x8e, 0x93, 0xa3,
+	0xf5, 0x42, 0x5a, 0x02, 0x4d, 0x55, 0xef, 0x5e,
+	0x39, 0x61, 0x9f, 0x2d, 0xd9, 0x51, 0xe4, 0x55,
 })
 
 // testnetGenesisMerkleRoot is the hash of the first transaction in the genesis block
@@ -206,9 +206,9 @@ var testnetGenesisBlock = externalapi.DomainBlock{
 		testnetGenesisMerkleRoot,
 		&externalapi.DomainHash{},
 		&externalapi.DomainHash{},
-		0x176eb9ddb6d,
+		0x177e83f864a,
 		0x1e7fffff,
-		0x5dba6,
+		0x4d8e0,
 	),
 	Transactions: []*externalapi.DomainTransaction{testnetGenesisCoinbaseTx},
 }

domain/dagconfig/params.go

@@ -256,11 +256,11 @@ var MainnetParams = Params{
 // TestnetParams defines the network parameters for the test Kaspa network.
 var TestnetParams = Params{
 	K:           defaultGHOSTDAGK,
-	Name:        "kaspa-testnet-2",
+	Name:        "kaspa-testnet-4",
 	Net:         appmessage.Testnet,
 	RPCPort:     "16210",
 	DefaultPort: "16211",
-	DNSSeeds:    []string{"testnet-2-dnsseed.daglabs-dev.com"},
+	DNSSeeds:    []string{"testnet-4-dnsseed.daglabs-dev.com"},
 
 	// DAG parameters
 	GenesisBlock:                   &testnetGenesisBlock,

infrastructure/config/config.go

@@ -25,6 +25,9 @@ import (
 	"github.com/pkg/errors"
 )
 
+// DelayDuration a duration for the delay, global for testing
+var DelayDuration time.Duration
+
 const (
 	defaultConfigFilename      = "kaspad.conf"
 	defaultDataDirname         = "data"
@@ -117,6 +120,7 @@ type Flags struct {
 	MaxUTXOCacheSize     uint64        `long:"maxutxocachesize" description:"Max size of loaded UTXO into ram from the disk in bytes"`
 	UTXOIndex            bool          `long:"utxoindex" description:"Enable the UTXO index"`
 	IsArchivalNode       bool          `long:"archival" description:"Run as an archival node: don't delete old block data when moving the pruning point (Warning: heavy disk usage)'"`
+	Delay                float32       `long:"delay" description:"Provide a delay in seconds as a floating point"`
 	NetworkFlags
 	ServiceOptions *ServiceOptions
 }
@@ -574,6 +578,7 @@ func LoadConfig() (*Config, error) {
 		log.Warnf("%s", configFileError)
 	}
 
+	DelayDuration = time.Duration(cfg.Delay * float32(time.Second))
 	return cfg, nil
 }
 

infrastructure/network/netadapter/netadapter.go

@@ -3,6 +3,7 @@ package netadapter
 import (
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/kaspanet/kaspad/app/appmessage"
 	"github.com/kaspanet/kaspad/infrastructure/config"
@@ -175,18 +176,20 @@ func (na *NetAdapter) ID() *id.ID {
 // P2PBroadcast sends the given `message` to every peer corresponding
 // to each NetConnection in the given netConnections
 func (na *NetAdapter) P2PBroadcast(netConnections []*NetConnection, message appmessage.Message) error {
-	na.p2pConnectionsLock.RLock()
-	defer na.p2pConnectionsLock.RUnlock()
-
-	for _, netConnection := range netConnections {
-		err := netConnection.router.OutgoingRoute().Enqueue(message)
-		if err != nil {
-			if errors.Is(err, routerpkg.ErrRouteClosed) {
-				log.Debugf("Cannot enqueue message to %s: router is closed", netConnection)
-				continue
+	go func() {
+		time.Sleep(config.DelayDuration)
+		na.p2pConnectionsLock.RLock()
+		defer na.p2pConnectionsLock.RUnlock()
+		for _, netConnection := range netConnections {
+			err := netConnection.router.OutgoingRoute().Enqueue(message)
+			if err != nil {
+				if errors.Is(err, routerpkg.ErrRouteClosed) {
+					log.Debugf("Cannot enqueue message to %s: router is closed", netConnection)
+					continue
+				}
+				log.Error(err)
 			}
-			return err
 		}
-	}
+	}()
 	return nil
 }

infrastructure/network/netadapter/router/router.go

@@ -35,13 +35,32 @@ func NewRouter() *Router {
 // be routed to the given `route`
 func (r *Router) AddIncomingRoute(messageTypes []appmessage.MessageCommand) (*Route, error) {
 	route := NewRoute()
+	err := r.initializeIncomingRoute(route, messageTypes)
+	if err != nil {
+		return nil, err
+	}
+	return route, nil
+}
+
+// AddIncomingRouteWithCapacity registers the messages of types `messageTypes` to
+// be routed to the given `route` with a capacity of `capacity`
+func (r *Router) AddIncomingRouteWithCapacity(capacity int, messageTypes []appmessage.MessageCommand) (*Route, error) {
+	route := newRouteWithCapacity(capacity)
+	err := r.initializeIncomingRoute(route, messageTypes)
+	if err != nil {
+		return nil, err
+	}
+	return route, nil
+}
+
+func (r *Router) initializeIncomingRoute(route *Route, messageTypes []appmessage.MessageCommand) error {
 	for _, messageType := range messageTypes {
 		if r.doesIncomingRouteExist(messageType) {
-			return nil, errors.Errorf("a route for '%s' already exists", messageType)
+			return errors.Errorf("a route for '%s' already exists", messageType)
 		}
 		r.setIncomingRoute(messageType, route)
 	}
-	return route, nil
+	return nil
 }
 
 // RemoveRoute unregisters the messages of types `messageTypes` from

version/version.go

@@ -11,7 +11,7 @@ const validCharacters = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrs
 const (
 	appMajor uint = 0
 	appMinor uint = 9
-	appPatch uint = 0
+	appPatch uint = 2
 )
 
 // appBuild is defined as a variable so it can be overridden during the build