Skip to content

Commit d633d6f

Browse files
GrapeBaBaGrapeBaBa
committed
[FAB-2205]Make gossip comm configuable
https://jira.hyperledger.org/browse/FAB-2205 Change-Id: Ifb12f5c9f269bd2bed5ee74d9aa84dee16b4ed4a Signed-off-by: grapebaba <[email protected]>
1 parent 9f561b8 commit d633d6f

File tree

6 files changed

+89
-47
lines changed

6 files changed

+89
-47
lines changed

‎gossip/comm/comm_impl.go

+14-13
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package comm
1919
import (
2020
"bytes"
2121
"crypto/tls"
22+
"errors"
2223
"fmt"
2324
"math/rand"
2425
"net"
@@ -33,6 +34,7 @@ import (
3334
"github.com/hyperledger/fabric/gossip/util"
3435
proto "github.com/hyperledger/fabric/protos/gossip"
3536
"github.com/op/go-logging"
37+
"github.com/spf13/viper"
3638
"golang.org/x/net/context"
3739
"google.golang.org/grpc"
3840
"google.golang.org/grpc/credentials"
@@ -47,16 +49,15 @@ const (
4749
sendOverflowErr = "Send buffer overflow"
4850
)
4951

50-
var errSendOverflow = fmt.Errorf(sendOverflowErr)
51-
var dialTimeout = defDialTimeout
52+
var errSendOverflow = errors.New(sendOverflowErr)
5253

5354
func init() {
5455
rand.Seed(42)
5556
}
5657

5758
// SetDialTimeout sets the dial timeout
5859
func SetDialTimeout(timeout time.Duration) {
59-
dialTimeout = timeout
60+
viper.Set("peer.gossip.dialTimeout", timeout)
6061
}
6162

6263
func (c *commImpl) SetDialOpts(opts ...grpc.DialOption) {
@@ -75,7 +76,7 @@ func NewCommInstanceWithServer(port int, idMapper identity.Mapper, peerIdentity
7576
var certHash []byte
7677

7778
if len(dialOpts) == 0 {
78-
dialOpts = []grpc.DialOption{grpc.WithTimeout(dialTimeout)}
79+
dialOpts = []grpc.DialOption{grpc.WithTimeout(util.GetDurationOrDefault("peer.gossip.dialTimeout", defDialTimeout))}
7980
}
8081

8182
if port > 0 {
@@ -168,7 +169,7 @@ func (c *commImpl) createConnection(endpoint string, expectedPKIID common.PKIidT
168169
defer c.logger.Debug("Exiting")
169170

170171
if c.isStopping() {
171-
return nil, fmt.Errorf("Stopping")
172+
return nil, errors.New("Stopping")
172173
}
173174
cc, err = grpc.Dial(endpoint, append(c.opts, grpc.WithBlock())...)
174175
if err != nil {
@@ -188,7 +189,7 @@ func (c *commImpl) createConnection(endpoint string, expectedPKIID common.PKIidT
188189
if expectedPKIID != nil && !bytes.Equal(pkiID, expectedPKIID) {
189190
// PKIID is nil when we don't know the remote PKI id's
190191
c.logger.Warning("Remote endpoint claims to be a different peer, expected", expectedPKIID, "but got", pkiID)
191-
return nil, fmt.Errorf("Authentication failure")
192+
return nil, errors.New("Authentication failure")
192193
}
193194
conn := newConnection(cl, cc, stream, nil)
194195
conn.pkiID = pkiID
@@ -275,7 +276,7 @@ func (c *commImpl) Probe(remotePeer *RemotePeer) error {
275276
endpoint := remotePeer.Endpoint
276277
pkiID := remotePeer.PKIID
277278
if c.isStopping() {
278-
return fmt.Errorf("Stopping")
279+
return errors.New("Stopping")
279280
}
280281
c.logger.Debug("Entering, endpoint:", endpoint, "PKIID:", pkiID)
281282
cc, err := grpc.Dial(remotePeer.Endpoint, append(c.opts, grpc.WithBlock())...)
@@ -407,15 +408,15 @@ func (c *commImpl) authenticateRemotePeer(stream stream) (common.PKIidType, erro
407408

408409
c.logger.Debug("Sending", cMsg, "to", remoteAddress)
409410
stream.Send(cMsg)
410-
m := readWithTimeout(stream, defConnTimeout)
411+
m := readWithTimeout(stream, util.GetDurationOrDefault("peer.gossip.connTimeout", defConnTimeout))
411412
if m == nil {
412413
c.logger.Warning("Timed out waiting for connection message from", remoteAddress)
413-
return nil, fmt.Errorf("Timed out")
414+
return nil, errors.New("Timed out")
414415
}
415416
receivedMsg := m.GetConn()
416417
if receivedMsg == nil {
417418
c.logger.Warning("Expected connection message but got", receivedMsg)
418-
return nil, fmt.Errorf("Wrong type")
419+
return nil, errors.New("Wrong type")
419420
}
420421

421422
if receivedMsg.PkiID == nil {
@@ -425,7 +426,7 @@ func (c *commImpl) authenticateRemotePeer(stream stream) (common.PKIidType, erro
425426

426427
if c.isPKIblackListed(receivedMsg.PkiID) {
427428
c.logger.Warning("Connection attempt from", remoteAddress, "but it is black-listed")
428-
return nil, fmt.Errorf("Black-listed")
429+
return nil, errors.New("Black-listed")
429430
}
430431
c.logger.Debug("Received", receivedMsg, "from", remoteAddress)
431432
err = c.idMapper.Put(receivedMsg.PkiID, receivedMsg.Cert)
@@ -456,7 +457,7 @@ func (c *commImpl) authenticateRemotePeer(stream stream) (common.PKIidType, erro
456457

457458
func (c *commImpl) GossipStream(stream proto.Gossip_GossipStreamServer) error {
458459
if c.isStopping() {
459-
return fmt.Errorf("Shutting down")
460+
return errors.New("Shutting down")
460461
}
461462
PKIID, err := c.authenticateRemotePeer(stream)
462463
if err != nil {
@@ -573,7 +574,7 @@ func createGRPCLayer(port int) (*grpc.Server, net.Listener, grpc.DialOption, []b
573574
}
574575

575576
if len(cert.Certificate) == 0 {
576-
panic(fmt.Errorf("Certificate chain is nil"))
577+
panic(errors.New("Certificate chain is nil"))
577578
}
578579

579580
returnedCertHash = certHashFromRawCert(cert.Certificate[0])

‎gossip/comm/comm_test.go

+31-5
Original file line numberDiff line numberDiff line change
@@ -22,14 +22,17 @@ import (
2222
"fmt"
2323
"math/rand"
2424
"os"
25+
"strings"
2526
"sync"
2627
"testing"
2728
"time"
2829

2930
"github.com/hyperledger/fabric/gossip/api"
3031
"github.com/hyperledger/fabric/gossip/common"
3132
"github.com/hyperledger/fabric/gossip/identity"
33+
"github.com/hyperledger/fabric/gossip/util"
3234
proto "github.com/hyperledger/fabric/protos/gossip"
35+
"github.com/spf13/viper"
3336
"github.com/stretchr/testify/assert"
3437
"golang.org/x/net/context"
3538
"google.golang.org/grpc"
@@ -38,7 +41,6 @@ import (
3841

3942
func init() {
4043
rand.Seed(42)
41-
SetDialTimeout(time.Duration(300) * time.Millisecond)
4244
}
4345

4446
func acceptAll(msg interface{}) bool {
@@ -149,6 +151,23 @@ func handshaker(endpoint string, comm Comm, t *testing.T, sigMutator func([]byte
149151
return acceptChan
150152
}
151153

154+
func TestViperConfig(t *testing.T) {
155+
viper.SetConfigName("core")
156+
viper.SetEnvPrefix("CORE")
157+
viper.AddConfigPath("./../../peer")
158+
viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
159+
viper.AutomaticEnv()
160+
err := viper.ReadInConfig()
161+
if err != nil { // Handle errors reading the config file
162+
panic(fmt.Errorf("Fatal error config file: %s \n", err))
163+
}
164+
165+
assert.Equal(t, time.Duration(2)*time.Second, util.GetDurationOrDefault("peer.gossip.connTimeout", 0))
166+
assert.Equal(t, time.Duration(300)*time.Millisecond, util.GetDurationOrDefault("peer.gossip.dialTimeout", 0))
167+
assert.Equal(t, 20, util.GetIntOrDefault("peer.gossip.recvBuffSize", 0))
168+
assert.Equal(t, 20, util.GetIntOrDefault("peer.gossip.sendBuffSize", 0))
169+
}
170+
152171
func TestHandshake(t *testing.T) {
153172
t.Parallel()
154173
comm, _ := newCommInstance(9611, naiveSec)
@@ -285,7 +304,7 @@ func TestParallelSend(t *testing.T) {
285304
defer comm1.Stop()
286305
defer comm2.Stop()
287306

288-
messages2Send := defRecvBuffSize
307+
messages2Send := util.GetIntOrDefault("peer.gossip.recvBuffSize", defRecvBuffSize)
289308

290309
wg := sync.WaitGroup{}
291310
go func() {
@@ -378,7 +397,7 @@ func TestAccept(t *testing.T) {
378397
var evenResults []uint64
379398
var oddResults []uint64
380399

381-
out := make(chan uint64, defRecvBuffSize)
400+
out := make(chan uint64, util.GetIntOrDefault("peer.gossip.recvBuffSize", defRecvBuffSize))
382401
sem := make(chan struct{}, 0)
383402

384403
readIntoSlice := func(a *[]uint64, ch <-chan proto.ReceivedMessage) {
@@ -392,11 +411,11 @@ func TestAccept(t *testing.T) {
392411
go readIntoSlice(&evenResults, evenNONCES)
393412
go readIntoSlice(&oddResults, oddNONCES)
394413

395-
for i := 0; i < defRecvBuffSize; i++ {
414+
for i := 0; i < util.GetIntOrDefault("peer.gossip.recvBuffSize", defRecvBuffSize); i++ {
396415
comm2.Send(createGossipMsg(), remotePeer(7611))
397416
}
398417

399-
waitForMessages(t, out, defRecvBuffSize, "Didn't receive all messages sent")
418+
waitForMessages(t, out, util.GetIntOrDefault("peer.gossip.recvBuffSize", defRecvBuffSize), "Didn't receive all messages sent")
400419

401420
comm1.Stop()
402421
comm2.Stop()
@@ -532,3 +551,10 @@ func waitForMessages(t *testing.T, msgChan chan uint64, count int, errMsg string
532551
}
533552
assert.Equal(t, count, c, errMsg)
534553
}
554+
555+
func TestMain(m *testing.M) {
556+
SetDialTimeout(time.Duration(300) * time.Millisecond)
557+
558+
ret := m.Run()
559+
os.Exit(ret)
560+
}

‎gossip/comm/conn.go

+4-3
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222
"sync/atomic"
2323

2424
"github.com/hyperledger/fabric/gossip/common"
25+
"github.com/hyperledger/fabric/gossip/util"
2526
proto "github.com/hyperledger/fabric/protos/gossip"
2627
"github.com/op/go-logging"
2728
"google.golang.org/grpc"
@@ -183,7 +184,7 @@ func (cs *connectionStore) closeByPKIid(pkiID common.PKIidType) {
183184

184185
func newConnection(cl proto.GossipClient, c *grpc.ClientConn, cs proto.Gossip_GossipStreamClient, ss proto.Gossip_GossipStreamServer) *connection {
185186
connection := &connection{
186-
outBuff: make(chan *msgSending, defSendBuffSize),
187+
outBuff: make(chan *msgSending, util.GetIntOrDefault("peer.gossip.sendBuffSize", defSendBuffSize)),
187188
cl: cl,
188189
conn: c,
189190
clientStream: cs,
@@ -242,7 +243,7 @@ func (conn *connection) send(msg *proto.GossipMessage, onErr func(error)) {
242243
conn.Lock()
243244
defer conn.Unlock()
244245

245-
if len(conn.outBuff) == defSendBuffSize {
246+
if len(conn.outBuff) == util.GetIntOrDefault("peer.gossip.sendBuffSize", defSendBuffSize) {
246247
go onErr(errSendOverflow)
247248
return
248249
}
@@ -257,7 +258,7 @@ func (conn *connection) send(msg *proto.GossipMessage, onErr func(error)) {
257258

258259
func (conn *connection) serviceConnection() error {
259260
errChan := make(chan error, 1)
260-
msgChan := make(chan *proto.GossipMessage, defRecvBuffSize)
261+
msgChan := make(chan *proto.GossipMessage, util.GetIntOrDefault("peer.gossip.recvBuffSize", defRecvBuffSize))
261262
defer close(msgChan)
262263

263264
// Call stream.Recv() asynchronously in readFromStream(),

‎gossip/integration/integration.go

+11-26
Original file line numberDiff line numberDiff line change
@@ -25,28 +25,13 @@ import (
2525
"github.com/hyperledger/fabric/gossip/api"
2626
"github.com/hyperledger/fabric/gossip/gossip"
2727
"github.com/hyperledger/fabric/gossip/identity"
28+
"github.com/hyperledger/fabric/gossip/util"
2829
"github.com/spf13/viper"
2930
"google.golang.org/grpc"
3031
)
3132

3233
// This file is used to bootstrap a gossip instance and/or leader election service instance
3334

34-
func getIntOrDefault(key string, defVal int) int {
35-
if viper.GetInt(key) == 0 {
36-
return defVal
37-
} else {
38-
return viper.GetInt(key)
39-
}
40-
}
41-
42-
func getDurationOrDefault(key string, defVal time.Duration) time.Duration {
43-
if viper.GetDuration(key) == 0 {
44-
return defVal
45-
} else {
46-
return viper.GetDuration(key)
47-
}
48-
}
49-
5035
func newConfig(selfEndpoint string, externalEndpoint string, bootPeers ...string) *gossip.Config {
5136
port, err := strconv.ParseInt(strings.Split(selfEndpoint, ":")[1], 10, 64)
5237
if err != nil {
@@ -65,18 +50,18 @@ func newConfig(selfEndpoint string, externalEndpoint string, bootPeers ...string
6550
BindPort: int(port),
6651
BootstrapPeers: bootPeers,
6752
ID: selfEndpoint,
68-
MaxBlockCountToStore: getIntOrDefault("peer.gossip.maxBlockCountToStore", 100),
69-
MaxPropagationBurstLatency: getDurationOrDefault("peer.gossip.maxPropagationBurstLatency", 10*time.Millisecond),
70-
MaxPropagationBurstSize: getIntOrDefault("peer.gossip.maxPropagationBurstSize", 10),
71-
PropagateIterations: getIntOrDefault("peer.gossip.propagateIterations", 1),
72-
PropagatePeerNum: getIntOrDefault("peer.gossip.propagatePeerNum", 3),
73-
PullInterval: getDurationOrDefault("peer.gossip.pullInterval", 4*time.Second),
74-
PullPeerNum: getIntOrDefault("peer.gossip.pullPeerNum", 3),
53+
MaxBlockCountToStore: util.GetIntOrDefault("peer.gossip.maxBlockCountToStore", 100),
54+
MaxPropagationBurstLatency: util.GetDurationOrDefault("peer.gossip.maxPropagationBurstLatency", 10*time.Millisecond),
55+
MaxPropagationBurstSize: util.GetIntOrDefault("peer.gossip.maxPropagationBurstSize", 10),
56+
PropagateIterations: util.GetIntOrDefault("peer.gossip.propagateIterations", 1),
57+
PropagatePeerNum: util.GetIntOrDefault("peer.gossip.propagatePeerNum", 3),
58+
PullInterval: util.GetDurationOrDefault("peer.gossip.pullInterval", 4*time.Second),
59+
PullPeerNum: util.GetIntOrDefault("peer.gossip.pullPeerNum", 3),
7560
InternalEndpoint: selfEndpoint,
7661
ExternalEndpoint: externalEndpoint,
77-
PublishCertPeriod: getDurationOrDefault("peer.gossip.publishCertPeriod", 10*time.Second),
78-
RequestStateInfoInterval: getDurationOrDefault("peer.gossip.requestStateInfoInterval", 4*time.Second),
79-
PublishStateInfoInterval: getDurationOrDefault("peer.gossip.publishStateInfoInterval", 4*time.Second),
62+
PublishCertPeriod: util.GetDurationOrDefault("peer.gossip.publishCertPeriod", 10*time.Second),
63+
RequestStateInfoInterval: util.GetDurationOrDefault("peer.gossip.requestStateInfoInterval", 4*time.Second),
64+
PublishStateInfoInterval: util.GetDurationOrDefault("peer.gossip.publishStateInfoInterval", 4*time.Second),
8065
SkipBlockVerification: viper.GetBool("peer.gossip.skipBlockVerification"),
8166
TLSServerCert: cert,
8267
}

‎gossip/util/misc.go

+21
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,9 @@ import (
2222
"reflect"
2323
"runtime"
2424
"sync"
25+
"time"
26+
27+
"github.com/spf13/viper"
2528
)
2629

2730
// Equals returns whether a and b are the same
@@ -146,3 +149,21 @@ func PrintStackTrace() {
146149
runtime.Stack(buf, true)
147150
fmt.Printf("%s", buf)
148151
}
152+
153+
// GetIntOrDefault returns the int value from config if present otherwise default value
154+
func GetIntOrDefault(key string, defVal int) int {
155+
if val := viper.GetInt(key); val != 0 {
156+
return val
157+
}
158+
159+
return defVal
160+
}
161+
162+
// GetIntOrDefault returns the Duration value from config if present otherwise default value
163+
func GetDurationOrDefault(key string, defVal time.Duration) time.Duration {
164+
if val := viper.GetDuration(key); val != 0 {
165+
return val
166+
}
167+
168+
return defVal
169+
}

‎peer/core.yaml

+8
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,14 @@ peer:
100100
skipBlockVerification: false
101101
# Should we ignore security or not
102102
ignoreSecurity: false
103+
# Dial timeout(unit: second)
104+
dialTimeout: 3s
105+
# Connection timeout(unit: second)
106+
connTimeout: 2s
107+
# Buffer size of received messages
108+
recvBuffSize: 20
109+
# Buffer size of sending messages
110+
sendBuffSize: 20
103111

104112
# This is an endpoint that is published to peers outside of the organization.
105113
# If this isn't set, the peer will not be known to other organizations.

0 commit comments

Comments
 (0)