socket: support network IO deadline (#339)

This allows connection-oriented protocols to recover from bad
network conditions, by enabling to set a limit on the time for
network I/O (#339).

conn.go

@@ -19,6 +19,7 @@ import (
 	"io"
 	"net"
 	"sync"
+	"time"
 )
 
 // conn implements the Pipe interface on top of net.Conn.  The
@@ -49,6 +50,10 @@ func (p *conn) Recv() (*Message, error) {
 	var err error
 	var msg *Message
 
+	if err := p.setNetworkTimeout(); err != nil {
+		return nil, err
+	}
+
 	if err = binary.Read(p.c, binary.BigEndian, &sz); err != nil {
 		return nil, err
 	}
@@ -78,6 +83,10 @@ func (p *conn) Send(msg *Message) error {
 		return nil
 	}
 
+	if err := p.setNetworkTimeout(); err != nil {
+		return err
+	}
+
 	// send length header
 	if err := binary.Write(p.c, binary.BigEndian, l); err != nil {
 		return err
@@ -181,6 +190,10 @@ func (p *conn) handshake(props []interface{}) error {
 		p.maxrx = int64(v.(int))
 	}
 
+	if err := p.setNetworkTimeout(); err != nil {
+		return err
+	}
+
 	h := connHeader{S: 'S', P: 'P', Proto: p.proto.Number()}
 	if err = binary.Write(p.c, binary.BigEndian, &h); err != nil {
 		return err
@@ -207,3 +220,13 @@ func (p *conn) handshake(props []interface{}) error {
 	p.open = true
 	return nil
 }
+
+func (p *conn) setNetworkTimeout() error {
+	if v, err := p.sock.GetOption(OptionNetworkIoDeadline); err == nil {
+		// Socket implementation ensures that option type is time.Duration.
+		if iotimeout := v.(time.Duration); iotimeout > 0 {
+			return p.c.SetDeadline(time.Now().Add(iotimeout))
+		}
+	}
+	return nil
+}

core.go

@@ -46,8 +46,9 @@ type socket struct {
 	recverr    error // error to return on attempts to Recv()
 	senderr    error // error to return on attempts to Send()
 
-	rdeadline  time.Duration
-	wdeadline  time.Duration
+	rdeadline  time.Duration // mangos socket read deadline
+	wdeadline  time.Duration // mangos socket write deadline
+	iodeadline time.Duration // IO timeout for connection used by pipe
 	reconntime time.Duration // reconnect time after error or disconnect
 	reconnmax  time.Duration // max reconnect interval
 	linger     time.Duration
@@ -437,6 +438,11 @@ func (sock *socket) SetOption(name string, value interface{}) error {
 		sock.wdeadline = value.(time.Duration)
 		sock.Unlock()
 		return nil
+	case OptionNetworkIoDeadline:
+		sock.Lock()
+		sock.iodeadline = value.(time.Duration)
+		sock.Unlock()
+		return nil
 	case OptionLinger:
 		sock.Lock()
 		sock.linger = value.(time.Duration)
@@ -523,6 +529,10 @@ func (sock *socket) GetOption(name string) (interface{}, error) {
 		sock.Lock()
 		defer sock.Unlock()
 		return sock.wdeadline, nil
+	case OptionNetworkIoDeadline:
+		sock.Lock()
+		defer sock.Unlock()
+		return sock.iodeadline, nil
 	case OptionLinger:
 		sock.Lock()
 		defer sock.Unlock()

options.go

@@ -39,6 +39,13 @@ const (
 	// non-blocking operation.  By default there is no timeout.
 	OptionSendDeadline = "SEND-DEADLINE"
 
+	// OptionNetworkIoDeadline enables recovery from bad connections for
+	// connection-based transports; by limiting the time until a (read or
+	// write) network operation times out.  The value is a time.Duration.
+	// Setting this option is recommended for connection-based protocols,
+	// it is disabled by default.
+	OptionNetworkIoDeadline = "NETWORK-IO-DEADLINE"
+
 	// OptionRetryTime is used by REQ.  The argument is a time.Duration.
 	// When a request has not been replied to within the given duration,
 	// the request will automatically be resent to an available peer.