ClickHouse#74 ClickHouse#86 read/write timeouts

Author: kshvakov

.travis.yml

@@ -1,4 +1,3 @@
-group: deprecated-2017Q4
 sudo: required
 language: go
 go:

bootstrap.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/kshvakov/clickhouse/lib/binary"
@@ -26,12 +27,25 @@ const (
 )
 
 var (
+	unixtime    int64
 	logOutput   io.Writer = os.Stdout
 	hostname, _           = os.Hostname()
 )
 
 func init() {
 	sql.Register("clickhouse", &bootstrap{})
+	go func() {
+		for tick := time.Tick(time.Second); ; {
+			select {
+			case <-tick:
+				atomic.AddInt64(&unixtime, int64(time.Second))
+			}
+		}
+	}()
+}
+
+func now() time.Time {
+	return time.Unix(atomic.LoadInt64(&unixtime), 0)
 }
 
 type bootstrap struct{}
@@ -112,11 +126,9 @@ func open(dsn string) (*clickhouse, error) {
 
 	var (
 		ch = clickhouse{
-			logf:         func(string, ...interface{}) {},
-			compress:     compress,
-			blockSize:    blockSize,
-			readTimeout:  readTimeout,
-			writeTimeout: writeTimeout,
+			logf:      func(string, ...interface{}) {},
+			compress:  compress,
+			blockSize: blockSize,
 			ServerInfo: data.ServerInfo{
 				Timezone: time.Local,
 			},
@@ -131,7 +143,7 @@ func open(dsn string) (*clickhouse, error) {
 		database,
 		username,
 	)
-	if ch.conn, err = dial(secure, skipVerify, hosts, noDelay, connOpenStrategy, ch.logf); err != nil {
+	if ch.conn, err = dial(secure, skipVerify, hosts, readTimeout, writeTimeout, noDelay, connOpenStrategy, ch.logf); err != nil {
 		return nil, err
 	}
 	logger.SetPrefix(fmt.Sprintf("[clickhouse][connect=%d]", ch.conn.ident))

clickhouse.go

@@ -43,8 +43,6 @@ type clickhouse struct {
 	compress      bool
 	blockSize     int
 	inTransaction bool
-	readTimeout   time.Duration
-	writeTimeout  time.Duration
 }
 
 func (ch *clickhouse) Prepare(query string) (driver.Stmt, error) {

clickhouse_read_block.go

@@ -1,17 +1,10 @@
 package clickhouse
 
 import (
-	"time"
-
 	"github.com/kshvakov/clickhouse/lib/data"
 )
 
 func (ch *clickhouse) readBlock() (*data.Block, error) {
-	{
-		ch.conn.SetReadDeadline(time.Now().Add(ch.readTimeout))
-		ch.conn.SetWriteDeadline(time.Now().Add(ch.writeTimeout))
-	}
-
 	if _, err := ch.decoder.String(); err != nil { // temporary table
 		return nil, err
 	}

clickhouse_send_query.go

@@ -1,18 +1,12 @@
 package clickhouse
 
 import (
-	"time"
-
 	"github.com/kshvakov/clickhouse/lib/data"
 	"github.com/kshvakov/clickhouse/lib/protocol"
 )
 
 func (ch *clickhouse) sendQuery(query string) error {
 	ch.logf("[send query] %s", query)
-	{
-		ch.conn.SetReadDeadline(time.Now().Add(ch.readTimeout))
-		ch.conn.SetWriteDeadline(time.Now().Add(ch.writeTimeout))
-	}
 	if err := ch.encoder.Uvarint(protocol.ClientQuery); err != nil {
 		return err
 	}

clickhouse_test.go

@@ -1002,7 +1002,7 @@ func Test_Context_Timeout(t *testing.T) {
 		{
 			ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*20)
 			defer cancel()
-			if row := connect.QueryRowContext(ctx, "SELECT 1, sleep(10)"); assert.NotNil(t, row) {
+			if row := connect.QueryRowContext(ctx, "SELECT 1, sleep(2)"); assert.NotNil(t, row) {
 				var a, b int
 				assert.Equal(t, driver.ErrBadConn, row.Scan(&a, &b))
 			}
@@ -1023,7 +1023,7 @@ func Test_Context_Timeout(t *testing.T) {
 func Test_Timeout(t *testing.T) {
 	if connect, err := sql.Open("clickhouse", "tcp://127.0.0.1:9000?debug=true&read_timeout=0.2"); assert.NoError(t, err) && assert.NoError(t, connect.Ping()) {
 		{
-			if row := connect.QueryRow("SELECT 1, sleep(10)"); assert.NotNil(t, row) {
+			if row := connect.QueryRow("SELECT 1, sleep(2)"); assert.NotNil(t, row) {
 				var a, b int
 				assert.Equal(t, driver.ErrBadConn, row.Scan(&a, &b))
 			}

clickhouse_write_block.go

@@ -1,19 +1,13 @@
 package clickhouse
 
 import (
-	"time"
-
 	"github.com/kshvakov/clickhouse/lib/data"
 	"github.com/kshvakov/clickhouse/lib/protocol"
 )
 
 func (ch *clickhouse) writeBlock(block *data.Block) error {
 	ch.Lock()
 	defer ch.Unlock()
-	{
-		ch.conn.SetReadDeadline(time.Now().Add(ch.readTimeout))
-		ch.conn.SetWriteDeadline(time.Now().Add(ch.writeTimeout))
-	}
 	if err := ch.encoder.Uvarint(protocol.ClientData); err != nil {
 		return err
 	}

connect.go

@@ -26,7 +26,7 @@ const (
 	connOpenInOrder
 )
 
-func dial(secure, skipVerify bool, hosts []string, noDelay bool, openStrategy openStrategy, logf func(string, ...interface{})) (*connect, error) {
+func dial(secure, skipVerify bool, hosts []string, readTimeout, writeTimeout time.Duration, noDelay bool, openStrategy openStrategy, logf func(string, ...interface{})) (*connect, error) {
 	var (
 		err error
 		abs = func(v int) int {
@@ -66,10 +66,12 @@ func dial(secure, skipVerify bool, hosts []string, noDelay bool, openStrategy op
 				tcp.SetNoDelay(noDelay) // Disable or enable the Nagle Algorithm for this tcp socket
 			}
 			return &connect{
-				Conn:   conn,
-				logf:   logf,
-				ident:  ident,
-				buffer: bufio.NewReaderSize(conn, 4*1024*1024),
+				Conn:         conn,
+				logf:         logf,
+				ident:        ident,
+				buffer:       bufio.NewReaderSize(conn, 4*1024*1024),
+				readTimeout:  readTimeout,
+				writeTimeout: writeTimeout,
 			}, nil
 		}
 	}
@@ -78,10 +80,14 @@ func dial(secure, skipVerify bool, hosts []string, noDelay bool, openStrategy op
 
 type connect struct {
 	net.Conn
-	logf   func(string, ...interface{})
-	ident  int
-	buffer *bufio.Reader
-	closed bool
+	logf                  func(string, ...interface{})
+	ident                 int
+	buffer                *bufio.Reader
+	closed                bool
+	readTimeout           time.Duration
+	writeTimeout          time.Duration
+	lastReadDeadlineTime  time.Time
+	lastWriteDeadlineTime time.Time
 }
 
 func (conn *connect) Read(b []byte) (int, error) {
@@ -91,10 +97,14 @@ func (conn *connect) Read(b []byte) (int, error) {
 		total  int
 		dstLen = len(b)
 	)
+	if currentTime := now(); conn.readTimeout != 0 && currentTime.Sub(conn.lastReadDeadlineTime) > (conn.readTimeout>>2) {
+		conn.SetReadDeadline(time.Now().Add(conn.readTimeout))
+		conn.lastReadDeadlineTime = currentTime
+	}
 	for total < dstLen {
 		if n, err = conn.buffer.Read(b[total:]); err != nil {
 			conn.logf("[connect] read error: %v", err)
-			conn.closed = true
+			conn.Close()
 			return n, driver.ErrBadConn
 		}
 		total += n
@@ -109,10 +119,14 @@ func (conn *connect) Write(b []byte) (int, error) {
 		total  int
 		srcLen = len(b)
 	)
+	if currentTime := now(); conn.writeTimeout != 0 && currentTime.Sub(conn.lastWriteDeadlineTime) > (conn.writeTimeout>>2) {
+		conn.SetWriteDeadline(time.Now().Add(conn.writeTimeout))
+		conn.lastWriteDeadlineTime = currentTime
+	}
 	for total < srcLen {
 		if n, err = conn.Conn.Write(b[total:]); err != nil {
 			conn.logf("[connect] write error: %v", err)
-			conn.closed = true
+			conn.Close()
 			return n, driver.ErrBadConn
 		}
 		total += n