make executor work chan smaller, add executor.Close

the size of the work chan probably doesn't matter... there is some
discussion of this on the associated PR
https://github.com/pilosa/pilosa/pull/2034
may test with an unbuffered channel as well.

Closing the executor avoids leaking goroutines which seems to be an
issue while running the test suite.
This commit is contained in:
Matt Jaffee 2019-07-15 07:56:57 -05:00
parent 4e55a1fd73
commit 7d7a5539ca
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
2 changed files with 19 additions and 3 deletions

View file

@ -72,9 +72,14 @@ func optExecutorInternalQueryClient(c InternalQueryClient) executorOption {
// 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(),
work: make(chan job, 2000),
// capacity of this channel is unlikely to affect much
work: make(chan job, workerPoolSize),
}
for _, opt := range opts {
err := opt(e)
@ -82,12 +87,17 @@ func newExecutor(opts ...executorOption) *executor {
panic(err)
}
}
for i := 0; i < runtime.NumCPU()+8; i++ {
for i := 0; i < workerPoolSize; i++ {
go worker(e.work)
}
return e
}
func (e *executor) Close() error {
close(e.work)
return nil
}
// Execute executes a PQL query.
func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) (QueryResponse, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.Execute")

View file

@ -426,6 +426,8 @@ func (s *Server) Open() error {
// Close closes the server and waits for it to shutdown.
func (s *Server) Close() error {
errE := s.executor.Close()
// Notify goroutines to stop.
close(s.closing)
s.wg.Wait()
@ -445,7 +447,11 @@ func (s *Server) Close() error {
if errh != nil {
return errors.Wrap(errh, "closing holder")
}
return errors.Wrap(errc, "closing cluster")
if errc != nil {
return errors.Wrap(errc, "closing cluster")
}
return errors.Wrap(errE, "closing executor")
}
// loadNodeID gets NodeID from disk, or creates a new value.