From fc231ff80248cff10ef82aed9b343e3d94bbd4fc Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Thu, 8 Nov 2018 17:01:05 +0300 Subject: [PATCH 01/19] Fixes #1731 --- cmd/import.go | 5 ++--- ctl/import.go | 16 +++++++++------- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/cmd/import.go b/cmd/import.go index 878aad286..81a12a47f 100644 --- a/cmd/import.go +++ b/cmd/import.go @@ -19,10 +19,8 @@ import ( "io" "github.com/pilosa/pilosa" - - "github.com/spf13/cobra" - "github.com/pilosa/pilosa/ctl" + "github.com/spf13/cobra" ) var Importer *ctl.ImportCommand @@ -55,6 +53,7 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM. flags.StringVarP(&Importer.Field, "field", "f", "", "Field to import into.") flags.BoolVar(&Importer.IndexOptions.Keys, "index-keys", false, "Specify keys=true when creating an index") flags.BoolVar(&Importer.FieldOptions.Keys, "field-keys", false, "Specify keys=true when creating a field") + flags.StringVar(&Importer.FieldOptions.Type, "field-type", "", "Specify the field type when creating a field. One of: set, int, time, bool, mutex") flags.Int64Var(&Importer.FieldOptions.Min, "field-min", 0, "Specify the minimum for an int field on creation") flags.Int64Var(&Importer.FieldOptions.Max, "field-max", 0, "Specify the maximum for an int field on creation") flags.StringVar(&Importer.FieldOptions.CacheType, "field-cache-type", pilosa.CacheTypeRanked, "Specify the cache type for a set field on creation. One of: none, lru, ranked") diff --git a/ctl/import.go b/ctl/import.go index d49b50eb3..ab19064ce 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -99,13 +99,15 @@ func (cmd *ImportCommand) Run(ctx context.Context) error { cmd.client = client if cmd.CreateSchema { - // set the correct type for the field - if cmd.FieldOptions.TimeQuantum != "" { - cmd.FieldOptions.Type = "time" - } else if cmd.FieldOptions.Min != 0 || cmd.FieldOptions.Max != 0 { - cmd.FieldOptions.Type = "int" - } else { - cmd.FieldOptions.Type = "set" + if cmd.FieldOptions.Type == "" { + // set the correct type for the field + if cmd.FieldOptions.TimeQuantum != "" { + cmd.FieldOptions.Type = "time" + } else if cmd.FieldOptions.Min != 0 || cmd.FieldOptions.Max != 0 { + cmd.FieldOptions.Type = "int" + } else { + cmd.FieldOptions.Type = "set" + } } err := cmd.ensureSchema(ctx) if err != nil { From 27c222f02dda7b681ceddd1caf76670b633ef913 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Thu, 8 Nov 2018 17:02:07 +0300 Subject: [PATCH 02/19] Refactored missing executeRequest bits; check resp is not nil --- http/client.go | 18 +++++++----------- 1 file changed, 7 insertions(+), 11 deletions(-) diff --git a/http/client.go b/http/client.go index 0d6df281f..3d19df227 100644 --- a/http/client.go +++ b/http/client.go @@ -91,9 +91,7 @@ func (c *InternalClient) maxShardByIndex(ctx context.Context) (map[string]uint64 defer resp.Body.Close() var rsp getShardsMaxResponse - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("http: status=%d", resp.StatusCode) - } else if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil { + if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil { return nil, fmt.Errorf("json decode: %s", err) } @@ -152,7 +150,7 @@ func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt pilo // Execute request against the host. resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { - if resp.StatusCode == http.StatusConflict { + if resp != nil && resp.StatusCode == http.StatusConflict { return pilosa.ErrIndexExists } return err @@ -258,8 +256,6 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pilosa.URI, index s body, err := ioutil.ReadAll(resp.Body) if err != nil { return nil, errors.Wrap(err, "reading") - } else if resp.StatusCode != http.StatusOK { - return nil, errors.New(string(body)) } qresp := &pilosa.QueryResponse{} @@ -689,7 +685,7 @@ func (c *InternalClient) backupShardNode(ctx context.Context, index, field strin // Execute request. resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { - if resp.StatusCode == http.StatusNotFound { + if resp != nil && resp.StatusCode == http.StatusNotFound { return nil, pilosa.ErrFragmentNotFound } return nil, err @@ -746,7 +742,7 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel // Execute request against the host. resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { - if resp.StatusCode == http.StatusConflict { + if resp != nil && resp.StatusCode == http.StatusConflict { return pilosa.ErrFieldExists } return err @@ -782,7 +778,7 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { // Return the appropriate error. - if resp.StatusCode == http.StatusNotFound { + if resp != nil && resp.StatusCode == http.StatusNotFound { return nil, pilosa.ErrFragmentNotFound } return nil, err @@ -825,7 +821,7 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { - if resp.StatusCode == http.StatusNotFound { + if resp != nil && resp.StatusCode == http.StatusNotFound { return nil, nil, nil } return nil, nil, err @@ -904,7 +900,7 @@ func (c *InternalClient) RowAttrDiff(ctx context.Context, uri *pilosa.URI, index // Execute request. resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { - if resp.StatusCode == http.StatusNotFound { + if resp != nil && resp.StatusCode == http.StatusNotFound { return nil, pilosa.ErrFieldNotFound } return nil, err From 70f85211d97118331ba95c06ec3af26f0fb40aff Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Thu, 15 Nov 2018 22:06:21 +0300 Subject: [PATCH 03/19] prevent panic in Bitmap.UnmarshalBinary when there is no data --- api.go | 4 ++++ fragment.go | 4 +--- roaring/roaring.go | 4 ++++ 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/api.go b/api.go index dcedc3d92..03b3bbcb0 100644 --- a/api.go +++ b/api.go @@ -268,6 +268,10 @@ func setUpImportOptions(opts ...ImportOption) (*ImportOptions, error) { // of the rows in this shard of this field concatenated together in one long // bitmap. func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, shard uint64, remote bool, data []byte, opts ...ImportOption) (err error) { + if len(data) == 0 { + return errors.New("no data to import") + } + if err = api.validate(apiField); err != nil { return errors.Wrap(err, "validating api method") } diff --git a/fragment.go b/fragment.go index dd1d874de..359628ab5 100644 --- a/fragment.go +++ b/fragment.go @@ -25,6 +25,7 @@ import ( "hash" "io" "io/ioutil" + "math" "os" "sort" "sync" @@ -33,9 +34,6 @@ import ( "unsafe" "github.com/cespare/xxhash" - - "math" - "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" "github.com/pilosa/pilosa/pql" diff --git a/roaring/roaring.go b/roaring/roaring.go index ed7be6be1..ba258526f 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -3465,6 +3465,10 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint // UnmarshalBinary decodes b from a binary-encoded byte slice. data can be in // either official roaring format or Pilosa's roaring format. func (b *Bitmap) UnmarshalBinary(data []byte) error { + if data == nil { + // Nothing to unmarshal + return nil + } fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2])) if fileMagic == magicNumber { // if pilosa roaring return errors.Wrap(b.unmarshalPilosaRoaring(data), "unmarshaling as pilosa roaring") From a203313143de80ea0c350e5a33881f0064298dd2 Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 14 Nov 2018 22:44:27 -0600 Subject: [PATCH 04/19] move Logger and Stats to their own packages I'd like to add stat tracking to Roaring, which means it has to be able to import the stats package, which means stats has to be a package rather than part of the pilosa package. If stats stops being in pilosa, it still needs a way to import logger, so logger also has to leave the pilosa package. Then everything using them needs to import them and use package selectors on their names. This doesn't actually add the stats support to roaring, it just makes it so there's a way to import the stats code from something in the roaring package. --- api.go | 3 +- cache.go | 27 +++++++++-------- cluster.go | 9 +++--- diagnostics.go | 5 ++-- field.go | 10 ++++--- fragment.go | 12 ++++---- gossip/gossip.go | 5 ++-- holder.go | 12 ++++---- http/handler.go | 7 +++-- index.go | 10 ++++--- logger.go => logger/logger.go | 6 ++-- server.go | 10 ++++--- server/server.go | 14 +++++---- stats.go => stats/stats.go | 12 ++++---- stats_test.go => stats/stats_test.go | 44 +++++++++++++++------------- statsd/statsd.go | 13 ++++---- translate.go | 7 +++-- view.go | 10 ++++--- 18 files changed, 121 insertions(+), 95 deletions(-) rename logger.go => logger/logger.go (91%) rename stats.go => stats/stats.go (96%) rename stats_test.go => stats/stats_test.go (80%) diff --git a/api.go b/api.go index 03b3bbcb0..7a95ce76e 100644 --- a/api.go +++ b/api.go @@ -28,6 +28,7 @@ import ( "github.com/pilosa/pilosa/pql" "github.com/pilosa/pilosa/roaring" + "github.com/pilosa/pilosa/stats" "github.com/pkg/errors" "golang.org/x/sync/errgroup" ) @@ -930,7 +931,7 @@ func (api *API) AvailableShardsByIndex(_ context.Context) map[string]*roaring.Bi // StatsWithTags returns an instance of whatever implementation of StatsClient // pilosa is using with the given tags. -func (api *API) StatsWithTags(tags []string) StatsClient { +func (api *API) StatsWithTags(tags []string) stats.StatsClient { if api.holder == nil || api.cluster == nil { return nil } diff --git a/cache.go b/cache.go index df82a5802..40509ab64 100644 --- a/cache.go +++ b/cache.go @@ -23,6 +23,7 @@ import ( "time" "github.com/pilosa/pilosa/lru" + "github.com/pilosa/pilosa/stats" ) const ( @@ -50,14 +51,14 @@ type cache interface { Top() []bitmapPair // SetStats defines the stats client used in the cache. - SetStats(s StatsClient) + SetStats(s stats.StatsClient) } // lruCache represents a least recently used Cache implementation. type lruCache struct { cache *lru.Cache counts map[uint64]uint64 - stats StatsClient + stats stats.StatsClient } // newLRUCache returns a new instance of LRUCache. @@ -65,7 +66,7 @@ func newLRUCache(maxEntries uint32) *lruCache { c := &lruCache{ cache: lru.New(int(maxEntries)), counts: make(map[uint64]uint64), - stats: NopStatsClient, + stats: stats.NopStatsClient, } c.cache.OnEvicted = c.onEvicted return c @@ -122,7 +123,7 @@ func (c *lruCache) Top() []bitmapPair { } // SetStats defines the stats client used in the cache. -func (c *lruCache) SetStats(s StatsClient) { +func (c *lruCache) SetStats(s stats.StatsClient) { c.stats = s } @@ -150,7 +151,7 @@ type rankCache struct { // thresholdValue is the value of the last item in the cache thresholdValue uint64 - stats StatsClient + stats stats.StatsClient } // NewRankCache returns a new instance of RankCache. @@ -159,7 +160,7 @@ func NewRankCache(maxEntries uint32) *rankCache { maxEntries: maxEntries, thresholdBuffer: int(thresholdFactor * float64(maxEntries)), entries: make(map[uint64]uint64), - stats: NopStatsClient, + stats: stats.NopStatsClient, } } @@ -279,7 +280,7 @@ func (c *rankCache) recalculate() { } // SetStats defines the stats client used in the cache. -func (c *rankCache) SetStats(s StatsClient) { +func (c *rankCache) SetStats(s stats.StatsClient) { c.stats = s } @@ -458,12 +459,12 @@ func (s *simpleCache) Add(id uint64, b *Row) { // nopCache represents a no-op Cache implementation. type nopCache struct { - stats StatsClient + stats stats.StatsClient } // Ensure NopCache implements Cache. var globalNopCache cache = nopCache{ - stats: NopStatsClient, + stats: stats.NopStatsClient, } func (c nopCache) Add(uint64, uint64) {} @@ -471,10 +472,10 @@ func (c nopCache) BulkAdd(uint64, uint64) {} func (c nopCache) Get(uint64) uint64 { return 0 } func (c nopCache) IDs() []uint64 { return []uint64{} } -func (c nopCache) Invalidate() {} -func (c nopCache) Len() int { return 0 } -func (c nopCache) Recalculate() {} -func (c nopCache) SetStats(StatsClient) {} +func (c nopCache) Invalidate() {} +func (c nopCache) Len() int { return 0 } +func (c nopCache) Recalculate() {} +func (c nopCache) SetStats(stats.StatsClient) {} func (c nopCache) Top() []bitmapPair { return []bitmapPair{} diff --git a/cluster.go b/cluster.go index 4b1ab9089..e22781a25 100644 --- a/cluster.go +++ b/cluster.go @@ -31,6 +31,7 @@ import ( "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" + "github.com/pilosa/pilosa/logger" "github.com/pilosa/pilosa/roaring" "github.com/pkg/errors" uuid "github.com/satori/go.uuid" @@ -216,7 +217,7 @@ type cluster struct { // nolint: maligned wg sync.WaitGroup closing chan struct{} - logger Logger + logger logger.Logger InternalClient InternalClient } @@ -235,7 +236,7 @@ func newCluster() *cluster { InternalClient: newNopInternalClient(), - logger: NopLogger, + logger: logger.NopLogger, } } @@ -1379,7 +1380,7 @@ type resizeJob struct { mu sync.RWMutex state string - Logger Logger + Logger logger.Logger } // newResizeJob returns a new instance of resizeJob. @@ -1411,7 +1412,7 @@ func newResizeJob(existingNodes []*Node, node *Node, action string) *resizeJob { IDs: ids, action: action, result: make(chan string), - Logger: NopLogger, + Logger: logger.NopLogger, } } diff --git a/diagnostics.go b/diagnostics.go index 673fb0c7f..5ed97940a 100644 --- a/diagnostics.go +++ b/diagnostics.go @@ -24,6 +24,7 @@ import ( "sync" "time" + "github.com/pilosa/pilosa/logger" "github.com/pkg/errors" ) @@ -51,7 +52,7 @@ type diagnosticsCollector struct { client *http.Client - Logger Logger + Logger logger.Logger server *Server } @@ -65,7 +66,7 @@ func newDiagnosticsCollector(host string) *diagnosticsCollector { // nolint: unp start: time.Now(), client: &http.Client{Timeout: 10 * time.Second}, metrics: make(map[string]interface{}), - Logger: NopLogger, + Logger: logger.NopLogger, } } diff --git a/field.go b/field.go index bc656fa54..4189e55a4 100644 --- a/field.go +++ b/field.go @@ -28,8 +28,10 @@ import ( "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" + "github.com/pilosa/pilosa/logger" "github.com/pilosa/pilosa/pql" "github.com/pilosa/pilosa/roaring" + "github.com/pilosa/pilosa/stats" "github.com/pkg/errors" ) @@ -69,7 +71,7 @@ type Field struct { rowAttrStore AttrStore broadcaster broadcaster - Stats StatsClient + Stats stats.StatsClient // Field options. options FieldOptions @@ -79,7 +81,7 @@ type Field struct { // Shards with data on any node in the cluster, according to this node. remoteAvailableShards *roaring.Bitmap - logger Logger + logger logger.Logger } // FieldOption is a functional option type for pilosa.fieldOptions. @@ -196,13 +198,13 @@ func newField(path, index, name string, opts FieldOption) (*Field, error) { rowAttrStore: nopStore, broadcaster: NopBroadcaster, - Stats: NopStatsClient, + Stats: stats.NopStatsClient, options: applyDefaultOptions(fo), remoteAvailableShards: roaring.NewBitmap(), - logger: NopLogger, + logger: logger.NopLogger, } return f, nil } diff --git a/fragment.go b/fragment.go index 359628ab5..805112039 100644 --- a/fragment.go +++ b/fragment.go @@ -36,8 +36,10 @@ import ( "github.com/cespare/xxhash" "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" + "github.com/pilosa/pilosa/logger" "github.com/pilosa/pilosa/pql" "github.com/pilosa/pilosa/roaring" + "github.com/pilosa/pilosa/stats" "github.com/pkg/errors" ) @@ -116,7 +118,7 @@ type fragment struct { MaxOpN int // Logger used for out-of-band log entries. - Logger Logger + Logger logger.Logger // Row attribute storage. // This is set by the parent field unless overridden for testing. @@ -126,7 +128,7 @@ type fragment struct { // existing value (to clear) prior to setting a new value. mutexVector vector - stats StatsClient + stats stats.StatsClient } // newFragment returns a new instance of Fragment. @@ -140,10 +142,10 @@ func newFragment(path, index, field, view string, shard uint64) *fragment { CacheType: DefaultCacheType, CacheSize: DefaultCacheSize, - Logger: NopLogger, + Logger: logger.NopLogger, MaxOpN: defaultFragmentMaxOpN, - stats: NopStatsClient, + stats: stats.NopStatsClient, } } @@ -1718,7 +1720,7 @@ func (f *fragment) Snapshot() error { defer f.mu.Unlock() return f.snapshot() } -func track(start time.Time, message string, stats StatsClient, logger Logger) { +func track(start time.Time, message string, stats stats.StatsClient, logger logger.Logger) { elapsed := time.Since(start) logger.Printf("%s took %s", message, elapsed) stats.Histogram("snapshot", elapsed.Seconds(), 1.0) diff --git a/gossip/gossip.go b/gossip/gossip.go index ecd663ecf..8f19203f5 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -29,6 +29,7 @@ import ( "github.com/hashicorp/memberlist" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/logger" "github.com/pilosa/pilosa/roaring" "github.com/pilosa/pilosa/toml" "github.com/pkg/errors" @@ -47,7 +48,7 @@ type memberSet struct { papi *pilosa.API config *config - Logger pilosa.Logger + Logger logger.Logger logger *log.Logger logOutput io.Writer @@ -170,7 +171,7 @@ func NewMemberSet(cfg Config, api *pilosa.API, options ...memberSetOption) (*mem host := api.Node().URI.Host g := &memberSet{ papi: api, - Logger: pilosa.NopLogger, + Logger: logger.NopLogger, } // options diff --git a/holder.go b/holder.go index 0a1af820b..68643915e 100644 --- a/holder.go +++ b/holder.go @@ -27,7 +27,9 @@ import ( "syscall" "time" + "github.com/pilosa/pilosa/logger" "github.com/pilosa/pilosa/roaring" + "github.com/pilosa/pilosa/stats" "github.com/pkg/errors" uuid "github.com/satori/go.uuid" ) @@ -66,7 +68,7 @@ type Holder struct { closing chan struct{} // Stats - Stats StatsClient + Stats stats.StatsClient // Data directory path. Path string @@ -74,7 +76,7 @@ type Holder struct { // The interval at which the cached row ids are persisted to disk. cacheFlushInterval time.Duration - Logger Logger + Logger logger.Logger } // NewHolder returns a new instance of Holder. @@ -89,13 +91,13 @@ func NewHolder() *Holder { NewPrimaryTranslateStore: newNopTranslateStore, broadcaster: NopBroadcaster, - Stats: NopStatsClient, + Stats: stats.NopStatsClient, NewAttrStore: newNopAttrStore, cacheFlushInterval: defaultCacheFlushInterval, - Logger: NopLogger, + Logger: logger.NopLogger, } } @@ -605,7 +607,7 @@ type holderSyncer struct { Cluster *cluster // Stats - Stats StatsClient + Stats stats.StatsClient // Signals that the sync should stop. Closing <-chan struct{} diff --git a/http/handler.go b/http/handler.go index 81e31a82b..1ecbd2fb1 100644 --- a/http/handler.go +++ b/http/handler.go @@ -36,6 +36,7 @@ import ( "github.com/gorilla/handlers" "github.com/gorilla/mux" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/logger" "github.com/pkg/errors" ) @@ -44,7 +45,7 @@ import ( type Handler struct { Handler http.Handler - logger pilosa.Logger + logger logger.Logger // Keeps the query argument validators for each handler validators map[string]*queryValidationSpec @@ -95,7 +96,7 @@ func OptHandlerAPI(api *pilosa.API) handlerOption { } } -func OptHandlerLogger(logger pilosa.Logger) handlerOption { +func OptHandlerLogger(logger logger.Logger) handlerOption { return func(h *Handler) error { h.logger = logger return nil @@ -121,7 +122,7 @@ func OptHandlerCloseTimeout(d time.Duration) handlerOption { // NewHandler returns a new instance of Handler with a default logger. func NewHandler(opts ...handlerOption) (*Handler, error) { handler := &Handler{ - logger: pilosa.NopLogger, + logger: logger.NopLogger, closeTimeout: time.Second * 30, } handler.Handler = newRouter(handler) diff --git a/index.go b/index.go index 297c02f20..e06ce1732 100644 --- a/index.go +++ b/index.go @@ -25,7 +25,9 @@ import ( "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" + "github.com/pilosa/pilosa/logger" "github.com/pilosa/pilosa/roaring" + "github.com/pilosa/pilosa/stats" "github.com/pkg/errors" ) @@ -49,9 +51,9 @@ type Index struct { columnAttrs AttrStore broadcaster broadcaster - Stats StatsClient + Stats stats.StatsClient - logger Logger + logger logger.Logger } // NewIndex returns a new instance of Index. @@ -70,8 +72,8 @@ func NewIndex(path, name string) (*Index, error) { columnAttrs: nopStore, broadcaster: NopBroadcaster, - Stats: NopStatsClient, - logger: NopLogger, + Stats: stats.NopStatsClient, + logger: logger.NopLogger, trackExistence: true, }, nil } diff --git a/logger.go b/logger/logger.go similarity index 91% rename from logger.go rename to logger/logger.go index 074da8a36..ed5a2dc26 100644 --- a/logger.go +++ b/logger/logger.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package pilosa +package logger import ( "io" @@ -39,7 +39,7 @@ func (n *nopLogger) Printf(format string, v ...interface{}) {} // Debugf is a no-op implementation of the Logger Debugf method. func (n *nopLogger) Debugf(format string, v ...interface{}) {} -// standardLogger is a basic implementation of pilosa.Logger based on log.Logger. +// standardLogger is a basic implementation of Logger based on log.Logger. type standardLogger struct { logger *log.Logger } @@ -60,7 +60,7 @@ func (s *standardLogger) Logger() *log.Logger { return s.logger } -// verboseLogger is an implementation of pilosa.Logger which includes debug messages. +// verboseLogger is an implementation of Logger which includes debug messages. type verboseLogger struct { logger *log.Logger } diff --git a/server.go b/server.go index 5e6f61087..24385c4ea 100644 --- a/server.go +++ b/server.go @@ -27,7 +27,9 @@ import ( "sync" "time" + "github.com/pilosa/pilosa/logger" "github.com/pilosa/pilosa/roaring" + "github.com/pilosa/pilosa/stats" "github.com/pkg/errors" "golang.org/x/sync/errgroup" ) @@ -58,7 +60,7 @@ type Server struct { // nolint: maligned // External systemInfo SystemInfo gcNotifier GCNotifier - logger Logger + logger logger.Logger nodeID string uri URI @@ -81,7 +83,7 @@ func (s *Server) Holder() *Holder { // ServerOption is a functional option type for pilosa.Server type ServerOption func(s *Server) error -func OptServerLogger(l Logger) ServerOption { +func OptServerLogger(l logger.Logger) ServerOption { return func(s *Server) error { s.logger = l return nil @@ -176,7 +178,7 @@ func OptServerPrimaryTranslateStoreFunc(tf func(interface{}) TranslateStore) Ser } } -func OptServerStatsClient(sc StatsClient) ServerOption { +func OptServerStatsClient(sc stats.StatsClient) ServerOption { return func(s *Server) error { s.holder.Stats = sc return nil @@ -258,7 +260,7 @@ func NewServer(opts ...ServerOption) (*Server, error) { metricInterval: 0, diagnosticInterval: 0, - logger: NopLogger, + logger: logger.NopLogger, } s.executor = newExecutor(optExecutorInternalQueryClient(s.defaultClient)) s.cluster.InternalClient = s.defaultClient diff --git a/server/server.go b/server/server.go index 6cdc43883..1e140d1f0 100644 --- a/server/server.go +++ b/server/server.go @@ -41,12 +41,14 @@ import ( "github.com/pilosa/pilosa/gopsutil" "github.com/pilosa/pilosa/gossip" "github.com/pilosa/pilosa/http" + "github.com/pilosa/pilosa/logger" + "github.com/pilosa/pilosa/stats" "github.com/pilosa/pilosa/statsd" "github.com/pkg/errors" ) type loggerLogger interface { - pilosa.Logger + logger.Logger Logger() *log.Logger } @@ -185,9 +187,9 @@ func (m *Command) setupLogger() error { } if m.Config.Verbose { - m.logger = pilosa.NewVerboseLogger(m.logOutput) + m.logger = logger.NewVerboseLogger(m.logOutput) } else { - m.logger = pilosa.NewStandardLogger(m.logOutput) + m.logger = logger.NewStandardLogger(m.logOutput) } return nil } @@ -375,14 +377,14 @@ func (m *Command) Close() error { } // newStatsClient creates a stats client from the config -func newStatsClient(name string, host string) (pilosa.StatsClient, error) { +func newStatsClient(name string, host string) (stats.StatsClient, error) { switch name { case "expvar": - return pilosa.NewExpvarStatsClient(), nil + return stats.NewExpvarStatsClient(), nil case "statsd": return statsd.NewStatsClient(host) case "nop", "none": - return pilosa.NopStatsClient, nil + return stats.NopStatsClient, nil default: return nil, errors.Errorf("'%v' not a valid stats client, choose from [expvar, statsd, none].", name) } diff --git a/stats.go b/stats/stats.go similarity index 96% rename from stats.go rename to stats/stats.go index 8f23c77aa..169df0d6d 100644 --- a/stats.go +++ b/stats/stats.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package pilosa +package stats import ( "expvar" @@ -20,6 +20,8 @@ import ( "strings" "sync" "time" + + "github.com/pilosa/pilosa/logger" ) // Expvar global expvar map. @@ -52,7 +54,7 @@ type StatsClient interface { Timing(name string, value time.Duration, rate float64) // SetLogger Set the logger output type - SetLogger(logger Logger) + SetLogger(logger logger.Logger) // Starts the service Open() @@ -74,7 +76,7 @@ func (c *nopStatsClient) Gauge(name string, value float64, rate float64) func (c *nopStatsClient) Histogram(name string, value float64, rate float64) {} func (c *nopStatsClient) Set(name string, value string, rate float64) {} func (c *nopStatsClient) Timing(name string, value time.Duration, rate float64) {} -func (c *nopStatsClient) SetLogger(logger Logger) {} +func (c *nopStatsClient) SetLogger(logger logger.Logger) {} func (c *nopStatsClient) Open() {} func (c *nopStatsClient) Close() error { return nil } @@ -149,7 +151,7 @@ func (c *expvarStatsClient) Timing(name string, value time.Duration, rate float6 } // SetLogger has no logger. -func (c *expvarStatsClient) SetLogger(logger Logger) { +func (c *expvarStatsClient) SetLogger(logger logger.Logger) { } // Open no-op. @@ -221,7 +223,7 @@ func (a MultiStatsClient) Timing(name string, value time.Duration, rate float64) } // SetLogger Sets the StatsD logger output type. -func (a MultiStatsClient) SetLogger(logger Logger) { +func (a MultiStatsClient) SetLogger(logger logger.Logger) { for _, c := range a { c.SetLogger(logger) } diff --git a/stats_test.go b/stats/stats_test.go similarity index 80% rename from stats_test.go rename to stats/stats_test.go index 067cfa991..3da83ce70 100644 --- a/stats_test.go +++ b/stats/stats_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package pilosa_test +package stats_test import ( "context" @@ -23,6 +23,8 @@ import ( "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/http" + "github.com/pilosa/pilosa/logger" + "github.com/pilosa/pilosa/stats" "github.com/pilosa/pilosa/test" ) @@ -32,51 +34,51 @@ func TestMultiStatClient_Expvar(t *testing.T) { hldr := test.MustOpenHolder() defer hldr.Close() - c := pilosa.NewExpvarStatsClient() - ms := make(pilosa.MultiStatsClient, 1) + c := stats.NewExpvarStatsClient() + ms := make(stats.MultiStatsClient, 1) ms[0] = c hldr.Stats = ms hldr.SetBit("d", "f", 0, 0) hldr.SetBit("d", "f", 0, 1) - hldr.SetBit("d", "f", 0, ShardWidth) - hldr.SetBit("d", "f", 0, ShardWidth+2) + hldr.SetBit("d", "f", 0, pilosa.ShardWidth) + hldr.SetBit("d", "f", 0, pilosa.ShardWidth+2) hldr.ClearBit("d", "f", 0, 1) - if pilosa.Expvar.String() != `{"index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}}` { - t.Fatalf("unexpected expvar : %s", pilosa.Expvar.String()) + if stats.Expvar.String() != `{"index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}}` { + t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } hldr.Stats.CountWithCustomTags("cc", 1, 1.0, []string{"foo:bar"}) - if pilosa.Expvar.String() != `{"cc": 1, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}}` { - t.Fatalf("unexpected expvar : %s", pilosa.Expvar.String()) + if stats.Expvar.String() != `{"cc": 1, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}}` { + t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } // Gauge creates a unique key, subsequent Gauge calls will overwrite hldr.Stats.Gauge("g", 5, 1.0) hldr.Stats.Gauge("g", 8, 1.0) - if pilosa.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}}` { - t.Fatalf("unexpected expvar : %s", pilosa.Expvar.String()) + if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}}` { + t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } // Set creates a unique key, subsequent sets will overwrite hldr.Stats.Set("s", "4", 1.0) hldr.Stats.Set("s", "7", 1.0) - if pilosa.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}, "s": "7"}` { - t.Fatalf("unexpected expvar : %s", pilosa.Expvar.String()) + if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}, "s": "7"}` { + t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } // Record timing duration and a uniquely Set key/value dur, _ := time.ParseDuration("123us") hldr.Stats.Timing("tt", dur, 1.0) - if pilosa.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}, "s": "7", "tt": 123µs}` { - t.Fatalf("unexpected expvar : %s", pilosa.Expvar.String()) + if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}, "s": "7", "tt": 123µs}` { + t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } // Expvar histogram is implemented as a gauge hldr.Stats.Histogram("hh", 3, 1.0) - if pilosa.Expvar.String() != `{"cc": 1, "g": 8, "hh": 3, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}, "s": "7", "tt": 123µs}` { - t.Fatalf("unexpected expvar : %s", pilosa.Expvar.String()) + if stats.Expvar.String() != `{"cc": 1, "g": 8, "hh": 3, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}, "s": "7", "tt": 123µs}` { + t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } // Expvar should ignore earlier set tags from setbit @@ -92,8 +94,8 @@ func TestStatsCount_TopN(t *testing.T) { hldr.SetBit("d", "f", 0, 0) hldr.SetBit("d", "f", 0, 1) - hldr.SetBit("d", "f", 0, ShardWidth) - hldr.SetBit("d", "f", 0, ShardWidth+2) + hldr.SetBit("d", "f", 0, pilosa.ShardWidth) + hldr.SetBit("d", "f", 0, pilosa.ShardWidth+2) // Execute query. called := false @@ -311,11 +313,11 @@ func (s *MockStats) CountWithCustomTags(name string, value int64, rate float64, } func (c *MockStats) Tags() []string { return nil } -func (c *MockStats) WithTags(tags ...string) pilosa.StatsClient { return c } +func (c *MockStats) WithTags(tags ...string) stats.StatsClient { return c } func (c *MockStats) Gauge(name string, value float64, rate float64) {} func (c *MockStats) Histogram(name string, value float64, rate float64) {} func (c *MockStats) Set(name string, value string, rate float64) {} func (c *MockStats) Timing(name string, value time.Duration, rate float64) {} -func (c *MockStats) SetLogger(logger pilosa.Logger) {} +func (c *MockStats) SetLogger(logger logger.Logger) {} func (c *MockStats) Open() {} func (c *MockStats) Close() error { return nil } diff --git a/statsd/statsd.go b/statsd/statsd.go index 7a9ae6c14..eaf8facf1 100644 --- a/statsd/statsd.go +++ b/statsd/statsd.go @@ -19,7 +19,8 @@ import ( "time" "github.com/DataDog/datadog-go/statsd" - "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/logger" + "github.com/pilosa/pilosa/stats" ) // StatsD protocol wrapper using the DataDog library that added Tags to the StatsD protocol @@ -34,13 +35,13 @@ const ( ) // Ensure client implements interface. -var _ pilosa.StatsClient = &statsClient{} +var _ stats.StatsClient = &statsClient{} // statsClient represents a StatsD implementation of pilosa.statsClient. type statsClient struct { client *statsd.Client tags []string - logger pilosa.Logger + logger logger.Logger } // NewStatsClient returns a new instance of StatsClient. @@ -52,7 +53,7 @@ func NewStatsClient(host string) (*statsClient, error) { return &statsClient{ client: c, - logger: pilosa.NopLogger, + logger: logger.NopLogger, }, nil } @@ -70,7 +71,7 @@ func (c *statsClient) Tags() []string { } // WithTags returns a new client with additional tags appended. -func (c *statsClient) WithTags(tags ...string) pilosa.StatsClient { +func (c *statsClient) WithTags(tags ...string) stats.StatsClient { return &statsClient{ client: c.client, tags: unionStringSlice(c.tags, tags), @@ -122,7 +123,7 @@ func (c *statsClient) Timing(name string, value time.Duration, rate float64) { } // SetLogger sets the logger for client. -func (c *statsClient) SetLogger(logger pilosa.Logger) { +func (c *statsClient) SetLogger(logger logger.Logger) { c.logger = logger } diff --git a/translate.go b/translate.go index 86ed6c914..669e1e323 100644 --- a/translate.go +++ b/translate.go @@ -15,6 +15,7 @@ import ( "time" "github.com/cespare/xxhash" + "github.com/pilosa/pilosa/logger" "github.com/pkg/errors" ) @@ -68,7 +69,7 @@ type TranslateFile struct { Path string mapSize int - logger Logger + logger logger.Logger // If non-nil, data is streamed from a primary and this is a read-only store. PrimaryTranslateStore TranslateStore primaryID string // unique ID used to identify the primary store @@ -89,7 +90,7 @@ func OptTranslateFileMapSize(mapSize int) TranslateFileOption { return nil } } -func OptTranslateFileLogger(l Logger) TranslateFileOption { +func OptTranslateFileLogger(l logger.Logger) TranslateFileOption { return func(s *TranslateFile) error { s.logger = l return nil @@ -116,7 +117,7 @@ func NewTranslateFile(opts ...TranslateFileOption) *TranslateFile { mapSize: defaultMapSize, - logger: NopLogger, + logger: logger.NopLogger, replicationClosing: make(chan struct{}), primaryStoreEvents: make(chan primaryStoreEvent), diff --git a/view.go b/view.go index a0abb0e9c..128e3b828 100644 --- a/view.go +++ b/view.go @@ -22,8 +22,10 @@ import ( "strings" "sync" + "github.com/pilosa/pilosa/logger" "github.com/pilosa/pilosa/pql" "github.com/pilosa/pilosa/roaring" + "github.com/pilosa/pilosa/stats" "github.com/pkg/errors" ) @@ -50,9 +52,9 @@ type view struct { fragments map[uint64]*fragment broadcaster broadcaster - stats StatsClient + stats stats.StatsClient rowAttrStore AttrStore - logger Logger + logger logger.Logger } // newView returns a new instance of View. @@ -70,8 +72,8 @@ func newView(path, index, field, name string, fieldOptions FieldOptions) *view { fragments: make(map[uint64]*fragment), broadcaster: NopBroadcaster, - stats: NopStatsClient, - logger: NopLogger, + stats: stats.NopStatsClient, + logger: logger.NopLogger, } } From 33add4f1e000343b4909c333350037ededdddd19 Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 14 Nov 2018 23:01:48 -0600 Subject: [PATCH 05/19] proof of concept for stats This commit adds some trivial stat-tracking which can be observed at localhost:10101/debug/vars. However, writes to a locking data structure aren't cheap, so the stat-tracking is by default not compiled. To build it, add the build tag `roaringstats`, which will cause the `statsHit` function to actually do something. Otherwise, it's an empty and inlineable function, meaning the compiler throws it away entirely. This would, in principle, let us get additional visibility into edge cases and which code paths are hot. This is not the same thing as profiling for overall performance; the stat counts aren't affected by whether a particular code path is using a large amount of CPU time, just reporting how often it happens at all. --- roaring/containers.go | 2 + roaring/roaring.go | 71 ++++++++++++++++++++++++++++++++++++ roaring/roaring_nop_stats.go | 8 ++++ roaring/roaring_stats.go | 15 ++++++++ 4 files changed, 96 insertions(+) create mode 100644 roaring/roaring_nop_stats.go create mode 100644 roaring/roaring_stats.go diff --git a/roaring/containers.go b/roaring/containers.go index ed745a915..3fe0814cc 100644 --- a/roaring/containers.go +++ b/roaring/containers.go @@ -64,6 +64,7 @@ func (sc *sliceContainers) PutContainerValues(key uint64, containerType byte, n } func (sc *sliceContainers) Remove(key uint64) { + statsHit("sliceContainers/Remove") i := search64(sc.keys, key) if i < 0 { return @@ -73,6 +74,7 @@ func (sc *sliceContainers) Remove(key uint64) { } func (sc *sliceContainers) insertAt(key uint64, c *Container, i int) { + statsHit("sliceContainers/insertAt") sc.keys = append(sc.keys, 0) copy(sc.keys[i+1:], sc.keys[i:]) sc.keys[i] = key diff --git a/roaring/roaring.go b/roaring/roaring.go index ba258526f..3a4050cea 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -1027,6 +1027,7 @@ func (iv interval16) runlen() int32 { // newContainer returns a new instance of container. func NewContainer() *Container { + statsHit("NewContainer") return &Container{containerType: containerArray} } @@ -1194,6 +1195,7 @@ func (c *Container) add(v uint16) (added bool) { func (c *Container) arrayAdd(v uint16) bool { // Optimize appending to the end of an array container. if c.n > 0 && c.n < ArrayMaxSize && c.isArray() && c.array[c.n-1] < v { + statsHit("arrayAdd/append") c.unmap() c.array = append(c.array, v) return true @@ -1207,11 +1209,13 @@ func (c *Container) arrayAdd(v uint16) bool { // Convert to a bitmap container if too many values are in an array container. if c.n >= ArrayMaxSize { + statsHit("arrayAdd/arrayToBitmap") c.arrayToBitmap() return c.bitmapAdd(v) } // Otherwise insert into array. + statsHit("arrayAdd/insert") c.unmap() i = -i - 1 c.array = append(c.array, 0) @@ -1325,6 +1329,7 @@ func (c *Container) countRuns() (r int32) { // amount of space. func (c *Container) optimize() { if c.n == 0 { + statsHit("optimize/empty") return } runs := c.countRuns() @@ -1341,21 +1346,33 @@ func (c *Container) optimize() { // Then convert accordingly. if c.isArray() { if newType == containerBitmap { + statsHit("optimize/arrayToBitmap") c.arrayToBitmap() } else if newType == containerRun { + statsHit("optimize/arrayToRun") c.arrayToRun() + } else { + statsHit("optimize/arrayUnchanged") } } else if c.isBitmap() { if newType == containerArray { + statsHit("optimize/bitmapToArray") c.bitmapToArray() } else if newType == containerRun { + statsHit("optimize/bitmapToRun") c.bitmapToRun() + } else { + statsHit("optimize/bitmapUnchanged") } } else if c.isRun() { if newType == containerBitmap { + statsHit("optimize/runToBitmap") c.runToBitmap() } else if newType == containerArray { + statsHit("optimize/runToArray") c.runToArray() + } else { + statsHit("optimize/runUnchanged") } } } @@ -1425,6 +1442,7 @@ func (c *Container) bitmapRemove(v uint16) bool { // Convert to array if we go below the threshold. if c.n == ArrayMaxSize { + statsHit("bitmapRemove/bitmapToArray") c.bitmapToArray() } return true @@ -1492,6 +1510,7 @@ func (c *Container) runMax() uint16 { // bitmapToArray converts from bitmap format to array format. func (c *Container) bitmapToArray() { + statsHit("bitmapToArray") c.array = make([]uint16, 0, c.n) c.containerType = containerArray @@ -1515,6 +1534,7 @@ func (c *Container) bitmapToArray() { // arrayToBitmap converts from array format to bitmap format. func (c *Container) arrayToBitmap() { + statsHit("arrayToBitmap") c.bitmap = make([]uint64, bitmapN) c.containerType = containerBitmap @@ -1534,6 +1554,7 @@ func (c *Container) arrayToBitmap() { // runToBitmap converts from RLE format to bitmap format. func (c *Container) runToBitmap() { + statsHit("runToBitmap") c.bitmap = make([]uint64, bitmapN) c.containerType = containerBitmap @@ -1557,6 +1578,7 @@ func (c *Container) runToBitmap() { // bitmapToRun converts from bitmap format to RLE format. func (c *Container) bitmapToRun() { + statsHit("bitmapToRun") c.containerType = containerRun // return early if empty if c.n == 0 { @@ -1613,6 +1635,7 @@ func (c *Container) bitmapToRun() { // arrayToRun converts from array format to RLE format. func (c *Container) arrayToRun() { + statsHit("arrayToRun") c.containerType = containerRun // return early if empty if c.n == 0 { @@ -1640,6 +1663,7 @@ func (c *Container) arrayToRun() { // runToArray converts from RLE format to array format. func (c *Container) runToArray() { + statsHit("runToArray") c.containerType = containerArray c.array = make([]uint16, 0, c.n) @@ -1661,16 +1685,20 @@ func (c *Container) runToArray() { // Clone returns a copy of c. func (c *Container) Clone() *Container { + statsHit("Container/Clone") other := &Container{n: c.n, containerType: c.containerType} switch c.containerType { case containerArray: + statsHit("Container/Clone/Array") other.array = make([]uint16, len(c.array)) copy(other.array, c.array) case containerBitmap: + statsHit("Container/Clone/Bitmap") other.bitmap = make([]uint64, len(c.bitmap)) copy(other.bitmap, c.bitmap) case containerRun: + statsHit("Container/Clone/Run") other.runs = make([]interval16, len(c.runs)) copy(other.runs, c.runs) } @@ -1689,6 +1717,7 @@ func (c *Container) WriteTo(w io.Writer) (n int64, err error) { } func (c *Container) arrayWriteTo(w io.Writer) (n int64, err error) { + statsHit("Container/arrayWriteTo") if len(c.array) == 0 { return 0, nil } @@ -1705,12 +1734,14 @@ func (c *Container) arrayWriteTo(w io.Writer) (n int64, err error) { } func (c *Container) bitmapWriteTo(w io.Writer) (n int64, err error) { + statsHit("Container/bitmapWriteTo") // Write sizeof(uint64) * bitmapN bytes. nn, err := w.Write((*[0xFFFFFFF]byte)(unsafe.Pointer(&c.bitmap[0]))[:(8 * bitmapN)]) return int64(nn), err } func (c *Container) runWriteTo(w io.Writer) (n int64, err error) { + statsHit("Container/runWriteTo") if len(c.runs) == 0 { return 0, nil } @@ -1815,6 +1846,7 @@ func flip(a *Container) *Container { // nolint: deadcode } func flipArray(b *Container) *Container { + statsHit("flipArray") // TODO: actually implement this x := b.Clone() x.arrayToBitmap() @@ -1822,6 +1854,7 @@ func flipArray(b *Container) *Container { } func flipBitmap(b *Container) *Container { + statsHit("flipBitmap") other := &Container{bitmap: make([]uint64, bitmapN), containerType: containerBitmap} for i, bitmap := range b.bitmap { @@ -1833,6 +1866,7 @@ func flipBitmap(b *Container) *Container { } func flipRun(b *Container) *Container { + statsHit("flipRun") // TODO: actually implement this x := b.Clone() x.runToBitmap() @@ -1868,6 +1902,7 @@ func intersectionCount(a, b *Container) int32 { } func intersectionCountArrayArray(a, b *Container) (n int32) { + statsHit("intersectionCount/ArrayArray") na, nb := len(a.array), len(b.array) for i, j := 0, 0; i < na && j < nb; { va, vb := a.array[i], b.array[j] @@ -1884,6 +1919,7 @@ func intersectionCountArrayArray(a, b *Container) (n int32) { } func intersectionCountArrayRun(a, b *Container) (n int32) { + statsHit("intersectionCount/ArrayRun") na, nb := len(a.array), len(b.runs) for i, j := 0, 0; i < na && j < nb; { va, vb := a.array[i], b.runs[j] @@ -1900,6 +1936,7 @@ func intersectionCountArrayRun(a, b *Container) (n int32) { } func intersectionCountRunRun(a, b *Container) (n int32) { + statsHit("intersectionCount/RunRun") na, nb := len(a.runs), len(b.runs) for i, j := 0, 0; i < na && j < nb; { va, vb := a.runs[i], b.runs[j] @@ -1931,6 +1968,7 @@ func intersectionCountRunRun(a, b *Container) (n int32) { } func intersectionCountBitmapRun(a, b *Container) (n int32) { + statsHit("intersectionCount/BitmapRun") for _, iv := range b.runs { n += a.bitmapCountRange(int32(iv.start), int32(iv.last)+1) } @@ -1938,6 +1976,7 @@ func intersectionCountBitmapRun(a, b *Container) (n int32) { } func intersectionCountArrayBitmap(a, b *Container) (n int32) { + statsHit("intersectionCount/ArrayBitmap") ln := len(b.bitmap) for _, val := range a.array { i := int(val >> 6) @@ -1951,6 +1990,7 @@ func intersectionCountArrayBitmap(a, b *Container) (n int32) { } func intersectionCountBitmapBitmap(a, b *Container) (n int32) { + statsHit("intersectionCount/BitmapBitmap") return int32(popcountAndSlice(a.bitmap, b.bitmap)) } @@ -1983,6 +2023,7 @@ func intersect(a, b *Container) *Container { } func intersectArrayArray(a, b *Container) *Container { + statsHit("intersect/ArrayArray") output := &Container{containerType: containerArray} na, nb := len(a.array), len(b.array) for i, j := 0, 0; i < na && j < nb; { @@ -2004,6 +2045,7 @@ func intersectArrayArray(a, b *Container) *Container { // container. The return is always an array container (since it's guaranteed to // be low-cardinality) func intersectArrayRun(a, b *Container) *Container { + statsHit("intersect/ArrayRun") output := &Container{containerType: containerArray} na, nb := len(a.array), len(b.runs) for i, j := 0, 0; i < na && j < nb; { @@ -2023,6 +2065,7 @@ func intersectArrayRun(a, b *Container) *Container { // intersectRunRun computes the intersect of two run containers. func intersectRunRun(a, b *Container) *Container { + statsHit("intersect/RunRun") output := &Container{containerType: containerRun} na, nb := len(a.runs), len(b.runs) for i, j := 0, 0; i < na && j < nb; { @@ -2062,6 +2105,7 @@ func intersectRunRun(a, b *Container) *Container { // intersectBitmapRun returns an array container if the run container's // cardinality is < ArrayMaxSize. Otherwise it returns a bitmap container. func intersectBitmapRun(a, b *Container) *Container { + statsHit("intersect/BitmapRun") var output *Container if b.n < ArrayMaxSize { // output is array container @@ -2125,6 +2169,7 @@ func intersectBitmapRun(a, b *Container) *Container { } func intersectArrayBitmap(a, b *Container) *Container { + statsHit("intersect/ArrayBitmap") output := &Container{containerType: containerArray} for _, va := range a.array { bmidx := va / 64 @@ -2140,6 +2185,7 @@ func intersectArrayBitmap(a, b *Container) *Container { } func intersectBitmapBitmap(a, b *Container) *Container { + statsHit("intersect/BitmapBitmap") // local variables added to prevent BCE checks in loop // see https://go101.org/article/bounds-check-elimination.html var ( @@ -2191,6 +2237,7 @@ func union(a, b *Container) *Container { } func unionArrayArray(a, b *Container) *Container { + statsHit("union/ArrayArray") output := &Container{containerType: containerArray} na, nb := len(a.array), len(b.array) for i, j := 0, 0; ; { @@ -2224,6 +2271,7 @@ func unionArrayArray(a, b *Container) *Container { // unionArrayRun optimistically assumes that the result will be a run container, // and converts to a bitmap or array container afterwards if necessary. func unionArrayRun(a, b *Container) *Container { + statsHit("union/ArrayRun") if b.n == maxContainerVal+1 { return b.Clone() } @@ -2281,6 +2329,7 @@ func (c *Container) runAppendInterval(v interval16) int32 { } func unionRunRun(a, b *Container) *Container { + statsHit("union/RunRun") if a.n == maxContainerVal+1 { return a.Clone() } @@ -2315,6 +2364,7 @@ func unionRunRun(a, b *Container) *Container { } func unionBitmapRun(a, b *Container) *Container { + statsHit("union/BitmapRun") if b.n == maxContainerVal+1 { return b.Clone() } @@ -2503,6 +2553,7 @@ func difference(a, b *Container) *Container { // differenceArrayArray computes the difference bween two arrays. func differenceArrayArray(a, b *Container) *Container { + statsHit("difference/ArrayArray") output := &Container{containerType: containerArray} na, nb := len(a.array), len(b.array) for i, j := 0, 0; i < na; { @@ -2528,6 +2579,7 @@ func differenceArrayArray(a, b *Container) *Container { // differenceArrayRun computes the difference of an array from a run. func differenceArrayRun(a, b *Container) *Container { + statsHit("difference/ArrayRun") // func (ac *arrayContainer) iandNotRun16(rc *runContainer16) container { if a.n == 0 || b.n == 0 { @@ -2585,6 +2637,7 @@ func differenceArrayRun(a, b *Container) *Container { // differenceBitmapRun computes the difference of an bitmap from a run. func differenceBitmapRun(a, b *Container) *Container { + statsHit("difference/BitmapRun") if a.n == 0 || b.n == 0 { return a.Clone() } @@ -2599,6 +2652,7 @@ func differenceBitmapRun(a, b *Container) *Container { // differenceRunArray subtracts the bits in an array container from a run // container. func differenceRunArray(a, b *Container) *Container { + statsHit("difference/RunArray") if a.n == 0 || b.n == 0 { return a.Clone() } @@ -2654,6 +2708,7 @@ RUNLOOP: // differenceRunBitmap computes the difference of an run from a bitmap. func differenceRunBitmap(a, b *Container) *Container { + statsHit("difference/RunBitmap") // If a is full, difference is the flip of b. if len(a.runs) > 0 && a.runs[0].start == 0 && a.runs[0].last == 65535 { return flipBitmap(b) @@ -2711,6 +2766,7 @@ func differenceRunBitmap(a, b *Container) *Container { // differenceRunRun computes the difference of two runs. func differenceRunRun(a, b *Container) *Container { + statsHit("difference/RunRun") if a.n == 0 || b.n == 0 { return a.Clone() } @@ -2774,6 +2830,7 @@ func differenceRunRun(a, b *Container) *Container { } func differenceArrayBitmap(a, b *Container) *Container { + statsHit("difference/ArrayBitmap") output := &Container{containerType: containerArray} for _, va := range a.array { bmidx := va / 64 @@ -2790,6 +2847,7 @@ func differenceArrayBitmap(a, b *Container) *Container { } func differenceBitmapArray(a, b *Container) *Container { + statsHit("difference/BitmapArray") output := a.Clone() for _, v := range b.array { @@ -2805,6 +2863,7 @@ func differenceBitmapArray(a, b *Container) *Container { } func differenceBitmapBitmap(a, b *Container) *Container { + statsHit("difference/BitmapBitmap") // local variables added to prevent BCE checks in loop // see https://go101.org/article/bounds-check-elimination.html @@ -2862,6 +2921,7 @@ func xor(a, b *Container) *Container { } func xorArrayArray(a, b *Container) *Container { + statsHit("xor/ArrayArray") output := &Container{containerType: containerArray} na, nb := len(a.array), len(b.array) for i, j := 0, 0; i < na || j < nb; { @@ -2891,6 +2951,7 @@ func xorArrayArray(a, b *Container) *Container { } func xorArrayBitmap(a, b *Container) *Container { + statsHit("xor/ArrayBitmap") output := b.Clone() for _, v := range a.array { if b.bitmapContains(v) { @@ -2910,6 +2971,7 @@ func xorArrayBitmap(a, b *Container) *Container { } func xorBitmapBitmap(a, b *Container) *Container { + statsHit("xor/BitmapBitmap") // local variables added to prevent BCE checks in loop // see https://go101.org/article/bounds-check-elimination.html @@ -2987,6 +3049,7 @@ func (op *op) UnmarshalBinary(data []byte) error { if len(data) < op.size() { return fmt.Errorf("op data out of bounds: len=%d", len(data)) } + statsHit("op/UnmarshalBinary") // Verify checksum. h := fnv.New32a() @@ -3011,6 +3074,7 @@ func lowbits(v uint64) uint16 { return uint16(v & 0xFFFF) } // search32 returns the index of value in a. If value is not found, it works the // same way as search64. func search32(a []uint16, value uint16) int32 { + statsHit("search32") // Optimize for elements and the last element. n := int32(len(a)) if n == 0 { @@ -3054,6 +3118,7 @@ func search32(a []uint16, value uint16) int32 { // since negative 0 is no different from positive 0, we offset the returned // negative indices by 1. See the test for this function for examples. func search64(a []uint64, value uint64) int { + statsHit("search64") // Optimize for elements and the last element. n := len(a) if n == 0 { @@ -3132,6 +3197,7 @@ func (a *ErrorList) AppendWithPrefix(err error, prefix string) { // xorArrayRun computes the exclusive or of an array and a run container. func xorArrayRun(a, b *Container) *Container { + statsHit("xor/ArrayRun") output := &Container{containerType: containerRun} na, nb := len(a.array), len(b.runs) var vb interval16 @@ -3290,6 +3356,7 @@ type xorstm struct { // xorRunRun computes the exclusive or of two run containers. func xorRunRun(a, b *Container) *Container { + statsHit("xor/RunRun") na, nb := len(a.runs), len(b.runs) if na == 0 { return b.Clone() @@ -3338,6 +3405,7 @@ func xorRunRun(a, b *Container) *Container { // xorRunRun computes the exclusive or of a bitmap and a run container. func xorBitmapRun(a, b *Container) *Container { + statsHit("xor/BitmapRun") output := a.Clone() for j := 0; j < len(b.runs); j++ { output.bitmapXorRange(uint64(b.runs[j].start), uint64(b.runs[j].last)+1) @@ -3352,6 +3420,7 @@ func xorBitmapRun(a, b *Container) *Container { } func bitmapsEqual(b, c *Bitmap) error { // nolint: deadcode + statsHit("bitmapsEqual") if b.OpWriter != c.OpWriter { return errors.New("opWriters not equal") } @@ -3404,6 +3473,7 @@ const ( ) func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint, card int) byte, header, pos int, haveRuns bool, err error) { + statsHit("readOfficialHeader") if len(buf) < 8 { err = fmt.Errorf("buffer too small, expecting at least 8 bytes, was %d", len(buf)) return size, containerTyper, header, pos, haveRuns, err @@ -3469,6 +3539,7 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error { // Nothing to unmarshal return nil } + statsHit("Bitmap/UnmarshalBinary") fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2])) if fileMagic == magicNumber { // if pilosa roaring return errors.Wrap(b.unmarshalPilosaRoaring(data), "unmarshaling as pilosa roaring") diff --git a/roaring/roaring_nop_stats.go b/roaring/roaring_nop_stats.go new file mode 100644 index 000000000..c9e029ab7 --- /dev/null +++ b/roaring/roaring_nop_stats.go @@ -0,0 +1,8 @@ +// +build !roaringstats + +package roaring + +// statsCount does nothing, because you aren't building with +// the "roaringstats" build tag. +func statsHit(string) { +} diff --git a/roaring/roaring_stats.go b/roaring/roaring_stats.go new file mode 100644 index 000000000..fd8ade91f --- /dev/null +++ b/roaring/roaring_stats.go @@ -0,0 +1,15 @@ +// +build roaringstats + +package roaring + +import ( + "github.com/pilosa/pilosa/stats" +) + +var statsEv = stats.NewExpvarStatsClient() + +// statsHit increments the given stat, so we can tell how often we've hit +// that particular event. +func statsHit(name string) { + statsEv.Count(name, 1, 1) +} From 8e270f9201822605ab1424254920547121dd8eb0 Mon Sep 17 00:00:00 2001 From: Seebs Date: Thu, 15 Nov 2018 14:58:07 -0600 Subject: [PATCH 06/19] provide commented-out test case for bug in dead code bitmapEquals isn't currently being called ever, but it has an arcane edge-case bug, so I've made the test case for it and commented it out for future reference. --- roaring/roaring_internal_test.go | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/roaring/roaring_internal_test.go b/roaring/roaring_internal_test.go index a4c162629..266bf39e7 100644 --- a/roaring/roaring_internal_test.go +++ b/roaring/roaring_internal_test.go @@ -3261,3 +3261,26 @@ func TestUnmarshalOfficialRoaring(t *testing.T) { } } + +/* +// This function exercises an arcane edge case in dead code. +// It doesn't need to be run right now. +func TestEquals(t *testing.T) { + bma := NewBitmap() + bmr := NewBitmap() + for i := uint64(0); i < 30; i++ { + bma.Add(i) + bmr.Add(i) + } + bmr.Optimize() + bmi := bma.Intersect(bmr) + err := bitmapsEqual(bmi, bma) + if err != nil { + t.Fatalf("expected intersection to equal array") + } + err = bitmapsEqual(bmi, bmr) + if err != nil { + t.Fatalf("expected intersection to equal run") + } +} +*/ From 08d7f656672a2731c1c2df619d0a9bc1e3565ec1 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Fri, 16 Nov 2018 13:01:21 -0600 Subject: [PATCH 07/19] increase the translate file size for tests/benchmarks --- translate_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/translate_test.go b/translate_test.go index 1aea26cfe..fe78108ff 100644 --- a/translate_test.go +++ b/translate_test.go @@ -809,7 +809,7 @@ func NewTranslateFile() *TranslateFile { } f.Close() - s := &TranslateFile{TranslateFile: pilosa.NewTranslateFile(pilosa.OptTranslateFileMapSize(2 << 25))} + s := &TranslateFile{TranslateFile: pilosa.NewTranslateFile(pilosa.OptTranslateFileMapSize(2 << 26))} s.Path = f.Name() return s } From c8e6fd2e43581e83dc357750c714e6d7fd36f529 Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 9 Nov 2018 22:44:37 -0600 Subject: [PATCH 08/19] improve type matrix for IntersectionCount benchmarks The circumstances under which bitmaps are converted between types are not 100% nailed down, and the IntersectionCount benchmark was actually using a bitmap for the "run" data set as well as for the "bitmap" data set. Fix that by using Optimize() explicitly. Also, add a second RLE set so we can compare the difference between "one run for the entire set" and "several runs". Also add array/array comparisons. We use two different lengths of arrays, because performance turns out to vary between "first array longer" and "second array longer". Also added a benchmark for getBenchData itself, since it's at least one possible use case for "creating a lot of containers". --- roaring/roaring_test.go | 117 ++++++++++++++++++++++++++++++++++------ 1 file changed, 101 insertions(+), 16 deletions(-) diff --git a/roaring/roaring_test.go b/roaring/roaring_test.go index 64cb45e81..6c31b7701 100644 --- a/roaring/roaring_test.go +++ b/roaring/roaring_test.go @@ -1056,19 +1056,39 @@ func TestBitmapBufIterator(t *testing.T) { } -var benchmarkBitmapIntersectionCountData struct { - a, b, r *roaring.Bitmap +// this data is used to test various operations across +// different types. +type benchmarkSampleData struct { + a1, a2, b, r1, r2 *roaring.Bitmap } -func getBenchData() *struct{ a, b, r *roaring.Bitmap } { - data := &benchmarkBitmapIntersectionCountData - if data.a == nil { +var sampleData benchmarkSampleData + +func isAllType(b *roaring.Bitmap, typ string) bool { + bi := b.Info() + for _, c := range bi.Containers { + if c.Type != typ { + return false + } + } + return true +} + +func getBenchData(b *testing.B) *benchmarkSampleData { + data := &sampleData + if data.a1 == nil { const max = (1 << 24) / 64 // Build bitmap with array container. - data.a = roaring.NewFileBitmap() - for i, n := 0, 2*roaring.ArrayMaxSize/3; i < n; i++ { - data.a.Add(uint64(rand.Intn(max))) + data.a1 = roaring.NewFileBitmap() + data.a2 = roaring.NewFileBitmap() + // two lists of different lengths + for i, n := 0, roaring.ArrayMaxSize/3; i < n; i++ { + data.a1.Add(uint64(rand.Intn(max))) + data.a2.Add(uint64(rand.Intn(max))) + } + for i, n := 0, roaring.ArrayMaxSize/3; i < n; i++ { + data.a1.Add(uint64(rand.Intn(max))) } // Build bitmap with bitmap container. @@ -1078,12 +1098,42 @@ func getBenchData() *struct{ a, b, r *roaring.Bitmap } { } // build bitmap with run container - data.r = roaring.NewFileBitmap() + data.r1 = roaring.NewFileBitmap() for i, n := 0, MaxContainerVal; i < n; i++ { - data.r.Add(uint64(i)) + data.r1.Add(uint64(i)) } + // build bitmap with multiple runs + data.r2 = roaring.NewFileBitmap() + for i, n := 0, MaxContainerVal; i < n; i++ { + data.r2.Add(uint64(i)) + // break the runs up, this should produce 16 runs, which + // is small enough to make RLE tempting + if i&0xfff == 0xfff { + i += 5 + } + } + data.a1.Optimize() + data.a2.Optimize() + data.b.Optimize() + data.r1.Optimize() + data.r2.Optimize() } + if !isAllType(data.a1, "array") { + b.Fatalf("expected data.a1 to be an array, it wasn't.") + } + if !isAllType(data.a2, "array") { + b.Fatalf("expected data.a2 to be an array, it wasn't.") + } + if !isAllType(data.b, "bitmap") { + b.Fatalf("expected data.b to be a bitmap, it wasn't.") + } + if !isAllType(data.r1, "run") { + b.Fatalf("expected data.r1 to be RLE, it wasn't.") + } + if !isAllType(data.r2, "run") { + b.Fatalf("expected data.r2 to be RLE, it wasn't.") + } return data } @@ -1138,30 +1188,65 @@ func TestBitmap_Intersect(t *testing.T) { } } +func BenchmarkGetBenchData(b *testing.B) { + for i := 0; i < b.N; i++ { + sampleData = benchmarkSampleData{} + getBenchData(b) + } +} + func BenchmarkBitmap_IntersectionCount_ArrayRun(b *testing.B) { - data := getBenchData() + data := getBenchData(b) // Reset timer & benchmark. b.ResetTimer() for i := 0; i < b.N; i++ { - data.a.IntersectionCount(data.r) + data.a1.IntersectionCount(data.r1) + } +} + +func BenchmarkBitmap_IntersectionCount_ArrayRuns(b *testing.B) { + data := getBenchData(b) + // Reset timer & benchmark. + b.ResetTimer() + for i := 0; i < b.N; i++ { + data.a1.IntersectionCount(data.r2) } } func BenchmarkBitmap_IntersectionCount_BitmapRun(b *testing.B) { - data := getBenchData() + data := getBenchData(b) // Reset timer & benchmark. b.ResetTimer() for i := 0; i < b.N; i++ { - data.b.IntersectionCount(data.r) + data.b.IntersectionCount(data.r1) + } +} + +func BenchmarkBitmap_IntersectionCount_BitmapRuns(b *testing.B) { + data := getBenchData(b) + // Reset timer & benchmark. + b.ResetTimer() + for i := 0; i < b.N; i++ { + data.b.IntersectionCount(data.r2) + } +} + +func BenchmarkBitmap_IntersectionCount_ArrayArray(b *testing.B) { + data := getBenchData(b) + // Reset timer & benchmark. + b.ResetTimer() + for i := 0; i < b.N; i++ { + data.a1.IntersectionCount(data.a2) + data.a2.IntersectionCount(data.a1) } } func BenchmarkBitmap_IntersectionCount_ArrayBitmap(b *testing.B) { - data := getBenchData() + data := getBenchData(b) // Reset timer & benchmark. b.ResetTimer() for i := 0; i < b.N; i++ { - data.a.IntersectionCount(data.b) + data.a1.IntersectionCount(data.b) } } From 32c4b3540f33d9384e291ce16e481e4384184482 Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 13 Nov 2018 12:12:13 -0600 Subject: [PATCH 09/19] simplify intersectBitmapRun output to remove a conversion If the total number of things returned was small enough to make an array, intersectBitmapRun converted to an array. This seems possibly-premature; future processing might well prefer a bitmap. We know everything gets optimized before being written out, let's not convert without a specific reason. But also, let's use an array no matter which container is small enough to prove that we can do so safely. Fixes #854. --- roaring/roaring.go | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/roaring/roaring.go b/roaring/roaring.go index 3a4050cea..ced003d37 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -2102,12 +2102,12 @@ func intersectRunRun(a, b *Container) *Container { return output } -// intersectBitmapRun returns an array container if the run container's -// cardinality is < ArrayMaxSize. Otherwise it returns a bitmap container. +// intersectBitmapRun returns an array container if either container's +// cardinality is <= ArrayMaxSize. Otherwise it returns a bitmap container. func intersectBitmapRun(a, b *Container) *Container { statsHit("intersect/BitmapRun") var output *Container - if b.n < ArrayMaxSize { + if b.n <= ArrayMaxSize || a.n <= ArrayMaxSize { // output is array container output = &Container{containerType: containerArray} for _, iv := range b.runs { @@ -2161,9 +2161,6 @@ func intersectBitmapRun(a, b *Container) *Container { valast = vastart + 63 } } - if output.n < ArrayMaxSize { - output.bitmapToArray() - } } return output } From d4364bea527f3a7d023e41974dc97b2c2ef8cc13 Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 12 Nov 2018 18:04:54 -0600 Subject: [PATCH 10/19] slightly streamline array/array comparison The net effect of this is to not recompute "the current value of the first array" on every loop, pretty much. However, the swap to make sure the inner loop is on the longer array seems to be significant for performance. On my system, this moves runtime from ~29us per op to ~17us per op. --- roaring/roaring.go | 28 +++++++++++++++++++--------- 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/roaring/roaring.go b/roaring/roaring.go index ced003d37..6cf7eda1c 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -1903,16 +1903,26 @@ func intersectionCount(a, b *Container) int32 { func intersectionCountArrayArray(a, b *Container) (n int32) { statsHit("intersectionCount/ArrayArray") - na, nb := len(a.array), len(b.array) - for i, j := 0, 0; i < na && j < nb; { - va, vb := a.array[i], b.array[j] - if va < vb { - i++ - } else if va > vb { - j++ - } else { + s1, s2 := a.array, b.array + if len(s1) == 0 || len(s2) == 0 { + return 0 + } + if len(s1) > len(s2) { + s1, s2 = s2, s1 + } + l2 := len(s2) + i2 := 0 + v2 := s2[0] + for _, v1 := range s1 { + for v2 < v1 { + i2++ + if i2 >= l2 { + return n + } + v2 = s2[i2] + } + if v2 == v1 { n++ - i, j = i+1, j+1 } } return n From 9b552ab5086ef9a9937baf06b773a427bf245ae9 Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 13 Nov 2018 12:21:47 -0600 Subject: [PATCH 11/19] enhance TestRunCountRange confirm that the number of runs comes out as expected, and add a couple of numbers out of order to verify that the 17-18-19 set gets coalesced into one run even if we add 17 and 19 before 18. --- roaring/roaring_internal_test.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/roaring/roaring_internal_test.go b/roaring/roaring_internal_test.go index 266bf39e7..78b686371 100644 --- a/roaring/roaring_internal_test.go +++ b/roaring/roaring_internal_test.go @@ -165,8 +165,8 @@ func TestRunCountRange(t *testing.T) { } c.add(17) - c.add(18) c.add(19) + c.add(18) cnt = c.runCountRange(1, 22) if cnt != 10 { @@ -180,6 +180,11 @@ func TestRunCountRange(t *testing.T) { if cnt != 9 { t.Fatalf("should get 9 from multiple ranges overlapping both sides, but got: %v", cnt) } + // verify that the disparate ops resulted in three separate runs + cnt = c.countRuns() + if cnt != 3 { + t.Fatalf("should get 3 total runs, but got: %v [%v]", cnt, c.runs) + } } func TestRunContains(t *testing.T) { From 1a8633f3a5eaafb03b2b5e384c7e70e04fec57d8 Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 13 Nov 2018 13:33:52 -0600 Subject: [PATCH 12/19] use roaring conventions for variable names Roaring likes to call things "a" and "b", not "1" and "2", and use "n" for length, not "l", etcetera. Adopt these conventions to make code more readable. Also drop the 'vb' value since it isn't expensive to compute and the compiler can figure out that it can reuse the value. --- roaring/roaring.go | 25 ++++++++++++------------- 1 file changed, 12 insertions(+), 13 deletions(-) diff --git a/roaring/roaring.go b/roaring/roaring.go index 6cf7eda1c..bb3c5da6e 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -1903,25 +1903,24 @@ func intersectionCount(a, b *Container) int32 { func intersectionCountArrayArray(a, b *Container) (n int32) { statsHit("intersectionCount/ArrayArray") - s1, s2 := a.array, b.array - if len(s1) == 0 || len(s2) == 0 { + ca, cb := a.array, b.array + na, nb := len(ca), len(cb) + if na == 0 || nb == 0 { return 0 } - if len(s1) > len(s2) { - s1, s2 = s2, s1 + if na > nb { + ca, cb = cb, ca + na, nb = nb, na } - l2 := len(s2) - i2 := 0 - v2 := s2[0] - for _, v1 := range s1 { - for v2 < v1 { - i2++ - if i2 >= l2 { + j := 0 + for _, va := range ca { + for cb[j] < va { + j++ + if j >= nb { return n } - v2 = s2[i2] } - if v2 == v1 { + if cb[j] == va { n++ } } From 7c82f4804604a12c1efa1c688110dc6aeead2ab0 Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 13 Nov 2018 14:06:15 -0600 Subject: [PATCH 13/19] improve testing for intersections of array/array pairs A transient bug introduced in intersectionCountArrayArray was not caught by the tests, because it would only manifest when two containers of different lengths were being compared. Also improve the testing for intersectArrayArray, even though that code hasn't been changed. --- roaring/roaring_test.go | 24 ++++++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/roaring/roaring_test.go b/roaring/roaring_test.go index 6c31b7701..f5c016ea3 100644 --- a/roaring/roaring_test.go +++ b/roaring/roaring_test.go @@ -411,13 +411,29 @@ func TestBitmap_Intersection_Empty(t *testing.T) { } func TestBitmap_IntersectArrayArray(t *testing.T) { - bm0 := roaring.NewFileBitmap(0, 1, 2683, 5005) + bm0 := roaring.NewFileBitmap(0, 1, 7, 9, 11, 2683, 5005) bm1 := roaring.NewFileBitmap(0, 2683, 2684, 5000) + expected := []uint64{0, 2683} result := bm0.Intersect(bm1) if n := result.Count(); n != 2 { t.Fatalf("unexpected n: %d", n) } + for _, e := range expected { + if !result.Contains(e) { + t.Fatalf("missing value %d", e) + } + } + // confirm that it also works going the other way + result = bm1.Intersect(bm0) + if n := result.Count(); n != 2 { + t.Fatalf("unexpected n: %d", n) + } + for _, e := range expected { + if !result.Contains(e) { + t.Fatalf("missing value %d", e) + } + } } func TestBitmap_IntersectBitmapBitmap(t *testing.T) { @@ -689,10 +705,10 @@ func TestBitmap_Flip_After(t *testing.T) { } -// Ensure bitmap can return the number of intersecting bits in two bitmaps. +// Ensure bitmap can return the number of intersecting bits in two arrays. func TestBitmap_IntersectionCount_ArrayArray(t *testing.T) { - bm0 := roaring.NewFileBitmap(0, 1, 1000001, 1000002, 1000003) - bm1 := roaring.NewFileBitmap(0, 50000, 1000001, 1000002) + bm0 := roaring.NewFileBitmap(0, 1000001, 1000002, 1000003) + bm1 := roaring.NewFileBitmap(0, 50000, 999998, 999999, 1000000, 1000001, 1000002) if n := bm0.IntersectionCount(bm1); n != 3 { t.Fatalf("unexpected n: %d", n) From e20671b2b4c8a542fd30ffb1829b836aa38dc5de Mon Sep 17 00:00:00 2001 From: Seebs Date: Tue, 13 Nov 2018 14:20:25 -0600 Subject: [PATCH 14/19] silence gometalinter I am aware that I don't actually ever use the length of a after this line of code, but if I don't correctly update it, any future change that needs that length will break mysteriously. We humbly ask gometalinter to consider the reply of counsel in _Arkell v. Pressdram_ (1971). --- roaring/roaring.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/roaring/roaring.go b/roaring/roaring.go index bb3c5da6e..9c4274df9 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -1910,7 +1910,7 @@ func intersectionCountArrayArray(a, b *Container) (n int32) { } if na > nb { ca, cb = cb, ca - na, nb = nb, na + na, nb = nb, na // nolint: ineffassign } j := 0 for _, va := range ca { From 65f478470f83d52418fa2de33541c6bd0a95b443 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 19 Nov 2018 15:00:09 -0600 Subject: [PATCH 15/19] logging cleanup - start with lowercase unless reporting error or warning --- api.go | 2 +- cluster.go | 19 +++++++------------ fragment.go | 1 - holder.go | 4 ++-- server.go | 6 ++---- server/server.go | 8 ++++---- translate.go | 7 ++----- 7 files changed, 18 insertions(+), 29 deletions(-) diff --git a/api.go b/api.go index 7a95ce76e..a55e23f7b 100644 --- a/api.go +++ b/api.go @@ -957,7 +957,7 @@ func (api *API) validateShardOwnership(indexName string, shard uint64) error { } func (api *API) indexField(indexName string, fieldName string, shard uint64) (*Index, *Field, error) { - api.server.logger.Printf("importing: %v %v %v", indexName, fieldName, shard) + api.server.logger.Debugf("importing: %v %v %v", indexName, fieldName, shard) // Find the Index. index := api.holder.Index(indexName) diff --git a/cluster.go b/cluster.go index e22781a25..fad19de1b 100644 --- a/cluster.go +++ b/cluster.go @@ -347,8 +347,6 @@ func (c *cluster) unprotectedUpdateCoordinator(n *Node) bool { // addNode adds a node to the Cluster and updates and saves the // new topology. unprotected. func (c *cluster) addNode(node *Node) error { - c.logger.Printf("add node %s to cluster on %s", node, c.Node) - // If the node being added is the coordinator, set it for this node. if node.IsCoordinator { c.Coordinator = node.ID @@ -481,7 +479,7 @@ func (c *cluster) setNodeState(state string) error { // nolint: unparam State: state, } - c.logger.Printf("Sending State %s (%s)", state, c.Coordinator) + c.logger.Printf("sending state %s (%s)", state, c.Coordinator) if err := c.sendTo(c.coordinatorNode(), ns); err != nil { return fmt.Errorf("sending node state error: err=%s", err) } @@ -970,7 +968,6 @@ func (c *cluster) close() error { } func (c *cluster) markAsJoined() { - c.logger.Printf("mark node as joined (received coordinator update)") if !c.joined { c.joined = true close(c.joining) @@ -1069,7 +1066,6 @@ func (c *cluster) unprotectedSetStateAndBroadcast(state string) error { } // Broadcast cluster status changes to the cluster. status := c.unprotectedStatus() - c.logger.Printf("broadcasting ClusterStatus: %s", status) return c.broadcaster.SendSync(status) // TODO fix c.Status } @@ -1246,7 +1242,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { return errors.Wrap(err, "merging cluster status") } - c.logger.Printf("MergeClusterStatus done, start goroutine") + c.logger.Printf("done MergeClusterStatus, start goroutine") // The actual resizing runs in a goroutine because we don't want to block // the distribution of other ResizeInstructions to the rest of the cluster. @@ -1266,7 +1262,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { if err := func() error { // Sync the schema received in the resize instruction. - c.logger.Printf("Holder ApplySchema") + c.logger.Debugf("holder applySchema") if err := c.holder.applySchema(instr.Schema); err != nil { return errors.Wrap(err, "applying schema") } @@ -1651,17 +1647,17 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { switch e.Event { case NodeJoin: - c.logger.Printf("nodeJoin of %s on %s", e.Node.URI, c.Node.URI) + c.logger.Debugf("nodeJoin of %s on %s", e.Node.URI, c.Node.URI) // Ignore the event if this is not the coordinator. if !c.isCoordinator() { return nil } return c.nodeJoin(e.Node) case NodeLeave: - c.logger.Printf("received node leave on %s: %s, uri: %v", c.Node, e.Node, e.Node.URI) c.mu.Lock() defer c.mu.Unlock() if c.unprotectedIsCoordinator() { + c.logger.Printf("received node leave: %v", e.Node) // if removeNodeBasicSorted succeeds, that means that the node was // not already removed by a removeNode request. We treat this as the // host being temporarily unavailable, and expect it to come back @@ -1673,7 +1669,6 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { err = c.unprotectedSetStateAndBroadcast(c.determineClusterState()) } } - c.logger.Printf("finished node leave on %s: %s, uri: %v", c.Node, e.Node, e.Node.URI) case NodeUpdate: c.logger.Printf("received node update event: id: %v, string: %v, uri: %v", e.Node.ID, e.Node.String(), e.Node.URI) // NodeUpdate is intentionally not implemented. @@ -1686,7 +1681,7 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { func (c *cluster) nodeJoin(node *Node) error { c.mu.Lock() defer c.mu.Unlock() - c.logger.Printf("NodeJoin event on coordinator, node: %s, id: %s", node.URI, node.ID) + c.logger.Printf("node join event on coordinator, node: %s, id: %s", node.URI, node.ID) if c.needTopologyAgreement() { // A host that is not part of the topology can't be added to the STARTING cluster. if !c.Topology.ContainsID(node.ID) { @@ -1726,7 +1721,7 @@ func (c *cluster) nodeJoin(node *Node) error { // the cluster. if cnode := c.unprotectedNodeByID(node.ID); cnode != nil { if cnode.URI != node.URI { - c.logger.Printf("Node: %v changed URI from %s to %s", cnode.ID, cnode.URI, node.URI) + c.logger.Printf("node: %v changed URI from %s to %s", cnode.ID, cnode.URI, node.URI) cnode.URI = node.URI } return c.unprotectedSetStateAndBroadcast(c.determineClusterState()) diff --git a/fragment.go b/fragment.go index 805112039..a4e4fd72d 100644 --- a/fragment.go +++ b/fragment.go @@ -1734,7 +1734,6 @@ func (f *fragment) snapshot() error { // f.mu must be locked when calling it. func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) error { // nolint: interfacer - f.Logger.Printf("fragment: snapshotting %s/%s/%s/%d", f.index, f.field, f.view, f.shard) completeMessage := fmt.Sprintf("fragment: snapshot complete %s/%s/%s/%d", f.index, f.field, f.view, f.shard) start := time.Now() defer track(start, completeMessage, f.stats, f.Logger) diff --git a/holder.go b/holder.go index 68643915e..ed8365a77 100644 --- a/holder.go +++ b/holder.go @@ -474,7 +474,7 @@ func (h *Holder) flushCaches() { } if err := fragment.FlushCache(); err != nil { - h.Logger.Printf("error flushing cache: err=%s, path=%s", err, fragment.cachePath()) + h.Logger.Printf("ERROR flushing cache: err=%s, path=%s", err, fragment.cachePath()) } } } @@ -535,7 +535,7 @@ func (h *Holder) setFileLimit() { h.Logger.Printf("ERROR checking open file limit: %s", err) } else { if oldLimit.Cur < fileLimit { - h.Logger.Printf("WARNING: Tried to set open file limit to %d, but it is %d. You may consider running \"sudo ulimit -n %d\" before starting Pilosa to avoid \"too many open files\" error. See https://www.pilosa.com/docs/administration/#open-file-limits for more information.", fileLimit, oldLimit.Cur, fileLimit) + h.Logger.Printf("WARNING: Tried to set open file limit to %d, but it is %d. You may consider running \"sudo ulimit -n %d\" before starting Pilosa to avoid \"too many open files\" error. See https://www.pilosa.com/docs/latest/administration/#open-file-limits for more information.", fileLimit, oldLimit.Cur, fileLimit) } } } diff --git a/server.go b/server.go index 24385c4ea..47dc63938 100644 --- a/server.go +++ b/server.go @@ -585,7 +585,6 @@ func (s *Server) SendSync(m Message) error { msg = append([]byte{getMessageType(m)}, msg...) for _, node := range s.cluster.nodes { node := node - s.logger.Printf("SendSync to: %s", node.URI) // Don't forward the message to ourselves. if s.uri == node.URI { continue @@ -606,7 +605,6 @@ func (s *Server) SendAsync(m Message) error { // SendTo represents an implementation of Broadcaster. func (s *Server) SendTo(to *Node, m Message) error { - s.logger.Printf("SendTo: %s", to.URI) msg, err := s.serializer.Marshal(m) if err != nil { return fmt.Errorf("marshaling message: %v", err) @@ -658,7 +656,7 @@ func (s *Server) mergeRemoteStatus(ns *NodeStatus) error { // if we don't know about a field locally, log an error because // fields should be created and synced prior to shard creation if f == nil { - s.logger.Printf("Local Field not found: %s/%s", is.Name, fs.Name) + s.logger.Printf("local field not found: %s/%s", is.Name, fs.Name) continue } if err := f.AddRemoteAvailableShards(fs.AvailableShards); err != nil { @@ -703,7 +701,7 @@ func (s *Server) monitorDiagnostics() { s.diagnostics.CheckVersion() err = s.diagnostics.Flush() if err != nil { - s.logger.Printf("Diagnostics error: %s", err) + s.logger.Printf("diagnostics error: %s", err) } } diff --git a/server/server.go b/server/server.go index 1e140d1f0..de0d3eae0 100644 --- a/server/server.go +++ b/server/server.go @@ -142,7 +142,7 @@ func (m *Command) Start() (err error) { go func() { err := m.Handler.Serve() if err != nil { - m.logger.Printf("Handler serve error: %v", err) + m.logger.Printf("handler serve error: %v", err) } }() @@ -151,7 +151,7 @@ func (m *Command) Start() (err error) { return errors.Wrap(err, "opening server") } - m.logger.Printf("Listening as %s\n", m.API.Node().URI) + m.logger.Printf("listening as %s\n", m.API.Node().URI) return nil } @@ -163,13 +163,13 @@ func (m *Command) Wait() error { signal.Notify(c, os.Interrupt, syscall.SIGTERM) select { case sig := <-c: - m.logger.Printf("Received %s; gracefully shutting down...\n", sig.String()) + m.logger.Printf("received signal '%s', gracefully shutting down...\n", sig.String()) // Second signal causes a hard shutdown. go func() { <-c; os.Exit(1) }() return errors.Wrap(m.Close(), "closing command") case <-m.done: - m.logger.Printf("Server closed externally") + m.logger.Printf("server closed externally") return nil } } diff --git a/translate.go b/translate.go index 669e1e323..2c3125ad9 100644 --- a/translate.go +++ b/translate.go @@ -195,12 +195,11 @@ func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error { } // Stop translate store replication. - s.logger.Printf("stop monitor replication") close(s.replicationClosing) s.repWG.Wait() // Set the primary node for translate store replication. - s.logger.Printf("set primary translate store to %s", ev.id) + s.logger.Debugf("set primary translate store to %s", ev.id) s.primaryID = ev.id if ev.id == "" { s.PrimaryTranslateStore = nil @@ -209,7 +208,6 @@ func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error { } // Start translate store replication. Stream from primary, if available. - s.logger.Printf("start monitor replication") if s.PrimaryTranslateStore != nil { s.replicationClosing = make(chan struct{}) s.repWG.Add(1) @@ -386,7 +384,6 @@ func (s *TranslateFile) monitorReplication() { // monitorPrimaryStoreEvents is executed in a separate goroutine and listens for changes // to the primary store assignment. func (s *TranslateFile) monitorPrimaryStoreEvents() { - s.logger.Printf("monitor primary store events") // Keep handling events until the store closes. for { select { @@ -404,7 +401,7 @@ func (s *TranslateFile) replicate(ctx context.Context) error { off := s.size() // Connect to remote primary. - s.logger.Printf("pilosa: replicating from offset %d", off) + s.logger.Debugf("pilosa: replicating from offset %d", off) rc, err := s.PrimaryTranslateStore.Reader(ctx, off) if err != nil { return err From 77598c2cc6f069a2acee6b667219051fecdec1c3 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 19 Nov 2018 16:15:33 -0600 Subject: [PATCH 16/19] dup log output onto stderr to catch panics in log file --- server/server.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/server/server.go b/server/server.go index de0d3eae0..41349b21d 100644 --- a/server/server.go +++ b/server/server.go @@ -176,14 +176,18 @@ func (m *Command) Wait() error { // setupLogger sets up the logger based on the configuration. func (m *Command) setupLogger() error { - var err error if m.Config.LogPath == "" { m.logOutput = m.Stderr } else { - m.logOutput, err = os.OpenFile(m.Config.LogPath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600) + f, err := os.OpenFile(m.Config.LogPath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600) if err != nil { return errors.Wrap(err, "opening file") } + m.logOutput = f + err = syscall.Dup2(int(f.Fd()), int(os.Stderr.Fd())) + if err != nil { + return errors.Wrap(err, "dup2ing stderr onto logfile") + } } if m.Config.Verbose { From 8bc110458568bb0e62dcb37e3a07d3c38ac03032 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 20 Nov 2018 08:56:12 -0600 Subject: [PATCH 17/19] fix fragment checksums race condition --- fragment.go | 6 +++--- fragment_internal_test.go | 17 +++++++++++++++++ 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/fragment.go b/fragment.go index a4e4fd72d..bab1f3ec0 100644 --- a/fragment.go +++ b/fragment.go @@ -1492,9 +1492,6 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor lastRowID = rowID rowSet[rowID] = struct{}{} } - - // Invalidate block checksum. - delete(f.checksums, int(rowID/HashBlockSize)) } f.mu.Lock() @@ -1518,6 +1515,9 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor // Update cache counts for all affected rows. for rowID := range rowSet { + // Invalidate block checksum. + delete(f.checksums, int(rowID/HashBlockSize)) + n := results.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth) f.cache.BulkAdd(rowID, n) } diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 4e3007957..b36bf3380 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -25,6 +25,8 @@ import ( "testing" "testing/quick" + "golang.org/x/sync/errgroup" + "github.com/davecgh/go-spew/spew" "github.com/pilosa/pilosa/pql" "github.com/pilosa/pilosa/roaring" @@ -1399,6 +1401,21 @@ func TestFragment_ImportSet(t *testing.T) { } } +func TestFragment_ConcurrentImport(t *testing.T) { + t.Run("bulkImportStandard", func(t *testing.T) { + f := mustOpenFragment("i", "f", viewStandard, 0, "") + defer f.Close() + + eg := errgroup.Group{} + eg.Go(func() error { return f.bulkImportStandard([]uint64{1, 2}, []uint64{1, 2}, &ImportOptions{}) }) + eg.Go(func() error { return f.bulkImportStandard([]uint64{3, 4}, []uint64{3, 4}, &ImportOptions{}) }) + err := eg.Wait() + if err != nil { + t.Fatalf("importing data to fragment: %v", err) + } + }) +} + // Ensure a fragment can import mutually exclusive values. func TestFragment_ImportMutex(t *testing.T) { tests := []struct { From 5458eb1656934ecd8cad846425bc6be8b907cb54 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 20 Nov 2018 13:04:08 -0600 Subject: [PATCH 18/19] fix holder.opened race with absurd lockedChan --- cluster.go | 2 +- executor.go | 2 +- holder.go | 36 ++++++++++++++++++++++++++++++++---- server.go | 2 +- 4 files changed, 35 insertions(+), 7 deletions(-) diff --git a/cluster.go b/cluster.go index fad19de1b..a70bddc77 100644 --- a/cluster.go +++ b/cluster.go @@ -1249,7 +1249,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { go func() { // Make sure the holder has opened. - <-c.holder.opened + c.holder.opened.Recv() // Prepare the return message. complete := &ResizeInstructionComplete{ diff --git a/executor.go b/executor.go index c4153ab61..9062d9176 100644 --- a/executor.go +++ b/executor.go @@ -2055,7 +2055,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64, if !opt.Remote { nodes = Nodes(e.Cluster.nodes).Clone() } else { - nodes = []*Node{e.Cluster.unprotectedNodeByID(e.Node.ID)} + nodes = []*Node{e.Cluster.nodeByID(e.Node.ID)} } // Start mapping across all primary owners. diff --git a/holder.go b/holder.go index ed8365a77..bb9e9394a 100644 --- a/holder.go +++ b/holder.go @@ -57,7 +57,7 @@ type Holder struct { NewPrimaryTranslateStore func(interface{}) TranslateStore // opened channel is closed once Open() completes. - opened chan struct{} + opened lockedChan broadcaster broadcaster @@ -79,13 +79,39 @@ type Holder struct { Logger logger.Logger } +// lockedChan looks a little ridiculous admittedly, but exists for good reason. +// The channel within is used (for example) to signal to other goroutines when +// the Holder has finished opening (via closing the channel). However, it is +// possible for the holder to be closed and then reopened, but a channel which +// is closed cannot be re-opened. We must create a new channel - this creates a +// data race with any goroutine which might be accessing the channel. To ensure +// that there is no data race on the value of the channel itself, we wrap any +// operation on it with an RWMutex so that we can guarantee that nothing is +// trying to listen on it when it gets swapped. +type lockedChan struct { + ch chan struct{} + mu sync.RWMutex +} + +func (lc *lockedChan) Close() { + lc.mu.RLock() + close(lc.ch) + lc.mu.RUnlock() +} + +func (lc *lockedChan) Recv() { + lc.mu.RLock() + <-lc.ch + lc.mu.RUnlock() +} + // NewHolder returns a new instance of Holder. func NewHolder() *Holder { return &Holder{ indexes: make(map[string]*Index), closing: make(chan struct{}), - opened: make(chan struct{}), + opened: lockedChan{ch: make(chan struct{})}, translateFile: NewTranslateFile(), NewPrimaryTranslateStore: newNopTranslateStore, @@ -159,7 +185,7 @@ func (h *Holder) Open() error { h.Stats.Open() - close(h.opened) + h.opened.Close() return nil } @@ -184,7 +210,9 @@ func (h *Holder) Close() error { } // Reset opened in case Holder needs to be reopened. - h.opened = make(chan struct{}) + h.opened.mu.Lock() + h.opened.ch = make(chan struct{}) + h.opened.mu.Unlock() return nil } diff --git a/server.go b/server.go index 47dc63938..7eb62de3e 100644 --- a/server.go +++ b/server.go @@ -628,7 +628,7 @@ func (s *Server) handleRemoteStatus(pb Message) { go func() { // Make sure the holder has opened. - <-s.holder.opened + s.holder.opened.Recv() err := s.mergeRemoteStatus(pb.(*NodeStatus)) if err != nil { From e1adb8ce5f58b1e004da085d58113c6b8222461f Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 20 Nov 2018 13:20:39 -0600 Subject: [PATCH 19/19] fix view.createFragment race --- executor.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/executor.go b/executor.go index 9062d9176..51cfae201 100644 --- a/executor.go +++ b/executor.go @@ -1664,7 +1664,7 @@ func (e *executor) executeSetRowShard(ctx context.Context, index string, c *pql. if err != nil { return false, errors.Wrap(err, "creating view") } - fragment, err = view.createFragmentIfNotExists(shard) + fragment, err = view.CreateFragmentIfNotExists(shard) if err != nil { return false, errors.Wrapf(err, "creating fragment: %d", shard) }