diff --git a/packages/ca/heartbeat_manager_events.go b/packages/ca/events.go
similarity index 51%
rename from packages/ca/heartbeat_manager_events.go
rename to packages/ca/events.go
index 01179eea90e18ee288133ef59ed5805a158467ab..8611cf42e9e99d1eb622c0e039b4d5d15f5e93a3 100644
--- a/packages/ca/heartbeat_manager_events.go
+++ b/packages/ca/events.go
@@ -10,6 +10,23 @@ type HeartbeatManagerEvents struct {
RemoveNeighbor *events.Event
}
+type NeighborManagerEvents struct {
+ AddNeighbor *events.Event
+ RemoveNeighbor *events.Event
+ NeighborActive *events.Event
+ NeighborIdle *events.Event
+ ChainReset *events.Event
+ StatementMissing *events.Event
+}
+
+type StatementChainEvents struct {
+ Reset *events.Event
+}
+
+func HashCaller(handler interface{}, params ...interface{}) {
+ handler.(func([]byte))(params[0].([]byte))
+}
+
func IdentityNeighborManagerCaller(handler interface{}, params ...interface{}) {
handler.(func(*identity.Identity, *NeighborManager))(params[0].(*identity.Identity), params[1].(*NeighborManager))
}
diff --git a/packages/ca/heartbeat/heartbeat.go b/packages/ca/heartbeat/heartbeat.go
index 0873ff994f7176bd383ed29e53e0c6bde60fb04c..fe7df8e5a2536c71c87bcbcfdda38d369e4bcef0 100644
--- a/packages/ca/heartbeat/heartbeat.go
+++ b/packages/ca/heartbeat/heartbeat.go
@@ -17,11 +17,13 @@ import (
type Heartbeat struct {
nodeId string
mainStatement *OpinionStatement
+ droppedNeighbors [][]byte
neighborStatements map[string][]*OpinionStatement
signature []byte
nodeIdMutex sync.RWMutex
mainStatementMutex sync.RWMutex
+ droppedNeighborsMutex sync.RWMutex
neighborStatementsMutex sync.RWMutex
signatureMutex sync.RWMutex
}
@@ -58,6 +60,20 @@ func (heartbeat *Heartbeat) SetMainStatement(mainStatement *OpinionStatement) {
heartbeat.mainStatement = mainStatement
}
+func (heartbeat *Heartbeat) GetDroppedNeighbors() [][]byte {
+ heartbeat.droppedNeighborsMutex.RLock()
+ defer heartbeat.droppedNeighborsMutex.RUnlock()
+
+ return heartbeat.droppedNeighbors
+}
+
+func (heartbeat *Heartbeat) SetDroppedNeighbors(droppedNeighbors [][]byte) {
+ heartbeat.droppedNeighborsMutex.Lock()
+ defer heartbeat.droppedNeighborsMutex.Unlock()
+
+ heartbeat.droppedNeighbors = droppedNeighbors
+}
+
func (heartbeat *Heartbeat) GetNeighborStatements() map[string][]*OpinionStatement {
heartbeat.neighborStatementsMutex.RLock()
defer heartbeat.neighborStatementsMutex.RUnlock()
@@ -141,6 +157,7 @@ func (heartbeat *Heartbeat) FromProto(proto proto.Message) {
heartbeat.nodeId = protoHeartbeat.NodeId
heartbeat.mainStatement = &mainStatement
+ heartbeat.droppedNeighbors = protoHeartbeat.DroppedNeighbors
heartbeat.neighborStatements = neighborStatements
heartbeat.signature = protoHeartbeat.Signature
}
@@ -158,6 +175,7 @@ func (heartbeat *Heartbeat) ToProto() proto.Message {
return &heartbeatProto.HeartBeat{
NodeId: heartbeat.nodeId,
+ DroppedNeighbors: heartbeat.droppedNeighbors,
MainStatement: heartbeat.mainStatement.ToProto().(*heartbeatProto.OpinionStatement),
NeighborStatements: neighborStatements,
Signature: heartbeat.signature,
@@ -176,6 +194,7 @@ func (heartbeat *Heartbeat) String() string {
return stringify.Struct("Heartbeat",
stringify.StructField("nodeId", heartbeat.nodeId),
stringify.StructField("mainStatement", heartbeat.mainStatement),
+ stringify.StructField("droppedNeighbors", heartbeat.droppedNeighbors),
stringify.StructField("neighborStatements", heartbeat.neighborStatements),
stringify.StructField("signature", heartbeat.signature),
)
diff --git a/packages/ca/heartbeat/proto/heartbeat.pb.go b/packages/ca/heartbeat/proto/heartbeat.pb.go
index ee8c13d28c54d29dd3590b7b9f489c0d527deda6..71d92c9b60ae353b54cc942e3bc30f4801e44b35 100644
--- a/packages/ca/heartbeat/proto/heartbeat.pb.go
+++ b/packages/ca/heartbeat/proto/heartbeat.pb.go
@@ -23,8 +23,9 @@ const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
type HeartBeat struct {
NodeId string `protobuf:"bytes,1,opt,name=nodeId,proto3" json:"nodeId,omitempty"`
MainStatement *OpinionStatement `protobuf:"bytes,2,opt,name=mainStatement,proto3" json:"mainStatement,omitempty"`
- NeighborStatements []*OpinionStatement `protobuf:"bytes,3,rep,name=neighborStatements,proto3" json:"neighborStatements,omitempty"`
- Signature []byte `protobuf:"bytes,4,opt,name=signature,proto3" json:"signature,omitempty"`
+ DroppedNeighbors [][]byte `protobuf:"bytes,3,rep,name=droppedNeighbors,proto3" json:"droppedNeighbors,omitempty"`
+ NeighborStatements []*OpinionStatement `protobuf:"bytes,4,rep,name=neighborStatements,proto3" json:"neighborStatements,omitempty"`
+ Signature []byte `protobuf:"bytes,5,opt,name=signature,proto3" json:"signature,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
@@ -69,6 +70,13 @@ func (m *HeartBeat) GetMainStatement() *OpinionStatement {
return nil
}
+func (m *HeartBeat) GetDroppedNeighbors() [][]byte {
+ if m != nil {
+ return m.DroppedNeighbors
+ }
+ return nil
+}
+
func (m *HeartBeat) GetNeighborStatements() []*OpinionStatement {
if m != nil {
return m.NeighborStatements
@@ -90,16 +98,18 @@ func init() {
func init() { proto.RegisterFile("heartbeat.proto", fileDescriptor_3c667767fb9826a9) }
var fileDescriptor_3c667767fb9826a9 = []byte{
- // 176 bytes of a gzipped FileDescriptorProto
+ // 200 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0xcf, 0x48, 0x4d, 0x2c,
0x2a, 0x49, 0x4a, 0x4d, 0x2c, 0xd1, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x62, 0x05, 0x53, 0x52,
0xe2, 0xf9, 0x05, 0x99, 0x79, 0x99, 0xf9, 0x79, 0xf1, 0xc5, 0x25, 0x89, 0x25, 0xa9, 0xb9, 0xa9,
- 0x79, 0x50, 0x79, 0xa5, 0x93, 0x8c, 0x5c, 0x9c, 0x1e, 0x20, 0x3d, 0x4e, 0xa9, 0x89, 0x25, 0x42,
+ 0x79, 0x50, 0x79, 0xa5, 0xaf, 0x8c, 0x5c, 0x9c, 0x1e, 0x20, 0x3d, 0x4e, 0xa9, 0x89, 0x25, 0x42,
0x62, 0x5c, 0x6c, 0x79, 0xf9, 0x29, 0xa9, 0x9e, 0x29, 0x12, 0x8c, 0x0a, 0x8c, 0x1a, 0x9c, 0x41,
0x50, 0x9e, 0x90, 0x2d, 0x17, 0x6f, 0x6e, 0x62, 0x66, 0x5e, 0x30, 0x4c, 0xb3, 0x04, 0x93, 0x02,
0xa3, 0x06, 0xb7, 0x91, 0x38, 0xc4, 0x10, 0x3d, 0x7f, 0x88, 0xe1, 0x70, 0xe9, 0x20, 0x54, 0xd5,
- 0x42, 0xee, 0x5c, 0x42, 0x79, 0xa9, 0x99, 0xe9, 0x19, 0x49, 0xf9, 0x45, 0x70, 0xc1, 0x62, 0x09,
- 0x66, 0x05, 0x66, 0x7c, 0x66, 0x60, 0xd1, 0x22, 0x24, 0xc3, 0xc5, 0x59, 0x9c, 0x99, 0x9e, 0x97,
- 0x58, 0x52, 0x5a, 0x94, 0x2a, 0xc1, 0xa2, 0xc0, 0xa8, 0xc1, 0x13, 0x84, 0x10, 0x48, 0x62, 0x03,
- 0x9b, 0x64, 0x0c, 0x08, 0x00, 0x00, 0xff, 0xff, 0xde, 0x82, 0x59, 0xb6, 0x05, 0x01, 0x00, 0x00,
+ 0x42, 0x5a, 0x5c, 0x02, 0x29, 0x45, 0xf9, 0x05, 0x05, 0xa9, 0x29, 0x7e, 0xa9, 0x99, 0xe9, 0x19,
+ 0x49, 0xf9, 0x45, 0xc5, 0x12, 0xcc, 0x0a, 0xcc, 0x1a, 0x3c, 0x41, 0x18, 0xe2, 0x42, 0xee, 0x5c,
+ 0x42, 0x79, 0x50, 0x0e, 0xdc, 0x80, 0x62, 0x09, 0x16, 0x05, 0x66, 0x7c, 0xf6, 0x61, 0xd1, 0x22,
+ 0x24, 0xc3, 0xc5, 0x59, 0x9c, 0x99, 0x9e, 0x97, 0x58, 0x52, 0x5a, 0x94, 0x2a, 0xc1, 0xaa, 0xc0,
+ 0xa8, 0xc1, 0x13, 0x84, 0x10, 0x48, 0x62, 0x03, 0x9b, 0x64, 0x0c, 0x08, 0x00, 0x00, 0xff, 0xff,
+ 0xc2, 0x6a, 0xf1, 0xb9, 0x31, 0x01, 0x00, 0x00,
}
diff --git a/packages/ca/heartbeat/proto/heartbeat.proto b/packages/ca/heartbeat/proto/heartbeat.proto
index afce2528ce4cf014837ce982e1b2e98b3a70b363..8ddc8d6349630c45e8789696b379aedd02874aa1 100644
--- a/packages/ca/heartbeat/proto/heartbeat.proto
+++ b/packages/ca/heartbeat/proto/heartbeat.proto
@@ -7,8 +7,8 @@ package proto;
message HeartBeat {
string nodeId = 1;
OpinionStatement mainStatement = 2;
- repeated OpinionStatement neighborStatements = 3;
- repeated string removedNeighbors = 4;
+ repeated bytes droppedNeighbors = 3;
+ repeated OpinionStatement neighborStatements = 4;
bytes signature = 5;
}
diff --git a/packages/ca/heartbeat_chain.go b/packages/ca/heartbeat_chain.go
deleted file mode 100644
index ed90265d718a989deebc518d00d449734a75c823..0000000000000000000000000000000000000000
--- a/packages/ca/heartbeat_chain.go
+++ /dev/null
@@ -1,16 +0,0 @@
-package ca
-
-import (
- "github.com/iotaledger/goshimmer/packages/ca/heartbeat"
-)
-
-type HeartbeatChain struct {
- missingHeartbeats map[string]bool
- heartbeats map[string]*heartbeat.Heartbeat
-}
-
-func NewHeartbeatChain() *HeartbeatChain {
- return &HeartbeatChain{
- heartbeats: make(map[string]*heartbeat.Heartbeat),
- }
-}
diff --git a/packages/ca/heartbeat_manager.go b/packages/ca/heartbeat_manager.go
index 16afa1db5e40fda4000af46e5e42c56acb619f54..00ae5b0bab2cd49d3ad2dfbd4120c4c295d986bf 100644
--- a/packages/ca/heartbeat_manager.go
+++ b/packages/ca/heartbeat_manager.go
@@ -19,10 +19,13 @@ type HeartbeatManager struct {
identity *identity.Identity
options *HeartbeatManagerOptions
statementChain *StatementChain
+ droppedNeighbors [][]byte
neighborManagers map[string]*NeighborManager
initialOpinions map[string]bool
+ droppedNeighborsMutex sync.RWMutex
neighborManagersMutex sync.RWMutex
+ initialOpinionsMutex sync.RWMutex
}
func NewHeartbeatManager(identity *identity.Identity, options ...HeartbeatManagerOption) *HeartbeatManager {
@@ -36,49 +39,96 @@ func NewHeartbeatManager(identity *identity.Identity, options ...HeartbeatManage
options: DEFAULT_OPTIONS.Override(options...),
statementChain: NewStatementChain(),
+ droppedNeighbors: make([][]byte, 0),
neighborManagers: make(map[string]*NeighborManager),
initialOpinions: make(map[string]bool),
}
}
func (heartbeatManager *HeartbeatManager) AddNeighbor(neighborIdentity *identity.Identity) {
+ heartbeatManager.neighborManagersMutex.RLock()
if _, exists := heartbeatManager.neighborManagers[neighborIdentity.StringIdentifier]; !exists {
- newNeighborManager := NewNeighborManager()
+ heartbeatManager.neighborManagersMutex.RUnlock()
+
+ heartbeatManager.neighborManagersMutex.Lock()
+ if _, exists := heartbeatManager.neighborManagers[neighborIdentity.StringIdentifier]; !exists {
+ newNeighborManager := NewNeighborManager()
- heartbeatManager.neighborManagers[neighborIdentity.StringIdentifier] = newNeighborManager
+ heartbeatManager.neighborManagers[neighborIdentity.StringIdentifier] = newNeighborManager
+ heartbeatManager.neighborManagersMutex.Unlock()
- heartbeatManager.Events.AddNeighbor.Trigger(neighborIdentity, newNeighborManager)
+ heartbeatManager.Events.AddNeighbor.Trigger(neighborIdentity, newNeighborManager)
+ } else {
+ heartbeatManager.neighborManagersMutex.Unlock()
+ }
+ } else {
+ heartbeatManager.neighborManagersMutex.RUnlock()
+ }
+}
+
+func (heartbeatManager *HeartbeatManager) RemoveNeighbor(neighborIdentity *identity.Identity) {
+ heartbeatManager.neighborManagersMutex.RLock()
+ if _, exists := heartbeatManager.neighborManagers[neighborIdentity.StringIdentifier]; exists {
+ heartbeatManager.neighborManagersMutex.RUnlock()
+
+ heartbeatManager.neighborManagersMutex.Lock()
+ if neighborManager, exists := heartbeatManager.neighborManagers[neighborIdentity.StringIdentifier]; exists {
+ delete(heartbeatManager.neighborManagers, neighborIdentity.StringIdentifier)
+ heartbeatManager.neighborManagersMutex.Unlock()
+
+ heartbeatManager.droppedNeighborsMutex.Lock()
+ heartbeatManager.droppedNeighbors = append(heartbeatManager.droppedNeighbors, neighborIdentity.Identifier)
+ heartbeatManager.droppedNeighborsMutex.Unlock()
+
+ heartbeatManager.Events.RemoveNeighbor.Trigger(neighborIdentity, neighborManager)
+ } else {
+ heartbeatManager.neighborManagersMutex.Unlock()
+ }
+ } else {
+ heartbeatManager.neighborManagersMutex.RUnlock()
}
}
func (heartbeatManager *HeartbeatManager) InitialDislike(transactionId []byte) {
+ heartbeatManager.initialOpinionsMutex.Lock()
heartbeatManager.initialOpinions[string(transactionId)] = false
+ heartbeatManager.initialOpinionsMutex.Unlock()
}
func (heartbeatManager *HeartbeatManager) InitialLike(transactionId []byte) {
+ heartbeatManager.initialOpinionsMutex.Lock()
heartbeatManager.initialOpinions[string(transactionId)] = true
+ heartbeatManager.initialOpinionsMutex.Unlock()
}
func (heartbeatManager *HeartbeatManager) GenerateHeartbeat() (result *heartbeat.Heartbeat, err errors.IdentifiableError) {
- if mainStatement, mainStatementErr := heartbeatManager.generateMainStatement(); mainStatementErr == nil {
- if neighborStatements, neighborStatementErr := heartbeatManager.generateNeighborStatements(); neighborStatementErr == nil {
- generatedHeartbeat := heartbeat.NewHeartbeat()
- generatedHeartbeat.SetNodeId(heartbeatManager.identity.StringIdentifier)
- generatedHeartbeat.SetMainStatement(mainStatement)
- generatedHeartbeat.SetNeighborStatements(neighborStatements)
-
- if signingErr := generatedHeartbeat.Sign(heartbeatManager.identity); signingErr == nil {
- result = generatedHeartbeat
- } else {
- err = signingErr
- }
- } else {
- err = neighborStatementErr
- }
- } else {
+ mainStatement, mainStatementErr := heartbeatManager.generateMainStatement()
+ if mainStatementErr != nil {
err = mainStatementErr
+
+ return
}
+ neighborStatements, neighborStatementErr := heartbeatManager.generateNeighborStatements()
+ if neighborStatementErr != nil {
+ err = neighborStatementErr
+
+ return
+ }
+
+ generatedHeartbeat := heartbeat.NewHeartbeat()
+ generatedHeartbeat.SetNodeId(heartbeatManager.identity.StringIdentifier)
+ generatedHeartbeat.SetMainStatement(mainStatement)
+ generatedHeartbeat.SetDroppedNeighbors(heartbeatManager.generateDroppedNeighbors())
+ generatedHeartbeat.SetNeighborStatements(neighborStatements)
+
+ signingErr := generatedHeartbeat.Sign(heartbeatManager.identity)
+ if signingErr != nil {
+ err = signingErr
+ }
+
+ result = generatedHeartbeat
+
return
}
@@ -112,7 +162,7 @@ func (heartbeatManager *HeartbeatManager) generateMainStatement() (result *heart
mainStatement.SetTime(uint64(time.Now().Unix()))
mainStatement.SetToggledTransactions(heartbeatManager.generateToggledTransactions())
- if lastAppliedStatement := heartbeatManager.statementChain.lastAppliedStatement; lastAppliedStatement != nil {
+ if lastAppliedStatement := heartbeatManager.statementChain.GetTail(); lastAppliedStatement != nil {
mainStatement.SetPreviousStatementHash(lastAppliedStatement.GetHash())
}
@@ -120,7 +170,7 @@ func (heartbeatManager *HeartbeatManager) generateMainStatement() (result *heart
result = mainStatement
heartbeatManager.resetInitialOpinions()
- heartbeatManager.statementChain.lastAppliedStatement = mainStatement
+ heartbeatManager.statementChain.tail = mainStatement
} else {
err = signingErr
}
@@ -128,6 +178,19 @@ func (heartbeatManager *HeartbeatManager) generateMainStatement() (result *heart
return
}
+func (heartbeatManager *HeartbeatManager) generateDroppedNeighbors() (result [][]byte) {
+ heartbeatManager.droppedNeighborsMutex.RLock()
+ result = make([][]byte, len(heartbeatManager.droppedNeighbors))
+ copy(result, heartbeatManager.droppedNeighbors)
+ heartbeatManager.droppedNeighborsMutex.RUnlock()
+
+ heartbeatManager.droppedNeighborsMutex.Lock()
+ heartbeatManager.droppedNeighbors = make([][]byte, 0)
+ heartbeatManager.droppedNeighborsMutex.Unlock()
+
+ return
+}
+
func (heartbeatManager *HeartbeatManager) generateNeighborStatements() (result map[string][]*heartbeat.OpinionStatement, err errors.IdentifiableError) {
result = make(map[string][]*heartbeat.OpinionStatement)
diff --git a/packages/ca/heartbeat_manager_test.go b/packages/ca/heartbeat_manager_test.go
index 5b90540a452c40b52a4f53971af36174aff2029f..83cd9c3edb918b71cf3ffa4397d6d85049d3bb10 100644
--- a/packages/ca/heartbeat_manager_test.go
+++ b/packages/ca/heartbeat_manager_test.go
@@ -20,11 +20,14 @@ func generateRandomTransactionId() (result []byte) {
func TestHeartbeatManager_GenerateHeartbeat(t *testing.T) {
ownIdentity := identity.GenerateRandomIdentity()
neighborIdentity := identity.GenerateRandomIdentity()
+ droppedNeighborIdentity := identity.GenerateRandomIdentity()
// generate first heartbeat ////////////////////////////////////////////////////////////////////////////////////////
heartbeatManager1 := NewHeartbeatManager(ownIdentity)
heartbeatManager1.AddNeighbor(neighborIdentity)
+ heartbeatManager1.AddNeighbor(droppedNeighborIdentity)
+ heartbeatManager1.RemoveNeighbor(droppedNeighborIdentity)
heartbeatManager1.InitialDislike(generateRandomTransactionId())
heartbeatManager1.InitialDislike(generateRandomTransactionId())
heartbeatManager1.InitialLike(generateRandomTransactionId())
diff --git a/packages/ca/neighbor_manager.go b/packages/ca/neighbor_manager.go
index d5b77118f866000b2e5f7988e9d89b4cd6822152..07e0936acf6b8988a1807ecfbd8cfd2282f62f5d 100644
--- a/packages/ca/neighbor_manager.go
+++ b/packages/ca/neighbor_manager.go
@@ -2,7 +2,7 @@ package ca
import (
"bytes"
- "sort"
+ "fmt"
"strconv"
"github.com/iotaledger/goshimmer/packages/typeutils"
@@ -29,12 +29,21 @@ type NeighborManager struct {
func NewNeighborManager(options ...NeighborManagerOption) *NeighborManager {
return &NeighborManager{
Events: NeighborManagerEvents{
+ AddNeighbor: events.NewEvent(func(handler interface{}, params ...interface{}) {
+
+ }),
+ RemoveNeighbor: events.NewEvent(func(handler interface{}, params ...interface{}) {
+
+ }),
ChainReset: events.NewEvent(events.CallbackCaller),
StatementMissing: events.NewEvent(HashCaller),
},
- options: DEFAULT_NEIGHBOR_MANAGER_OPTIONS.Override(options...),
- mainChain: NewStatementChain(),
- neighborChains: make(map[string]*StatementChain),
+ options: DEFAULT_NEIGHBOR_MANAGER_OPTIONS.Override(options...),
+ mainChain: NewStatementChain(),
+ missingHeartbeats: make(map[string]bool),
+ pendingHeartbeats: make(map[string]*heartbeat.Heartbeat),
+ heartbeats: make(map[string]*heartbeat.Heartbeat),
+ neighborChains: make(map[string]*StatementChain),
}
}
@@ -75,6 +84,10 @@ func (neighborManager *NeighborManager) storeHeartbeat(heartbeat *heartbeat.Hear
}
}
+ // endregion ///////////////////////////////////////////////////////////////////////////////////////////////////////
+
+ // region store heartbeat and update lists /////////////////////////////////////////////////////////////////////////
+
heartbeatHash := typeutils.BytesToString(mainStatement.GetHash())
if neighborManager.lastReceivedHeartbeat == nil || mainStatement.GetTime() > neighborManager.lastReceivedHeartbeat.GetMainStatement().GetTime() {
@@ -92,67 +105,141 @@ func (neighborManager *NeighborManager) storeHeartbeat(heartbeat *heartbeat.Hear
func (neighborManager *NeighborManager) applyPendingHeartbeats() (err errors.IdentifiableError) {
if len(neighborManager.missingHeartbeats) == 0 && len(neighborManager.pendingHeartbeats) >= 1 {
- // cycle through heartbeats and apply them one by one
+ sortedPendingHeartbeats, sortingErr := neighborManager.getPendingHeartbeatsSorted()
+ if sortingErr != nil {
+ err = sortingErr
+
+ return
+ }
+
+ for _, sortedHeartbeat := range sortedPendingHeartbeats {
+ if applicationErr := neighborManager.applyHeartbeat(sortedHeartbeat); applicationErr != nil {
+ err = applicationErr
+
+ return
+ }
+ }
}
return
}
-func (neighborManager *NeighborManager) ApplyHeartbeat(heartbeat *heartbeat.Heartbeat) (err errors.IdentifiableError) {
- if storeErr := neighborManager.storeHeartbeat(heartbeat); storeErr != nil {
- err = storeErr
+func (neighborManager *NeighborManager) getPendingHeartbeatsSorted() (result []*heartbeat.Heartbeat, err errors.IdentifiableError) {
+ pendingHeartbeatCount := len(neighborManager.pendingHeartbeats)
+ result = make([]*heartbeat.Heartbeat, pendingHeartbeatCount)
- return
+ processedHeartbeats := 0
+ currentHeartbeat := neighborManager.lastReceivedHeartbeat
+ for currentHeartbeat != nil && len(neighborManager.pendingHeartbeats) >= 1 {
+ mainStatement := currentHeartbeat.GetMainStatement()
+ if mainStatement == nil {
+ result = nil
+ err = ErrInternalError.Derive("missing main statement in heartbeat")
+
+ return
+ }
+
+ currentHeartbeatHash := typeutils.BytesToString(mainStatement.GetHash())
+ previousHeartbeatHash := typeutils.BytesToString(mainStatement.GetPreviousStatementHash())
+
+ if _, exists := neighborManager.pendingHeartbeats[currentHeartbeatHash]; !exists {
+ result = nil
+ err = ErrInternalError.Derive("pending heartbeats list is out of sync")
+
+ return
+ }
+ delete(neighborManager.pendingHeartbeats, currentHeartbeatHash)
+
+ result[pendingHeartbeatCount-processedHeartbeats-1] = currentHeartbeat
+
+ currentHeartbeat = neighborManager.heartbeats[previousHeartbeatHash]
+ processedHeartbeats++
}
- // region mark idle neighbors //////////////////////////////////////////////////////////////////////////////////////
+ return
+}
+
+// this method removes dropped neighbors from the neighborChains
+func (neighborManager *NeighborManager) removeDroppedNeighbors(droppedNeighbors [][]byte) {
+ for _, droppedNeighbor := range droppedNeighbors {
+ neighborIdString := typeutils.BytesToString(droppedNeighbor)
+
+ if _, exists := neighborManager.neighborChains[neighborIdString]; exists {
+ delete(neighborManager.neighborChains, neighborIdString)
+
+ neighborManager.Events.RemoveNeighbor.Trigger(droppedNeighbor)
+ }
+ }
+}
- existingNeighbors := make(map[string]*StatementChain)
+func (neighborManager *NeighborManager) markIdleNeighbors(neighborStatements map[string][]*heartbeat.OpinionStatement) {
+ idleNeighbors := make(map[string]*StatementChain)
for neighborId, neighborChain := range neighborManager.neighborChains {
- existingNeighbors[neighborId] = neighborChain
+ idleNeighbors[neighborId] = neighborChain
}
for neighborId := range neighborStatements {
- if neighborStatementChain, neighborExists := existingNeighbors[neighborId]; neighborExists {
- neighborStatementChain.ResetIdleCounter()
+ if _, neighborExists := idleNeighbors[neighborId]; neighborExists {
+ // TRIGGER ACTIVE
- delete(existingNeighbors, neighborId)
+ delete(idleNeighbors, neighborId)
}
}
- for _, neighborStatementChain := range existingNeighbors {
- neighborStatementChain.IncreaseIdleCounter()
+ for _, x := range idleNeighbors {
+ // TRIGGER IDLE
+ if false {
+ fmt.Println(x)
+ }
}
+}
- // endregion ///////////////////////////////////////////////////////////////////////////////////////////////////////
-
- // region add received statements //////////////////////////////////////////////////////////////////////////////////
-
- applyStatements := neighborManager.mainChain.AddStatement(mainStatement)
+func (neighborManager *NeighborManager) updateStatementChains(mainStatement *heartbeat.OpinionStatement, neighborStatements map[string][]*heartbeat.OpinionStatement) (err errors.IdentifiableError) {
+ neighborManager.mainChain.AddStatement(mainStatement)
- // add neighbor statements to chain
for neighborId, statementsOfNeighbor := range neighborStatements {
- neighborChain, exists := neighborManager.neighborChains[neighborId]
- if !exists {
- neighborChain = neighborManager.addNeighborChain(neighborId)
+ neighborChain, neighborChainErr := neighborManager.addNeighborChain(neighborId)
+ if neighborChainErr != nil {
+ err = neighborChainErr
+
+ return
}
- if neighborChain != nil {
- for _, neighborStatement := range statementsOfNeighbor {
- neighborChain.AddStatement(neighborStatement)
- }
+ for _, neighborStatement := range statementsOfNeighbor {
+ neighborChain.AddStatement(neighborStatement)
}
}
- // endregion ///////////////////////////////////////////////////////////////////////////////////////////////////////
+ return
+}
+
+func (neighborManager *NeighborManager) applyHeartbeat(heartbeat *heartbeat.Heartbeat) (err errors.IdentifiableError) {
+ mainStatement := heartbeat.GetMainStatement()
+ neighborStatements := heartbeat.GetNeighborStatements()
+
+ neighborManager.removeDroppedNeighbors(heartbeat.GetDroppedNeighbors())
- // region apply pending statements /////////////////////////////////////////////////////////////////////////////////
+ neighborManager.markIdleNeighbors(neighborStatements)
- if applyStatements {
- neighborManager.mainChain.lastAppliedStatement = mainStatement
+ if err = neighborManager.updateStatementChains(mainStatement, neighborStatements); err != nil {
+ return
}
- // endregion ///////////////////////////////////////////////////////////////////////////////////////////////////////
+ return
+}
+
+func (neighborManager *NeighborManager) ApplyHeartbeat(heartbeat *heartbeat.Heartbeat) (err errors.IdentifiableError) {
+ if storeErr := neighborManager.storeHeartbeat(heartbeat); storeErr != nil {
+ err = storeErr
+
+ return
+ }
+
+ if applicationErr := neighborManager.applyPendingHeartbeats(); applicationErr != nil {
+ err = applicationErr
+
+ return
+ }
return
}
@@ -160,7 +247,7 @@ func (neighborManager *NeighborManager) ApplyHeartbeat(heartbeat *heartbeat.Hear
func (neighborManager *NeighborManager) GenerateHeartbeatStatements() (result []*heartbeat.OpinionStatement) {
result = make([]*heartbeat.OpinionStatement, 0)
- if lastAppliedStatement := neighborManager.mainChain.GetLastAppliedStatement(); lastAppliedStatement != nil {
+ if lastAppliedStatement := neighborManager.mainChain.GetTail(); lastAppliedStatement != nil {
currentStatement := lastAppliedStatement
for currentStatement != nil && !bytes.Equal(currentStatement.GetHash(), neighborManager.previouslyReportedHeartbeatHash) {
result = append([]*heartbeat.OpinionStatement{currentStatement}, result...)
@@ -174,49 +261,23 @@ func (neighborManager *NeighborManager) GenerateHeartbeatStatements() (result []
return
}
-func (neighborManager *NeighborManager) addNeighborChain(neighborId string) *StatementChain {
- if len(neighborManager.neighborChains) < MAX_NEIGHBOR_COUNT {
- newNeighborChain := NewStatementChain()
- neighborManager.neighborChains[neighborId] = newNeighborChain
+func (neighborManager *NeighborManager) addNeighborChain(neighborId string) (result *StatementChain, err errors.IdentifiableError) {
+ if existingNeighborChain, exists := neighborManager.neighborChains[neighborId]; exists {
+ result = existingNeighborChain
- return newNeighborChain
- }
-
- neighbors := make([]string, 0, len(neighborManager.neighborChains))
- for neighborId := range neighborManager.neighborChains {
- neighbors = append(neighbors, neighborId)
+ return
}
- sort.Slice(neighbors, func(i, j int) bool {
- neighborChainI := neighborManager.neighborChains[neighbors[i]]
- neighborChainJ := neighborManager.neighborChains[neighbors[i]]
-
- switch true {
- case neighborChainI.GetIdleCounter() < neighborChainJ.GetIdleCounter():
- return true
- default:
- return false
-
- }
- })
- newNeighborChain := NewStatementChain()
+ if len(neighborManager.neighborChains) >= MAX_NEIGHBOR_COUNT {
+ err = ErrTooManyNeighbors.Derive("failed to add new neighbor: too many neighbors")
- neighborManager.neighborChains[neighborId] = newNeighborChain
+ return
+ }
- return newNeighborChain
-}
+ result = NewStatementChain()
+ neighborManager.neighborChains[neighborId] = result
-type NeighborManagerOptions struct {
- maxNeighborChains int
-}
+ neighborManager.Events.AddNeighbor.Trigger(neighborId, result)
-func (options NeighborManagerOptions) Override(optionalOptions ...NeighborManagerOption) *NeighborManagerOptions {
- result := &options
- for _, option := range optionalOptions {
- option(result)
- }
-
- return result
+ return
}
-
-type NeighborManagerOption func(*NeighborManagerOptions)
diff --git a/packages/ca/neighbor_manager_events.go b/packages/ca/neighbor_manager_events.go
deleted file mode 100644
index 753b80fc2df5ecf19f0fd1f7051437c476c10a92..0000000000000000000000000000000000000000
--- a/packages/ca/neighbor_manager_events.go
+++ /dev/null
@@ -1,10 +0,0 @@
-package ca
-
-import (
- "github.com/iotaledger/goshimmer/packages/events"
-)
-
-type NeighborManagerEvents struct {
- ChainReset *events.Event
- StatementMissing *events.Event
-}
diff --git a/packages/ca/neighbor_manager_options.go b/packages/ca/neighbor_manager_options.go
index 3e23aff02cb58f2637553385f8365ce0e560488b..7f650e08f1639ab04fb93dd34436a8e7c87acdee 100644
--- a/packages/ca/neighbor_manager_options.go
+++ b/packages/ca/neighbor_manager_options.go
@@ -3,3 +3,18 @@ package ca
var DEFAULT_NEIGHBOR_MANAGER_OPTIONS = &NeighborManagerOptions{
maxNeighborChains: 8,
}
+
+type NeighborManagerOptions struct {
+ maxNeighborChains int
+}
+
+func (options NeighborManagerOptions) Override(optionalOptions ...NeighborManagerOption) *NeighborManagerOptions {
+ result := &options
+ for _, option := range optionalOptions {
+ option(result)
+ }
+
+ return result
+}
+
+type NeighborManagerOption func(*NeighborManagerOptions)
diff --git a/packages/ca/statement_chain.go b/packages/ca/statement_chain.go
index c04af8cf56d806328f8a53a86a7c3bb573101aeb..400332f3f6c70e66cfcfc3e8000602d8700abeb6 100644
--- a/packages/ca/statement_chain.go
+++ b/packages/ca/statement_chain.go
@@ -3,77 +3,83 @@ package ca
import (
"bytes"
+ "github.com/iotaledger/goshimmer/packages/errors"
+
"github.com/iotaledger/goshimmer/packages/typeutils"
"github.com/iotaledger/goshimmer/packages/events"
"github.com/iotaledger/goshimmer/packages/ca/heartbeat"
- "github.com/iotaledger/goshimmer/packages/errors"
)
type StatementChain struct {
Events StatementChainEvents
- statements map[string]*heartbeat.OpinionStatement
- missingStatements map[string]bool
- idleCounter int
- lastAppliedStatement *heartbeat.OpinionStatement
- lastReceivedStatement *heartbeat.OpinionStatement
+ pendingTransactionStatuses map[string]bool
+ transactionStatuses map[string]bool
+ statements map[string]*heartbeat.OpinionStatement
+ tail *heartbeat.OpinionStatement
}
func NewStatementChain() *StatementChain {
return &StatementChain{
Events: StatementChainEvents{
- Reset: events.NewEvent(events.CallbackCaller),
- StatementMissing: events.NewEvent(HashCaller),
+ Reset: events.NewEvent(events.CallbackCaller),
},
- statements: make(map[string]*heartbeat.OpinionStatement),
- missingStatements: make(map[string]bool),
+ pendingTransactionStatuses: make(map[string]bool),
+ transactionStatuses: make(map[string]bool),
+ statements: make(map[string]*heartbeat.OpinionStatement),
}
}
-func (statementChain *StatementChain) IncreaseIdleCounter() {
- statementChain.idleCounter++
-}
+func (statementChain *StatementChain) getTransactionStatus(transactionId string) (result bool, exists bool) {
+ if result, exists = statementChain.pendingTransactionStatuses[transactionId]; exists {
+ return
+ }
-func (statementChain *StatementChain) ResetIdleCounter() {
- statementChain.idleCounter = 0
-}
+ result, exists = statementChain.transactionStatuses[transactionId]
-func (statementChain *StatementChain) GetIdleCounter() int {
- return statementChain.idleCounter
+ return
}
-func (statementChain *StatementChain) AddStatement(statement *heartbeat.OpinionStatement) bool {
+func (statementChain *StatementChain) AddStatement(statement *heartbeat.OpinionStatement) errors.IdentifiableError {
previousStatementHash := statement.GetPreviousStatementHash()
- lastAppliedMainStatement := statementChain.lastAppliedStatement
+ lastAppliedStatement := statementChain.tail
- if len(previousStatementHash) == 0 {
+ if len(previousStatementHash) == 0 || lastAppliedStatement != nil && !bytes.Equal(lastAppliedStatement.GetHash(), previousStatementHash) {
statementChain.Reset()
- } else if lastAppliedMainStatement != nil && !bytes.Equal(lastAppliedMainStatement.GetHash(), previousStatementHash) {
- if (statement.GetTime() - lastAppliedMainStatement.GetTime()) >= MAX_STATEMENT_TIMEOUT {
- statementChain.Reset()
- } else if !statementChain.StatementExists(previousStatementHash) {
- statementChain.addMissingStatement(previousStatementHash)
- statementChain.addStatement(statement)
-
- return false
- }
}
- statementChain.addStatement(statement)
-
- return true
-}
+ for _, toggledTransaction := range statement.GetToggledTransactions() {
+ transactionId := typeutils.BytesToString(toggledTransaction.GetTransactionId())
+
+ if toggledTransaction.IsInitialStatement() {
+ if _, exists := statementChain.getTransactionStatus(transactionId); exists {
+ return ErrMalformedHeartbeat.Derive("two initial statements for the same transaction")
+ }
+
+ statementChain.pendingTransactionStatuses[transactionId] = false
+ } else if toggledTransaction.IsFinalStatement() {
+ // finalize -> clean up
+ } else {
+ if currentValue, exists := statementChain.getTransactionStatus(transactionId); exists {
+ statementChain.pendingTransactionStatuses[transactionId] = !currentValue
+ }
+ }
+ }
-func (statementChain *StatementChain) addStatement(statement *heartbeat.OpinionStatement) {
statementChain.statements[typeutils.BytesToString(statement.GetHash())] = statement
+ statementChain.tail = statement
+
+ return nil
}
-func (statementChain *StatementChain) addMissingStatement(statementHash []byte) {
- statementChain.missingStatements[typeutils.BytesToString(statementHash)] = true
+func (statementChain *StatementChain) ApplyPendingTransactionStatusChanges() {
+ for transactionId, value := range statementChain.pendingTransactionStatuses {
+ statementChain.transactionStatuses[transactionId] = value
+ }
- statementChain.Events.StatementMissing.Trigger(statementHash)
+ statementChain.pendingTransactionStatuses = make(map[string]bool)
}
func (statementChain *StatementChain) GetStatement(statementHash []byte) *heartbeat.OpinionStatement {
@@ -81,21 +87,12 @@ func (statementChain *StatementChain) GetStatement(statementHash []byte) *heartb
}
func (statementChain *StatementChain) Reset() {
- statementChain.Events.Reset.Trigger()
-}
+ statementChain.statements = make(map[string]*heartbeat.OpinionStatement)
+ statementChain.tail = nil
-func (statementChain *StatementChain) Apply(statement *heartbeat.OpinionStatement) (err errors.IdentifiableError) {
- return
-}
-
-func (statementChain *StatementChain) SetLastReceivedStatement(statement *heartbeat.OpinionStatement) {
- statementChain.lastReceivedStatement = statement
-}
-
-func (statementChain *StatementChain) GetLastAppliedStatement() *heartbeat.OpinionStatement {
- return statementChain.lastAppliedStatement
+ statementChain.Events.Reset.Trigger()
}
-func (statementChain *StatementChain) StatementExists(statementHash []byte) bool {
- return true
+func (statementChain *StatementChain) GetTail() *heartbeat.OpinionStatement {
+ return statementChain.tail
}
diff --git a/packages/ca/statement_chain_events.go b/packages/ca/statement_chain_events.go
deleted file mode 100644
index 21ec16ee77d03d212be09c3961c38719541d02dc..0000000000000000000000000000000000000000
--- a/packages/ca/statement_chain_events.go
+++ /dev/null
@@ -1,14 +0,0 @@
-package ca
-
-import (
- "github.com/iotaledger/goshimmer/packages/events"
-)
-
-type StatementChainEvents struct {
- Reset *events.Event
- StatementMissing *events.Event
-}
-
-func HashCaller(handler interface{}, params ...interface{}) {
- handler.(func([]byte))(params[0].([]byte))
-}