diff --git a/client/conn.go b/client/conn.go index 54453e6f8..d0b2d3f47 100644 --- a/client/conn.go +++ b/client/conn.go @@ -54,6 +54,7 @@ type conn struct { // pconn represents a connection to a peer. type pconn struct { + wg *sync.WaitGroup parent *conn ackCh chan *types.Ack pCaller rpc.PCaller @@ -147,8 +148,10 @@ func newConn(cfg *Config) (c *conn, err error) { } func (c *pconn) startAckWorkers(workerCount int) (err error) { + c.wg = &sync.WaitGroup{} c.ackCh = make(chan *types.Ack, workerCount*4) for i := 0; i < workerCount; i++ { + c.wg.Add(1) go c.ackWorker() } return @@ -156,15 +159,13 @@ func (c *pconn) startAckWorkers(workerCount int) (err error) { func (c *pconn) stopAckWorkers() { if c.ackCh != nil { - select { - case <-c.ackCh: - default: - close(c.ackCh) - } + close(c.ackCh) } } func (c *pconn) ackWorker() { + defer c.wg.Done() + var ( oneTime sync.Once pc rpc.PCaller @@ -202,6 +203,7 @@ ackWorkerLoop: func (c *pconn) close() error { c.stopAckWorkers() + c.wg.Wait() if c.pCaller != nil { c.pCaller.Close() } diff --git a/cmd/cql-minerd/integration_test.go b/cmd/cql-minerd/integration_test.go index ceb079a8c..3dd048e24 100644 --- a/cmd/cql-minerd/integration_test.go +++ b/cmd/cql-minerd/integration_test.go @@ -47,6 +47,7 @@ import ( "github.com/CovenantSQL/CovenantSQL/naconn" "github.com/CovenantSQL/CovenantSQL/proto" "github.com/CovenantSQL/CovenantSQL/route" + rrpc "github.com/CovenantSQL/CovenantSQL/rpc" rpc "github.com/CovenantSQL/CovenantSQL/rpc/mux" "github.com/CovenantSQL/CovenantSQL/test" "github.com/CovenantSQL/CovenantSQL/types" @@ -745,13 +746,6 @@ func benchDB(b *testing.B, db *sql.DB, createDB bool) { }) }) - routineCount := runtime.NumGoroutine() - if routineCount > 500 { - b.Errorf("go routine count: %d", routineCount) - } else { - log.Infof("go routine count: %d", routineCount) - } - rowCount := db.QueryRow("SELECT COUNT(1) FROM " + TABLENAME) var count int64 err = rowCount.Scan(&count) @@ -787,13 +781,6 @@ func benchDB(b *testing.B, db *sql.DB, createDB bool) { }) }) - routineCount = runtime.NumGoroutine() - if routineCount > 500 { - b.Errorf("go routine count: %d", routineCount) - } else { - log.Infof("go routine count: %d", routineCount) - } - //row := db.QueryRow("SELECT nonIndexedColumn FROM test LIMIT 1") //var result int @@ -803,6 +790,13 @@ func benchDB(b *testing.B, db *sql.DB, createDB bool) { err = db.Close() So(err, ShouldBeNil) + + routineCount := runtime.NumGoroutine() + if routineCount > 500 { + b.Errorf("go routine count: %d", routineCount) + } else { + log.Infof("go routine count: %d", routineCount) + } } func benchMiner(b *testing.B, minerCount uint16) { @@ -962,7 +956,7 @@ func benchOutsideMinerWithTargetMinerList( Node: minerCount, UseEventualConsistency: benchEventualConsistency, IsolationLevel: int(sql.LevelReadUncommitted), - AdvancePayment: 1000000000, + AdvancePayment: testAdvancePayment, } // wait for chain service var ctx1, cancel1 = context.WithTimeout(context.Background(), 1*time.Minute) @@ -996,6 +990,8 @@ func benchOutsideMinerWithTargetMinerList( So(err, ShouldBeNil) benchDB(b, db, minerCount > 0) + rrpc.Stat() + rpc.Stat() } func BenchmarkClientOnly(b *testing.B) { diff --git a/rpc/mux/pool.go b/rpc/mux/pool.go index ef52df35f..40b46feef 100644 --- a/rpc/mux/pool.go +++ b/rpc/mux/pool.go @@ -20,6 +20,7 @@ import ( "net" "strings" "sync" + "sync/atomic" "github.com/pkg/errors" mux "github.com/xtaci/smux" @@ -27,6 +28,13 @@ import ( "github.com/CovenantSQL/CovenantSQL/conf" "github.com/CovenantSQL/CovenantSQL/proto" "github.com/CovenantSQL/CovenantSQL/rpc" + "github.com/CovenantSQL/CovenantSQL/utils/log" +) + +var ( + openSession uint32 + openStreamAttempt uint32 + openStreamSucc uint32 ) // Session is the Session type of SessionPool. @@ -91,6 +99,7 @@ func (s *Session) Get() (conn rpc.Client, err error) { if err != nil { return } + atomic.AddUint32(&openSession, 1) s.sess = append(s.sess, sess) s.offset = len(s.sess) - 1 } else { @@ -98,9 +107,11 @@ func (s *Session) Get() (conn rpc.Client, err error) { } // open connection + atomic.AddUint32(&openStreamAttempt, 1) stream, err = sess.OpenStream() if err != nil { // invalidate session + log.WithError(err).Debug("fail to open stream") sessions = nil sessions = append(sessions, s.sess[0:s.offset]...) sessions = append(sessions, s.sess[s.offset+1:]...) @@ -108,10 +119,19 @@ func (s *Session) Get() (conn rpc.Client, err error) { continue } + atomic.AddUint32(&openStreamSucc, 1) return rpc.NewClient(stream), nil } } +func Stat() { + log.WithFields(log.Fields{ + "open_session": openSession, + "open_stream_attempt": openStreamAttempt, + "open_stream_succ": openStreamSucc, + }).Info("session pool stat") +} + // Len returns physical connection count. func (s *Session) Len() int { s.RLock() @@ -203,10 +223,13 @@ func (p *SessionPool) GetEx(id proto.NodeID, isAnonymous bool) (conn rpc.Client, if sess, err = newSession(id, true); err != nil { return } + atomic.AddUint32(&openSession, 1) + atomic.AddUint32(&openStreamAttempt, 1) if stream, err = sess.OpenStream(); err != nil { err = errors.Wrapf(err, "open new session to %s failed", id) return } + atomic.AddUint32(&openStreamSucc, 1) return rpc.NewClient(&oneOffMuxConn{ sess: sess, Stream: stream, diff --git a/rpc/pcaller.go b/rpc/pcaller.go index e89e25638..66b132f64 100644 --- a/rpc/pcaller.go +++ b/rpc/pcaller.go @@ -21,12 +21,19 @@ import ( "net/rpc" "strings" "sync" + "sync/atomic" "time" "github.com/pkg/errors" "github.com/CovenantSQL/CovenantSQL/proto" "github.com/CovenantSQL/CovenantSQL/route" + "github.com/CovenantSQL/CovenantSQL/utils/log" +) + +var ( + dialCount uint32 + closeCount uint32 ) // PersistentCaller is a wrapper for session pooling and RPC calling. @@ -57,11 +64,23 @@ func (c *PersistentCaller) initClient(isAnonymous bool) (err error) { err = errors.Wrap(err, "dial to node failed") return } + log.WithFields(log.Fields{ + "is_anonymous": isAnonymous, + "node": c.TargetID, + }).Info("dial node with pool") + atomic.AddUint32(&dialCount, 1) //c.TargetAddr = conn.RemoteAddr().String() } return } +func Stat() { + log.WithFields(log.Fields{ + "dial_count": dialCount, + "close_count": closeCount, + }).Info("pcaller stat") +} + // Call invokes the named function, waits for it to complete, and returns its error status. func (c *PersistentCaller) Call(method string, args interface{}, reply interface{}) (err error) { startTime := time.Now() @@ -100,6 +119,10 @@ func (c *PersistentCaller) Call(method string, args interface{}, reply interface func (c *PersistentCaller) ResetClient() (err error) { c.Lock() if c.client != nil { + log.WithFields(log.Fields{ + "node": c.TargetID, + }).Info("close node") + atomic.AddUint32(&closeCount, 1) _ = c.client.Close() } c.client = nil @@ -112,6 +135,10 @@ func (c *PersistentCaller) Close() { c.Lock() defer c.Unlock() if c.client != nil { + log.WithFields(log.Fields{ + "node": c.TargetID, + }).Info("close node") + atomic.AddUint32(&closeCount, 1) _ = c.client.Close() } }