diff --git a/cluster/cluster.go b/cluster/cluster.go index d028e95..f143661 100644 --- a/cluster/cluster.go +++ b/cluster/cluster.go @@ -7,6 +7,7 @@ import ( "os" "sort" "strings" + "sync" "time" "github.com/duanhf2012/origin/rpc" @@ -33,6 +34,7 @@ type CCluster struct { writer net.Conn LocalRpcClient *rpc.Client + localRpcLocker sync.Mutex innerLocalServiceList map[string]bool } @@ -186,7 +188,7 @@ func (slf *CCluster) ConnService() error { if slf.LocalRpcClient.IsClosed() { slf.ReSetLocalRpcClient() } - time.Sleep(time.Second * 4) + time.Sleep(time.Second * 2) } return nil diff --git a/rpc/client.go b/rpc/client.go index 7865491..4cef3e0 100644 --- a/rpc/client.go +++ b/rpc/client.go @@ -51,6 +51,7 @@ type Client struct { pending map[uint64]*Call closing bool // user has called Close shutdown bool // server has told us to stop + bPipe bool } // A ClientCodec implements writing of RPC requests and @@ -151,6 +152,9 @@ func (client *Client) input() { if err != nil { call.Error = errors.New("reading body " + err.Error()) } + if client.bPipe { + err = nil + } call.done() } } @@ -199,18 +203,19 @@ func (call *Call) done() { // so no interlocking is required. However each half may be accessed // concurrently so the implementation of conn should protect against // concurrent reads or concurrent writes. -func NewClient(conn io.ReadWriteCloser) *Client { +func NewClient(conn io.ReadWriteCloser, isPipe bool) *Client { encBuf := bufio.NewWriter(conn) client := &gobClientCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(encBuf), encBuf} - return NewClientWithCodec(client) + return NewClientWithCodec(client, isPipe) } // NewClientWithCodec is like NewClient but uses the specified // codec to encode requests and decode responses. -func NewClientWithCodec(codec ClientCodec) *Client { +func NewClientWithCodec(codec ClientCodec, isPipe bool) *Client { client := &Client{ codec: codec, pending: make(map[uint64]*Call), + bPipe: isPipe, } go client.input() return client @@ -269,7 +274,7 @@ func DialHTTPPath(network, address, path string) (*Client, error) { // before switching to RPC protocol. resp, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: "CONNECT"}) if err == nil && resp.Status == connected { - return NewClient(conn), nil + return NewClient(conn, false), nil } if err == nil { err = errors.New("unexpected HTTP response: " + resp.Status) @@ -291,7 +296,7 @@ func Dial(network, address string) (*Client, error) { } tcpconn, _ := conn.(*net.TCPConn) tcpconn.SetNoDelay(true) - return NewClient(conn), nil + return NewClient(conn, false), nil } func DialTimeOut(network, address string, timeout time.Duration) (*Client, error) { @@ -301,7 +306,7 @@ func DialTimeOut(network, address string, timeout time.Duration) (*Client, error } tcpconn, _ := conn.(*net.TCPConn) tcpconn.SetNoDelay(true) - return NewClient(conn), nil + return NewClient(conn, false), nil } // Close calls the underlying codec's Close method. If the connection is already