Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Feat/en 777 refactor p2p #58

Merged
merged 26 commits into from
Feb 26, 2019
Merged

Feat/en 777 refactor p2p #58

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
26 commits
Select commit Hold shift + click to select a range
e335df7
p2p: done complete refactoring of the package
Feb 6, 2019
8cea60f
p2p: done complete refactoring of the package
Feb 7, 2019
00c3f5d
p2p:
Feb 8, 2019
a95cb70
p2p, process:
Feb 8, 2019
ed2f388
integration tests: changed timeout on bulk tx generation to 1min
Feb 8, 2019
a1d71ac
p2p:
Feb 11, 2019
d5d1f80
Merge branch 'master' into feat/EN-777-refactor-p2p
Feb 12, 2019
91aa0ed
p2p:
Feb 12, 2019
cd252a3
p2p:
Feb 12, 2019
b2809af
p2p - fix after review:
Feb 14, 2019
f2bed84
p2p - fix after review:
Feb 14, 2019
ababa7e
Merge branch 'feat/EN-777-refactor-p2p' of https://github.com/ElrondN…
Feb 14, 2019
ceb1280
p2p:
Feb 20, 2019
168fdaf
Gopkg.lock file
Feb 20, 2019
9124587
process:
Feb 21, 2019
39c4a64
p2p:
Feb 21, 2019
89c0675
Merge branch 'master' into feat/EN-777-refactor-p2p
Feb 21, 2019
c7ef353
Merged with master modifications
Feb 22, 2019
de14c57
p2p:
Feb 22, 2019
8a00585
p2p:
Feb 22, 2019
5df08c4
p2p:
Feb 24, 2019
3188087
node:
Feb 24, 2019
3092015
integrationTests:
Feb 24, 2019
1add3a7
integrationTests:
Feb 25, 2019
f638ddb
Merge branch 'master' into feat/EN-777-refactor-p2p
Feb 26, 2019
7771431
Merged with master
Feb 26, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions Gopkg.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

