Add a separate write timeout

Using the same write timeout and read timeout means that the query
can take cfg.Timeout * 2 time. We want a shorter write timeout.

cluster.go

@@ -48,6 +48,7 @@ type ClusterConfig struct {
 	ProtoVersion       int
 	Timeout            time.Duration                            // connection timeout (default: 600ms)
 	ConnectTimeout     time.Duration                            // initial connection timeout, used during initial dial to server (default: 600ms)
+	WriteTimeout       time.Duration                            // timeout for writing a query. defaults to Timeout if not specified.
 	Port               int                                      // port (default: 9042)
 	Keyspace           string                                   // initial keyspace (optional)
 	NumConns           int                                      // number of connections per host (default: 2)

conn.go

@@ -120,6 +120,7 @@ type ConnConfig struct {
 	ProtoVersion   int
 	CQLVersion     string
 	Timeout        time.Duration
+	WriteTimeout   time.Duration
 	ConnectTimeout time.Duration
 	Dialer         Dialer
 	Compressor     Compressor
@@ -166,6 +167,7 @@ type Conn struct {
 	w    contextWriter
 
 	timeout        time.Duration
+	writeTimeout   time.Duration
 	cfg            *ConnConfig
 	frameObserver  FrameHeaderObserver
 	streamObserver StreamObserver
@@ -272,6 +274,11 @@ func (s *Session) dialWithoutObserver(ctx context.Context, host *HostInfo, cfg *
 		conn = tconn
 	}
 
+	writeTimeout := cfg.Timeout
+	if cfg.WriteTimeout > 0 {
+		writeTimeout = cfg.WriteTimeout
+	}
+
 	ctx, cancel := context.WithCancel(ctx)
 	c := &Conn{
 		conn:          conn,
@@ -288,14 +295,15 @@ func (s *Session) dialWithoutObserver(ctx context.Context, host *HostInfo, cfg *
 		frameObserver: s.frameObserver,
 		w: &deadlineContextWriter{
 			w:         conn,
-			timeout:   cfg.Timeout,
+			timeout:   writeTimeout,
 			semaphore: make(chan struct{}, 1),
 			quit:      make(chan struct{}),
 		},
 		ctx:            ctx,
 		cancel:         cancel,
 		logger:         cfg.logger(),
 		streamObserver: s.streamObserver,
+		writeTimeout:   writeTimeout,
 	}
 
 	if err := c.init(ctx); err != nil {
@@ -332,7 +340,7 @@ func (c *Conn) init(ctx context.Context) error {
 
 	// dont coalesce startup frames
 	if c.session.cfg.WriteCoalesceWaitTime > 0 && !c.cfg.disableCoalesce {
-		c.w = newWriteCoalescer(c.conn, c.timeout, c.session.cfg.WriteCoalesceWaitTime, ctx.Done())
+		c.w = newWriteCoalescer(c.conn, c.writeTimeout, c.session.cfg.WriteCoalesceWaitTime, ctx.Done())
 	}
 
 	go c.serve(ctx)

connectionpool.go

@@ -107,6 +107,7 @@ func connConfig(cfg *ClusterConfig) (*ConnConfig, error) {
 		ProtoVersion:    cfg.ProtoVersion,
 		CQLVersion:      cfg.CQLVersion,
 		Timeout:         cfg.Timeout,
+		WriteTimeout:    cfg.WriteTimeout,
 		ConnectTimeout:  cfg.ConnectTimeout,
 		Dialer:          cfg.Dialer,
 		Compressor:      cfg.Compressor,