optimize meta executor and shard writer

cmd/influxd/run/server.go

@@ -204,7 +204,8 @@ func NewServer(c *Config, buildInfo *BuildInfo) (*Server, error) {
 	s.PointsWriter.Subscriber = s.Subscriber
 
 	// Initialize meta executor.
-	s.MetaExecutor = coordinator.NewMetaExecutor(time.Duration(c.Coordinator.QueryTimeout), time.Duration(c.Coordinator.DialTimeout))
+	s.MetaExecutor = coordinator.NewMetaExecutor(time.Duration(c.Coordinator.ShardReaderTimeout), time.Duration(c.Coordinator.DialTimeout),
+		time.Duration(c.Coordinator.PoolMaxIdleTime), c.Coordinator.PoolMaxIdleStreams)
 	s.MetaExecutor.MetaClient = s.MetaClient
 	s.MetaExecutor.TLSConfig = tlsClientConfig
 

coordinator/client_pool.go

@@ -1,10 +1,13 @@
 package coordinator
 
 import (
+	"errors"
 	"net"
 	"sync"
 )
 
+var ErrClientClosed = errors.New("client already closed")
+
 type clientPool struct {
 	mu   sync.RWMutex
 	pool map[uint64]Pool

coordinator/config.go

@@ -26,6 +26,9 @@ const (
 	// remains idle in the connection pool.
 	DefaultPoolMaxIdleTime = time.Minute
 
+	// DefaultShardReaderTimeout is the default timeout set on shard readers.
+	DefaultShardReaderTimeout = time.Duration(0)
+
 	// DefaultWriteTimeout is the default timeout for a complete write to succeed.
 	DefaultWriteTimeout = 10 * time.Second
 
@@ -78,6 +81,7 @@ func NewConfig() Config {
 		DialTimeout:          toml.Duration(DefaultDialTimeout),
 		PoolMaxIdleStreams:   DefaultPoolMaxIdleStreams,
 		PoolMaxIdleTime:      toml.Duration(DefaultPoolMaxIdleTime),
+		ShardReaderTimeout:   toml.Duration(DefaultShardReaderTimeout),
 		WriteTimeout:         toml.Duration(DefaultWriteTimeout),
 		QueryTimeout:         toml.Duration(query.DefaultQueryTimeout),
 		MaxConcurrentQueries: DefaultMaxConcurrentQueries,