41 changes: 21 additions & 20 deletions cmd/bootnode/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,8 +42,7 @@ import (
"github.com/ElrondNetwork/elrond-go-sandbox/p2p/libp2p"
"github.com/ElrondNetwork/elrond-go-sandbox/process/block"
"github.com/ElrondNetwork/elrond-go-sandbox/process/factory"
"github.com/ElrondNetwork/elrond-go-sandbox/process/interceptor"
"github.com/ElrondNetwork/elrond-go-sandbox/process/resolver"
"github.com/ElrondNetwork/elrond-go-sandbox/process/factory/containers"
sync2 "github.com/ElrondNetwork/elrond-go-sandbox/process/sync"
"github.com/ElrondNetwork/elrond-go-sandbox/process/transaction"
"github.com/ElrondNetwork/elrond-go-sandbox/sharding"
Expand Down Expand Up @@ -332,22 +331,24 @@ func createNode(ctx *cli.Context, cfg *config.Config, genesisConfig *genesis, sy
return nil, err
}

interceptorsContainer := interceptor.NewContainer()
resolversContainer := resolver.NewContainer()

processorFactory, err := factory.NewProcessorsCreator(factory.ProcessorsCreatorConfig{
InterceptorContainer: interceptorsContainer,
ResolverContainer: resolversContainer,
Messenger: netMessenger,
Blockchain: blkc,
DataPool: datapool,
ShardCoordinator: shardCoordinator,
AddrConverter: addressConverter,
Hasher: hasher,
Marshalizer: marshalizer,
SingleSignKeyGen: keyGen,
Uint64ByteSliceConverter: uint64ByteSliceConverter,
})
interceptorsContainer := containers.NewObjectsContainer()
resolversContainer := containers.NewResolversContainer()

processorFactory, err := factory.NewInterceptorsResolversCreator(
raduchis marked this conversation as resolved.
Show resolved Hide resolved
factory.InterceptorsResolversConfig{

InterceptorContainer: interceptorsContainer,
ResolverContainer: resolversContainer,
Messenger: netMessenger,
Blockchain: blkc,
DataPool: datapool,
ShardCoordinator: shardCoordinator,
AddrConverter: addressConverter,
Hasher: hasher,
Marshalizer: marshalizer,
SingleSignKeyGen: keyGen,
Uint64ByteSliceConverter: uint64ByteSliceConverter,
})
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -412,7 +413,7 @@ func createNode(ctx *cli.Context, cfg *config.Config, genesisConfig *genesis, sy
node.WithPublicKey(pubKey),
node.WithPrivateKey(privKey),
node.WithForkDetector(forkDetector),
node.WithProcessorCreator(processorFactory),
node.WithInterceptorsResolversFactory(processorFactory),
)

if err != nil {
Expand All @@ -429,7 +430,7 @@ func createNode(ctx *cli.Context, cfg *config.Config, genesisConfig *genesis, sy

func createRequestTransactionHandler(txResolver *transaction.TxResolver, log *logger.Logger) func(destShardID uint32, txHash []byte) {
return func(destShardID uint32, txHash []byte) {
_ = txResolver.RequestTransactionFromHash(txHash)
_ = txResolver.RequestHash(txHash)
raduchis marked this conversation as resolved.
Show resolved Hide resolved
log.Debug(fmt.Sprintf("Requested tx for shard %d with hash %s from network\n", destShardID, toB64(txHash)))
}
}
Expand Down
9 changes: 0 additions & 9 deletions cmd/facade/mock/nodeMock.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ import (

"github.com/ElrondNetwork/elrond-go-sandbox/data/state"
"github.com/ElrondNetwork/elrond-go-sandbox/data/transaction"
"github.com/ElrondNetwork/elrond-go-sandbox/process"
)

type NodeMock struct {
Expand Down Expand Up @@ -69,14 +68,6 @@ func (nm *NodeMock) SendTransaction(nonce uint64, sender string, receiver string
return nm.SendTransactionHandler(nonce, sender, receiver, value, transactionData, signature)
}

func (nm *NodeMock) GetInterceptors() []process.Interceptor {
return nil
}

func (nm *NodeMock) GetResolvers() []process.Resolver {
return nil
}

func (nm *NodeMock) GetCurrentPublicKey() string {
return nm.GetCurrentPublicKeyHandler()
}
Expand Down
4 changes: 2 additions & 2 deletions consensus/spos/sposConsensusWorker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1208,8 +1208,8 @@ func (sposWorker *SPOSConsensusWorker) createEmptyBlock() bool {
return true
}

// ReceivedMessage method redirects the received message to the channel which should handle it
func (sposWorker *SPOSConsensusWorker) ReceivedMessage(message p2p.MessageP2P) error {
// Validate method redirects the received message to the channel which should handle it
func (sposWorker *SPOSConsensusWorker) Validate(message p2p.MessageP2P) error {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why Validate if it actually does something else?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

From the P2P perspective, it needs to validate the message. If the Validate func returns error, messenger won't broadcast the message to its peers. Now, since we already parse the message (in case of all interceptors) and check different things like nil fields, signature, etc, why not store the produced object? Otherwise we should have another registered object/function just for storing.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If it's called Validate it should not do anything else. The flow is validate -> do something if the data was correct. If somebody sees a public Validate method and wants to validate that some data is correct it will end up writing in a channel. Now he must look for whoever reads from that channel and sees if it's a desired behaviour or not.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed to ProcessReceivedMessage

if message == nil {
return ErrNilMessage
}
Expand Down
44 changes: 10 additions & 34 deletions consensus/spos/sposConsensusWorker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1283,7 +1283,7 @@ func TestMessage_ExtendEndRound(t *testing.T) {
cnWorkers[0].ExtendEndRound()
}

func TestMessage_ReceivedMessageTxBlockBody(t *testing.T) {
func TestMessage_ReceivedMessageTxBlockBodyFromSelfShouldErr(t *testing.T) {
cnWorkers := InitMessage()

blk := &block.TxBlockBody{}
Expand All @@ -1301,33 +1301,9 @@ func TestMessage_ReceivedMessageTxBlockBody(t *testing.T) {

buffCnsDta, _ := mock.MarshalizerMock{}.Marshal(cnsDta)

cnWorkers[0].ReceivedMessage(&mock.P2PMessageMock{DataField: buffCnsDta})
}

func TestMessage_ReceivedMessageUnknown(t *testing.T) {
cnWorkers := InitMessage()

hdr := &block.Header{}
hdr.Nonce = 1
hdr.TimeStamp = cnWorkers[0].GetRoundTime()

message, _ := mock.MarshalizerMock{}.Marshal(hdr)
hdr.BlockBodyHash = mock.HasherMock{}.Compute(string(message))
message, _ = mock.MarshalizerMock{}.Marshal(hdr)

cnsDta := spos.NewConsensusData(
message,
nil,
[]byte(cnWorkers[0].Cns.SelfPubKey()),
[]byte("sig"),
spos.MtUnknown,
cnWorkers[0].GetRoundTime(),
0,
)

buffCnsDta, _ := mock.MarshalizerMock{}.Marshal(cnsDta)
err := cnWorkers[0].Validate(&mock.P2PMessageMock{DataField: buffCnsDta})
raduchis marked this conversation as resolved.
Show resolved Hide resolved

cnWorkers[0].ReceivedMessage(&mock.P2PMessageMock{DataField: buffCnsDta})
assert.Equal(t, spos.ErrMessageSentFromSelf, err)
}

func TestMessage_DecodeBlockBody(t *testing.T) {
Expand Down Expand Up @@ -3074,7 +3050,7 @@ func TestReceivedMessage_ShouldReturnWhenIsCanceled(t *testing.T) {

cnWorker.Cns.Chr.SetSelfSubround(-1)
buffCnsDta, _ := mock.MarshalizerMock{}.Marshal(cnsDta)
cnWorker.ReceivedMessage(&mock.P2PMessageMock{DataField: buffCnsDta})
cnWorker.Validate(&mock.P2PMessageMock{DataField: buffCnsDta})

assert.Equal(t, 0, len(cnWorker.ReceivedMessages[spos.MtBlockBody]))
}
Expand Down Expand Up @@ -3115,7 +3091,7 @@ func TestReceivedMessage_ShouldReturnWhenDataReceivedIsInvalid(t *testing.T) {
pubKeyMock,
)

cnWorker.ReceivedMessage(nil)
cnWorker.Validate(nil)

assert.Equal(t, 0, len(cnWorker.ReceivedMessages[spos.MtBlockBody]))
}
Expand Down Expand Up @@ -3170,7 +3146,7 @@ func TestReceivedMessage_ShouldReturnWhenNodeIsNotInTheConsensusGroup(t *testing
)

buffCnsDta, _ := mock.MarshalizerMock{}.Marshal(cnsDta)
cnWorker.ReceivedMessage(&mock.P2PMessageMock{DataField: buffCnsDta})
cnWorker.Validate(&mock.P2PMessageMock{DataField: buffCnsDta})

assert.Equal(t, 0, len(cnWorker.ReceivedMessages[spos.MtBlockBody]))
}
Expand Down Expand Up @@ -3225,7 +3201,7 @@ func TestReceivedMessage_ShouldReturnWhenShouldDropMessage(t *testing.T) {
)

buffCnsDta, _ := mock.MarshalizerMock{}.Marshal(cnsDta)
cnWorker.ReceivedMessage(&mock.P2PMessageMock{DataField: buffCnsDta})
cnWorker.Validate(&mock.P2PMessageMock{DataField: buffCnsDta})

assert.Equal(t, 0, len(cnWorker.ReceivedMessages[spos.MtBlockBody]))
}
Expand Down Expand Up @@ -3280,7 +3256,7 @@ func TestReceivedMessage_ShouldReturnWhenReceivedMessageIsFromSelf(t *testing.T)
)

buffCnsDta, _ := mock.MarshalizerMock{}.Marshal(cnsDta)
cnWorker.ReceivedMessage(&mock.P2PMessageMock{DataField: buffCnsDta})
cnWorker.Validate(&mock.P2PMessageMock{DataField: buffCnsDta})

assert.Equal(t, 0, len(cnWorker.ReceivedMessages[spos.MtBlockBody]))
}
Expand Down Expand Up @@ -3335,7 +3311,7 @@ func TestReceivedMessage_ShouldReturnWhenSignatureIsInvalid(t *testing.T) {
)

buffCnsDta, _ := mock.MarshalizerMock{}.Marshal(cnsDta)
cnWorker.ReceivedMessage(&mock.P2PMessageMock{DataField: buffCnsDta})
cnWorker.Validate(&mock.P2PMessageMock{DataField: buffCnsDta})

assert.Equal(t, 0, len(cnWorker.ReceivedMessages[spos.MtBlockBody]))
}
Expand Down Expand Up @@ -3390,7 +3366,7 @@ func TestReceivedMessage_ShouldSendReceivedMesageOnChannel(t *testing.T) {
)

buffCnsDta, _ := mock.MarshalizerMock{}.Marshal(cnsDta)
cnWorker.ReceivedMessage(&mock.P2PMessageMock{DataField: buffCnsDta})
cnWorker.Validate(&mock.P2PMessageMock{DataField: buffCnsDta})

time.Sleep(100 * time.Millisecond)

Expand Down
15 changes: 7 additions & 8 deletions integrationTests/block/common.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,7 @@ import (
"github.com/ElrondNetwork/elrond-go-sandbox/p2p/libp2p"
"github.com/ElrondNetwork/elrond-go-sandbox/process"
"github.com/ElrondNetwork/elrond-go-sandbox/process/factory"
"github.com/ElrondNetwork/elrond-go-sandbox/process/interceptor"
"github.com/ElrondNetwork/elrond-go-sandbox/process/resolver"
"github.com/ElrondNetwork/elrond-go-sandbox/process/factory/containers"
"github.com/ElrondNetwork/elrond-go-sandbox/sharding"
"github.com/ElrondNetwork/elrond-go-sandbox/storage"
"github.com/ElrondNetwork/elrond-go-sandbox/storage/memorydb"
Expand Down Expand Up @@ -87,7 +86,7 @@ func createTestDataPool() data.TransientDataHolder {
return dPool
}

func adbCreateAccountsDB() *state.AccountsDB {
func createAccountsDB() *state.AccountsDB {
marsh := &marshal.JsonMarshalizer{}

dbw, _ := trie.NewDBWriteCache(createMemUnit())
Expand All @@ -101,7 +100,7 @@ func createNetNode(port int, dPool data.TransientDataHolder, accntAdapter state.
*node.Node,
p2p.Messenger,
crypto.PrivateKey,
process.ProcessorFactory) {
process.InterceptorsResolversFactory) {

hasher := sha256.Sha256{}
marshalizer := &marshal.JsonMarshalizer{}
Expand All @@ -116,9 +115,9 @@ func createNetNode(port int, dPool data.TransientDataHolder, accntAdapter state.
shardCoordinator := &sharding.OneShardCoordinator{}
uint64Converter := uint64ByteSlice.NewBigEndianConverter()

pFactory, _ := factory.NewProcessorsCreator(factory.ProcessorsCreatorConfig{
InterceptorContainer: interceptor.NewContainer(),
ResolverContainer: resolver.NewContainer(),
pFactory, _ := factory.NewInterceptorsResolversCreator(factory.InterceptorsResolversConfig{
InterceptorContainer: containers.NewObjectsContainer(),
ResolverContainer: containers.NewResolversContainer(),
Messenger: messenger,
Blockchain: blkc,
DataPool: dPool,
Expand All @@ -144,7 +143,7 @@ func createNetNode(port int, dPool data.TransientDataHolder, accntAdapter state.
node.WithUint64ByteSliceConverter(uint64Converter),
node.WithPrivateKey(sk),
node.WithPublicKey(pk),
node.WithProcessorCreator(pFactory),
node.WithInterceptorsResolversFactory(pFactory),
)

_ = pFactory.CreateInterceptors()
Expand Down
10 changes: 5 additions & 5 deletions integrationTests/block/interceptedRequestHdr_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ import (
"github.com/ElrondNetwork/elrond-go-sandbox/data/block"
"github.com/ElrondNetwork/elrond-go-sandbox/hashing/sha256"
"github.com/ElrondNetwork/elrond-go-sandbox/marshal"
block2 "github.com/ElrondNetwork/elrond-go-sandbox/process/block"
"github.com/ElrondNetwork/elrond-go-sandbox/process/block/resolvers"
"github.com/ElrondNetwork/elrond-go-sandbox/process/factory"
"github.com/stretchr/testify/assert"
)
Expand All @@ -26,10 +26,10 @@ func TestNode_GenerateSendInterceptHeaderByNonceWithMemMessenger(t *testing.T) {
dPoolResolver := createTestDataPool()

fmt.Println("Requestor:")
nRequestor, mesRequestor, _, pFactoryReq := createNetNode(32000, dPoolRequestor, adbCreateAccountsDB())
nRequestor, mesRequestor, _, pFactoryReq := createNetNode(32000, dPoolRequestor, createAccountsDB())

fmt.Println("Resolver:")
nResolver, mesResolver, _, _ := createNetNode(32001, dPoolResolver, adbCreateAccountsDB())
nResolver, mesResolver, _, _ := createNetNode(32001, dPoolResolver, createAccountsDB())

nRequestor.Start()
nResolver.Start()
Expand Down Expand Up @@ -84,8 +84,8 @@ func TestNode_GenerateSendInterceptHeaderByNonceWithMemMessenger(t *testing.T) {
//Step 4. request header
res, err := pFactoryReq.ResolverContainer().Get(string(factory.HeadersTopic))
assert.Nil(t, err)
hdrResolver := res.(*block2.HeaderResolver)
hdrResolver.RequestHeaderFromNonce(0)
hdrResolver := res.(*resolvers.HeaderResolver)
hdrResolver.RequestNonce(0)

select {
case <-chanDone:
Expand Down
10 changes: 4 additions & 6 deletions integrationTests/block/interceptedRequestTxBlockBody_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@ import (
"github.com/ElrondNetwork/elrond-go-sandbox/data/block"
"github.com/ElrondNetwork/elrond-go-sandbox/hashing/sha256"
"github.com/ElrondNetwork/elrond-go-sandbox/marshal"
block2 "github.com/ElrondNetwork/elrond-go-sandbox/process/block"
"github.com/ElrondNetwork/elrond-go-sandbox/process/factory"
"github.com/stretchr/testify/assert"
)
Expand All @@ -27,10 +26,10 @@ func TestNode_GenerateSendInterceptTxBlockBodyWithNetMessenger(t *testing.T) {
dPoolResolver := createTestDataPool()

fmt.Println("Requestor:")
nRequestor, mesRequestor, _, pFactoryReq := createNetNode(32000, dPoolRequestor, adbCreateAccountsDB())
nRequestor, mesRequestor, _, pFactoryReq := createNetNode(32000, dPoolRequestor, createAccountsDB())

fmt.Println("Resolver:")
nResolver, mesResolver, _, pFactoryRes := createNetNode(32001, dPoolResolver, adbCreateAccountsDB())
nResolver, mesResolver, _, pFactoryRes := createNetNode(32001, dPoolResolver, createAccountsDB())

_ = pFactoryReq.CreateInterceptors()
_ = pFactoryReq.CreateResolvers()
Expand Down Expand Up @@ -89,9 +88,8 @@ func TestNode_GenerateSendInterceptTxBlockBodyWithNetMessenger(t *testing.T) {
})

//Step 4. request tx block body
res, _ := pFactoryReq.ResolverContainer().Get(string(factory.TxBlockBodyTopic))
txBlockBodyRequestor := res.(*block2.GenericBlockBodyResolver)
txBlockBodyRequestor.RequestBlockBodyFromHash(txBlockBodyHash)
txBlockBodyRequestor, _ := pFactoryReq.ResolverContainer().Get(string(factory.TxBlockBodyTopic))
txBlockBodyRequestor.RequestHash(txBlockBodyHash)

select {
case <-chanDone:
Expand Down
15 changes: 7 additions & 8 deletions integrationTests/transaction/common.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,7 @@ import (
"github.com/ElrondNetwork/elrond-go-sandbox/p2p/libp2p"
"github.com/ElrondNetwork/elrond-go-sandbox/process"
"github.com/ElrondNetwork/elrond-go-sandbox/process/factory"
"github.com/ElrondNetwork/elrond-go-sandbox/process/interceptor"
"github.com/ElrondNetwork/elrond-go-sandbox/process/resolver"
"github.com/ElrondNetwork/elrond-go-sandbox/process/factory/containers"
"github.com/ElrondNetwork/elrond-go-sandbox/sharding"
"github.com/ElrondNetwork/elrond-go-sandbox/storage"
"github.com/ElrondNetwork/elrond-go-sandbox/storage/memorydb"
Expand Down Expand Up @@ -105,7 +104,7 @@ func createDummyHexAddress(chars int) string {
return string(buff)
}

func adbCreateAccountsDB() *state.AccountsDB {
func createAccountsDB() *state.AccountsDB {
marsh := &marshal.JsonMarshalizer{}

dbw, _ := trie.NewDBWriteCache(createMemUnit())
Expand All @@ -119,7 +118,7 @@ func createNetNode(port int, dPool data.TransientDataHolder, accntAdapter state.
*node.Node,
p2p.Messenger,
crypto.PrivateKey,
process.ProcessorFactory) {
process.InterceptorsResolversFactory) {

hasher := sha256.Sha256{}
marshalizer := &marshal.JsonMarshalizer{}
Expand All @@ -134,9 +133,9 @@ func createNetNode(port int, dPool data.TransientDataHolder, accntAdapter state.
shardCoordinator := &sharding.OneShardCoordinator{}
uint64Converter := uint64ByteSlice.NewBigEndianConverter()

pFactory, _ := factory.NewProcessorsCreator(factory.ProcessorsCreatorConfig{
InterceptorContainer: interceptor.NewContainer(),
ResolverContainer: resolver.NewContainer(),
pFactory, _ := factory.NewInterceptorsResolversCreator(factory.InterceptorsResolversConfig{
InterceptorContainer: containers.NewObjectsContainer(),
ResolverContainer: containers.NewResolversContainer(),
Messenger: messenger,
Blockchain: blkc,
DataPool: dPool,
Expand All @@ -162,7 +161,7 @@ func createNetNode(port int, dPool data.TransientDataHolder, accntAdapter state.
node.WithUint64ByteSliceConverter(uint64Converter),
node.WithPrivateKey(sk),
node.WithPublicKey(pk),
node.WithProcessorCreator(pFactory),
node.WithInterceptorsResolversFactory(pFactory),
)

_ = pFactory.CreateInterceptors()
Expand Down
Loading