Merge pull request #174 from prysmaticlabs/p2p-feed-api

Peer to Peer Feed API

Former-commit-id: 7a38354c89c2c0d85f52c5bfb2f59b11c8615f1f [formerly 7c73afa67bf0296a565991f85f207e857385e4f7]
Former-commit-id: d85cde67fdc27e55ecd3962108771c320b14c41a

cmd/utils/flags.go

@@ -55,7 +55,7 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/nat"
 	"github.com/ethereum/go-ethereum/p2p/netutil"
 	"github.com/ethereum/go-ethereum/params"
-	"github.com/ethereum/go-ethereum/sharding"
+	shardparams "github.com/ethereum/go-ethereum/sharding/params"
 	whisper "github.com/ethereum/go-ethereum/whisper/whisperv6"
 	"gopkg.in/urfave/cli.v1"
 )
@@ -536,7 +536,7 @@ var (
 	// Sharding Settings
 	DepositFlag = cli.BoolFlag{
 		Name:  "deposit",
-		Usage: "To become a notary in a sharding node, " + new(big.Int).Div(sharding.NotaryDeposit, new(big.Int).Exp(big.NewInt(10), big.NewInt(18), nil)).String() + " ETH will be deposited into SMC",
+		Usage: "To become a notary in a sharding node, " + new(big.Int).Div(shardparams.DefaultShardConfig.NotaryDeposit, new(big.Int).Exp(big.NewInt(10), big.NewInt(18), nil)).String() + " ETH will be deposited into SMC",
 	}
 	ActorFlag = cli.StringFlag{
 		Name:  "actor",

sharding/config.go

@@ -1,39 +0,0 @@
-package sharding
-
-import (
-	"math/big"
-
-	"github.com/ethereum/go-ethereum/common"
-)
-
-var (
-	// ShardCount is the number of network shards.
-	ShardCount = int64(100)
-	// ShardingManagerAddress is the address of the sharding manager contract.
-	ShardingManagerAddress = common.HexToAddress("0x0") // TODO
-	// SigGasLimit for verifying signatures.
-	SigGasLimit = 40000
-	// PeriodLength is num of blocks in period.
-	PeriodLength = int64(5)
-	// LookaheadPeriods is the number of periods ahead of current period
-	// which the contract is able to return the notary of that period.
-	LookaheadPeriods = 4
-	// NotaryDeposit is a required deposit size in wei.
-	NotaryDeposit = new(big.Int).Exp(big.NewInt(10), big.NewInt(21), nil) // 1000 ETH
-	// ProposerDeposit is a required deposit size in wei.
-	ProposerDeposit = new(big.Int).Exp(big.NewInt(10), big.NewInt(18), nil) // 1 ETH
-	// MinProposerBalance of proposer where bids are deducted.
-	MinProposerBalance = new(big.Int).Exp(big.NewInt(10), big.NewInt(17), nil) // 0.1 ETH
-	// ContractGasLimit to create contract.
-	ContractGasLimit = uint64(4700000) // Max is 4712388.
-	// NotaryLockupLength to lockup notary deposit from time of deregistration.
-	NotaryLockupLength = int64(16128)
-	// ProposerLockupLength to lockup proposer deposit from time of deregistration.
-	ProposerLockupLength = int64(48)
-	// NotarySubsidy is ETH awarded to notary after collation included in canonical chain.
-	NotarySubsidy = new(big.Int).Exp(big.NewInt(10), big.NewInt(15), nil) // 0.001 ETH.
-	// NotaryCommitSize sampled per block from the notaries pool per period per shard.
-	NotaryCommitSize = int64(135)
-	// NotaryQuorumSize votes the collation needs to get accepted to the canonical chain.
-	NotaryQuorumSize = int64(90)
-)

sharding/config_test.go

@@ -1,46 +0,0 @@
-package sharding
-
-import (
-	"math/big"
-	"testing"
-)
-
-func TestNotaryDeposit(t *testing.T) {
-	want, err := new(big.Int).SetString("1000000000000000000000", 10) // 1000 ETH
-	if !err {
-		t.Fatalf("Failed to setup test")
-	}
-	if NotaryDeposit.Cmp(want) != 0 {
-		t.Errorf("Notary deposit size incorrect. Wanted %d, got %d", want, NotaryDeposit)
-	}
-}
-
-func TestProposerDeposit(t *testing.T) {
-	want, err := new(big.Int).SetString("1000000000000000000", 10) // 1 ETH
-	if !err {
-		t.Fatalf("Failed to setup test")
-	}
-	if ProposerDeposit.Cmp(want) != 0 {
-		t.Errorf("Proposer deposit size incorrect. Wanted %d, got %d", want, ProposerDeposit)
-	}
-}
-
-func TestMinProposerBalance(t *testing.T) {
-	want, err := new(big.Int).SetString("100000000000000000", 10) // 0.1 ETH
-	if !err {
-		t.Fatalf("Failed to setup test")
-	}
-	if MinProposerBalance.Cmp(want) != 0 {
-		t.Errorf("Min proposer balance incorrect. Wanted %d, got %d", want, MinProposerBalance)
-	}
-}
-
-func TestNotarySubsidy(t *testing.T) {
-	want, err := new(big.Int).SetString("1000000000000000", 10) // 0.001 ETH
-	if !err {
-		t.Fatalf("Failed to setup test")
-	}
-	if NotarySubsidy.Cmp(want) != 0 {
-		t.Errorf("Notary subsidy size incorrect. Wanted %d, got %d", want, NotarySubsidy)
-	}
-}

sharding/contracts/sharding_manager.sol

@@ -50,26 +50,26 @@ contract SMC {
     uint nextPeriodNotarySampleSize;
     uint sampleSizeLastUpdatedPeriod;
 
+    // Number of shards
+    // TODO: Setting default as 100. This will be a dynamic when we introduce random beacon
+    uint public shardCount = 100;
+
     // Constant values
     // Length of challenge period for notary's proof of custody
     uint public constant CHALLENGE_PERIOD = 25;
     // Number of blocks per period
     uint constant PERIOD_LENGTH = 5;
-    // Number of shards
-    uint constant SHARD_COUNT = 100;
     // The minimum deposit size for a notary
     uint constant NOTARY_DEPOSIT = 1000 ether;
-    // The reward for notary on voting for a collation
-    uint constant NOTARY_REWARD = 0.001 ether;
     // Time the ether is locked by notaries
     uint constant NOTARY_LOCKUP_LENGTH = 16128;
-    // Number of periods ahead of current period, which the contract
-    // is able to return the notary of that period
-    uint constant LOOKAHEAD_LENGTH = 4;
     // Number of notaries to select from notary pool for each shard in each period
     uint constant COMMITTEE_SIZE = 135;
     // Threshold(number of notaries in committee) for a proposal to be accepted
     uint constant QUORUM_SIZE = 90;
+    // Number of periods ahead of current period, which the contract
+    // is able to return the notary of that period
+    uint constant LOOKAHEAD_LENGTH = 4;
 
     /// Checks if a notary with given shard id and period has been chosen as
     /// a committee member to vote for header added on to the main chain
@@ -172,7 +172,7 @@ contract SMC {
         uint _period,
         bytes32 _chunkRoot
         ) public {
-        require((_shardId >= 0) && (_shardId < SHARD_COUNT));
+        require((_shardId >= 0) && (_shardId < shardCount));
         require(_period == block.number / PERIOD_LENGTH);
         require(_period > lastSubmittedCollation[_shardId]);
 
@@ -198,7 +198,7 @@ contract SMC {
         uint _index,
         bytes32 _chunkRoot        
     ) public {
-        require((_shardId >= 0) && (_shardId < SHARD_COUNT));
+        require((_shardId >= 0) && (_shardId < shardCount));
         require(_period == block.number / PERIOD_LENGTH);
         require(_period == lastSubmittedCollation[_shardId]);
         require(_index < COMMITTEE_SIZE);

sharding/contracts/sharding_manager_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
-	"github.com/ethereum/go-ethereum/sharding"
+	"github.com/ethereum/go-ethereum/sharding/params"
 )
 
 type smcTestHelper struct {
@@ -68,7 +68,7 @@ func deploySMCContract(backend *backends.SimulatedBackend, key *ecdsa.PrivateKey
 
 // fastForward is a helper function to skip through n period.
 func (s *smcTestHelper) fastForward(p int) {
-	for i := 0; i < p*int(sharding.PeriodLength); i++ {
+	for i := 0; i < p*int(params.DefaultShardConfig.PeriodLength); i++ {
 		s.backend.Commit()
 	}
 }
@@ -168,7 +168,7 @@ func (s *smcTestHelper) addHeader(a *testAccount, shard *big.Int, period *big.In
 
 	// Filter SMC logs by headerAdded.
 	shardIndex := []*big.Int{shard}
-	logPeriod := uint64(period.Int64() * sharding.PeriodLength)
+	logPeriod := uint64(period.Int64() * params.DefaultShardConfig.PeriodLength)
 	log, err := s.smc.FilterHeaderAdded(&bind.FilterOpts{Start: logPeriod}, shardIndex)
 	if err != nil {
 		return err
@@ -200,7 +200,7 @@ func (s *smcTestHelper) submitVote(a *testAccount, shard *big.Int, period *big.I
 	}
 	// Filter SMC logs by submitVote.
 	shardIndex := []*big.Int{shard}
-	logPeriod := uint64(period.Int64() * sharding.PeriodLength)
+	logPeriod := uint64(period.Int64() * params.DefaultShardConfig.PeriodLength)
 	log, err := s.smc.FilterVoteSubmitted(&bind.FilterOpts{Start: logPeriod}, shardIndex)
 	if err != nil {
 		return err
@@ -377,7 +377,7 @@ func TestNotaryRelease(t *testing.T) {
 	}
 
 	// Fast forward until lockup ends.
-	s.fastForward(int(sharding.NotaryLockupLength + 1))
+	s.fastForward(int(params.DefaultShardConfig.NotaryLockupLength + 1))
 
 	// Notary 0 releases.
 	_, err = s.smc.ReleaseNotary(s.testAccounts[0].txOpts)

sharding/mainchain/smc_client.go

@@ -43,6 +43,7 @@ type Client interface {
 	SetDepositFlag(deposit bool)
 	DataDirPath() string
 	Sign(hash common.Hash) ([]byte, error)
+	GetShardCount() (int64, error)
 }
 
 // SMCClient defines a struct that interacts with a
@@ -225,3 +226,13 @@ func (s *SMCClient) Sign(hash common.Hash) ([]byte, error) {
 	account := s.Account()
 	return s.keystore.SignHash(*account, hash.Bytes())
 }
+
+// GetShardCount gets the count of the total shards
+// currently operating in the sharded universe.
+func (s *SMCClient) GetShardCount() (int64, error) {
+	shardCount, err := s.SMCCaller().ShardCount(&bind.CallOpts{})
+	if err != nil {
+		return 0, err
+	}
+	return shardCount.Int64(), nil
+}

sharding/mainchain/utils.go

@@ -9,8 +9,8 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/rpc"
-	"github.com/ethereum/go-ethereum/sharding"
 	"github.com/ethereum/go-ethereum/sharding/contracts"
+	"github.com/ethereum/go-ethereum/sharding/params"
 )
 
 // dialRPC endpoint to node.
@@ -24,16 +24,16 @@ func dialRPC(endpoint string) (*rpc.Client, error) {
 // initSMC initializes the sharding manager contract bindings.
 // If the SMC does not exist, it will be deployed.
 func initSMC(s *SMCClient) (*contracts.SMC, error) {
-	b, err := s.client.CodeAt(context.Background(), sharding.ShardingManagerAddress, nil)
+	b, err := s.client.CodeAt(context.Background(), params.DefaultShardConfig.SMCAddress, nil)
 	if err != nil {
-		return nil, fmt.Errorf("unable to get contract code at %s: %v", sharding.ShardingManagerAddress.Hex(), err)
+		return nil, fmt.Errorf("unable to get contract code at %s: %v", params.DefaultShardConfig.SMCAddress.Hex(), err)
 	}
 
 	// Deploy SMC for development only.
 	// TODO: Separate contract deployment from the sharding node. It would only need to be deployed
 	// once on the mainnet, so this code would not need to ship with the node.
 	if len(b) == 0 {
-		log.Info(fmt.Sprintf("No sharding manager contract found at %s. Deploying new contract.", sharding.ShardingManagerAddress.Hex()))
+		log.Info(fmt.Sprintf("No sharding manager contract found at %s. Deploying new contract.", params.DefaultShardConfig.SMCAddress.Hex()))
 
 		txOps, err := s.CreateTXOpts(big.NewInt(0))
 		if err != nil {
@@ -56,5 +56,5 @@ func initSMC(s *SMCClient) (*contracts.SMC, error) {
 		return contract, nil
 	}
 
-	return contracts.NewSMC(sharding.ShardingManagerAddress, s.client)
+	return contracts.NewSMC(params.DefaultShardConfig.SMCAddress, s.client)
 }

sharding/node/backend.go

@@ -29,7 +29,7 @@ import (
 	"github.com/ethereum/go-ethereum/sharding/database"
 	"github.com/ethereum/go-ethereum/sharding/params"
 	"github.com/ethereum/go-ethereum/sharding/txpool"
-	cli "gopkg.in/urfave/cli.v1"
+	"gopkg.in/urfave/cli.v1"
 )
 
 const shardChainDbName = "shardchaindata"
@@ -38,7 +38,7 @@ const shardChainDbName = "shardchaindata"
 // it contains APIs and fields that handle the different components of the sharded
 // Ethereum network.
 type ShardEthereum struct {
-	shardConfig  *params.ShardConfig  // Holds necessary information to configure shards.
+	shardConfig  *params.ShardConfig  // Holds necessary information to configure shard node.
 	txPool       *txpool.ShardTXPool  // Defines the sharding-specific txpool. To be designed.
 	actor        sharding.Actor       // Either notary, proposer, or observer.
 	shardChainDb ethdb.Database       // Access to the persistent db to store shard data.
@@ -89,6 +89,9 @@ func New(ctx *cli.Context) (*ShardEthereum, error) {
 	// Adds the initialized SMCClient to the ShardEthereum instance.
 	shardEthereum.smcClient = smcClient
 
+	// Configure shardConfig by loading the default.
+	shardEthereum.shardConfig = params.DefaultShardConfig
+
 	// Adds the initialized shardChainDb to the ShardEthereum instance.
 	shardEthereum.shardChainDb = shardChainDb
 
@@ -100,7 +103,7 @@ func New(ctx *cli.Context) (*ShardEthereum, error) {
 		return nil, err
 	}
 
-	if err := shardEthereum.registerActorService(actorFlag, shardIDFlag); err != nil {
+	if err := shardEthereum.registerActorService(shardEthereum.shardConfig, actorFlag, shardIDFlag); err != nil {
 		return nil, err
 	}
 
@@ -210,18 +213,18 @@ func (s *ShardEthereum) registerTXPool(actor string) error {
 }
 
 // Registers the actor according to CLI flags. Either notary/proposer/observer.
-func (s *ShardEthereum) registerActorService(actor string, shardID int) error {
+func (s *ShardEthereum) registerActorService(config *params.ShardConfig, actor string, shardID int) error {
 	return s.Register(func(ctx *sharding.ServiceContext) (sharding.Service, error) {
 
 		var p2p *shardp2p.Server
 		ctx.RetrieveService(&p2p)
 
 		if actor == "notary" {
-			return notary.NewNotary(s.smcClient, p2p, s.shardChainDb)
+			return notary.NewNotary(config, s.smcClient, p2p, s.shardChainDb)
 		} else if actor == "proposer" {
 			var txPool *txpool.ShardTXPool
 			ctx.RetrieveService(&txPool)
-			return proposer.NewProposer(s.smcClient, p2p, txPool, s.shardChainDb, shardID)
+			return proposer.NewProposer(config, s.smcClient, p2p, txPool, s.shardChainDb, shardID)
 		}
 		return observer.NewObserver(p2p, s.shardChainDb, shardID)
 	})

sharding/notary/notary.go

@@ -10,8 +10,8 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
-	"github.com/ethereum/go-ethereum/sharding"
 	"github.com/ethereum/go-ethereum/sharding/mainchain"
+	sparams "github.com/ethereum/go-ethereum/sharding/params"
 )
 
 // SubscribeBlockHeaders checks incoming block headers and determines if
@@ -55,7 +55,11 @@ func subscribeBlockHeaders(client mainchain.Client) error {
 // conditions are met.
 func checkSMCForNotary(client mainchain.Client, head *types.Header) error {
 	log.Info("Checking if we are an eligible collation notary for a shard...")
-	for s := int64(0); s < sharding.ShardCount; s++ {
+	shardCount, err := client.GetShardCount()
+	if err != nil {
+		return fmt.Errorf("can't get shard count from smc: %v", err)
+	}
+	for s := int64(0); s < shardCount; s++ {
 		// Checks if we are an eligible notary according to the SMC.
 		addr, err := client.SMCCaller().GetNotaryInCommittee(&bind.CallOpts{}, big.NewInt(s))
 
@@ -132,7 +136,7 @@ func submitCollation(shardID int64) error {
 // joinNotaryPool checks if the deposit flag is true and the account is a
 // notary in the SMC. If the account is not in the set, it will deposit ETH
 // into contract.
-func joinNotaryPool(client mainchain.Client) error {
+func joinNotaryPool(config *sparams.ShardConfig, client mainchain.Client) error {
 	if !client.DepositFlag() {
 		return errors.New("joinNotaryPool called when deposit flag was not set")
 	}
@@ -142,7 +146,7 @@ func joinNotaryPool(client mainchain.Client) error {
 	}
 
 	log.Info("Joining notary pool")
-	txOps, err := client.CreateTXOpts(sharding.NotaryDeposit)
+	txOps, err := client.CreateTXOpts(config.NotaryDeposit)
 	if err != nil {
 		return fmt.Errorf("unable to initiate the deposit transaction: %v", err)
 	}
@@ -151,7 +155,7 @@ func joinNotaryPool(client mainchain.Client) error {
 	if err != nil {
 		return fmt.Errorf("unable to deposit eth and become a notary: %v", err)
 	}
-	log.Info(fmt.Sprintf("Deposited %dETH into contract with transaction hash: %s", new(big.Int).Div(sharding.NotaryDeposit, big.NewInt(params.Ether)), tx.Hash().String()))
+	log.Info(fmt.Sprintf("Deposited %dETH into contract with transaction hash: %s", new(big.Int).Div(config.NotaryDeposit, big.NewInt(params.Ether)), tx.Hash().String()))
 
 	return nil
 }

sharding/notary/service.go

@@ -9,20 +9,22 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/sharding"
 	"github.com/ethereum/go-ethereum/sharding/mainchain"
+	"github.com/ethereum/go-ethereum/sharding/params"
 )
 
 // Notary holds functionality required to run a collation notary
 // in a sharded system. Must satisfy the Service interface defined in
 // sharding/service.go.
 type Notary struct {
+	config       *params.ShardConfig
 	smcClient    *mainchain.SMCClient
 	shardp2p     sharding.ShardP2P
 	shardChainDb ethdb.Database
 }
 
 // NewNotary creates a new notary instance.
-func NewNotary(smcClient *mainchain.SMCClient, shardp2p sharding.ShardP2P, shardChainDb ethdb.Database) (*Notary, error) {
-	return &Notary{smcClient, shardp2p, shardChainDb}, nil
+func NewNotary(config *params.ShardConfig, smcClient *mainchain.SMCClient, shardp2p sharding.ShardP2P, shardChainDb ethdb.Database) (*Notary, error) {
+	return &Notary{config, smcClient, shardp2p, shardChainDb}, nil
 }
 
 // Start the main routine for a notary.
@@ -41,7 +43,7 @@ func (n *Notary) notarizeCollations() {
 	// TODO: handle this better through goroutines. Right now, these methods
 	// are blocking.
 	if n.smcClient.DepositFlag() {
-		if err := joinNotaryPool(n.smcClient); err != nil {
+		if err := joinNotaryPool(n.config, n.smcClient); err != nil {
 			log.Error(fmt.Sprintf("Could not fetch current block number: %v", err))
 			return
 		}

sharding/notary/service_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/sharding"
 	"github.com/ethereum/go-ethereum/sharding/contracts"
+	"github.com/ethereum/go-ethereum/sharding/params"
 	cli "gopkg.in/urfave/cli.v1"
 )
 
@@ -93,6 +94,10 @@ func (s *smcClient) DataDirPath() string {
 	return "/tmp/datadir"
 }
 
+func (s *smcClient) GetShardCount() (int64, error) {
+	return 100, nil
+}
+
 // Helper/setup methods.
 // TODO: consider moving these to common sharding testing package as the notary and smc tests
 // use them.
@@ -122,7 +127,7 @@ func TestIsAccountInNotaryPool(t *testing.T) {
 
 	txOpts := transactOpts()
 	// deposit in notary pool, then it should return true.
-	txOpts.Value = sharding.NotaryDeposit
+	txOpts.Value = params.DefaultShardConfig.NotaryDeposit
 	if _, err := smc.RegisterNotary(txOpts); err != nil {
 		t.Fatalf("Failed to deposit: %v", err)
 	}
@@ -138,7 +143,7 @@ func TestIsAccountInNotaryPool(t *testing.T) {
 
 func TestJoinNotaryPool(t *testing.T) {
 	backend, smc := setup()
-	client := &smcClient{smc: smc, depositFlag: false, t: t}
+	client := &smcClient{smc: smc, t: t}
 	// There should be no notary initially.
 	numNotaries, err := smc.NotaryPoolLength(&bind.CallOpts{})
 	if err != nil {
@@ -149,13 +154,13 @@ func TestJoinNotaryPool(t *testing.T) {
 	}
 
 	client.SetDepositFlag(false)
-	err = joinNotaryPool(client)
+	err = joinNotaryPool(params.DefaultShardConfig, client)
 	if err == nil {
 		t.Error("Joined notary pool while --deposit was not present")
 	}
 
 	client.SetDepositFlag(true)
-	err = joinNotaryPool(client)
+	err = joinNotaryPool(params.DefaultShardConfig, client)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -171,7 +176,7 @@ func TestJoinNotaryPool(t *testing.T) {
 	}
 
 	// Trying to join while deposited should do nothing
-	err = joinNotaryPool(client)
+	err = joinNotaryPool(params.DefaultShardConfig, client)
 	if err != nil {
 		t.Error(err)
 	}

sharding/p2p/feed.go

@@ -0,0 +1,33 @@
+package p2p
+
+import (
+	"reflect"
+
+	"github.com/ethereum/go-ethereum/event"
+)
+
+// P2P feed is a one to many subscription feed of the argument type.
+//
+// Messages received via p2p protocol are sent to subscribers by these event
+// feeds. Message consumers should not use event feeds to reply to or broadcast
+// messages. The p2p server will not relay them to peers. Rather, use the
+// Send() or Broadcast() method on p2p.Server.
+//
+// Event feeds from p2p will always be of type p2p.Message. The message
+// contains information about the sender, aka the peer, and the message payload
+// itself.
+//
+//   feed, err := ps.Feed(MyMessage{})
+//   ch := make(chan p2p.Message, 100) // Choose a reasonable buffer size!
+//   sub := feed.Subscribe(ch)
+//
+//   // Wait until my message comes from a peer.
+//   msg := <- ch
+//   fmt.Printf("Message received: %v", msg.Data)
+func (s *Server) Feed(msg interface{}) (*event.Feed, error) {
+	t := reflect.TypeOf(msg)
+	if s.feeds[t] == nil {
+		s.feeds[t] = new(event.Feed)
+	}
+	return s.feeds[t], nil
+}

sharding/p2p/feed_example_test.go

@@ -0,0 +1,46 @@
+package p2p
+
+import "fmt"
+
+// Feeds can be use to subscribe to any type of message.
+func ExampleServer_Feed() {
+	s, err := NewServer()
+	if err != nil {
+		panic(err)
+	}
+
+	// Let's wait for a puzzle from our peers then try to solve it.
+	type Puzzle struct {
+		Challenge string
+		Answer    string
+	}
+
+	feed, err := s.Feed(Puzzle{})
+	if err != nil {
+		// This shouldn't happen, but we should handle it anyway.
+		panic(err)
+	}
+
+	ch := make(chan Message, 5) // Small buffer size. I don't expect many puzzles.
+	sub := feed.Subscribe(ch)
+
+	// Always close these resources.
+	defer sub.Unsubscribe()
+	defer close(ch)
+
+	// Wait until we have a puzzle to solve.
+	msg := <-ch
+	puzzle, ok := msg.Data.(Puzzle)
+
+	if !ok {
+		panic("Received a message that wasn't a puzzle!")
+	}
+
+	fmt.Printf("Received puzzle %s from peer %v\n", puzzle, msg.Peer)
+
+	if puzzle.Answer == "fourteen" {
+		fmt.Println("I solved the puzzle!")
+	} else {
+		fmt.Println("The answer isn't \"fourteen\"... giving up")
+	}
+}

sharding/p2p/feed_test.go

@@ -0,0 +1,33 @@
+package p2p
+
+import "testing"
+
+func TestFeed_ReturnsSameFeed(t *testing.T) {
+	tests := []struct {
+		a    interface{}
+		b    interface{}
+		want bool
+	}{
+		// Equality tests
+		{a: 1, b: 2, want: true},
+		{a: 'a', b: 'b', want: true},
+		{a: struct{ c int }{c: 1}, b: struct{ c int }{c: 2}, want: true},
+		{a: struct{ c string }{c: "a"}, b: struct{ c string }{c: "b"}, want: true},
+		// Inequality tests
+		{a: 1, b: '2', want: false},
+		{a: 'a', b: 1, want: false},
+		{a: struct{ c int }{c: 1}, b: struct{ c int64 }{c: 2}, want: false},
+		{a: struct{ c string }{c: "a"}, b: struct{ c float64 }{c: 3.4}, want: false},
+	}
+
+	s, _ := NewServer()
+
+	for _, tt := range tests {
+		feed1, _ := s.Feed(tt.a)
+		feed2, _ := s.Feed(tt.b)
+
+		if (feed1 == feed2) != tt.want {
+			t.Errorf("Expected %v == %v to be %t", feed1, feed2, tt.want)
+		}
+	}
+}

sharding/p2p/message.go

@@ -0,0 +1,9 @@
+package p2p
+
+// Message represents a message received from an external peer.
+type Message struct {
+	// Peer represents the sender of the message.
+	Peer Peer
+	// Data can be any type of message found in sharding/p2p/messages package.
+	Data interface{}
+}

sharding/p2p/peer.go

@@ -0,0 +1,7 @@
+package p2p
+
+// Peer
+// TODO - Design and implement.
+// See design doc: https://docs.google.com/document/d/1cthKuGPreOSQH96Ujt7sArcT-IRICk6b-QcdD0EnLsI/edit
+// https://github.com/prysmaticlabs/geth-sharding/issues/175
+type Peer struct{}

sharding/p2p/service.go

@@ -2,24 +2,43 @@
 package p2p
 
 import (
+	"reflect"
+
+	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 )
 
-// Server is a placeholder for a shardp2p service. To be designed.
-type Server struct{}
-
-// NewServer creates a new shardp2p service instance.
-func NewServer() (*Server, error) {
-	return &Server{}, nil
+// Server is a placeholder for a p2p service. To be designed.
+type Server struct {
+	feeds map[reflect.Type]*event.Feed
 }
 
-// Start the main routine for an shardp2p server.
+// NewServer creates a new p2p server instance.
+func NewServer() (*Server, error) {
+	return &Server{
+		feeds: make(map[reflect.Type]*event.Feed),
+	}, nil
+}
+
+// Start the main routine for an p2p server.
 func (s *Server) Start() {
 	log.Info("Starting shardp2p server")
 }
 
-// Stop the main shardp2p loop..
+// Stop the main p2p loop.
 func (s *Server) Stop() error {
 	log.Info("Stopping shardp2p server")
 	return nil
 }
+
+// Send a message to a specific peer.
+func (s *Server) Send(msg interface{}, peer Peer) {
+	// TODO
+	// https://github.com/prysmaticlabs/geth-sharding/issues/175
+}
+
+// Broadcast a message to the world.
+func (s *Server) Broadcast(msg interface{}) {
+	// TODO
+	// https://github.com/prysmaticlabs/geth-sharding/issues/176
+}

sharding/p2p/service_test.go

@@ -1,6 +1,8 @@
 package p2p
 
-import "github.com/ethereum/go-ethereum/sharding"
+import (
+	"github.com/ethereum/go-ethereum/sharding"
+)
 
 // Verifies that Server implements the ShardP2P interface.
 var _ = sharding.ShardP2P(&Server{})

sharding/params/chainConfig.go

@@ -1,10 +0,0 @@
-// Package params defines important configuration options to be used when instantiating
-// objects within the sharding package. For example, it defines objects such as a
-// ShardConfig that will be useful when creating new shard instances.
-package params
-
-import "github.com/ethereum/go-ethereum/common"
-
-type ShardConfig struct {
-	SMCAddress common.Address
-}

sharding/params/config.go

@@ -0,0 +1,40 @@
+// Package params defines important configuration options to be used when instantiating
+// objects within the sharding package. For example, it defines objects such as a
+// ShardConfig that will be useful when creating new shard instances.
+package params
+
+import (
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common"
+)
+
+// DefaultShardConfig contains default configs for node to use in the sharded universe
+var DefaultShardConfig = &ShardConfig{
+	SMCAddress:            common.HexToAddress("0x0"),
+	PeriodLength:          5,
+	NotaryDeposit:         new(big.Int).Exp(big.NewInt(10), big.NewInt(21), nil), // 1000 ETH
+	NotaryLockupLength:    16128,
+	ProposerLockupLength:  48,
+	NotaryCommitteeSize:   135,
+	NotaryQuorumSize:      90,
+	NotaryChallengePeriod: 25,
+}
+
+// DefaultShardChainConfig contains default chain configs of an individual shard.
+var DefaultShardChainConfig = &ShardChainConfig{}
+
+// ShardConfig contains configs for node to participate in the sharded universe.
+type ShardConfig struct {
+	SMCAddress            common.Address // SMCAddress is the address of SMC in mainchain.
+	PeriodLength          int64          // PeriodLength is num of blocks in period.
+	NotaryDeposit         *big.Int       // NotaryDeposit is a required deposit size in wei.
+	NotaryLockupLength    int64          // NotaryLockupLength to lockup notary deposit from time of deregistration.
+	ProposerLockupLength  int64          // ProposerLockupLength to lockup proposer deposit from time of deregistration.
+	NotaryCommitteeSize   int64          // NotaryCommitSize sampled per block from the notaries pool per period per shard.
+	NotaryQuorumSize      int64          // NotaryQuorumSize votes the collation needs to get accepted to the canonical chain.
+	NotaryChallengePeriod int64          // NotaryChallengePeriod is the duration a notary has to store collations for.
+}
+
+// ShardChainConfig contains chain config of an individual shard. Still to be designed.
+type ShardChainConfig struct{}

sharding/params/config_test.go

@@ -0,0 +1,52 @@
+package params
+
+import (
+	"math/big"
+	"testing"
+)
+
+func TestNotaryDeposit(t *testing.T) {
+	want, err := new(big.Int).SetString("1000000000000000000000", 10) // 1000 ETH
+	if !err {
+		t.Fatalf("Failed to setup test")
+	}
+	if DefaultShardConfig.NotaryDeposit.Cmp(want) != 0 {
+		t.Errorf("Notary deposit size incorrect. Wanted %d, got %d", want, DefaultShardConfig.NotaryDeposit)
+	}
+}
+
+func TestPeriodLength(t *testing.T) {
+	if DefaultShardConfig.PeriodLength != 5 {
+		t.Errorf("Shard count incorrect. Wanted %d, got %d", 5, DefaultShardConfig.PeriodLength)
+	}
+}
+
+func TestNotaryLockupLength(t *testing.T) {
+	if DefaultShardConfig.NotaryLockupLength != 16128 {
+		t.Errorf("Shard count incorrect. Wanted %d, got %d", 16128, DefaultShardConfig.NotaryLockupLength)
+	}
+}
+
+func TestProposerLockupLength(t *testing.T) {
+	if DefaultShardConfig.ProposerLockupLength != 48 {
+		t.Errorf("Shard count incorrect. Wanted %d, got %d", 48, DefaultShardConfig.ProposerLockupLength)
+	}
+}
+
+func TestNotaryCommitteeSize(t *testing.T) {
+	if DefaultShardConfig.NotaryCommitteeSize != 135 {
+		t.Errorf("Shard count incorrect. Wanted %d, got %d", 135, DefaultShardConfig.NotaryCommitteeSize)
+	}
+}
+
+func TestNotaryQuorumSize(t *testing.T) {
+	if DefaultShardConfig.NotaryQuorumSize != 90 {
+		t.Errorf("Shard count incorrect. Wanted %d, got %d", 90, DefaultShardConfig.NotaryQuorumSize)
+	}
+}
+
+func TestNotaryChallengePeriod(t *testing.T) {
+	if DefaultShardConfig.NotaryChallengePeriod != 25 {
+		t.Errorf("Shard count incorrect. Wanted %d, got %d", 25, DefaultShardConfig.NotaryChallengePeriod)
+	}
+}

sharding/proposer/proposer.go

@@ -15,14 +15,18 @@ import (
 // and body. Header consists of shardID, ChunkRoot, period,
 // proposer addr and signatures. Body contains serialized blob
 // of a collations transactions.
-func createCollation(client mainchain.Client, shardId *big.Int, period *big.Int, txs []*types.Transaction) (*sharding.Collation, error) {
+func createCollation(client mainchain.Client, shardID *big.Int, period *big.Int, txs []*types.Transaction) (*sharding.Collation, error) {
 	// shardId has to be within range
-	if shardId.Cmp(big.NewInt(0)) < 0 || shardId.Cmp(big.NewInt(sharding.ShardCount)) > 0 {
-		return nil, fmt.Errorf("can't create collation for shard %v. Must be between 0 and %v", shardId, sharding.ShardCount)
+	shardCount, err := client.GetShardCount()
+	if err != nil {
+		return nil, fmt.Errorf("can't get shard count from smc: %v", err)
+	}
+	if shardID.Cmp(big.NewInt(0)) < 0 || shardID.Cmp(big.NewInt(shardCount)) > 0 {
+		return nil, fmt.Errorf("can't create collation for shard %v. Must be between 0 and %v", shardID, shardCount)
 	}
 
 	// check with SMC to see if we can add the header.
-	if a, _ := checkHeaderAdded(client, shardId, period); !a {
+	if a, _ := checkHeaderAdded(client, shardID, period); !a {
 		return nil, fmt.Errorf("can't create collation, collation with same period has already been added")
 	}
 
@@ -34,7 +38,7 @@ func createCollation(client mainchain.Client, shardId *big.Int, period *big.Int,
 
 	// construct the header, leave chunkRoot and signature fields empty, to be filled later.
 	addr := client.Account().Address
-	header := sharding.NewCollationHeader(shardId, nil, period, &addr, nil)
+	header := sharding.NewCollationHeader(shardID, nil, period, &addr, nil)
 
 	// construct the body with header, blobs(serialized txs) and txs.
 	collation := sharding.NewCollation(header, blobs, txs)
@@ -78,9 +82,9 @@ func addHeader(client mainchain.Client, collation *sharding.Collation) error {
 // submitted to the main chain. There can only be one header per shard
 // per period, proposer should check if a header's already submitted,
 // checkHeaderAdded returns true if it's available, false if it's unavailable.
-func checkHeaderAdded(client mainchain.Client, shardId *big.Int, period *big.Int) (bool, error) {
+func checkHeaderAdded(client mainchain.Client, shardID *big.Int, period *big.Int) (bool, error) {
 	// Get the period of the last header.
-	lastPeriod, err := client.SMCCaller().LastSubmittedCollation(&bind.CallOpts{}, shardId)
+	lastPeriod, err := client.SMCCaller().LastSubmittedCollation(&bind.CallOpts{}, shardID)
 	if err != nil {
 		return false, fmt.Errorf("unable to get the period of last submitted collation: %v", err)
 	}

sharding/proposer/service.go

@@ -14,12 +14,14 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/sharding"
 	"github.com/ethereum/go-ethereum/sharding/mainchain"
+	"github.com/ethereum/go-ethereum/sharding/params"
 )
 
 // Proposer holds functionality required to run a collation proposer
 // in a sharded system. Must satisfy the Service interface defined in
 // sharding/service.go.
 type Proposer struct {
+	config       *params.ShardConfig
 	client       *mainchain.SMCClient
 	shardp2p     sharding.ShardP2P
 	txpool       sharding.TXPool
@@ -30,8 +32,8 @@ type Proposer struct {
 // NewProposer creates a struct instance of a proposer service.
 // It will have access to a mainchain client, a shardp2p network,
 // and a shard transaction pool.
-func NewProposer(client *mainchain.SMCClient, shardp2p sharding.ShardP2P, txpool sharding.TXPool, shardChainDb ethdb.Database, shardID int) (*Proposer, error) {
-	return &Proposer{client, shardp2p, txpool, shardChainDb, shardID}, nil
+func NewProposer(config *params.ShardConfig, client *mainchain.SMCClient, shardp2p sharding.ShardP2P, txpool sharding.TXPool, shardChainDb ethdb.Database, shardID int) (*Proposer, error) {
+	return &Proposer{config, client, shardp2p, txpool, shardChainDb, shardID}, nil
 }
 
 // Start the main loop for proposing collations.
@@ -63,7 +65,7 @@ func (p *Proposer) proposeCollations() {
 		log.Error(fmt.Sprintf("Could not fetch current block number: %v", err))
 		return
 	}
-	period := new(big.Int).Div(blockNumber.Number(), big.NewInt(sharding.PeriodLength))
+	period := new(big.Int).Div(blockNumber.Number(), big.NewInt(p.config.PeriodLength))
 
 	// Create collation.
 	collation, err := createCollation(p.client, big.NewInt(int64(p.shardID)), period, txs)

sharding/proposer/service_test.go

@@ -14,8 +14,8 @@ import (
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
-	"github.com/ethereum/go-ethereum/sharding"
 	"github.com/ethereum/go-ethereum/sharding/contracts"
+	"github.com/ethereum/go-ethereum/sharding/params"
 	"gopkg.in/urfave/cli.v1"
 )
 
@@ -89,10 +89,14 @@ func (m *mockNode) Close() {
 	m.t.Fatal("Close called")
 }
 
-func (s *mockNode) DataDirPath() string {
+func (m *mockNode) DataDirPath() string {
 	return "/tmp/datadir"
 }
 
+func (m *mockNode) GetShardCount() (int64, error) {
+	return 100, nil
+}
+
 func transactOpts() *bind.TransactOpts {
 	return bind.NewKeyedTransactor(key)
 }
@@ -124,7 +128,7 @@ func TestCreateCollation(t *testing.T) {
 	}
 
 	// fast forward to 2nd period.
-	for i := 0; i < 2*int(sharding.PeriodLength); i++ {
+	for i := 0; i < 2*int(params.DefaultShardConfig.PeriodLength); i++ {
 		backend.Commit()
 	}
 
@@ -182,7 +186,7 @@ func TestAddCollation(t *testing.T) {
 	}
 
 	// fast forward to next period.
-	for i := 0; i < int(sharding.PeriodLength); i++ {
+	for i := 0; i < int(params.DefaultShardConfig.PeriodLength); i++ {
 		backend.Commit()
 	}
 
@@ -229,7 +233,7 @@ func TestCheckCollation(t *testing.T) {
 		t.Errorf("Create collation failed: %v", err)
 	}
 
-	for i := 0; i < int(sharding.PeriodLength); i++ {
+	for i := 0; i < int(params.DefaultShardConfig.PeriodLength); i++ {
 		backend.Commit()
 	}