diff --git a/datanode/server.go b/datanode/server.go index 3bfaeb8e9..fbd753e6e 100644 --- a/datanode/server.go +++ b/datanode/server.go @@ -381,6 +381,10 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) { log.LogErrorf("parseConfig: masters addrs format err[%v]", masters) return err } + poolSize := cfg.GetInt64(proto.CfgHttpPoolSize) + syslog.Printf("parseConfig: http pool size %d", poolSize) + MasterClient.SetTransport(proto.GetHttpTransporter(&proto.HttpCfg{PoolSize: int(poolSize)})) + if err = MasterClient.Start(); err != nil { return err } diff --git a/master/cluster.go b/master/cluster.go index d88c12cfa..29c08cf45 100644 --- a/master/cluster.go +++ b/master/cluster.go @@ -386,6 +386,9 @@ func newCluster(name string, leaderInfo *LeaderInfo, fsm *MetadataFsm, partition c.DecommissionLimit = defaultDecommissionParallelLimit c.checkAutoCreateDataPartition = false c.masterClient = masterSDK.NewMasterClient(nil, false) + c.masterClient.SetTransport(proto.GetHttpTransporter(&proto.HttpCfg{ + PoolSize: int(cfg.httpPoolSize), + })) c.inodeCountNotEqualMP = new(sync.Map) c.maxInodeNotEqualMP = new(sync.Map) c.dentryCountNotEqualMP = new(sync.Map) diff --git a/master/config.go b/master/config.go index 59190bc1d..c8e0c4007 100644 --- a/master/config.go +++ b/master/config.go @@ -55,6 +55,8 @@ const ( cfgVolForceDeletion = "volForceDeletion" cfgVolDeletionDentryThreshold = "volDeletionDentryThreshold" + + cfgHttpReversePoolSize = "httpReversePoolSize" ) // default value @@ -98,7 +100,7 @@ const ( defaultMaxConcurrentLcNodes = 3 metaPartitionInodeUsageThreshold float64 = 0.75 // inode usage threshold on a meta partition lowerLimitRWMetaPartition = 3 // lower limit of RW meta partition, equal defaultReplicaNum - // defaultIntervalToCheckDelVerTaskExpiration = 3 + defaultHttpReversePoolSize = 1024 ) // AddrDatabase is a map that stores the address of a given host (e.g., the leader) @@ -114,6 +116,8 @@ type clusterConfig struct { IntervalToAlarmMissingDataPartition int64 PeriodToLoadALLDataPartitions int64 metaNodeReservedMem uint64 + httpProxyPoolSize uint64 + httpPoolSize uint64 IntervalToCheckDataPartition int // seconds IntervalToCheckQos int // seconds numberOfDataPartitionsToFree int diff --git a/master/http_server.go b/master/http_server.go index ce65885e2..5afb31dd8 100644 --- a/master/http_server.go +++ b/master/http_server.go @@ -880,10 +880,17 @@ func ErrResponse(w http.ResponseWriter, err error) { } func (m *Server) newReverseProxy() *httputil.ReverseProxy { - return &httputil.ReverseProxy{Director: func(request *http.Request) { - request.URL.Scheme = "http" - request.URL.Host = m.leaderInfo.addr - }} + tr := proto.GetHttpTransporter(&proto.HttpCfg{ + PoolSize: int(m.config.httpProxyPoolSize), + }) + + return &httputil.ReverseProxy{ + Director: func(request *http.Request) { + request.URL.Scheme = "http" + request.URL.Host = m.leaderInfo.addr + }, + Transport: tr, + } } func (m *Server) proxy(w http.ResponseWriter, r *http.Request) { diff --git a/master/server.go b/master/server.go index c5716d293..10957f3af 100644 --- a/master/server.go +++ b/master/server.go @@ -136,11 +136,12 @@ func (m *Server) Start(cfg *config.Config) (err error) { m.config = newClusterConfig() gConfig = m.config m.leaderInfo = &LeaderInfo{} - m.reverseProxy = m.newReverseProxy() + if err = m.checkConfig(cfg); err != nil { log.LogError(errors.Stack(err)) return } + m.reverseProxy = m.newReverseProxy() if m.rocksDBStore, err = raftstore_db.NewRocksDBStoreAndRecovery(m.storeDir, LRUCacheSize, WriteBufferSize); err != nil { return @@ -269,6 +270,14 @@ func (m *Server) checkConfig(cfg *config.Config) (err error) { m.config.nodeSetCapacity = defaultNodeSetCapacity } + m.config.httpProxyPoolSize = uint64(cfg.GetInt64(cfgHttpReversePoolSize)) + if m.config.httpProxyPoolSize < defaultHttpReversePoolSize { + m.config.httpProxyPoolSize = defaultHttpReversePoolSize + } + m.config.httpPoolSize = uint64(cfg.GetInt64(proto.CfgHttpPoolSize)) + + syslog.Printf("http reverse pool size %d, http pool size %d\n", m.config.httpProxyPoolSize, m.config.httpPoolSize) + m.config.DefaultNormalZoneCnt = defaultNodeSetGrpBatchCnt m.config.DomainBuildAsPossible = cfg.GetBoolWithDefault(cfgDomainBuildAsPossible, false) domainBatchGrpCnt := cfg.GetString(cfgDomainBatchGrpCnt) diff --git a/metanode/metanode.go b/metanode/metanode.go index 67552fc58..6dd94f644 100644 --- a/metanode/metanode.go +++ b/metanode/metanode.go @@ -329,6 +329,10 @@ func (m *MetaNode) parseConfig(cfg *config.Config) (err error) { log.LogErrorf("parseConfig: masters addrs format err[%v]", masters) return err } + poolSize := cfg.GetInt64(proto.CfgHttpPoolSize) + syslog.Printf("parseConfig: http pool size %d", poolSize) + masterClient.SetTransport(proto.GetHttpTransporter(&proto.HttpCfg{PoolSize: int(poolSize)})) + if err = masterClient.Start(); err != nil { return err } diff --git a/objectnode/server.go b/objectnode/server.go index 186a23643..d330d8be7 100644 --- a/objectnode/server.go +++ b/objectnode/server.go @@ -292,6 +292,10 @@ func (o *ObjectNode) loadConfig(cfg *config.Config) (err error) { o.disableCreateBucketByS3 = cfg.GetBool(disableCreateBucketByS3) o.mc = master.NewMasterClient(masters, false) + poolSize := cfg.GetInt64(proto.CfgHttpPoolSize) + log.LogWarnf("loadConfig: http pool size %d", poolSize) + o.mc.SetTransport(proto.GetHttpTransporter(&proto.HttpCfg{PoolSize: int(poolSize)})) + o.vm = NewVolumeManager(masters, strict) o.userStore = NewUserInfoStore(masters, strict) diff --git a/proto/admin_proto.go b/proto/admin_proto.go index b2a734e11..88aa3ecdb 100644 --- a/proto/admin_proto.go +++ b/proto/admin_proto.go @@ -17,6 +17,8 @@ package proto import ( "encoding/json" "fmt" + "net" + "net/http" "strconv" "time" @@ -378,6 +380,35 @@ const ( MinDirChildrenNumLimit = 1000000 ) +const ( + CfgHttpPoolSize = "httpPoolSize" + defaultHttpPoolSize = 128 +) + +type HttpCfg struct { + PoolSize int +} + +func GetHttpTransporter(cfg *HttpCfg) *http.Transport { + if cfg.PoolSize < defaultHttpPoolSize { + cfg.PoolSize = defaultHttpPoolSize + } + + return &http.Transport{ + Proxy: http.ProxyFromEnvironment, + DialContext: (&net.Dialer{ + Timeout: 30 * time.Second, + KeepAlive: 30 * time.Second, + }).DialContext, + ForceAttemptHTTP2: true, + MaxIdleConns: cfg.PoolSize, + MaxIdleConnsPerHost: cfg.PoolSize / 2, + IdleConnTimeout: 90 * time.Second, + TLSHandshakeTimeout: 10 * time.Second, + ExpectContinueTimeout: 1 * time.Second, + } +} + // HTTPReply uniform response structure type HTTPReply struct { Code int32 `json:"code"` @@ -1030,7 +1061,7 @@ const ( ) const ( - QosDefaultBurst = 16000000 + QosDefaultBurst = 1600000 QosDefaultClientCnt uint32 = 100 QosDefaultDiskMaxFLowLimit int = 0x7FFFFFFF QosDefaultDiskMaxIoLimit int = 100000 diff --git a/sdk/master/client.go b/sdk/master/client.go index 732c94664..2bc01789b 100644 --- a/sdk/master/client.go +++ b/sdk/master/client.go @@ -59,6 +59,7 @@ type MasterClient struct { leaderAddr string timeout time.Duration clientIDKey string + client *http.Client adminAPI *AdminAPI clientAPI *ClientAPI @@ -66,6 +67,12 @@ type MasterClient struct { userAPI *UserAPI } +func (c *MasterClient) SetTransport(tr http.RoundTripper) { + c.client = &http.Client{ + Transport: tr, + } +} + func (c *MasterClient) ReplaceMasterAddresses(addrs []string) { c.Lock() defer c.Unlock() @@ -248,6 +255,13 @@ func (c *MasterClient) prepareRequest() (addr string, nodes []string) { func (c *MasterClient) httpRequest(method, url string, r *request) (resp *http.Response, err error) { client := http.DefaultClient + if c.client != nil { + if log.EnableDebug() { + log.LogDebug("use client config from config") + } + client = c.client + } + if !r.noTimeout { client.Timeout = c.timeout }