From 9a453ef51ab6db68312eaadc3ee6883e0644edbf Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 15 Jul 2019 13:55:44 -0500 Subject: [PATCH] expose worker pool size to config, so we can set it lower in tests we are experiencing issues with CI where it fails with race: limit on 8128 simultaneously alive goroutines is exceeded, dying this, despite the fact that closing the executor should clean up all worker goroutines. Apparently in CircleCI runtime.NumCPU() reports 36, so the goroutines added up quickly. --- executor.go | 29 ++++++++++++++++++----------- server.go | 32 +++++++++++++++++++++++--------- server/config.go | 13 ++++++++++++- server/server.go | 1 + test/pilosa.go | 1 + 5 files changed, 55 insertions(+), 21 deletions(-) diff --git a/executor.go b/executor.go index bd48033da..2110a92c6 100644 --- a/executor.go +++ b/executor.go @@ -18,7 +18,6 @@ import ( "context" "encoding/json" "fmt" - "runtime" "sort" "sync" "time" @@ -58,8 +57,9 @@ type executor struct { // Stores key/id translation data. TranslateStore TranslateStore - workersWG sync.WaitGroup - work chan job + workersWG sync.WaitGroup + workerPoolSize int + work chan job } // executorOption is a functional option type for pilosa.Executor @@ -72,16 +72,18 @@ func optExecutorInternalQueryClient(c InternalQueryClient) executorOption { } } +func optExecutorWorkerPoolSize(size int) executorOption { + return func(e *executor) error { + e.workerPoolSize = size + return nil + } +} + // newExecutor returns a new instance of Executor. func newExecutor(opts ...executorOption) *executor { - // this is somewhat arbitrary, though going less than - // runtime.NumCPU() would likely result in a loss of throughput. - workerPoolSize := runtime.NumCPU() + 8 e := &executor{ - client: newNopInternalQueryClient(), - - // capacity of this channel is unlikely to affect much - work: make(chan job, workerPoolSize), + client: newNopInternalQueryClient(), + workerPoolSize: 2, } for _, opt := range opts { err := opt(e) @@ -89,7 +91,12 @@ func newExecutor(opts ...executorOption) *executor { panic(err) } } - for i := 0; i < workerPoolSize; i++ { + // this channel cap doesn't necessarily have to be the same as + // workerPoolSize... any larger doesn't seem to have an effect in + // the few tests we've done at scale with concurrent query + // workloads. Possible that it could be smaller. + e.work = make(chan job, e.workerPoolSize) + for i := 0; i < e.workerPoolSize; i++ { e.workersWG.Add(1) go func() { defer e.workersWG.Done() diff --git a/server.go b/server.go index 2b67c2b78..6eee374d1 100644 --- a/server.go +++ b/server.go @@ -49,13 +49,14 @@ type Server struct { // nolint: maligned closing chan struct{} // Internal - holder *Holder - cluster *cluster - diagnostics *diagnosticsCollector - executor *executor - hosts []string - clusterDisabled bool - serializer Serializer + holder *Holder + cluster *cluster + diagnostics *diagnosticsCollector + executor *executor + executorPoolSize int + hosts []string + clusterDisabled bool + serializer Serializer // External systemInfo SystemInfo @@ -179,13 +180,19 @@ func OptServerGCNotifier(gcn GCNotifier) ServerOption { // used to set the implementation of InternalClient. func OptServerInternalClient(c InternalClient) ServerOption { return func(s *Server) error { - s.executor = newExecutor(optExecutorInternalQueryClient(c)) s.defaultClient = c s.cluster.InternalClient = c return nil } } +func OptServerExecutorPoolSize(size int) ServerOption { + return func(s *Server) error { + s.executorPoolSize = size + return nil + } +} + // OptServerPrimaryTranslateStore has been deprecated. func OptServerPrimaryTranslateStore(store TranslateStore) ServerOption { return func(s *Server) error { @@ -306,7 +313,6 @@ func NewServer(opts ...ServerOption) (*Server, error) { logger: logger.NopLogger, } - s.executor = newExecutor(optExecutorInternalQueryClient(s.defaultClient)) s.cluster.InternalClient = s.defaultClient s.diagnostics.server = s @@ -317,6 +323,14 @@ func NewServer(opts ...ServerOption) (*Server, error) { return nil, errors.Wrap(err, "applying option") } } + + // set up executor after server opts have been processed + executorOpts := []executorOption{optExecutorInternalQueryClient(s.defaultClient)} + if s.executorPoolSize > 0 { + executorOpts = append(executorOpts, optExecutorWorkerPoolSize(s.executorPoolSize)) + } + s.executor = newExecutor(executorOpts...) + s.holder.translateFile.logger = s.logger path, err := expandDirName(s.dataDir) diff --git a/server/config.go b/server/config.go index c42d41b6c..b0db09ba7 100644 --- a/server/config.go +++ b/server/config.go @@ -19,6 +19,7 @@ import ( "fmt" "log" "net" + "runtime" "strconv" "strings" "time" @@ -85,6 +86,13 @@ type Config struct { // TLS TLS TLSConfig `toml:"tls"` + // WorkerPoolSize controls how many goroutines are created for + // processing queries. Defaults to runtime.NumCPU(). It is + // intentionally not defined as a flag... only exposed here so + // that we can limit the size while running tests in CI so we + // don't exhaust the goroutine limit. + WorkerPoolSize int + Cluster struct { // Disabled controls whether clustering functionality is enabled. Disabled bool `toml:"disabled"` @@ -151,7 +159,10 @@ func NewConfig() *Config { // a bit below your system limits. MaxMapCount: 1000000, MaxFileCount: 1000000, - TLS: TLSConfig{}, + + TLS: TLSConfig{}, + + WorkerPoolSize: runtime.NumCPU(), } // Cluster config. diff --git a/server/server.go b/server/server.go index 743accf95..db367f418 100644 --- a/server/server.go +++ b/server/server.go @@ -284,6 +284,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerMaxWritesPerRequest(m.Config.MaxWritesPerRequest), pilosa.OptServerMetricInterval(time.Duration(m.Config.Metric.PollInterval)), pilosa.OptServerDiagnosticsInterval(diagnosticsInterval), + pilosa.OptServerExecutorPoolSize(m.Config.WorkerPoolSize), pilosa.OptServerLogger(m.logger), pilosa.OptServerAttrStoreFunc(boltdb.NewAttrStore), diff --git a/test/pilosa.go b/test/pilosa.go index 31e93bb1d..4a2cd2973 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -72,6 +72,7 @@ func newCommand(opts ...server.CommandOption) *Command { m.Config.Bind = "http://localhost:0" m.Config.Cluster.Disabled = true m.Config.Translation.MapSize = 140000 + m.Config.WorkerPoolSize = 2 if testing.Verbose() { m.Command.Stdout = os.Stdout