-
Notifications
You must be signed in to change notification settings - Fork 721
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #48 from pingcap/qiuyesuifeng/node-conns
add keep alive connection support for sending command to raft server
- Loading branch information
Showing
8 changed files
with
275 additions
and
42 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| package server | ||
|
|
||
| import ( | ||
| "net" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/juju/errors" | ||
| "github.com/ngaut/log" | ||
| "github.com/ngaut/sync2" | ||
| ) | ||
|
|
||
| const ( | ||
| connectTimeout = 3 * time.Second | ||
| idleTimeout = 30 * time.Second | ||
| ) | ||
|
|
||
| type nodeConn struct { | ||
| conn net.Conn | ||
| touchedTime time.Time | ||
| } | ||
|
|
||
| func (nc *nodeConn) close() error { | ||
| return errors.Trace(nc.conn.Close()) | ||
| } | ||
|
|
||
| func newNodeConn(addr string) (*nodeConn, error) { | ||
| conn, err := net.DialTimeout("tcp", addr, connectTimeout) | ||
| if err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| return &nodeConn{ | ||
| conn: conn, | ||
| touchedTime: time.Now()}, nil | ||
| } | ||
|
|
||
| type createConnFunc func(addr string) (*nodeConn, error) | ||
|
|
||
| var defaultConnFunc = newNodeConn | ||
|
|
||
| type nodeConns struct { | ||
| m sync.Mutex | ||
| conns map[string]*nodeConn | ||
| idleTimeout sync2.AtomicDuration | ||
| f createConnFunc | ||
| } | ||
|
|
||
| // newNodeConns creates a new node conns. | ||
| func newNodeConns(f createConnFunc) *nodeConns { | ||
| ncs := new(nodeConns) | ||
| ncs.f = f | ||
| ncs.conns = make(map[string]*nodeConn) | ||
| return ncs | ||
| } | ||
|
|
||
| // This function is not thread-safed. | ||
| func (ncs *nodeConns) createNewConn(addr string) (*nodeConn, error) { | ||
| conn, err := ncs.f(addr) | ||
| if err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| ncs.conns[addr] = conn | ||
| return conn, nil | ||
| } | ||
|
|
||
| // SetIdleTimeout sets idleTimeout of each conn. | ||
| func (ncs *nodeConns) SetIdleTimeout(idleTimeout time.Duration) { | ||
| ncs.idleTimeout.Set(idleTimeout) | ||
| } | ||
|
|
||
| // GetConn gets the conn by addr. | ||
| func (ncs *nodeConns) GetConn(addr string) (*nodeConn, error) { | ||
| ncs.m.Lock() | ||
| defer ncs.m.Unlock() | ||
|
|
||
| conn, ok := ncs.conns[addr] | ||
| if !ok { | ||
| return ncs.createNewConn(addr) | ||
| } | ||
|
|
||
| timeout := ncs.idleTimeout.Get() | ||
| if timeout > 0 && conn.touchedTime.Add(timeout).Sub(time.Now()) < 0 { | ||
| err := conn.close() | ||
| if err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| return ncs.createNewConn(addr) | ||
| } | ||
|
|
||
| conn.touchedTime = time.Now() | ||
| return conn, nil | ||
| } | ||
|
|
||
| // RemoveConn removes the conn by addr. | ||
| func (ncs *nodeConns) RemoveConn(addr string) { | ||
| ncs.m.Lock() | ||
| defer ncs.m.Unlock() | ||
|
|
||
| conn, ok := ncs.conns[addr] | ||
| if !ok { | ||
| return | ||
| } | ||
|
|
||
| err := conn.close() | ||
| if err != nil { | ||
| log.Warnf("close node conn failed - %v", err) | ||
| } | ||
| delete(ncs.conns, addr) | ||
| } | ||
|
|
||
| // Close closes the conns. | ||
| func (ncs *nodeConns) Close() { | ||
| ncs.m.Lock() | ||
| defer ncs.m.Unlock() | ||
|
|
||
| for _, conn := range ncs.conns { | ||
| err := conn.close() | ||
| if err != nil { | ||
| log.Warnf("close node conn failed - %v", err) | ||
| } | ||
| } | ||
|
|
||
| ncs.conns = map[string]*nodeConn{} | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| package server | ||
|
|
||
| import ( | ||
| "net" | ||
| "time" | ||
|
|
||
| . "github.com/pingcap/check" | ||
| ) | ||
|
|
||
| type testNodeConnSuite struct { | ||
| } | ||
|
|
||
| var _ = Suite(&testNodeConnSuite{}) | ||
|
|
||
| type testConn struct { | ||
| } | ||
|
|
||
| func (c *testConn) Read(b []byte) (n int, err error) { return len(b), nil } | ||
| func (c *testConn) Write(b []byte) (n int, err error) { return len(b), nil } | ||
| func (c *testConn) Close() error { return nil } | ||
| func (c *testConn) LocalAddr() net.Addr { return nil } | ||
| func (c *testConn) RemoteAddr() net.Addr { return nil } | ||
| func (c *testConn) SetDeadline(t time.Time) error { return nil } | ||
| func (c *testConn) SetReadDeadline(t time.Time) error { return nil } | ||
| func (c *testConn) SetWriteDeadline(t time.Time) error { return nil } | ||
|
|
||
| func testNodeConn(addr string) (*nodeConn, error) { | ||
| return &nodeConn{ | ||
| conn: &testConn{}, | ||
| touchedTime: time.Now()}, nil | ||
| } | ||
|
|
||
| func (s *testNodeConnSuite) TestNodeConns(c *C) { | ||
| conns := newNodeConns(testNodeConn) | ||
| c.Assert(conns.conns, HasLen, 0) | ||
|
|
||
| addr1 := "127.0.0.1:1" | ||
| oldConn, err := conns.GetConn(addr1) | ||
| c.Assert(err, IsNil) | ||
| c.Assert(conns.conns, HasLen, 1) | ||
| c.Assert(conns.conns, HasKey, addr1) | ||
|
|
||
| newConn, err := conns.GetConn(addr1) | ||
| c.Assert(err, IsNil) | ||
| c.Assert(conns.conns, HasLen, 1) | ||
| c.Assert(conns.conns, HasKey, addr1) | ||
|
|
||
| c.Assert(oldConn, Equals, newConn) | ||
|
|
||
| conns.RemoveConn(addr1) | ||
| c.Assert(conns.conns, HasLen, 0) | ||
|
|
||
| addr2 := "127.0.0.1:2" | ||
| conns.GetConn(addr2) | ||
| c.Assert(conns.conns, HasLen, 1) | ||
| c.Assert(conns.conns, HasKey, addr2) | ||
|
|
||
| conns.Close() | ||
| c.Assert(conns.conns, HasLen, 0) | ||
|
|
||
| // Test with idleTimeout conn. | ||
| idleTimeout := 100 * time.Millisecond | ||
| conns.SetIdleTimeout(idleTimeout) | ||
|
|
||
| addr3 := "127.0.0.1:3" | ||
| oldConn, err = conns.GetConn(addr3) | ||
| c.Assert(err, IsNil) | ||
| c.Assert(conns.conns, HasLen, 1) | ||
| c.Assert(conns.conns, HasKey, addr3) | ||
|
|
||
| time.Sleep(2 * idleTimeout) | ||
|
|
||
| c.Assert(conns.conns, HasLen, 1) | ||
| c.Assert(conns.conns, HasKey, addr3) | ||
|
|
||
| newConn, err = conns.GetConn(addr3) | ||
| c.Assert(err, IsNil) | ||
| c.Assert(conns.conns, HasLen, 1) | ||
| c.Assert(conns.conns, HasKey, addr3) | ||
|
|
||
| c.Assert(oldConn, Not(Equals), newConn) | ||
|
|
||
| addr4 := "127.0.0.1:4" | ||
| conns.GetConn(addr4) | ||
| c.Assert(conns.conns, HasLen, 2) | ||
| c.Assert(conns.conns, HasKey, addr4) | ||
|
|
||
| conns.Close() | ||
| c.Assert(conns.conns, HasLen, 0) | ||
| } |
Oops, something went wrong.