Merge branch 'master' into minor_sync_issue

examples/getting_started/main.go

@@ -31,8 +31,7 @@ func main() {
 
 	keys := ed25519.RandomKeyPair()
 
-	log.Info().Str("private_key", keys.PrivateKeyHex()).Msg("")
-	log.Info().Str("public_key", keys.PublicKeyHex()).Msg("")
+	log.Info().Str("private_key", keys.PrivateKeyHex()).Str("public_key", keys.PublicKeyHex()).Msg("Generated keypair.")
 
 	builder := network.NewBuilder()
 	builder.SetKeys(keys)

examples/request_benchmark/main.go

@@ -20,8 +20,8 @@ import (
 )
 
 const (
-	defaultNumNodes      = 5
-	defaultNumReqPerNode = 50
+	defaultNumNodes      = 8
+	defaultNumReqPerNode = 100
 	host                 = "localhost"
 	startPort            = 23000
 )
@@ -109,7 +109,7 @@ func setupNetworks(host string, startPort int, numNodes int) []*network.Network
 	}
 
 	// Wait for all nodes to finish discovering other peers.
-	time.Sleep(1 * time.Second)
+	time.Sleep(5 * time.Second)
 
 	return nodes
 }

log/log.go

@@ -10,7 +10,7 @@ import (
 )
 
 var (
-	logger = zerolog.New(os.Stderr).With().Timestamp().Caller().Logger()
+	logger = zerolog.New(os.Stderr).With().Timestamp().Logger()
 )
 
 func init() {

network/client.go

@@ -14,12 +14,10 @@ import (
 	"github.com/pkg/errors"
 )
 
-const (
-	defaultRequestTimeout = 5 * time.Second
-)
-
 // PeerClient represents a single incoming peers client.
 type PeerClient struct {
+	sync.Once
+
 	Network *Network
 
 	ID      *peer.ID

network/discovery/plugin.go

@@ -62,7 +62,7 @@ func (state *Plugin) Receive(ctx *network.PluginContext) error {
 
 		log.Info().
 			Strs("peers", state.Routes.GetPeerAddresses()).
-			Msg("bootstrapped w/ peer(s)")
+			Msg("Bootstrapped w/ peer(s).")
 	case *protobuf.LookupNodeRequest:
 		if state.DisableLookup {
 			break
@@ -84,7 +84,7 @@ func (state *Plugin) Receive(ctx *network.PluginContext) error {
 
 		log.Info().
 			Strs("peers", state.Routes.GetPeerAddresses()).
-			Msg("connected to peer(s)")
+			Msg("Connected to peer(s).")
 	}
 
 	return nil
@@ -100,10 +100,10 @@ func (state *Plugin) PeerDisconnect(client *network.PeerClient) {
 		if state.Routes.PeerExists(*client.ID) {
 			state.Routes.RemovePeer(*client.ID)
 
-			log.Info().
+			log.Debug().
 				Str("address", client.Network.ID.Address).
 				Str("peer_address", client.ID.Address).
-				Msg("peer has disconnected")
+				Msg("Peer has disconnected.")
 		}
 	}
 }

network/network.go

@@ -3,7 +3,6 @@ package network
 import (
 	"bufio"
 	"context"
-	"fmt"
 	"math/rand"
 	"net"
 	"sync"
@@ -138,15 +137,15 @@ func (n *Network) dispatchMessage(client *PeerClient, msg *protobuf.Message) {
 	code := opcode.Opcode(msg.Opcode)
 	switch code {
 	case opcode.BytesCode:
-		ptr = &protobuf.Bytes{}
+		ptr = new(protobuf.Bytes)
 	case opcode.PingCode:
-		ptr = &protobuf.Ping{}
+		ptr = new(protobuf.Ping)
 	case opcode.PongCode:
-		ptr = &protobuf.Pong{}
+		ptr = new(protobuf.Pong)
 	case opcode.LookupNodeRequestCode:
-		ptr = &protobuf.LookupNodeRequest{}
+		ptr = new(protobuf.LookupNodeRequest)
 	case opcode.LookupNodeResponseCode:
-		ptr = &protobuf.LookupNodeResponse{}
+		ptr = new(protobuf.LookupNodeResponse)
 	case opcode.UnregisteredCode:
 		log.Error().Msg("network: message received had no opcode")
 		return
@@ -158,9 +157,10 @@ func (n *Network) dispatchMessage(client *PeerClient, msg *protobuf.Message) {
 			return
 		}
 	}
+
 	if len(msg.Message) > 0 {
 		if err := proto.Unmarshal(msg.Message, ptr); err != nil {
-			log.Error().Err(err).Msg("")
+			log.Error().Msgf("%v", err)
 			return
 		}
 	}
@@ -200,7 +200,6 @@ func (n *Network) dispatchMessage(client *PeerClient, msg *protobuf.Message) {
 
 // Listen starts listening for peers on a port.
 func (n *Network) Listen() {
-
 	// Handle 'network starts listening' callback for plugins.
 	n.plugins.Each(func(plugin PluginInterface) {
 		plugin.Startup(n)
@@ -234,7 +233,7 @@ func (n *Network) Listen() {
 
 	log.Info().
 		Str("address", n.Address).
-		Msg("listening for peers")
+		Msg("Listening for peers.")
 
 	// handle server shutdowns
 	go func() {
@@ -249,23 +248,21 @@ func (n *Network) Listen() {
 	for {
 		if conn, err := listener.Accept(); err == nil {
 			go n.Accept(conn)
-
 		} else {
 			// if the Shutdown flag is set, no need to continue with the for loop
 			select {
 			case <-n.kill:
-				log.Info().Msgf("shutting down server on %s.", n.Address)
+				log.Info().Msgf("Shutting down server %s.", n.Address)
 				return
 			default:
-				log.Error().Err(err).Msg("")
+				log.Error().Msgf("%v", err)
 			}
 		}
 	}
 }
 
-// getOrSetPeerClient either returns a cached peer client or creates a new one given a net.Conn
-// or dials the client if no net.Conn is provided.
-func (n *Network) getOrSetPeerClient(address string, conn net.Conn) (*PeerClient, error) {
+// Client either creates or returns a cached peer client given its host address.
+func (n *Network) Client(address string) (*PeerClient, error) {
 	address, err := ToUnifiedAddress(address)
 	if err != nil {
 		return nil, err
@@ -275,23 +272,6 @@ func (n *Network) getOrSetPeerClient(address string, conn net.Conn) (*PeerClient
 		return nil, errors.New("network: peer should not dial itself")
 	}
 
-	// if conn is not nil, check that the sender host matches the net.Conn remote host address
-	if conn != nil {
-		addrInfo, err := ParseAddress(address)
-		if err != nil {
-			return nil, err
-		}
-
-		remoteAddrInfo, err := ParseAddress(fmt.Sprintf("%s://%s", conn.RemoteAddr().Network(), conn.RemoteAddr().String()))
-		if err != nil {
-			return nil, err
-		}
-
-		if addrInfo.Host != remoteAddrInfo.Host {
-			return nil, errors.New("network: sender address did not match connection remote address")
-		}
-	}
-
 	clientNew, err := createPeerClient(n, address)
 	if err != nil {
 		return nil, err
@@ -313,12 +293,10 @@ func (n *Network) getOrSetPeerClient(address string, conn net.Conn) (*PeerClient
 		client.setOutgoingReady()
 	}()
 
-	if conn == nil {
-		conn, err = n.Dial(address)
-		if err != nil {
-			n.peers.Delete(address)
-			return nil, err
-		}
+	conn, err := n.Dial(address)
+	if err != nil {
+		n.peers.Delete(address)
+		return nil, err
 	}
 
 	n.connections.Store(address, &ConnState{
@@ -332,11 +310,6 @@ func (n *Network) getOrSetPeerClient(address string, conn net.Conn) (*PeerClient
 	return client, nil
 }
 
-// Client either creates or returns a cached peer client given its host address.
-func (n *Network) Client(address string) (*PeerClient, error) {
-	return n.getOrSetPeerClient(address, nil)
-}
-
 // ConnectionStateExists returns true if network has a connection on a given address.
 func (n *Network) ConnectionStateExists(address string) bool {
 	_, ok := n.connections.Load(address)
@@ -414,16 +387,12 @@ func (n *Network) Dial(address string) (net.Conn, error) {
 		return nil, err
 	}
 
-	// use the connection for also receiving messages
-	go n.Accept(conn)
-
 	return conn, nil
 }
 
 // Accept handles peer registration and processes incoming message streams.
 func (n *Network) Accept(incoming net.Conn) {
 	var client *PeerClient
-	var clientInit sync.Once
 
 	recvWindow := NewRecvWindow(n.opts.recvWindowSize)
 
@@ -444,18 +413,21 @@ func (n *Network) Accept(incoming net.Conn) {
 		msg, err := n.receiveMessage(incoming)
 		if err != nil {
 			if err != errEmptyMsg {
-				log.Error().Err(err).Msg("")
+				log.Error().Msgf("%v", err)
 			}
 			break
 		}
 
 		// Initialize client if not exists.
-		clientInit.Do(func() {
-			client, err = n.getOrSetPeerClient(msg.Sender.Address, incoming)
+		if client == nil {
+			client, err = n.Client(msg.Sender.Address)
+
 			if err != nil {
 				return
 			}
+		}
 
+		client.Do(func() {
 			client.ID = (*peer.ID)(msg.Sender)
 
 			if !n.ConnectionStateExists(client.ID.Address) {
@@ -476,7 +448,7 @@ func (n *Network) Accept(incoming net.Conn) {
 				log.Error().
 					Interface("peer_id", peer.ID(*msg.Sender)).
 					Interface("client_id", client.ID).
-					Msg("message signed by peer does not match client ID")
+					Msg("Message signed by peer does not match client ID.")
 				return
 			}