v1.4.18-patch-2

Author: CassOnMars

nekryptology/pkg/vdf/wesolowski.go

@@ -89,7 +89,10 @@ func GenerateVDFWithStopChan(seed []byte, iterations, int_size_bits uint32, stop
 func GenerateVDFIteration(seed, x_blob []byte, iterations, int_size_bits uint32) ([]byte, []byte) {
 	int_size := (int_size_bits + 16) >> 4
 	D := iqc.CreateDiscriminant(seed, int_size_bits)
-	x, _ := iqc.NewClassGroupFromBytesDiscriminant(x_blob[:(2*int_size)], D)
+	x, ok := iqc.NewClassGroupFromBytesDiscriminant(x_blob[:(2*int_size)], D)
+	if !ok {
+		return nil, nil
+	}
 
 	y, proof := calculateVDF(D, x, iterations, int_size_bits, nil)
 
@@ -105,18 +108,36 @@ func VerifyVDF(seed, proof_blob []byte, iterations, int_size_bits uint32) bool {
 
 	D := iqc.CreateDiscriminant(seed, int_size_bits)
 	x := iqc.NewClassGroupFromAbDiscriminant(big.NewInt(2), big.NewInt(1), D)
-	y, _ := iqc.NewClassGroupFromBytesDiscriminant(proof_blob[:(2*int_size)], D)
-	proof, _ := iqc.NewClassGroupFromBytesDiscriminant(proof_blob[2*int_size:], D)
+	y, ok := iqc.NewClassGroupFromBytesDiscriminant(proof_blob[:(2*int_size)], D)
+	if !ok {
+		return false
+	}
+
+	proof, ok := iqc.NewClassGroupFromBytesDiscriminant(proof_blob[2*int_size:], D)
+	if !ok {
+		return false
+	}
 
 	return verifyProof(x, y, proof, iterations)
 }
 
 func VerifyVDFIteration(seed, x_blob, proof_blob []byte, iterations, int_size_bits uint32) bool {
 	int_size := (int_size_bits + 16) >> 4
 	D := iqc.CreateDiscriminant(seed, int_size_bits)
-	x, _ := iqc.NewClassGroupFromBytesDiscriminant(x_blob[:(2*int_size)], D)
-	y, _ := iqc.NewClassGroupFromBytesDiscriminant(proof_blob[:(2*int_size)], D)
-	proof, _ := iqc.NewClassGroupFromBytesDiscriminant(proof_blob[2*int_size:], D)
+	x, ok := iqc.NewClassGroupFromBytesDiscriminant(x_blob[:(2*int_size)], D)
+	if !ok {
+		return false
+	}
+
+	y, ok := iqc.NewClassGroupFromBytesDiscriminant(proof_blob[:(2*int_size)], D)
+	if !ok {
+		return false
+	}
+
+	proof, ok := iqc.NewClassGroupFromBytesDiscriminant(proof_blob[2*int_size:], D)
+	if !ok {
+		return false
+	}
 
 	return verifyProof(x, y, proof, iterations)
 }

node/config/engine.go

@@ -8,6 +8,14 @@ type EngineConfig struct {
 	PendingCommitWorkers int64  `yaml:"pendingCommitWorkers"`
 	MinimumPeersRequired int    `yaml:"minimumPeersRequired"`
 	StatsMultiaddr       string `yaml:"statsMultiaddr"`
+	// Sets the fmt.Sprintf format string to use as the listen multiaddrs for
+	// data worker processes
+	DataWorkerBaseListenMultiaddr string `yaml:"dataWorkerBaseListenMultiaddr"`
+	// Sets the starting port number to use as the listen port for data worker
+	// processes, incrementing by 1 until n-1, n = cores. (Example: a 4 core
+	// system, base listen port of 40000 will listen on 40000, 40001, 40002)
+	DataWorkerBaseListenPort uint16 `yaml:"dataWorkerBaseListenPort"`
+	DataWorkerMemoryLimit    int64  `yaml:"dataWorkerMemoryLimit"`
 
 	// Values used only for testing – do not override these in production, your
 	// node will get kicked out

node/config/version.go

@@ -29,5 +29,5 @@ func FormatVersion(version []byte) string {
 }
 
 func GetPatchNumber() byte {
-	return 0x01
+	return 0x02
 }

node/consensus/master/broadcast_messaging.go

@@ -116,15 +116,19 @@ func (e *MasterClockConsensusEngine) handleSelfTestReport(
 	peerID []byte,
 	any *anypb.Any,
 ) error {
-	if bytes.Equal(peerID, e.pubSub.GetPeerID()) {
-		return nil
-	}
-
 	report := &protobufs.SelfTestReport{}
 	if err := any.UnmarshalTo(report); err != nil {
 		return errors.Wrap(err, "handle self test report")
 	}
 
+	if bytes.Equal(peerID, e.pubSub.GetPeerID()) {
+		info := e.peerInfoManager.GetPeerInfo(peerID)
+		info.LastSeen = time.Now().UnixMilli()
+		info.DifficultyMetric = report.DifficultyMetric
+		info.MasterHeadFrame = report.MasterHeadFrame
+		return nil
+	}
+
 	// minimum proof size is one timestamp, one vdf proof, must match one fewer
 	// than core count
 	if len(report.Proof) < 516+8 ||

node/consensus/master/master_clock_consensus_engine.go

@@ -7,6 +7,7 @@ import (
 	"crypto/rand"
 	"encoding/binary"
 	"encoding/hex"
+	"fmt"
 	"io"
 	"math/big"
 	"sync"
@@ -15,9 +16,12 @@ import (
 	"github.com/iden3/go-iden3-crypto/poseidon"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/mr-tron/base58"
+	"github.com/multiformats/go-multiaddr"
+	mn "github.com/multiformats/go-multiaddr/net"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 	"source.quilibrium.com/quilibrium/monorepo/node/config"
 	"source.quilibrium.com/quilibrium/monorepo/node/consensus"
 	qtime "source.quilibrium.com/quilibrium/monorepo/node/consensus/time"
@@ -68,6 +72,7 @@ type MasterClockConsensusEngine struct {
 	verifyTestCh                chan verifyChallenge
 	currentReceivingSyncPeers   int
 	currentReceivingSyncPeersMx sync.Mutex
+	engineConfig                *config.EngineConfig
 }
 
 var _ consensus.ConsensusEngine = (*MasterClockConsensusEngine)(nil)
@@ -131,6 +136,7 @@ func NewMasterClockConsensusEngine(
 		frameValidationCh:   make(chan *protobufs.ClockFrame),
 		bandwidthTestCh:     make(chan []byte),
 		verifyTestCh:        make(chan verifyChallenge),
+		engineConfig:        engineConfig,
 	}
 
 	e.addPeerManifestReport(e.pubSub.GetPeerID(), report)
@@ -241,6 +247,16 @@ func (e *MasterClockConsensusEngine) Start() <-chan error {
 		time.Sleep(30 * time.Second)
 		difficultyMetric := int64(100000)
 		skew := (difficultyMetric * 12) / 10
+		parallelism := e.report.Cores - 1
+
+		if parallelism < 3 {
+			panic("invalid system configuration, minimum system configuration must be four cores")
+		}
+
+		clients, err := e.createParallelDataClients(int(parallelism))
+		if err != nil {
+			panic(err)
+		}
 
 		for {
 			head, err := e.masterTimeReel.Head()
@@ -250,23 +266,49 @@ func (e *MasterClockConsensusEngine) Start() <-chan error {
 
 			e.report.MasterHeadFrame = head.FrameNumber
 			e.report.DifficultyMetric = difficultyMetric
-			parallelism := e.report.Cores - 1
 
 			challenge := binary.BigEndian.AppendUint64(
 				[]byte{},
 				e.report.MasterHeadFrame,
 			)
 			challenge = append(challenge, e.pubSub.GetPeerID()...)
 
-			ts, proofs, nextDifficultyMetric, err :=
-				e.frameProver.CalculateChallengeProof(
-					challenge,
-					parallelism,
-					skew,
-				)
-			if err != nil {
-				panic(err)
+			proofs := make([][]byte, parallelism)
+			nextMetrics := make([]int64, parallelism)
+
+			wg := sync.WaitGroup{}
+			wg.Add(int(parallelism))
+
+			ts := time.Now().UnixMilli()
+			for i := uint32(0); i < parallelism; i++ {
+				i := i
+				go func() {
+					resp, err :=
+						clients[i].CalculateChallengeProof(
+							context.Background(),
+							&protobufs.ChallengeProofRequest{
+								Challenge: challenge,
+								Core:      i,
+								Skew:      skew,
+								NowMs:     ts,
+							},
+						)
+					if err != nil {
+						panic(err)
+					}
+
+					proofs[i], nextMetrics[i] = resp.Output, resp.NextSkew
+					wg.Done()
+				}()
 			}
+			wg.Wait()
+			nextDifficultySum := uint64(0)
+			for i := 0; i < int(parallelism); i++ {
+				nextDifficultySum += uint64(nextMetrics[i])
+			}
+
+			nextDifficultyMetric := int64(nextDifficultySum / uint64(parallelism))
+
 			e.logger.Info(
 				"recalibrating difficulty metric",
 				zap.Int64("previous_difficulty_metric", difficultyMetric),
@@ -336,6 +378,64 @@ func (e *MasterClockConsensusEngine) Start() <-chan error {
 	return errChan
 }
 
+func (e *MasterClockConsensusEngine) createParallelDataClients(
+	paralellism int,
+) ([]protobufs.DataIPCServiceClient, error) {
+	e.logger.Info(
+		"connecting to data worker processes",
+		zap.Int("parallelism", paralellism),
+	)
+
+	if e.engineConfig.DataWorkerBaseListenMultiaddr == "" {
+		e.engineConfig.DataWorkerBaseListenMultiaddr = "/ip4/127.0.0.1/tcp/%d"
+	}
+
+	if e.engineConfig.DataWorkerBaseListenPort == 0 {
+		e.engineConfig.DataWorkerBaseListenPort = 40000
+	}
+
+	clients := make([]protobufs.DataIPCServiceClient, paralellism)
+
+	for i := 0; i < paralellism; i++ {
+		ma, err := multiaddr.NewMultiaddr(
+			fmt.Sprintf(
+				e.engineConfig.DataWorkerBaseListenMultiaddr,
+				int(e.engineConfig.DataWorkerBaseListenPort)+i,
+			),
+		)
+		if err != nil {
+			panic(err)
+		}
+
+		_, addr, err := mn.DialArgs(ma)
+		if err != nil {
+			panic(err)
+		}
+
+		conn, err := grpc.Dial(
+			addr,
+			grpc.WithTransportCredentials(
+				insecure.NewCredentials(),
+			),
+			grpc.WithDefaultCallOptions(
+				grpc.MaxCallSendMsgSize(10*1024*1024),
+				grpc.MaxCallRecvMsgSize(10*1024*1024),
+			),
+		)
+		if err != nil {
+			panic(err)
+		}
+
+		clients[i] = protobufs.NewDataIPCServiceClient(conn)
+	}
+
+	e.logger.Info(
+		"connected to data worker processes",
+		zap.Int("parallelism", paralellism),
+	)
+	return clients, nil
+}
+
 func (e *MasterClockConsensusEngine) PerformValidation(
 	ctx context.Context,
 	msg *protobufs.ValidationMessage,

node/crypto/frame_prover.go

@@ -53,9 +53,10 @@ type FrameProver interface {
 	) bool
 	CalculateChallengeProof(
 		challenge []byte,
-		parallelism uint32,
+		core uint32,
 		skew int64,
-	) (int64, [][]byte, int64, error)
+		nowMs int64,
+	) ([]byte, int64, error)
 	VerifyChallengeProof(
 		challenge []byte,
 		timestamp int64,

node/crypto/wesolowski_frame_prover.go

@@ -6,7 +6,6 @@ import (
 	"crypto/rand"
 	"encoding/binary"
 	"math/big"
-	"sync"
 	"time"
 
 	"github.com/cloudflare/circl/sign/ed448"
@@ -601,43 +600,31 @@ func (w *WesolowskiFrameProver) VerifyWeakRecursiveProof(
 
 func (w *WesolowskiFrameProver) CalculateChallengeProof(
 	challenge []byte,
-	parallelism uint32,
+	core uint32,
 	skew int64,
-) (int64, [][]byte, int64, error) {
-	now := time.Now()
-	nowMs := now.UnixMilli()
+	nowMs int64,
+) ([]byte, int64, error) {
 	input := binary.BigEndian.AppendUint64([]byte{}, uint64(nowMs))
 	input = append(input, challenge...)
-	outputs := make([][]byte, parallelism)
-
-	wg := sync.WaitGroup{}
-	wg.Add(int(parallelism))
 
 	// 4.5 minutes = 270 seconds, one increment should be ten seconds
 	proofDuration := 270 * 1000
 	calibratedDifficulty := (int64(proofDuration) * 10000) / skew
-	for i := uint32(0); i < parallelism; i++ {
-		i := i
-		go func() {
-			instanceInput := binary.BigEndian.AppendUint32([]byte{}, i)
-			instanceInput = append(instanceInput, input...)
-			b := sha3.Sum256(instanceInput)
-			v := vdf.New(uint32(calibratedDifficulty), b)
-
-			v.Execute()
-			o := v.GetOutput()
+	instanceInput := binary.BigEndian.AppendUint32([]byte{}, core)
+	instanceInput = append(instanceInput, input...)
+	b := sha3.Sum256(instanceInput)
+	v := vdf.New(uint32(calibratedDifficulty), b)
 
-			outputs[i] = make([]byte, 516)
-			copy(outputs[i][:], o[:])
-			wg.Done()
-		}()
-	}
+	v.Execute()
+	o := v.GetOutput()
 
-	wg.Wait()
+	output := make([]byte, 516)
+	copy(output[:], o[:])
+	now := time.UnixMilli(nowMs)
 	after := time.Since(now)
 	nextSkew := (skew * after.Milliseconds()) / int64(proofDuration)
 
-	return nowMs, outputs, nextSkew, nil
+	return output, nextSkew, nil
 }
 
 func (w *WesolowskiFrameProver) VerifyChallengeProof(

node/crypto/wesolowski_frame_prover_test.go

@@ -30,10 +30,12 @@ func TestMasterProve(t *testing.T) {
 func TestChallengeProof(t *testing.T) {
 	l, _ := zap.NewProduction()
 	w := crypto.NewWesolowskiFrameProver(l)
-	now, proofs, nextSkew, err := w.CalculateChallengeProof([]byte{0x01, 0x02, 0x03}, 3, 120000)
+	now := time.Now().UnixMilli()
+	proofs, nextSkew, err := w.CalculateChallengeProof([]byte{0x01, 0x02, 0x03}, 0, 120000, now)
 	assert.NoError(t, err)
-	assert.True(t, w.VerifyChallengeProof([]byte{0x01, 0x02, 0x03}, now, 100000, proofs))
-	now, proofs, _, err = w.CalculateChallengeProof([]byte{0x01, 0x02, 0x03}, 3, nextSkew*12/10)
+	assert.True(t, w.VerifyChallengeProof([]byte{0x01, 0x02, 0x03}, now, 100000, [][]byte{proofs}))
+	now = time.Now().UnixMilli()
+	proofs, _, err = w.CalculateChallengeProof([]byte{0x01, 0x02, 0x03}, 0, nextSkew*12/10, now)
 	assert.NoError(t, err)
-	assert.True(t, w.VerifyChallengeProof([]byte{0x01, 0x02, 0x03}, now, nextSkew, proofs))
+	assert.True(t, w.VerifyChallengeProof([]byte{0x01, 0x02, 0x03}, now, nextSkew, [][]byte{proofs}))
 }