diff --git a/api.go b/api.go index 8b6f3249d..23c36cb0b 100644 --- a/api.go +++ b/api.go @@ -1515,6 +1515,10 @@ func (api *API) Info() serverInfo { } } +func (api *API) Inspect(ctx context.Context, req *InspectRequest) (*HolderInfo, error) { + return api.holder.Inspect(ctx, req) +} + // GetTranslateEntryReader provides an entry reader for key translation logs starting at offset. func (api *API) GetTranslateEntryReader(ctx context.Context, offsets TranslateOffsetMap) (_ TranslateEntryReader, err error) { span, ctx := tracing.StartSpanFromContext(ctx, "API.GetTranslateEntryReader") diff --git a/cluster_internal_test.go b/cluster_internal_test.go index b9e714326..059053423 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -96,7 +96,7 @@ func newIndexWithTempPath(name string) *Index { if err != nil { panic(err) } - index, err := NewIndex(path, name, DefaultPartitionN) + index, err := NewIndex(NewHolder(DefaultPartitionN), path, name) if err != nil { panic(err) } diff --git a/cmd/inspect.go b/cmd/inspect.go index 4a2e852f6..86452905d 100644 --- a/cmd/inspect.go +++ b/cmd/inspect.go @@ -45,5 +45,12 @@ Inspects a data file and provides stats. return inspector.Run(context.Background()) }, } + flags := inspectCmd.Flags() + flags.BoolVarP(&inspector.Quiet, "quiet", "q", false, "don't list details of containers") + flags.IntVarP(&inspector.Max, "max", "n", 0, "list at most max items (0 = unlimited)") + flags.StringVarP(&inspector.InspectOpts.Indexes, "index", "i", "", "filter indexes") + flags.StringVarP(&inspector.InspectOpts.Views, "view", "v", "", "filter views") + flags.StringVarP(&inspector.InspectOpts.Fields, "field", "f", "", "filter fields") + flags.StringVarP(&inspector.InspectOpts.Shards, "shard", "s", "", "filter shards") return inspectCmd } diff --git a/ctl/inspect.go b/ctl/inspect.go index 48924b89d..eb24a0597 100644 --- a/ctl/inspect.go +++ b/ctl/inspect.go @@ -16,15 +16,23 @@ package ctl import ( "context" + "encoding/binary" "fmt" + "hash/fnv" "io" + "io/ioutil" "os" + "path/filepath" + "sort" + "strconv" + "strings" "syscall" "text/tabwriter" "time" - "unsafe" + "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/internal" "github.com/pilosa/pilosa/v2/roaring" "github.com/pkg/errors" ) @@ -33,6 +41,12 @@ import ( type InspectCommand struct { // Path to data file Path string + // don't list details of objects + Quiet bool + // list only this many objects + Max int + // Filters: + InspectOpts pilosa.InspectRequest // Standard input/output *pilosa.CmdIO @@ -45,8 +59,119 @@ func NewInspectCommand(stdin io.Reader, stdout, stderr io.Writer) *InspectComman } } +type pointerContext struct { + from, to uintptr +} + +func (p *pointerContext) pretty(c roaring.ContainerInfo) string { + var pointer string + if c.Mapped { + if c.Pointer >= p.from && c.Pointer < p.to { + pointer = fmt.Sprintf("@+0x%x", c.Pointer-p.from) + } else { + pointer = fmt.Sprintf("!0x%x!", c.Pointer) + } + } else { + pointer = fmt.Sprintf("0x%x", c.Pointer) + } + return fmt.Sprintf("%s \t%d \t%d \t%s ", c.Type, c.N, c.Alloc, pointer) +} + +func (cmd *InspectCommand) PrintOps(info roaring.BitmapInfo) { + fmt.Fprintln(cmd.Stdout, " Ops:") + tw := tabwriter.NewWriter(cmd.Stdout, 0, 8, 0, '\t', 0) + fmt.Fprintf(tw, " \t%s\t%s\t%s\t\n", "TYPE", "OpN", "SIZE") + printed := 0 + for _, op := range info.OpDetails { + fmt.Fprintf(tw, "\t%s\t%d\t%d\t\n", op.Type, op.OpN, op.Size) + printed++ + if cmd.Max != 0 && printed >= cmd.Max { + break + } + } + tw.Flush() +} + +func (cmd *InspectCommand) PrintContainers(info roaring.BitmapInfo, pC pointerContext) { + fmt.Fprintln(cmd.Stdout, " Containers:") + tw := tabwriter.NewWriter(cmd.Stdout, 0, 8, 0, '\t', 0) + fmt.Fprintf(tw, " \t\tRoaring\t\t\t\tOps\t\t\t\tFlags\t\n") + fmt.Fprintf(tw, "\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t\n", "KEY", "TYPE", "N", "ALLOC", "OFFSET", "TYPE", "N", "ALLOC", "OFFSET", "FLAGS") + c1s := info.Containers + c2s := info.OpContainers + l1 := len(c1s) + l2 := len(c2s) + i1 := 0 + i2 := 0 + var c1, c2 roaring.ContainerInfo + c1.Key = ^uint64(0) + c2.Key = ^uint64(0) + c1e := false + c2e := false + if i1 < l1 { + c1 = c1s[i1] + i1++ + c1e = true + } + if i2 < l2 { + c2 = c2s[i2] + i2++ + c2e = true + } + printed := 0 + for c1e || c2e { + c1used := false + c2used := false + var key uint64 + c1fmt := "-\t\t\t" + c2fmt := "-\t\t\t" + // If c2 exists, we'll always prefer its flags, + // if it doesn't, this gets overwritten. + flags := c2.Flags + if !c2e || (c1e && c1.Key < c2.Key) { + c1fmt = pC.pretty(c1) + key = c1.Key + c1used = true + flags = c1.Flags + } else if !c1e || (c2e && c2.Key < c1.Key) { + c2fmt = pC.pretty(c2) + key = c2.Key + c2used = true + } else { + // c1e and c2e both set, and neither key is < the other. + c1fmt = pC.pretty(c1) + c2fmt = pC.pretty(c2) + key = c1.Key + c1used = true + c2used = true + } + if c1used { + if i1 < l1 { + c1 = c1s[i1] + i1++ + } else { + c1e = false + } + } + if c2used { + if i2 < l2 { + c2 = c2s[i2] + i2++ + } else { + c2e = false + } + } + fmt.Fprintf(tw, "\t%d\t%s\t%s\t%s\t\n", key, c1fmt, c2fmt, flags) + printed++ + if cmd.Max > 0 && printed >= cmd.Max { + break + } + } + tw.Flush() +} + // Run executes the inspect command. -func (cmd *InspectCommand) Run(_ context.Context) error { +func (cmd *InspectCommand) Run(ctx context.Context) error { // Open file handle. f, err := os.Open(cmd.Path) if err != nil { @@ -58,7 +183,173 @@ func (cmd *InspectCommand) Run(_ context.Context) error { if err != nil { return errors.Wrap(err, "statting file") } + if fi.IsDir() { + total := 0 + infos, err := f.Readdir(0) + if err != nil { + return err + } + if len(infos) == 0 { + return errors.New("directory contains no files") + } + names := make([]string, len(infos)) + nameToInfo := make(map[string]os.FileInfo, len(infos)) + // find numeric-only names; we'll operate on + // either those, or the whole holder if we find + // a .topology file. + n := 0 + for _, fi := range infos { + name := fi.Name() + if name == ".topology" { + return cmd.InspectHolder(ctx, cmd.Path) + } + if _, err := strconv.Atoi(name); err == nil { + names[n] = name + nameToInfo[name] = fi + n++ + } + } + if n == 0 { + return fmt.Errorf("directory contains no fragments (looking for numeric names)") + } + names = names[:n] + fmt.Fprintf(cmd.Stdout, "%s contains %d fragments:\n", cmd.Path, n) + for _, name := range names { + f2, err := os.Open(filepath.Join(cmd.Path, name)) + if err != nil { + return fmt.Errorf("opening %q: %v", name, err) + } + fmt.Fprintf(cmd.Stdout, "%s/%s:\n", cmd.Path, name) + err = cmd.InspectFile(f2, nameToInfo[name]) + total++ + f2.Close() + if err != nil { + return fmt.Errorf("inspecting %q: %v", name, err) + } + } + return nil + } + return cmd.InspectFile(f, fi) +} + +// loadTopology is copied almost exactly from pilosa/cluster.go. +func loadTopology(path string) (topology internal.Topology, myID string, err error) { + buf, err := ioutil.ReadFile(filepath.Join(path, ".topology")) + if os.IsNotExist(err) { + return topology, myID, err + } else if err != nil { + return topology, myID, errors.Wrap(err, "reading file") + } + if err := proto.Unmarshal(buf, &topology); err != nil { + return topology, myID, errors.Wrap(err, "unmarshalling") + } + sort.Slice(topology.NodeIDs, + func(i, j int) bool { + return topology.NodeIDs[i] < topology.NodeIDs[j] + }) + buf, err = ioutil.ReadFile(filepath.Join(path, ".id")) + if os.IsNotExist(err) { + return topology, myID, err + } else if err != nil { + return topology, myID, nil + } + myID = strings.TrimSpace(string(buf)) + return topology, myID, nil +} + +var partitions = make(map[string]map[uint64]int) + +func findPartition(index string, shard uint64, partitionN int) (partition int) { + var shardMap map[uint64]int + var ok bool + if shardMap, ok = partitions[index]; !ok { + shardMap = make(map[uint64]int) + partitions[index] = shardMap + } + if partition, ok = shardMap[shard]; !ok { + var buf [8]byte + binary.BigEndian.PutUint64(buf[:], shard) + + // Hash the bytes and mod by partition count. + h := fnv.New64a() + _, _ = h.Write([]byte(index)) + _, _ = h.Write(buf[:]) + partition = int(h.Sum64() % uint64(partitionN)) + shardMap[shard] = partition + } + return partition +} + +func findPartitionPath(path string, partitionN int) (int, error) { + parts := strings.Split(path, "/") + shard, err := strconv.ParseUint(parts[len(parts)-1], 10, 64) + if err != nil { + return 0, err + } + return findPartition(parts[0], shard, partitionN), nil +} + +func (cmd *InspectCommand) InspectHolder(ctx context.Context, path string) error { + holder := pilosa.NewHolder(pilosa.DefaultPartitionN) + holder.Path = path + holder.Opts.Inspect = true + holder.Opts.ReadOnly = true + err := holder.Open() + if err != nil { + return fmt.Errorf("%s: holder open: %v", path, err) + } + holderInfo, err := holder.Inspect(ctx, &cmd.InspectOpts) + if err != nil { + return fmt.Errorf("%s: inspect: %v", path, err) + } + myPartition := 0 + topology, myID, err := loadTopology(path) + if err == nil { + fmt.Fprintf(cmd.Stdout, "Cluster ID: %q\n", topology.ClusterID) + if len(topology.NodeIDs) > 1 { + fmt.Fprintf(cmd.Stdout, "Cluster of %d nodes, this node %q\n", len(topology.NodeIDs), myID) + } else { + fmt.Fprintf(cmd.Stdout, "Cluster has only one node: %q\n", myID) + } + found := false + for i := range topology.NodeIDs { + if topology.NodeIDs[i] == myID { + found = true + myPartition = i + break + } + } + if !found { + fmt.Fprintf(cmd.Stdout, "Warning: node ID %q not found in topology (%q)\n", myID, topology.NodeIDs) + } + } else { + fmt.Fprintf(cmd.Stdout, "warning: reading topology failed: %v\n", err) + } + for _, name := range holderInfo.FragmentNames { + partition, err := findPartitionPath(name, len(topology.NodeIDs)) + if err != nil { + fmt.Fprintf(cmd.Stdout, "%s: [can't find partition: %v]\n", name, err) + } else { + if partition == myPartition { + fmt.Fprintf(cmd.Stdout, "%s:\n", name) + } else { + fmt.Fprintf(cmd.Stdout, "%s: [primary node %q]\n", name, topology.NodeIDs[partition]) + } + } + details := holderInfo.FragmentInfo[name] + cmd.DisplayInfo(details.BitmapInfo) + if details.BlockChecksums != nil { + fmt.Fprintf(cmd.Stdout, " Checksums [%d total]:\n", len(details.BlockChecksums)) + for _, block := range details.BlockChecksums { + fmt.Fprintf(cmd.Stdout, " %8d: %x\n", block.ID, block.Checksum) + } + } + } + return nil +} + +func (cmd *InspectCommand) InspectFile(f *os.File, fi os.FileInfo) error { // Memory map the file. data, err := syscall.Mmap(int(f.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED) if err != nil { @@ -72,39 +363,37 @@ func (cmd *InspectCommand) Run(_ context.Context) error { }() // Attach the mmap file to the bitmap. t := time.Now() - fmt.Fprintf(cmd.Stderr, "unmarshalling bitmap...") - bm := roaring.NewBitmap() - if err := bm.UnmarshalBinary(data); err != nil { - return errors.Wrap(err, "unmarshalling") + fmt.Fprintf(cmd.Stderr, "inspecting bitmap...") + var info roaring.BitmapInfo + _, _, err = roaring.InspectBinary(data, true, &info) + fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t)) + cmd.DisplayInfo(info) + if err != nil { + return errors.Wrap(err, "inspecting") } - fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t)) + return nil +} - // Retrieve stats. - t = time.Now() - fmt.Fprintf(cmd.Stderr, "calculating stats...") - info := bm.Info() - fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t)) +func (cmd *InspectCommand) DisplayInfo(info roaring.BitmapInfo) { + pC := pointerContext{ + from: info.From, + to: info.To, + } // Print top-level info. - fmt.Fprintf(cmd.Stdout, "== Bitmap Info ==\n") - fmt.Fprintf(cmd.Stdout, "Containers: %d\n", len(info.Containers)) - fmt.Fprintf(cmd.Stdout, "Operations: %d\n", info.OpN) + fmt.Fprintf(cmd.Stdout, " Bitmap Info:\n") + fmt.Fprintf(cmd.Stdout, " Bits: %d\n", info.BitCount) + fmt.Fprintf(cmd.Stdout, " Containers: %d (%d roaring)\n", info.ContainerCount, len(info.Containers)) + fmt.Fprintf(cmd.Stdout, " Operations: %d (%d bits)\n", info.Ops, info.OpN) fmt.Fprintln(cmd.Stdout, "") // Print info for each container. - fmt.Fprintln(cmd.Stdout, "== Containers ==") - tw := tabwriter.NewWriter(cmd.Stdout, 0, 8, 0, '\t', 0) - fmt.Fprintf(tw, "%s\t%s\t% 8s \t% 8s\t%s\n", "KEY", "TYPE", "N", "ALLOC", "OFFSET") - for _, ci := range info.Containers { - fmt.Fprintf(tw, "%d\t%s\t% 8d \t% 8d \t0x%08x\n", - ci.Key, - ci.Type, - ci.N, - ci.Alloc, - uintptr(ci.Pointer)-uintptr(unsafe.Pointer(&data[0])), - ) + if !cmd.Quiet { + if info.ContainerCount > 0 { + cmd.PrintContainers(info, pC) + } + if info.Ops > 0 { + cmd.PrintOps(info) + } } - tw.Flush() - - return nil } diff --git a/ctl/inspect_test.go b/ctl/inspect_test.go index 1ff852712..2698dcc03 100644 --- a/ctl/inspect_test.go +++ b/ctl/inspect_test.go @@ -41,7 +41,7 @@ func TestInspectCommand_Run(t *testing.T) { file.Close() cm.Path = file.Name() err = cm.Run(context.Background()) - expectedError := "unmarshalling: " + expectedError := "inspecting: " if !strings.Contains(err.Error(), expectedError) { t.Fatalf("expected error '%s', got '%v'", expectedError, err) } @@ -52,7 +52,7 @@ func TestInspectCommand_Run(t *testing.T) { if err != nil { t.Fatalf("copying data: %v", err) } - if !strings.Contains(buf.String(), "unmarshalling bitmap...") { + if !strings.Contains(buf.String(), "inspecting bitmap...") { t.Fatalf("Inspect doesn't work: %s", err) } diff --git a/executor.go b/executor.go index 6891072ee..57bc2abca 100644 --- a/executor.go +++ b/executor.go @@ -381,6 +381,16 @@ func (e *executor) handlePreCalls(ctx context.Context, index string, c *pql.Call return nil } +// dumpPrecomputedCalls throws away precomputed call data. this is used so we +// can drop any large data associated with a call once we've processed +// the call. +func (e *executor) dumpPrecomputedCalls(ctx context.Context, c *pql.Call) { + for _, call := range c.Children { + e.dumpPrecomputedCalls(ctx, call) + } + c.Precomputed = nil +} + // handlePreCallChildren handles any pre-calls in the children of a given call. func (e *executor) handlePreCallChildren(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) error { for i := range c.Children { @@ -433,7 +443,7 @@ func (e *executor) execute(ctx context.Context, index string, q *pql.Query, shar // Execute each call serially. results := make([]interface{}, 0, len(q.Calls)) - for _, call := range q.Calls { + for i, call := range q.Calls { if err := validateQueryContext(ctx); err != nil { return nil, err } @@ -463,6 +473,11 @@ func (e *executor) execute(ctx context.Context, index string, q *pql.Query, shar return nil, err } results = append(results, v) + // Some Calls can have significant data associated with them + // that gets generated during processing, such as Precomputed + // values. Dumping the precomputed data, if any, lets the GC + // free the memory before we get there. + e.dumpPrecomputedCalls(ctx, q.Calls[i]) } return results, nil } diff --git a/field.go b/field.go index b746e1800..f82f03543 100644 --- a/field.go +++ b/field.go @@ -31,7 +31,6 @@ import ( "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/v2/internal" - "github.com/pilosa/pilosa/v2/logger" "github.com/pilosa/pilosa/v2/pql" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/stats" @@ -117,10 +116,6 @@ type Field struct { // Shards with data on any node in the cluster, according to this node. remoteAvailableShards *roaring.Bitmap - logger logger.Logger - - snapshotQueue snapshotQueue - translateStore TranslateStore // Instantiates new translation stores @@ -338,17 +333,17 @@ func OptFieldTypeBool() FieldOption { // that it's of the type `OptFieldType*`). This means // this function couldn't be used to set, for example, // `FieldOptions.Keys`. -func NewField(path, index, name string, opts FieldOption) (*Field, error) { +func NewField(holder *Holder, path, index, name string, opts FieldOption) (*Field, error) { err := validateName(name) if err != nil { return nil, errors.Wrap(err, "validating name") } - return newField(path, index, name, opts) + return newField(holder, path, index, name, opts) } // newField returns a new instance of field (without name validation). -func newField(path, index, name string, opts FieldOption) (*Field, error) { +func newField(holder *Holder, path, index, name string, opts FieldOption) (*Field, error) { // Apply functional option. fo := FieldOptions{} err := opts(&fo) @@ -372,7 +367,7 @@ func newField(path, index, name string, opts FieldOption) (*Field, error) { remoteAvailableShards: roaring.NewBitmap(), - logger: logger.NopLogger, + holder: holder, OpenTranslateStore: OpenInMemTranslateStore, } @@ -448,7 +443,7 @@ func (f *Field) loadAvailableShards() error { } // some other problem: if err != nil { - f.logger.Printf("available shards file present but unreadable, discarding: %v", err) + f.holder.Logger.Printf("available shards file present but unreadable, discarding: %v", err) err = os.Remove(path) if err != nil { return errors.Wrap(err, "deleting corrupt available shards list") @@ -457,7 +452,7 @@ func (f *Field) loadAvailableShards() error { } bm := roaring.NewBitmap() if err = bm.UnmarshalBinary(buf); err != nil { - f.logger.Printf("available shards file corrupt, discarding: %v", err) + f.holder.Logger.Printf("available shards file corrupt, discarding: %v", err) err = os.Remove(path) if err != nil { return errors.Wrap(err, "deleting corrupt available shards list") @@ -548,17 +543,17 @@ func (f *Field) Options() FieldOptions { func (f *Field) Open() error { if err := func() (err error) { // Ensure the field's path exists. - f.logger.Debugf("ensure field path exists: %s", f.path) + f.holder.Logger.Debugf("ensure field path exists: %s", f.path) if err := os.MkdirAll(f.path, 0777); err != nil { return errors.Wrap(err, "creating field dir") } - f.logger.Debugf("load meta file for index/field: %s/%s", f.index, f.name) + f.holder.Logger.Debugf("load meta file for index/field: %s/%s", f.index, f.name) if err := f.loadMeta(); err != nil { return errors.Wrap(err, "loading meta") } - f.logger.Debugf("load available shards for index/field: %s/%s", f.index, f.name) + f.holder.Logger.Debugf("load available shards for index/field: %s/%s", f.index, f.name) if err := f.loadAvailableShards(); err != nil { return errors.Wrap(err, "loading available shards") } @@ -570,17 +565,17 @@ func (f *Field) Open() error { } // Apply the field options loaded from meta (or set via setOptions()). - f.logger.Debugf("apply options for index/field: %s/%s", f.index, f.name) + f.holder.Logger.Debugf("apply options for index/field: %s/%s", f.index, f.name) if err := f.applyOptions(f.options); err != nil { return errors.Wrap(err, "applying options") } - f.logger.Debugf("open views for index/field: %s/%s", f.index, f.name) + f.holder.Logger.Debugf("open views for index/field: %s/%s", f.index, f.name) if err := f.openViews(); err != nil { return errors.Wrap(err, "opening views") } - f.logger.Debugf("open row attribute store for index/field: %s/%s", f.index, f.name) + f.holder.Logger.Debugf("open row attribute store for index/field: %s/%s", f.index, f.name) if err := f.rowAttrStore.Open(); err != nil { return errors.Wrap(err, "opening attrstore") } @@ -607,7 +602,7 @@ func (f *Field) Open() error { return err } - f.logger.Debugf("successfully opened field index/field: %s/%s", f.index, f.name) + f.holder.Logger.Debugf("successfully opened field index/field: %s/%s", f.index, f.name) return nil } func blockingWriteAvailableShards(fieldPath string, availableShardBytes []byte) { @@ -748,7 +743,7 @@ fileLoop: <-fieldQueue }() name := filepath.Base(fi.Name()) - f.logger.Debugf("open index/field/view: %s/%s/%s", f.index, f.name, fi.Name()) + f.holder.Logger.Debugf("open index/field/view: %s/%s/%s", f.index, f.name, fi.Name()) view := f.newView(f.viewPath(name), name) if err := view.open(); err != nil { return fmt.Errorf("opening view: view=%s, err=%s", view.name, err) @@ -770,7 +765,7 @@ fileLoop: } view.rowAttrStore = f.rowAttrStore - f.logger.Debugf("add index/field/view to field.viewMap: %s/%s/%s", f.index, f.name, view.name) + f.holder.Logger.Debugf("add index/field/view to field.viewMap: %s/%s/%s", f.index, f.name, view.name) mu.Lock() f.viewMap[view.name] = view mu.Unlock() @@ -1183,14 +1178,10 @@ func (f *Field) createViewIfNotExistsBase(name string) (*view, bool, error) { } func (f *Field) newView(path, name string) *view { - view := newView(path, f.index, f.name, name, f.options) - view.logger = f.logger + view := newView(f.holder, path, f.index, f.name, name, f.options) view.rowAttrStore = f.rowAttrStore view.stats = f.Stats view.broadcaster = f.broadcaster - if f.snapshotQueue != nil { - view.snapshotQueue = f.snapshotQueue - } return view } diff --git a/field_internal_test.go b/field_internal_test.go index db4a3b5e7..296e54be0 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -205,7 +205,7 @@ func NewTestField(t *testing.T, opts FieldOption) *TestField { if err != nil { t.Fatal(err) } - field, err := NewField(path, "i", "f", opts) + field, err := NewField(NewHolder(DefaultPartitionN), path, "i", "f", opts) if err != nil { t.Fatal(err) } @@ -235,7 +235,7 @@ func (f *TestField) Reopen() error { } path, index, name := f.Path(), f.Index(), f.Name() - f.Field, err = NewField(path, index, name, OptFieldTypeDefault()) + f.Field, err = NewField(NewHolder(DefaultPartitionN), path, index, name, OptFieldTypeDefault()) if err != nil { return err } @@ -730,7 +730,7 @@ func TestDecimalField_MinMaxBoundaries(t *testing.T) { }, } { t.Run("minmax"+strconv.Itoa(i), func(t *testing.T) { - _, err := NewField("no-path", "i", "f", OptFieldTypeDecimal(test.scale, test.min, test.max)) + _, err := NewField(NewHolder(DefaultPartitionN), "no-path", "i", "f", OptFieldTypeDecimal(test.scale, test.min, test.max)) if err != nil && test.expErr { if !strings.Contains(err.Error(), "is not supported") { t.Fatal(err) diff --git a/field_test.go b/field_test.go index 40b64728a..ab24b466f 100644 --- a/field_test.go +++ b/field_test.go @@ -144,7 +144,7 @@ func TestField_NameRestriction(t *testing.T) { if err != nil { panic(err) } - field, err := pilosa.NewField(path, "i", ".meta", pilosa.OptFieldTypeDefault()) + field, err := pilosa.NewField(pilosa.NewHolder(pilosa.DefaultPartitionN), path, "i", ".meta", pilosa.OptFieldTypeDefault()) if field != nil { t.Fatalf("unexpected field name %s", err) } @@ -177,13 +177,13 @@ func TestField_NameValidation(t *testing.T) { panic(err) } for _, name := range validFieldNames { - _, err := pilosa.NewField(path, "i", name, pilosa.OptFieldTypeDefault()) + _, err := pilosa.NewField(pilosa.NewHolder(pilosa.DefaultPartitionN), path, "i", name, pilosa.OptFieldTypeDefault()) if err != nil { t.Fatalf("unexpected field name: %s %s", name, err) } } for _, name := range invalidFieldNames { - _, err := pilosa.NewField(path, "i", name, pilosa.OptFieldTypeDefault()) + _, err := pilosa.NewField(pilosa.NewHolder(pilosa.DefaultPartitionN), path, "i", name, pilosa.OptFieldTypeDefault()) if err == nil { t.Fatalf("expected error on field name: %s", name) } diff --git a/fragment.go b/fragment.go index 3258d6f00..61ac78313 100644 --- a/fragment.go +++ b/fragment.go @@ -106,6 +106,10 @@ type fragment struct { field string view string shard uint64 + + // parent holder, used to find snapshot queue, etc. + holder *Holder + // debugging tool: addresses of current and previous maps prevdata, currdata struct{ from, to uintptr } @@ -120,6 +124,7 @@ type fragment struct { snapshotCond sync.Cond snapshotErr error // error yielded by the last snapshot operation snapshotStamp time.Time // timestamp of last snapshot + open bool // is this fragment actually open? // Cache for row counts. CacheType string // passed in by field @@ -153,11 +158,11 @@ type fragment struct { stats stats.StatsClient - snapshotQueue snapshotQueue + bitmapInfo *roaring.BitmapInfo } // newFragment returns a new instance of Fragment. -func newFragment(path, index, field, view string, shard uint64, flags byte) *fragment { +func newFragment(holder *Holder, path, index, field, view string, shard uint64, flags byte) *fragment { f := &fragment{ path: path, index: index, @@ -168,11 +173,10 @@ func newFragment(path, index, field, view string, shard uint64, flags byte) *fra CacheType: DefaultCacheType, CacheSize: DefaultCacheSize, - Logger: logger.NopLogger, + holder: holder, MaxOpN: defaultFragmentMaxOpN, - stats: stats.NopStatsClient, - snapshotQueue: defaultSnapshotQueue, + stats: stats.NopStatsClient, } f.snapshotCond = sync.Cond{L: &f.mu} return f @@ -181,6 +185,23 @@ func newFragment(path, index, field, view string, shard uint64, flags byte) *fra // cachePath returns the path to the fragment's cache data. func (f *fragment) cachePath() string { return f.path + cacheExt } +type FragmentInfo struct { + BitmapInfo roaring.BitmapInfo + BlockChecksums []FragmentBlock `json:"BlockChecksums,omitempty"` +} + +func (f *fragment) inspect(params InspectRequestParams) (fi FragmentInfo) { + if f.bitmapInfo == nil { + fi.BitmapInfo = f.storage.Info(params.Containers) + } else { + fi.BitmapInfo = *f.bitmapInfo + } + if params.Checksum { + fi.BlockChecksums = f.Blocks() + } + return fi +} + // Open opens the underlying storage. func (f *fragment) Open() error { f.mu.Lock() @@ -188,13 +209,13 @@ func (f *fragment) Open() error { if err := func() error { // Initialize storage in a function so we can close if anything goes wrong. - f.Logger.Debugf("open storage for index/field/view/fragment: %s/%s/%s/%d", f.index, f.field, f.view, f.shard) + f.holder.Logger.Debugf("open storage for index/field/view/fragment: %s/%s/%s/%d", f.index, f.field, f.view, f.shard) if err := f.openStorage(true); err != nil { return errors.Wrap(err, "opening storage") } // Fill cache with rows persisted to disk. - f.Logger.Debugf("open cache for index/field/view/fragment: %s/%s/%s/%d", f.index, f.field, f.view, f.shard) + f.holder.Logger.Debugf("open cache for index/field/view/fragment: %s/%s/%s/%d", f.index, f.field, f.view, f.shard) if err := f.openCache(); err != nil { e2 := f.closeStorage() if e2 != nil { @@ -213,8 +234,9 @@ func (f *fragment) Open() error { f.close() return err } + f.open = true - f.Logger.Debugf("successfully opened index/field/view/fragment: %s/%s/%s/%d", f.index, f.field, f.view, f.shard) + f.holder.Logger.Debugf("successfully opened index/field/view/fragment: %s/%s/%s/%d", f.index, f.field, f.view, f.shard) return nil } @@ -222,6 +244,9 @@ func (f *fragment) Open() error { // get no data. It tries to write the current storage to the provided file, // which is assumed to be the file they didn't get any data from. func (f *fragment) emptyStorage(file *os.File) (bool, error) { + if f.holder.Opts.ReadOnly { + return false, errors.New("can't flush/create storage for read-only holder") + } // No data. We'll mark this for no mapping, clear any existing // mapped containers, and set the Source to nil. We also have no // ops. @@ -273,9 +298,12 @@ func (f *fragment) importStorage(data []byte, file *os.File, newGen generation, } return false, fmt.Errorf("unmarshal storage: file=%s, err=%s", file.Name(), err) } - f.Logger.Printf("warning: unmarshal storage, file=%s, err=%v", file.Name(), err) + f.holder.Logger.Printf("warning: unmarshal storage, file=%s, err=%v", file.Name(), err) trunc, ok := cause.(roaring.FileShouldBeTruncatedError) - if ok { + if ok && !f.holder.Opts.ReadOnly { + // if the holder is ReadOnly, we silently ignore the "advisory" + // error. This may be a bad idea. + // generation code looks for a FileShouldBeTruncatedError return false, trunc } @@ -299,7 +327,7 @@ func (f *fragment) applyStorage(data []byte, file *os.File, newGen generation, m if file != nil { fi, err := file.Stat() if err != nil { - f.Logger.Printf("trying to apply new storage to existing bitmap, stat failed: %v", err) + f.holder.Logger.Printf("trying to apply new storage to existing bitmap, stat failed: %v", err) } if err == nil && fi != nil && fi.Size() == 0 { return f.emptyStorage(file) @@ -335,6 +363,12 @@ func (f *fragment) applyStorage(data []byte, file *os.File, newGen generation, m return mapped, err } +func (f *fragment) inspectStorage(data []byte, file *os.File, newGen generation, mapped bool) (didMap bool, err error) { + f.bitmapInfo = &roaring.BitmapInfo{} + f.storage, didMap, err = roaring.InspectBinary(data, mapped, f.bitmapInfo) + return didMap, err +} + // openStorage opens the storage bitmap. // // This has been massively reworked recently, and now hands a lot of @@ -353,13 +387,20 @@ func (f *fragment) openStorage(unmarshalData bool) error { } f.rowCache = &simpleCache{make(map[uint64]*Row)} var storageOp func([]byte, *os.File, generation, bool) (bool, error) - if unmarshalData { - storageOp = f.importStorage + if f.holder.Opts.Inspect { + // note that this will unmarshal even if we already have + // storage; when Inspect is on for a holder, we actually want + // to be able to report this. + storageOp = f.inspectStorage } else { - storageOp = f.applyStorage + if unmarshalData { + storageOp = f.importStorage + } else { + storageOp = f.applyStorage + } } var err error - f.gen, err = newGeneration(f.gen, f.path, unmarshalData, storageOp, f.Logger) + f.gen, err = newGeneration(f.gen, f.path, unmarshalData, storageOp, f.holder.Logger) if f.gen != nil { scratchData := f.gen.Bytes() f.prevdata = f.currdata @@ -407,7 +448,7 @@ func (f *fragment) openCache() error { // Unmarshal cache data. var pb internal.Cache if err := proto.Unmarshal(buf, &pb); err != nil { - f.Logger.Printf("error unmarshaling cache data, skipping: path=%s, err=%s", path, err) + f.holder.Logger.Printf("error unmarshaling cache data, skipping: path=%s, err=%s", path, err) return nil } @@ -429,19 +470,22 @@ func (f *fragment) Close() error { for f.snapshotPending { f.snapshotCond.Wait() } + // Note: snapshots won't progress on a closed fragment, so we + // wait until after a possible pending snapshot to close. + f.open = false return f.close() } func (f *fragment) close() error { // Flush cache if closing gracefully. if err := f.flushCache(); err != nil { - f.Logger.Printf("fragment: error flushing cache on close: err=%s, path=%s", err, f.path) + f.holder.Logger.Printf("fragment: error flushing cache on close: err=%s, path=%s", err, f.path) return errors.Wrap(err, "flushing cache") } // Close underlying storage. if err := f.closeStorage(); err != nil { - f.Logger.Printf("fragment: error closing storage: err=%s, path=%s", err, f.path) + f.holder.Logger.Printf("fragment: error closing storage: err=%s, path=%s", err, f.path) return errors.Wrap(err, "closing storage") } @@ -688,7 +732,8 @@ func (f *fragment) unprotectedSetRow(row *Row, rowID uint64) (changed bool, err f.rowCache.Add(rowID, nil) // Snapshot storage. - f.snapshotQueue.Enqueue(f) + f.holder.SnapshotQueue.Enqueue(f) + f.stats.Count("setRow", 1, 1.0) return changed, nil } @@ -728,7 +773,7 @@ func (f *fragment) unprotectedClearRow(rowID uint64) (changed bool, err error) { f.rowCache.Add(rowID, nil) // Snapshot storage. - f.snapshotQueue.Enqueue(f) + f.holder.SnapshotQueue.Enqueue(f) return changed, nil } @@ -2006,11 +2051,11 @@ func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct // we got an error. it's possible that the error indicates that something went wrong. mappedIn, mappedOut, unmappedIn, errs, e2 := f.storage.SanityCheckMapping(f.currdata.from, f.currdata.to) if errs != 0 { - f.Logger.Printf("transaction failed on %s. storage has %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total, last %v", + f.holder.Logger.Printf("transaction failed on %s. storage has %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total, last %v", f.path, mappedIn, mappedOut, unmappedIn, errs, e2) if f.prevdata.from != f.currdata.from { mappedIn, mappedOut, unmappedIn, errs, e2 = f.storage.SanityCheckMapping(f.prevdata.from, f.prevdata.to) - f.Logger.Printf("with previous map, storage would have %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total, last %v", + f.holder.Logger.Printf("with previous map, storage would have %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total, last %v", mappedIn, mappedOut, unmappedIn, errs, e2) } } @@ -2164,7 +2209,7 @@ func (f *fragment) importValue(columnIDs []uint64, values []int64, bitDepth uint // in theory, this should probably have been queued anyway, but if enough // of the bits matched existing bits, we'll be under our opN estimate, and // we want to ensure that the snapshot happens. - return f.snapshotQueue.Immediate(f) + return f.holder.SnapshotQueue.Immediate(f) } // importRoaring imports from the official roaring data format defined at @@ -2252,7 +2297,7 @@ func (f *fragment) incrementOpN(changed int) { f.opN += changed f.ops++ if f.opN > f.MaxOpN { - f.snapshotQueue.Enqueue(f) + f.holder.SnapshotQueue.Enqueue(f) } } @@ -2275,6 +2320,9 @@ func track(start time.Time, message string, stats stats.StatsClient, logger logg // snapshot does the actual snapshot operation. it does not check or care // about f.snapshotPending. func (f *fragment) snapshot() (err error) { + if !f.open { + return errors.New("snapshot request on closed fragment") + } wouldPanic := debug.SetPanicOnFault(true) defer func() { debug.SetPanicOnFault(wouldPanic) @@ -2286,7 +2334,7 @@ func (f *fragment) snapshot() (err error) { // we can't see the actual values that were used to generate this, probably. if e2.Error() == "runtime error: invalid memory address or nil pointer dereference" { mappedIn, mappedOut, unmappedIn, errs, _ := f.storage.SanityCheckMapping(f.currdata.from, f.currdata.to) - f.Logger.Printf("transaction failed on %s. storage has %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total", + f.holder.Logger.Printf("transaction failed on %s. storage has %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total", f.path, mappedIn, mappedOut, unmappedIn, errs) } } else { @@ -2306,7 +2354,7 @@ func (f *fragment) snapshot() (err error) { func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) (n int64, err error) { // nolint: interfacer 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) + defer track(start, completeMessage, f.stats, f.holder.Logger) // Create a temporary file to snapshot to. snapshotPath := f.path + snapshotExt @@ -3096,7 +3144,7 @@ func (s *fragmentSyncer) syncBlockFromPrimary(id int) error { // the primary node. nodes := s.Cluster.shardNodes(f.index, f.shard) if s.Node.ID != nodes[0].ID { - f.Logger.Debugf("non-primary replica expecting sync from primary: %s, index=%s, field=%s, shard=%d", nodes[0].ID, f.index, f.field, f.shard) + f.holder.Logger.Debugf("non-primary replica expecting sync from primary: %s, index=%s, field=%s, shard=%d", nodes[0].ID, f.index, f.field, f.shard) return nil } diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 5912eb132..fa1ffbeec 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -1477,7 +1477,7 @@ func BenchmarkFragment_Blocks(b *testing.B) { } // Open the fragment specified by the path. - f := newFragment(*FragmentPath, "i", "f", viewStandard, 0, 0) + f := newFragment(NewHolder(DefaultPartitionN), *FragmentPath, "i", "f", viewStandard, 0, 0) if err := f.Open(); err != nil { b.Fatal(err) } @@ -2006,7 +2006,7 @@ func BenchmarkFragment_Snapshot(b *testing.B) { b.ReportAllocs() // Open the fragment specified by the path. - f := newFragment(*FragmentPath, "i", "f", viewStandard, 0, 0) + f := newFragment(NewHolder(DefaultPartitionN), *FragmentPath, "i", "f", viewStandard, 0, 0) if err := f.Open(); err != nil { b.Fatal(err) } @@ -2121,7 +2121,7 @@ func BenchmarkImportRoaring(b *testing.B) { // care whether this succeeds, // but if it's happening we want // it to be done. - _ = f.snapshotQueue.Await(f) + _ = defaultSnapshotQueue.Await(f) f.Clean(b) b.Fatalf("import error: %v", err) } @@ -2162,7 +2162,7 @@ func BenchmarkImportRoaringConcurrent(b *testing.B) { err := frags[j].importRoaringT(data[j], false) // error unimportant if it happened, but we want // any snapshots to have finished. - _ = frags[j].snapshotQueue.Await(frags[j]) + _ = defaultSnapshotQueue.Await(frags[j]) return err }) } @@ -2203,7 +2203,7 @@ func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) { if err != nil { b.Fatalf("importing roaring: %v", err) } - err = frags[j].snapshotQueue.Immediate(frags[j]) + err = defaultSnapshotQueue.Immediate(frags[j]) if err != nil { b.Fatalf("snapshot after import: %v", err) } @@ -2214,7 +2214,7 @@ func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) { j := j eg.Go(func() error { err := frags[j].importRoaringT(updata, false) - err2 := frags[j].snapshotQueue.Await(frags[j]) + err2 := defaultSnapshotQueue.Await(frags[j]) if err == nil { err = err2 } @@ -2282,7 +2282,7 @@ func BenchmarkImportRoaringUpdate(b *testing.B) { if err != nil { b.Errorf("import error: %v", err) } - err = f.snapshotQueue.Immediate(f) + err = defaultSnapshotQueue.Immediate(f) if err != nil { b.Errorf("snapshot after import error: %v", err) } @@ -2292,7 +2292,7 @@ func BenchmarkImportRoaringUpdate(b *testing.B) { f.Clean(b) b.Errorf("import error: %v", err) } - err = f.snapshotQueue.Await(f) + err = defaultSnapshotQueue.Await(f) if err != nil { b.Errorf("snapshot after import error: %v", err) } @@ -2391,7 +2391,7 @@ func BenchmarkImportIntoLargeFragment(b *testing.B) { } origF.Close() fi.Close() - nf := newFragment(fi.Name(), "i", "f", viewStandard, 0, 0) + nf := newFragment(NewHolder(DefaultPartitionN), fi.Name(), "i", "f", viewStandard, 0, 0) err = nf.Open() if err != nil { b.Fatalf("opening fragment: %v", err) @@ -2428,7 +2428,7 @@ func BenchmarkImportRoaringIntoLargeFragment(b *testing.B) { } origF.Close() fi.Close() - nf := newFragment(fi.Name(), "i", "f", viewStandard, 0, 0) + nf := newFragment(NewHolder(DefaultPartitionN), fi.Name(), "i", "f", viewStandard, 0, 0) err = nf.Open() if err != nil { b.Fatalf("opening fragment: %v", err) @@ -2607,17 +2607,23 @@ func (f *fragment) sanityCheck(t testing.TB) { func (f *fragment) Clean(t testing.TB) { f.mu.Lock() - err := f.snapshotQueue.Await(f) - f.mu.Unlock() - if err != nil { - t.Fatalf("snapshot failed before sanity check: %v", err) - } - f.sanityCheck(t) - if f.storage != nil && f.storage.Source != nil { - if f.storage.Source.Dead() { - t.Fatalf("cleaning up fragment %s, source %s, source already dead", f.path, f.storage.Source.ID()) + // we need to ensure that we unlock the mutex before terminating + // the clean operation, but we need it held during the sanity + // check or else, in some cases, the background snapshot queue + // can decide to pick it up. + func() { + defer f.mu.Unlock() + err := defaultSnapshotQueue.Await(f) + if err != nil { + t.Fatalf("snapshot failed before sanity check: %v", err) } - } + f.sanityCheck(t) + if f.storage != nil && f.storage.Source != nil { + if f.storage.Source.Dead() { + t.Fatalf("cleaning up fragment %s, source %s, source already dead", f.path, f.storage.Source.ID()) + } + } + }() errc := f.Close() // prevent double-closes of generation during testing. f.gen = nil @@ -2626,10 +2632,6 @@ func (f *fragment) Clean(t testing.TB) { if errc != nil || errf != nil { t.Fatal("cleaning up fragment: ", errc, errf, errp) } - if f.snapshotQueue != nil { - f.snapshotQueue.Stop() - f.snapshotQueue = nil - } // not all fragments have cache files if errp != nil && !os.IsNotExist(errp) { t.Fatalf("cleaning up fragment cache: %v", errp) @@ -2649,10 +2651,6 @@ func (f *fragment) CleanKeep(t testing.TB) { if errc != nil { t.Fatal("closing fragment: ", errc, errp) } - if f.snapshotQueue != nil { - f.snapshotQueue.Stop() - f.snapshotQueue = nil - } // not all fragments have cache files if errp != nil && !os.IsNotExist(errp) { t.Fatalf("cleaning up fragment cache: %v", errp) @@ -2668,6 +2666,12 @@ func mustOpenBSIFragment(index, field, view string, shard uint64) *fragment { return mustOpenFragmentFlags(index, field, view, shard, "", 1) } +var testHolder = NewHolder(DefaultPartitionN) + +func init() { + testHolder.SnapshotQueue = newSnapshotQueue(1, 1, nil) +} + // mustOpenFragment returns a new instance of Fragment with a temporary path. func mustOpenFragmentFlags(index, field, view string, shard uint64, cacheType string, flags byte) *fragment { file, err := ioutil.TempFile(*TempDir, "pilosa-fragment-") @@ -2680,12 +2684,12 @@ func mustOpenFragmentFlags(index, field, view string, shard uint64, cacheType st cacheType = DefaultCacheType } - f := newFragment(file.Name(), index, field, view, shard, flags) + f := newFragment(testHolder, file.Name(), index, field, view, shard, flags) + f.CacheType = cacheType f.RowAttrStore = &memAttrStore{ store: make(map[uint64]map[string]interface{}), } - f.snapshotQueue = newSnapshotQueue(1, 1, nil) if err := f.Open(); err != nil { panic(err) @@ -3179,7 +3183,7 @@ func TestUnionInPlaceMapped(t *testing.T) { // it's used only in computation of things that usually don't go to // disk, which is why we handle this specially in testing and not // generically. - err = f.snapshotQueue.Immediate(f) + err = defaultSnapshotQueue.Immediate(f) if err != nil { t.Fatalf("snapshot after union-in-place: %v", err) } @@ -3332,7 +3336,6 @@ func TestImportClearRestart(t *testing.T) { cols: []uint64{1, 1, 1, 1, 1, 1}, }, } - for i, test := range tests { for _, maxOpN := range []int{0, 10000} { t.Run(fmt.Sprintf("%dMaxOpN%d", i, maxOpN), func(t *testing.T) { @@ -3385,7 +3388,7 @@ func TestImportClearRestart(t *testing.T) { check(t, f, exp) - f2 := newFragment(f.path, "i", "f", viewStandard, 0, 0) + f2 := newFragment(NewHolder(DefaultPartitionN), f.path, "i", "f", viewStandard, 0, 0) f2.MaxOpN = maxOpN f2.CacheType = f.CacheType @@ -3419,7 +3422,7 @@ func TestImportClearRestart(t *testing.T) { check(t, f2, exp) - f3 := newFragment(f2.path, "i", "f", viewStandard, 0, 0) + f3 := newFragment(NewHolder(DefaultPartitionN), f2.path, "i", "f", viewStandard, 0, 0) f3.MaxOpN = maxOpN f3.CacheType = f.CacheType diff --git a/generation.go b/generation.go index 7c1361b1d..f3cc6ae88 100644 --- a/generation.go +++ b/generation.go @@ -263,7 +263,7 @@ func (m *mmapGeneration) openFile() (shouldClose bool, err error) { } // do we actually want this in every openFile? I don't know. if err := syscall.Flock(int(m.file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { - m.file.Close() + _ = syswrap.CloseFile(m.file) m.file = nil return false, fmt.Errorf("flock: %s", err) } @@ -333,6 +333,7 @@ func newGeneration(existing generation, path string, readData bool, setup func([ m := mmapGeneration{path: path, logger: logger} if existing != nil { m.generation = existing.Generation() + 1 + m.retries = existing.(*mmapGeneration).retries // we might keep a previous generation around just for its generation count. if !existing.Dead() { defer existing.Done() diff --git a/generation_test.go b/generation_test.go index 75444fcf7..7df71fc98 100644 --- a/generation_test.go +++ b/generation_test.go @@ -39,14 +39,14 @@ func TestGenerationPanic(t *testing.T) { } prevData = f.gen.(*mmapGeneration).data f.mu.Lock() - _ = f.snapshotQueue.Immediate(f) + _ = defaultSnapshotQueue.Immediate(f) f.mu.Unlock() runtime.GC() for i := 0; i < (f.MaxOpN / 2); i++ { _, _ = f.setBit(0, uint64(i*32)+23) } f.mu.Lock() - f.snapshotQueue.Await(f) + defaultSnapshotQueue.Await(f) f.mu.Unlock() runtime.GC() newData := f.gen.(*mmapGeneration).data diff --git a/handler.go b/handler.go index 17156245b..bb5aa245d 100644 --- a/handler.go +++ b/handler.go @@ -287,3 +287,31 @@ type TranslateIDsRequest struct { type TranslateIDsResponse struct { Keys []string } + +// InspectRequestParams represents the parts of an InspectRequest that +// aren't generic holder filtering attributes. +type InspectRequestParams struct { + Containers bool // include container details + Checksum bool // perform checksums +} + +// InspectRequest represents a request for a possibly-partial +// holder inspection, using a provided holder filter and inspect-specific +// parameters. +type InspectRequest struct { + HolderFilterParams + InspectRequestParams +} + +// InspectResponse contains the structured results for an InspectRequest. +// It may some day be expanded to include metadata about views or indexes. +type InspectResponse struct { + Fragments []struct { + Index string + Field string + View string + Shard int64 + Path string + Info *FragmentInfo + } +} diff --git a/holder.go b/holder.go index de3cd0701..d1f457371 100644 --- a/holder.go +++ b/holder.go @@ -21,7 +21,9 @@ import ( "os" "path" "path/filepath" + "regexp" "sort" + "strconv" "strings" "sync" "syscall" @@ -77,9 +79,8 @@ type Holder struct { // The interval at which the cached row ids are persisted to disk. cacheFlushInterval time.Duration - Logger logger.Logger - - snapshotQueue snapshotQueue + Logger logger.Logger + SnapshotQueue SnapshotQueue // Instantiates new translation stores OpenTranslateStore OpenTranslateStoreFunc @@ -102,6 +103,18 @@ type Holder struct { // needs to be queued and completed after all indexes // have opened. opening bool + + Opts HolderOpts +} + +type HolderOpts struct { + // ReadOnly indicates that this holder's contents should not produce + // disk writes under any circumstances. It must be set before Open + // is called, and changing it is not supported. + ReadOnly bool + // If Inspect is set, we'll try to obtain additional information + // about fragments when opening them. + Inspect bool } func (h *Holder) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error) { @@ -169,9 +182,285 @@ func NewHolder(partitionN int) *Holder { translationSyncer: NopTranslationSyncer, Logger: logger.NopLogger, + + SnapshotQueue: defaultSnapshotQueue, } } +type HolderInfo struct { + FragmentInfo map[string]FragmentInfo + FragmentNames []string +} + +type regexpList []*regexp.Regexp + +func newRegexpList(regexes string) (results regexpList, err error) { + if regexes == "" { + return nil, nil + } + for _, sub := range strings.Split(regexes, ",") { + re, err := regexp.Compile(sub) + if err != nil { + return nil, err + } + results = append(results, re) + } + return results, nil +} + +func (rl regexpList) Match(haystack string) bool { + if rl == nil { + return true + } + for _, re := range rl { + if re.MatchString(haystack) { + return true + } + } + return false +} + +// shardRange represents a series of shards +type shardRange struct { + min, max uint64 +} + +type shardRangeList []shardRange + +func newShardRangeList(shards string) (results shardRangeList, err error) { + if shards == "" { + return nil, nil + } + for _, sub := range strings.Split(shards, ",") { + var sr shardRange + minMax := strings.Split(sub, "-") + if len(minMax) > 2 { + return nil, fmt.Errorf("invalid range %q", sub) + } + sr.min, err = strconv.ParseUint(minMax[0], 10, 64) + if err != nil { + return nil, err + } + sr.max = sr.min + if len(minMax) == 2 { + sr.max, err = strconv.ParseUint(minMax[0], 10, 64) + if err != nil { + return nil, err + } + } + if sr.max < sr.min { + return nil, fmt.Errorf("invalid range %q: max < min", sub) + } + results = append(results, sr) + } + return results, nil +} + +func (sl shardRangeList) Match(shard uint64) bool { + if sl == nil { + return true + } + for _, sr := range sl { + if shard >= sr.min && shard <= sr.max { + return true + } + } + return false +} + +// HolderFilter represents something that potentially filters out +// parts of a holder, indicating whether or not to process them, +// or recurse into them. It is permissible to recurse a thing +// without processing it, or process it without recursing it. +// For instance, something looking to accumulate statistics +// about views might return (true, false) from CheckView, +// while a fragment scanning operation would return (false, true) +// from everything above CheckFrag. +type HolderFilter interface { + CheckIndex(iname string) (process bool, recurse bool) + CheckField(iname, fname string) (process bool, recurse bool) + CheckView(iname, fname, vname string) (process bool, recurse bool) + CheckFragment(iname, fname, vname string, shard uint64) (process bool) +} + +// HolderFilterAll is a placeholder type which always returns true for the +// check functions. You can embed it to make a HolderOperator which processes +// everything. +type HolderFilterAll struct{} + +func (HolderFilterAll) CheckIndex(string) (bool, bool) { + return true, true +} + +func (HolderFilterAll) CheckField(string, string) (bool, bool) { + return true, true +} + +func (HolderFilterAll) CheckView(string, string, string) (bool, bool) { + return true, true +} + +func (HolderFilterAll) CheckFragment(string, string, string, uint64) bool { + return true +} + +// HolderProcessNone is a placeholder type which does nothing for the +// process functions. You can embed it to make a HolderOperator which +// does nothing, or embed it and provide your own ProcessFragment to +// do just that. +type HolderProcessNone struct{} + +func (HolderProcessNone) ProcessIndex(*Index) error { + return nil +} + +func (HolderProcessNone) ProcessField(*Field) error { + return nil +} + +func (HolderProcessNone) ProcessView(*view) error { + return nil +} + +func (HolderProcessNone) ProcessFragment(*fragment) error { + return nil +} + +// HolderProcess represents something that has operations which can be +// performed on indexes, fields, views, and/or fragments. +type HolderProcess interface { + ProcessIndex(*Index) error + ProcessField(*Field) error + ProcessView(*view) error + ProcessFragment(*fragment) error +} + +// HolderOperator is both a filter and a process. This is the general +// form of "I want to do something to some part of a holder." +type HolderOperator interface { + HolderFilter + HolderProcess +} + +var _ HolderOperator = (*holderInspector)(nil) + +type HolderFilterParams struct { + Indexes string + Fields string + Views string + Shards string +} + +type holderFilterFull struct { + HolderFilterParams + indexRegexps regexpList + fieldRegexps regexpList + viewRegexps regexpList + shardRanges shardRangeList +} + +type inspectRequestFull struct { + HolderFilter + params InspectRequestParams +} + +func (i *holderFilterFull) CheckIndex(iname string) (process, recurse bool) { + return true, i.indexRegexps.Match(iname) +} + +func (i *holderFilterFull) CheckField(iname, fname string) (process, recurse bool) { + return true, i.fieldRegexps.Match(fname) +} + +func (i *holderFilterFull) CheckView(iname, fname, vname string) (process, recurse bool) { + return true, i.viewRegexps.Match(vname) +} + +func (i *holderFilterFull) CheckFragment(iname, fname, vname string, shard uint64) (process bool) { + return i.shardRanges.Match(shard) +} + +func NewHolderFilter(params HolderFilterParams) (result HolderFilter, err error) { + filter := &holderFilterFull{ + HolderFilterParams: params, + } + filter.indexRegexps, err = newRegexpList(params.Indexes) + if err != nil { + return nil, err + } + filter.fieldRegexps, err = newRegexpList(params.Fields) + if err != nil { + return nil, err + } + filter.viewRegexps, err = newRegexpList(params.Views) + if err != nil { + return nil, err + } + filter.shardRanges, err = newShardRangeList(params.Shards) + if err != nil { + return nil, err + } + return filter, nil +} + +func expandInspectRequest(req *InspectRequest) (*inspectRequestFull, error) { + filter, err := NewHolderFilter(req.HolderFilterParams) + if err != nil { + return nil, err + } + irf := &inspectRequestFull{ + HolderFilter: filter, + params: req.InspectRequestParams, + } + return irf, nil +} + +type holderInspector struct { + *inspectRequestFull + pathParts [3]string + path string + hi *HolderInfo +} + +func (h *holderInspector) ProcessIndex(i *Index) error { + h.pathParts[0] = i.name + return nil +} + +func (h *holderInspector) ProcessField(f *Field) error { + h.pathParts[1] = f.name + return nil +} + +func (h *holderInspector) ProcessView(v *view) error { + h.pathParts[2] = v.name + h.path = strings.Join(h.pathParts[:], "/") + return nil +} + +func (h *holderInspector) ProcessFragment(f *fragment) error { + path := h.path + "/" + strconv.FormatUint(f.shard, 10) + h.hi.FragmentInfo[path] = f.inspect(h.inspectRequestFull.params) + h.hi.FragmentNames = append(h.hi.FragmentNames, path) + return nil +} + +func (h *Holder) Inspect(ctx context.Context, req *InspectRequest) (*HolderInfo, error) { + fullReq, err := expandInspectRequest(req) + if err != nil { + return nil, err + } + inspector := &holderInspector{ + inspectRequestFull: fullReq, + hi: &HolderInfo{ + FragmentInfo: make(map[string]FragmentInfo), + }, + } + err = h.Process(ctx, inspector) + sort.Strings(inspector.hi.FragmentNames) + return inspector.hi, err +} + // Open initializes the root data directory for the holder. func (h *Holder) Open() error { h.opening = true @@ -213,11 +502,6 @@ func (h *Holder) Open() error { return errors.Wrap(err, "reading directory") } - // Run snapshots asynchronously. The snapshotQueue will have a background - // task associated with it which flushes it and waits until this channel - // is closed, so we should always close this channel when done. - h.snapshotQueue = newSnapshotQueue(10, 2, h.Logger) - for _, fi := range fis { // Skip files or hidden directories. if !fi.IsDir() || strings.HasPrefix(fi.Name(), ".") { @@ -261,18 +545,25 @@ func (h *Holder) Open() error { h.Logger.Printf("open holder: complete") - // Periodically flush cache. - h.wg.Add(1) - go func() { defer h.wg.Done(); h.monitorCacheFlush() }() - h.Stats.Open() - h.snapshotQueue.ScanHolder(h) h.opened.Close() return nil } +// Activate runs the background tasks relevant to keeping a holder in a stable +// state, such as scanning it for needed snapshots, or flushing caches. This +// is separate from opening because, while a server would nearly always want +// to do this, other use cases (like consistency checks of a data directory) +// need to avoid it even getting started. +func (h *Holder) Activate() { + // Periodically flush cache. + h.wg.Add(2) + go func() { defer h.wg.Done(); h.monitorCacheFlush() }() + go func() { defer h.wg.Done(); h.SnapshotQueue.ScanHolder(h, h.closing) }() +} + // checkForeignIndex is a check before applying a foreign // index to a field; if the index is not yet available, // (because holder is still opening and may not have opened @@ -313,10 +604,6 @@ func (h *Holder) Close() error { return errors.Wrap(err, "closing index") } } - if h.snapshotQueue != nil { - h.snapshotQueue.Stop() - h.snapshotQueue = nil - } // Reset opened in case Holder needs to be reopened. h.opened.mu.Lock() @@ -587,19 +874,16 @@ func (h *Holder) createIndex(name string, opt IndexOptions) (*Index, error) { } func (h *Holder) newIndex(path, name string) (*Index, error) { - index, err := NewIndex(path, name, h.partitionN) + index, err := NewIndex(h, path, name) if err != nil { return nil, err } - index.logger = h.Logger index.Stats = h.Stats.WithTags(fmt.Sprintf("index:%s", index.Name())) index.broadcaster = h.broadcaster index.newAttrStore = h.NewAttrStore index.columnAttrs = h.NewAttrStore(filepath.Join(index.path, ".data")) - index.snapshotQueue = h.snapshotQueue index.OpenTranslateStore = h.OpenTranslateStore index.translationSyncer = h.translationSyncer - index.holder = h return index, nil } @@ -1368,3 +1652,132 @@ func uint64InSlice(i uint64, s []uint64) bool { } return false } + +// Process loops through a holder based on the Check functions in op, calling +// the Process functions in op when indicated. +func (h *Holder) Process(ctx context.Context, op HolderOperator) (err error) { + var indexNames, fieldNames, viewNames []string + var fragNums []uint64 + + h.mu.Lock() + for indexName := range h.indexes { + indexNames = append(indexNames, indexName) + } + h.mu.Unlock() + for _, indexName := range indexNames { + if err = ctx.Err(); err != nil { + return err + } + process, recurse := op.CheckIndex(indexName) + if !process && !recurse { + continue + } + h.mu.Lock() + index := h.indexes[indexName] + h.mu.Unlock() + if index == nil { + continue + } + if err = ctx.Err(); err != nil { + return err + } + if process { + err = op.ProcessIndex(index) + if err != nil { + return err + } + } + if !recurse { + continue + } + fieldNames = fieldNames[:0] + index.mu.Lock() + for fieldName := range index.fields { + fieldNames = append(fieldNames, fieldName) + } + index.mu.Unlock() + for _, fieldName := range fieldNames { + if err = ctx.Err(); err != nil { + return err + } + process, recurse := op.CheckField(indexName, fieldName) + if !process && !recurse { + continue + } + index.mu.Lock() + field := index.fields[fieldName] + index.mu.Unlock() + if field == nil { + continue + } + if err = ctx.Err(); err != nil { + return err + } + if process { + err = op.ProcessField(field) + if err != nil { + return err + } + } + if !recurse { + continue + } + viewNames = viewNames[:0] + field.mu.Lock() + for viewName := range field.viewMap { + viewNames = append(viewNames, viewName) + } + field.mu.Unlock() + for _, viewName := range viewNames { + if err = ctx.Err(); err != nil { + return err + } + process, recurse := op.CheckView(indexName, fieldName, viewName) + if !process && !recurse { + continue + } + field.mu.Lock() + view := field.viewMap[viewName] + field.mu.Unlock() + if view == nil { + continue + } + if err = ctx.Err(); err != nil { + return err + } + if process { + err = op.ProcessView(view) + if err != nil { + return err + } + } + if !recurse { + continue + } + fragNums := fragNums[:0] + view.mu.Lock() + for fragNum := range view.fragments { + fragNums = append(fragNums, fragNum) + } + view.mu.Unlock() + for _, fragNum := range fragNums { + if err = ctx.Err(); err != nil { + return err + } + process := op.CheckFragment(indexName, fieldName, viewName, fragNum) + if !process { + continue + } + view.mu.Lock() + frag := view.fragments[fragNum] + view.mu.Unlock() + err = op.ProcessFragment(frag) + if err != nil { + return err + } + } + } + } + } + return nil +} diff --git a/holder_internal_test.go b/holder_internal_test.go index ccc52c5b0..5ae365362 100644 --- a/holder_internal_test.go +++ b/holder_internal_test.go @@ -1,4 +1,4 @@ -// Copyright 2017 Pilosa Corp. +// Copyright 2020 Pilosa Corp. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -15,313 +15,163 @@ package pilosa import ( + "context" "io/ioutil" "os" - "path/filepath" - "reflect" - "strings" "testing" - "time" - - "github.com/pilosa/pilosa/v2/roaring" ) -type tHolder struct { - *Holder +type testHolderOperator struct { + indexSeen, indexProcessed int + fieldSeen, fieldProcessed int + viewSeen, viewProcessed int + fragmentSeen, fragmentProcessed int + waitHere chan struct{} } -// Close closes the holder and removes all underlying data. -func (h *tHolder) Close() error { - defer os.RemoveAll(h.Path) - return h.Holder.Close() +func (t *testHolderOperator) CheckIndex(string) (bool, bool) { + t.indexSeen++ + return true, true } -// Reopen instantiates and opens a new holder. -// Note that the holder must be Closed first. -func (h *tHolder) Reopen() error { - path, logger := h.Path, h.Holder.Logger - h.Holder = NewHolder(DefaultPartitionN) - h.Holder.Path = path - h.Holder.Logger = logger - return h.Holder.Open() +func (t *testHolderOperator) CheckField(string, string) (bool, bool) { + t.fieldSeen++ + return true, true } -func newHolder() *tHolder { - path, err := ioutil.TempDir(*TempDir, "pilosa-") +func (t *testHolderOperator) CheckView(string, string, string) (bool, bool) { + t.viewSeen++ + return true, true +} + +func (t *testHolderOperator) CheckFragment(string, string, string, uint64) bool { + t.fragmentSeen++ + return true +} + +func (t *testHolderOperator) ProcessIndex(*Index) error { + t.indexProcessed++ + return nil +} + +func (t *testHolderOperator) ProcessField(*Field) error { + t.fieldProcessed++ + return nil +} + +func (t *testHolderOperator) ProcessView(*view) error { + t.viewProcessed++ + return nil +} + +func (t *testHolderOperator) ProcessFragment(*fragment) error { + if t.waitHere != nil { + <-t.waitHere + } + t.fragmentProcessed++ + return nil +} + +func makeHolder() (*Holder, string, error) { + path, err := ioutil.TempDir("", "pilosa-") if err != nil { - panic(err) + return nil, "", err } - - h := &tHolder{Holder: NewHolder(DefaultPartitionN)} - h.Path = path - return h -} - -// MustCreateFieldIfNotExists returns a given field. Panic on error. -func (h *tHolder) MustCreateFieldIfNotExists(index, field string) *Field { - f, err := h.MustCreateIndexIfNotExists(index, IndexOptions{}).CreateFieldIfNotExists(field, OptFieldTypeDefault()) - if err != nil { - panic(err) - } - return f -} - -// MustCreateIndexIfNotExists returns a given index. Panic on error. -func (h *tHolder) MustCreateIndexIfNotExists(index string, opt IndexOptions) *Index { - idx, err := h.Holder.CreateIndexIfNotExists(index, opt) - if err != nil { - panic(err) - } - return idx -} - -// SetBit clears a bit on the given field. -func (h *tHolder) SetBit(index, field string, rowID, columnID uint64) { - f := h.MustCreateFieldIfNotExists(index, field) - _, err := f.SetBit(rowID, columnID, nil) - if err != nil { - panic(err) - } -} - -// Row returns a Row for a given field. -func (h *tHolder) Row(index, field string, rowID uint64) *Row { - f := h.MustCreateFieldIfNotExists(index, field) - row, err := f.Row(rowID) - if err != nil { - panic(err) - } - return row -} - -func TestHolder_Optn(t *testing.T) { - t.Run("ErrViewPermission", func(t *testing.T) { - if os.Geteuid() == 0 { - t.Skip("Skipping permissions test since user is root.") - } - availableShardFileFlushDuration.Set(100 * time.Millisecond) - h := newHolder() - defer h.Close() - - if idx, err := h.CreateIndex("foo", IndexOptions{}); err != nil { - t.Fatal(err) - } else if field, err := idx.CreateField("bar", OptFieldTypeDefault()); err != nil { - t.Fatal(err) - } else if _, err := field.createViewIfNotExists(viewStandard); err != nil { - t.Fatal(err) - } else if err := h.Holder.Close(); err != nil { - t.Fatal(err) - } else if err := os.Chmod(filepath.Join(h.Path, "foo", "bar", "views", "standard"), 0000); err != nil { - t.Fatal(err) - } - defer func() { - // we don't care about a failure here - _ = os.Chmod(filepath.Join(h.Path, "foo", "bar", "views", "standard"), 0755) - }() - if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "permission denied") { - t.Fatalf("unexpected error: %s", err) - } - }) - t.Run("ErrViewFragmentsMkdir", func(t *testing.T) { - if os.Geteuid() == 0 { - t.Skip("Skipping permissions test since user is root.") - } - h := newHolder() - defer h.Close() - - if idx, err := h.CreateIndex("foo", IndexOptions{}); err != nil { - t.Fatal(err) - } else if field, err := idx.CreateField("bar", OptFieldTypeDefault()); err != nil { - t.Fatal(err) - } else if _, err := field.createViewIfNotExists(viewStandard); err != nil { - t.Fatal(err) - } else if err := h.Holder.Close(); err != nil { - t.Fatal(err) - } else if err := os.Chmod(filepath.Join(h.Path, "foo", "bar", "views", "standard", "fragments"), 0000); err != nil { - t.Fatal(err) - } - defer func() { - // we don't care about a failure here - _ = os.Chmod(filepath.Join(h.Path, "foo", "bar", "views", "standard", "fragments"), 0755) - }() - - if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "permission denied") { - t.Fatalf("unexpected error: %s", err) - } - }) - - t.Run("ErrFragmentCachePermission", func(t *testing.T) { - if os.Geteuid() == 0 { - t.Skip("Skipping permissions test since user is root.") - } - h := newHolder() - defer h.Close() - - if idx, err := h.CreateIndex("foo", IndexOptions{}); err != nil { - t.Fatal(err) - } else if field, err := idx.CreateField("bar", OptFieldTypeDefault()); err != nil { - t.Fatal(err) - } else if view, err := field.createViewIfNotExists(viewStandard); err != nil { - t.Fatal(err) - } else if _, err := field.SetBit(0, 0, nil); err != nil { - t.Fatal(err) - } else if err := view.Fragment(0).FlushCache(); err != nil { - t.Fatal(err) - } else if err := h.Holder.Close(); err != nil { - t.Fatal(err) - } else if err := os.Chmod(filepath.Join(h.Path, "foo", "bar", "views", "standard", "fragments", "0.cache"), 0000); err != nil { - t.Fatal(err) - } - defer func() { - _ = os.Chmod(filepath.Join(h.Path, "foo", "bar", "views", "standard", "fragments", "0.cache"), 0644) - }() - if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "permission denied") { - t.Fatalf("unexpected error: %s", err) - } - }) - -} - -// Ensure holder can clean up orphaned fragments. -func TestHolderCleaner_CleanHolder(t *testing.T) { - availableShardFileFlushDuration.Set(100 * time.Millisecond) //shorten the default time to force a file write - cluster := NewTestCluster(2) - - // Create a local holder. - hldr0 := newHolder() - defer hldr0.Close() - - // Mock 2-node, fully replicated cluster. - cluster.ReplicaN = 2 - - cluster.nodes[0].URI = NewTestURIFromHostPort("localhost", 0) - - // Create fields on nodes. - for _, hldr := range []*tHolder{hldr0} { - hldr.MustCreateFieldIfNotExists("i", "f") - hldr.MustCreateFieldIfNotExists("i", "f0") - hldr.MustCreateFieldIfNotExists("y", "z") - } - - // Set data on the local holder. - hldr0.SetBit("i", "f", 0, 10) - hldr0.SetBit("i", "f", 0, 4000) - hldr0.SetBit("i", "f", 2, 20) - hldr0.SetBit("i", "f", 3, 10) - hldr0.SetBit("i", "f", 120, 10) - hldr0.SetBit("i", "f", 200, 4) - - hldr0.SetBit("i", "f0", 9, ShardWidth+5) - - hldr0.SetBit("y", "z", 10, (2*ShardWidth)+4) - hldr0.SetBit("y", "z", 10, (2*ShardWidth)+5) - hldr0.SetBit("y", "z", 10, (2*ShardWidth)+7) - - // Set highest shard. - err := hldr0.Field("i", "f").AddRemoteAvailableShards(roaring.NewBitmap(0, 1)) - if err != nil { - t.Fatalf("adding remote shards: %v", err) - } - err = hldr0.Field("y", "z").AddRemoteAvailableShards(roaring.NewBitmap(0, 1, 2)) - if err != nil { - t.Fatalf("adding remote shards: %v", err) - } - time.Sleep(2 * availableShardFileFlushDuration.Get()) - - // Keep replication the same and ensure we get the expected results. - cluster.ReplicaN = 2 - - // Set up cleaner for replication 2. - cleaner2 := holderCleaner{ - Node: cluster.nodes[0], - Holder: hldr0.Holder, - Cluster: cluster, - } - - if err := cleaner2.CleanHolder(); err != nil { - t.Fatal(err) - } - - // Verify data is the same on both nodes. - for i, hldr := range []*tHolder{hldr0} { - if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{10, 4000}) { - t.Fatalf("unexpected columns(%d/0): %+v", i, a) - } else if a := hldr.Row("i", "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) { - t.Fatalf("unexpected columns(%d/2): %+v", i, a) - } else if a := hldr.Row("i", "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) { - t.Fatalf("unexpected columns(%d/3): %+v", i, a) - } else if a := hldr.Row("i", "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) { - t.Fatalf("unexpected columns(%d/120): %+v", i, a) - } else if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) { - t.Fatalf("unexpected columns(%d/200): %+v", i, a) - } - - if a := hldr.Row("i", "f0", 9).Columns(); !reflect.DeepEqual(a, []uint64{ShardWidth + 5}) { - t.Fatalf("unexpected columns(%d/d/f0): %+v", i, a) - } - - if a := hldr.Row("y", "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(2 * ShardWidth) + 4, (2 * ShardWidth) + 5, (2 * ShardWidth) + 7}) { - t.Fatalf("unexpected columns(%d/y/z): %+v", i, a) - } - } - - // Change replication factor to ensure we have fragments to remove. - cluster.ReplicaN = 1 - - // Set up cleaner for replication 1. - cleaner1 := holderCleaner{ - Node: cluster.nodes[0], - Holder: hldr0.Holder, - Cluster: cluster, - } - - if err := cleaner1.CleanHolder(); err != nil { - t.Fatal(err) - } - - // Verify data is the same on both nodes. - for i, hldr := range []*tHolder{hldr0} { - if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{10, 4000}) { - t.Fatalf("unexpected columns(%d/0): %+v", i, a) - } else if a := hldr.Row("i", "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) { - t.Fatalf("unexpected columns(%d/2): %+v", i, a) - } else if a := hldr.Row("i", "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) { - t.Fatalf("unexpected columns(%d/3): %+v", i, a) - } else if a := hldr.Row("i", "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) { - t.Fatalf("unexpected columns(%d/120): %+v", i, a) - } else if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) { - t.Fatalf("unexpected columns(%d/200): %+v", i, a) - } - - f := hldr.fragment("i", "f0", viewStandard, 1) - if f != nil { - t.Fatalf("expected fragment to be deleted: (%d/i/f0): %+v", i, f) - } - - if a := hldr.Row("y", "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(2 * ShardWidth) + 4, (2 * ShardWidth) + 5, (2 * ShardWidth) + 7}) { - t.Fatalf("unexpected columns(%d/y/z): %+v", i, a) - } - } -} - -// Ensure holder can reopen. -func TestHolderCleaner_Reopen(t *testing.T) { h := NewHolder(DefaultPartitionN) - h.Path = "path" - err := h.Open() + + return h, path, nil +} + +func testSetBit(t *testing.T, h *Holder, index, field string, rowID, columnID uint64) { + idx, err := h.CreateIndexIfNotExists(index, IndexOptions{}) if err != nil { - t.Fatalf("couldn't open holder: %v", err) + t.Fatalf("creating index: %v", err) } - err = h.Close() + f, err := idx.CreateFieldIfNotExists(field, OptFieldTypeDefault()) if err != nil { - t.Fatalf("couldn't close holder: %v", err) + t.Fatalf("setting bit: %v", err) } - err = h.Open() + _, err = f.SetBit(rowID, columnID, nil) if err != nil { - t.Fatalf("couldn't open holder: %v", err) - } - err = h.Close() - if err != nil { - t.Fatalf("couldn't close holder: %v", err) + t.Fatalf("setting bit: %v", err) + } +} + +func TestHolderOperatorProcess(t *testing.T) { + h, path, err := makeHolder() + if err != nil { + t.Fatalf("creating holder: %v", err) + } + defer os.RemoveAll(path) + defer h.Close() + + // Write bits to separate indexes. + testSetBit(t, h, "i0", "f", 100, 200) + testSetBit(t, h, "i1", "f", 100, 200) + testSetBit(t, h, "i1", "f", 100, 12345678) + + testOp := testHolderOperator{} + ctx := context.Background() + err = h.Process(ctx, &testOp) + if err != nil { + t.Fatalf("processing holder: %v", err) + } + expected := testHolderOperator{ + indexSeen: 2, indexProcessed: 2, + fieldSeen: 2, fieldProcessed: 2, + viewSeen: 2, viewProcessed: 2, + fragmentSeen: 3, fragmentProcessed: 3, + } + if testOp != expected { + t.Fatalf("holder processor did not process as expected. expected %#v, got %#v", expected, testOp) + } +} + +func TestHolderOperatorCancel(t *testing.T) { + h, path, err := makeHolder() + if err != nil { + t.Fatalf("creating holder: %v", err) + } + defer os.RemoveAll(path) + defer h.Close() + + // Write bits to separate indexes. + testSetBit(t, h, "i0", "f", 100, 200) + testSetBit(t, h, "i1", "f", 100, 200) + testSetBit(t, h, "i1", "f", 100, 12345678) + + // Here, we want to ensure that the operation gets cancelled + // successfully. In practice we expect it to process one fragment, then + // end up blocked on the waitHere, then get cancelled... But the + // waitHere blockage isn't really something holder.Process can do + // anything about, so we close the channel, so two fragments are + // processed. But in theory you could end up with only one fragment + // processed if this goroutine managed to cancel before the processor + // gets to the next fragment. Point is, it shouldn't hit all three, + // because the checks against the cancellation should fire before it + // gets there. + testOp := testHolderOperator{waitHere: make(chan struct{})} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + err = h.Process(ctx, &testOp) + close(done) + }() + testOp.waitHere <- struct{}{} + cancel() + close(testOp.waitHere) + <-done + if err != context.Canceled { + t.Fatalf("processing holder: expected context.Canceled, got %v", err) + } + testOp.waitHere = nil + expected := testHolderOperator{ + indexSeen: 2, indexProcessed: 2, + fieldSeen: 2, fieldProcessed: 2, + viewSeen: 2, viewProcessed: 2, + fragmentSeen: 3, fragmentProcessed: 3, + } + if testOp == expected { + t.Fatalf("holder processor did not cancel. expected something other than %#v", expected) } } diff --git a/http/handler.go b/http/handler.go index 67be9e297..06a62ebd7 100644 --- a/http/handler.go +++ b/http/handler.go @@ -221,6 +221,8 @@ func (h *Handler) populateValidators() { h.validators["GetTransaction"] = queryValidationSpecRequired() h.validators["PostTransaction"] = queryValidationSpecRequired() h.validators["PostFinishTransaction"] = queryValidationSpecRequired() + h.validators["Inspect"] = queryValidationSpecRequired().Optional("indexes", "fields", "views", "shards", "checksum", "containers") + } type contextKeyQuery int @@ -352,6 +354,7 @@ func newRouter(handler *Handler) *mux.Router { router.HandleFunc("/index/{index}/field/{field}/import-roaring/{shard}", handler.handlePostImportRoaring).Methods("POST").Name("PostImportRoaring") router.HandleFunc("/index/{index}/query", handler.handlePostQuery).Methods("POST").Name("PostQuery") router.HandleFunc("/info", handler.handleGetInfo).Methods("GET").Name("GetInfo") + router.HandleFunc("/inspect", handler.handleInspect).Methods("GET").Name("Inspect") router.HandleFunc("/recalculate-caches", handler.handleRecalculateCaches).Methods("POST").Name("RecalculateCaches") router.HandleFunc("/schema", handler.handleGetSchema).Methods("GET").Name("GetSchema") router.HandleFunc("/schema", handler.handlePostSchema).Methods("POST").Name("PostSchema") @@ -590,6 +593,37 @@ func (h *Handler) handleGetInfo(w http.ResponseWriter, r *http.Request) { } } +func (h *Handler) handleInspect(w http.ResponseWriter, r *http.Request) { + if !validHeaderAcceptJSON(r.Header) { + http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) + return + } + q := r.URL.Query() + _, checksum := q["checksum"] + _, containers := q["containers"] + req := pilosa.InspectRequest{ + HolderFilterParams: pilosa.HolderFilterParams{ + Indexes: q.Get("indexes"), + Fields: q.Get("fields"), + Views: q.Get("views"), + Shards: q.Get("shards"), + }, + InspectRequestParams: pilosa.InspectRequestParams{ + Checksum: checksum, + Containers: containers, + }, + } + info, err := h.api.Inspect(r.Context(), &req) + if err != nil { + http.Error(w, fmt.Sprintf("inspect request: %v", err), http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(info); err != nil { + h.logger.Printf("write inspect response error: %s", err) + } +} + type getSchemaResponse struct { Indexes []*pilosa.IndexInfo `json:"indexes"` } diff --git a/index.go b/index.go index 481d22cf4..4a7ee3f5f 100644 --- a/index.go +++ b/index.go @@ -27,7 +27,6 @@ import ( "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/v2/internal" - "github.com/pilosa/pilosa/v2/logger" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/stats" "github.com/pkg/errors" @@ -46,9 +45,6 @@ type Index struct { trackExistence bool existenceFld *Field - // Partitions used by translation. - partitionN int - // Fields by name. fields map[string]*Field @@ -60,9 +56,6 @@ type Index struct { broadcaster broadcaster Stats stats.StatsClient - logger logger.Logger - snapshotQueue snapshotQueue - // Passed to field for foreign-index lookup. holder *Holder @@ -76,24 +69,23 @@ type Index struct { } // NewIndex returns a new instance of Index. -func NewIndex(path, name string, partitionN int) (*Index, error) { +func NewIndex(holder *Holder, path, name string) (*Index, error) { err := validateName(name) if err != nil { return nil, errors.Wrap(err, "validating name") } return &Index{ - path: path, - name: name, - partitionN: partitionN, - fields: make(map[string]*Field), + path: path, + name: name, + fields: make(map[string]*Field), newAttrStore: newNopAttrStore, columnAttrs: nopStore, broadcaster: NopBroadcaster, Stats: stats.NopStatsClient, - logger: logger.NopLogger, + holder: holder, trackExistence: true, translateStores: make(map[int]TranslateStore), @@ -155,18 +147,18 @@ func (i *Index) OpenWithTimestamp() error { return i.open(true) } func (i *Index) open(withTimestamp bool) (err error) { // Ensure the path exists. - i.logger.Debugf("ensure index path exists: %s", i.path) + i.holder.Logger.Debugf("ensure index path exists: %s", i.path) if err := os.MkdirAll(i.path, 0777); err != nil { return errors.Wrap(err, "creating directory") } // Read meta file. - i.logger.Debugf("load meta file for index: %s", i.name) + i.holder.Logger.Debugf("load meta file for index: %s", i.name) if err := i.loadMeta(); err != nil { return errors.Wrap(err, "loading meta file") } - i.logger.Debugf("open fields for index: %s", i.name) + i.holder.Logger.Debugf("open fields for index: %s", i.name) if err := i.openFields(withTimestamp); err != nil { return errors.Wrap(err, "opening fields") } @@ -181,15 +173,15 @@ func (i *Index) open(withTimestamp bool) (err error) { return errors.Wrap(err, "opening attrstore") } - i.logger.Debugf("open translate store for index: %s", i.name) + i.holder.Logger.Debugf("open translate store for index: %s", i.name) var g errgroup.Group var mu sync.Mutex - for partitionID := 0; partitionID < i.partitionN; partitionID++ { + for partitionID := 0; partitionID < i.holder.partitionN; partitionID++ { partitionID := partitionID g.Go(func() error { - store, err := i.OpenTranslateStore(i.TranslateStorePath(partitionID), i.name, "", partitionID, i.partitionN) + store, err := i.OpenTranslateStore(i.TranslateStorePath(partitionID), i.name, "", partitionID, i.holder.partitionN) if err != nil { return errors.Wrapf(err, "opening index translate store: partition=%d", partitionID) } @@ -239,7 +231,7 @@ fileLoop: defer func() { <-indexQueue }() - i.logger.Debugf("open field: %s", fi.Name()) + i.holder.Logger.Debugf("open field: %s", fi.Name()) mu.Lock() fld, err := i.newField(i.fieldPath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) if withTimestamp { @@ -257,7 +249,7 @@ fileLoop: if err := fld.Open(); err != nil { return fmt.Errorf("open field: name=%s, err=%s", fld.Name(), err) } - i.logger.Debugf("add field to index.fields: %s", fi.Name()) + i.holder.Logger.Debugf("add field to index.fields: %s", fi.Name()) mu.Lock() i.fields[fld.Name()] = fld mu.Unlock() @@ -512,17 +504,13 @@ func (i *Index) createField(name string, opt *FieldOptions) (*Field, error) { } func (i *Index) newField(path, name string) (*Field, error) { - f, err := newField(path, i.name, name, OptFieldTypeDefault()) + f, err := newField(i.holder, path, i.name, name, OptFieldTypeDefault()) if err != nil { return nil, err } - f.logger = i.logger f.Stats = i.Stats f.broadcaster = i.broadcaster f.rowAttrStore = i.newAttrStore(filepath.Join(f.path, ".data")) - if i.snapshotQueue != nil { - f.snapshotQueue = i.snapshotQueue - } f.OpenTranslateStore = i.OpenTranslateStore return f, nil } diff --git a/index_internal_test.go b/index_internal_test.go index 026607da4..5b5a5b5a3 100644 --- a/index_internal_test.go +++ b/index_internal_test.go @@ -25,7 +25,7 @@ func mustOpenIndex(opt IndexOptions) *Index { if err != nil { panic(err) } - index, err := NewIndex(path, "i", DefaultPartitionN) + index, err := NewIndex(NewHolder(1), path, "i") if err != nil { panic(err) } diff --git a/index_test.go b/index_test.go index 63e9b6179..bfebee7f7 100644 --- a/index_test.go +++ b/index_test.go @@ -242,7 +242,7 @@ func TestIndex_InvalidName(t *testing.T) { if err != nil { panic(err) } - index, err := pilosa.NewIndex(path, "ABC", pilosa.DefaultPartitionN) + index, err := pilosa.NewIndex(pilosa.NewHolder(pilosa.DefaultPartitionN), path, "ABC") if err == nil { t.Fatalf("should have gotten an error on index name with caps") } diff --git a/roaring/container_stash.go b/roaring/container_stash.go index 64a6d217e..a43b3eee8 100644 --- a/roaring/container_stash.go +++ b/roaring/container_stash.go @@ -45,35 +45,47 @@ type Container struct { type containerFlags uint8 +var containerFlagStrings = [...]string{ + "", + "mapped", + "frozen", + "frozen/mapped", + "pristine", + "pristine/mapped", + "pristine/frozen", + "pristine/frozen/mapped", +} + +func (f containerFlags) String() string { + return containerFlagStrings[f&7] +} + const ( flagMapped = containerFlags(1 << iota) flagFrozen + flagPristine ) func (c *Container) String() string { if c == nil { return "" } - froze := "" - switch c.flags { - case flagFrozen: - froze = "frozen " - case flagMapped: - froze = "mapped " - case flagFrozen | flagMapped: - froze = "frozen/mapped" + var space, froze string + if c.flags != 0 { + space = " " + froze = c.flags.String() } switch c.typeID { case containerArray: - return fmt.Sprintf("<%sarray container, N=%d>", froze, c.N()) + return fmt.Sprintf("<%s%sarray container, N=%d>", froze, space, c.N()) case containerBitmap: - return fmt.Sprintf("<%sbitmap container, N=%d, len %dx uint64>", - froze, c.N(), len(c.bitmap())) + return fmt.Sprintf("<%s%sbitmap container, N=%d, len %dx uint64>", + froze, space, c.N(), len(c.bitmap())) case containerRun: - return fmt.Sprintf("<%srun container, N=%d, len %dx interval>", - froze, c.N(), len(c.runs())) + return fmt.Sprintf("<%s%srun container, N=%d, len %dx interval>", + froze, space, c.N(), len(c.runs())) default: - return fmt.Sprintf("", froze, c.typeID, c.N()) + return fmt.Sprintf("", froze, space, c.typeID, c.N()) } } @@ -292,6 +304,7 @@ func (c *Container) unmapOrClone() *Container { return c.Clone() } c.flags &^= flagMapped + c.flags &^= flagPristine // mapped: we want to unmap the storage. switch c.typeID { case containerArray: @@ -368,6 +381,7 @@ func (c *Container) setArrayMaybeCopy(array []uint16, doCopy bool) { if len(array) > 1<<16 { panic("impossibly large array") } + c.flags &^= flagPristine // array we can fit in data store: if len(array) <= stashedArraySize { copy(c.data[:stashedArraySize], array) @@ -497,6 +511,7 @@ func (c *Container) setBitmap(bitmap []uint64) { panic("illegal bitmap length") } c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&bitmap[0])), bitmapN, bitmapN + c.flags &^= flagPristine } // runs yields the data viewed as a slice of intervals. @@ -531,6 +546,7 @@ func (c *Container) setRunsMaybeCopy(runs []interval16, doCopy bool) { if len(runs) > 1<<15 { panic("impossibly large run set") } + c.flags &^= flagPristine // array we can fit in data store: if len(runs) <= stashedRunSize { newRuns := (*[stashedRunSize]interval16)(unsafe.Pointer(&c.data))[:len(runs)] diff --git a/roaring/roaring.go b/roaring/roaring.go index cbb7dc583..b87f4d6af 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -1721,11 +1721,14 @@ func (b *Bitmap) writeToUnoptimized(w io.Writer) (n int64, err error) { // bitmap and yield information about containers, including type, size, and // the location of their data structures. type roaringIterator interface { + // Len reports the number of containers total. + Len() (count int64) // Next yields the information about the next container Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error) // Remaining yields the bytes left over past the end of the roaring data, - // which is typically an ops log in our case. - Remaining() []byte + // which is typically an ops log in our case, and also its offset in case + // we need to talk about truncation. + Remaining() ([]byte, int64) } // baseRoaringIterator holds values used by both Pilosa and official Roaring @@ -1884,11 +1887,16 @@ func (r *baseRoaringIterator) Done(err error) { r.currentDataOffset = 0 } -func (r *baseRoaringIterator) Remaining() []byte { +// Len() indicates the total number of containers the iterator expects to have. +func (r *baseRoaringIterator) Len() int64 { + return r.keys +} + +func (r *baseRoaringIterator) Remaining() ([]byte, int64) { if r.lastDataOffset == 0 { - return nil + return nil, 0 } - return r.data[r.lastDataOffset:] + return r.data[r.lastDataOffset:], r.lastDataOffset } func (r *pilosaRoaringIterator) Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error) { @@ -2435,19 +2443,24 @@ func (b *Bitmap) roaringSize() (int64, int64) { } // Info returns stats for the bitmap. -func (b *Bitmap) Info() bitmapInfo { - info := bitmapInfo{ - OpN: b.opN, - Ops: b.ops, - Containers: make([]containerInfo, 0, b.Containers.Size()), +func (b *Bitmap) Info(includeContainers bool) BitmapInfo { + info := BitmapInfo{ + OpN: b.opN, + Ops: b.ops, + ContainerCount: b.Containers.Size(), + } + if includeContainers { + info.Containers = make([]ContainerInfo, 0, info.ContainerCount) } - citer, _ := b.Containers.Iterator(0) for citer.Next() { k, c := citer.Value() ci := c.info() ci.Key = k - info.Containers = append(info.Containers, ci) + info.BitCount += uint64(c.N()) + if includeContainers { + info.Containers = append(info.Containers, ci) + } } return info } @@ -2504,11 +2517,16 @@ func (b *Bitmap) Flip(start, end uint64) *Bitmap { return result } -// bitmapInfo represents a point-in-time snapshot of bitmap stats. -type bitmapInfo struct { - OpN int - Ops int - Containers []containerInfo +// BitmapInfo represents a point-in-time snapshot of bitmap stats. +type BitmapInfo struct { + OpN int + Ops int + OpDetails []OpInfo `json:"OpDetails,omitempty"` + BitCount uint64 + ContainerCount int + Containers []ContainerInfo `json:"Containers,omitempty"` // The containers found in the bitmap originally + OpContainers []ContainerInfo `json:"OpContainers,omitempty"` // The containers resulting from ops log changes. + From, To uintptr // if set, indicates the address range used when unpacking } // Iterator represents an iterator over a Bitmap. @@ -3705,13 +3723,14 @@ func (c *Container) size() int { } // info returns the current stats about the container. -func (c *Container) info() containerInfo { - info := containerInfo{N: c.N()} +func (c *Container) info() ContainerInfo { + info := ContainerInfo{N: c.N(), Mapped: c.Mapped()} if c == nil { info.Type = "nil" info.Alloc = 0 return info } + info.Flags = c.flags.String() if c.isArray() { info.Type = "array" @@ -3723,17 +3742,7 @@ func (c *Container) info() containerInfo { info.Type = "bitmap" info.Alloc = len(c.bitmap()) * 8 // sizeof(uint64) } - - if c.Mapped() { - if c.isArray() { - info.Pointer = unsafe.Pointer(&c.array()[0]) - } else if c.isRun() { - info.Pointer = unsafe.Pointer(&c.runs()[0]) - } else { - info.Pointer = unsafe.Pointer(&c.bitmap()[0]) - } - } - + info.Pointer = uintptr(unsafe.Pointer(c.pointer)) return info } @@ -3799,13 +3808,15 @@ func (c *Container) bitmapRepair() { c.setN(n) } -// containerInfo represents a point-in-time snapshot of container stats. -type containerInfo struct { - Key uint64 // container key - Type string // container type (array, bitmap, or run) - N int32 // number of bits - Alloc int // memory used - Pointer unsafe.Pointer // offset within the mmap +// ContainerInfo represents a point-in-time snapshot of container stats. +type ContainerInfo struct { + Key uint64 // container key + Type string // container type (array, bitmap, or run) + Flags string // flag state + N int32 // number of bits + Alloc int // memory used + Pointer uintptr // address + Mapped bool // whether this container thinks it is mmapped } // flip returns a new container containing the inverse of all @@ -5267,6 +5278,15 @@ const ( opTypeRemoveRoaring = opType(5) ) +var opTypes = []string{ + "add", + "remove", + "addN", + "removeN", + "addRoaring", + "removeRoaring", +} + // op represents an operation on the bitmap. type op struct { typ opType @@ -5276,6 +5296,24 @@ type op struct { roaring []byte } +// OpInfo is a description of an op. +type OpInfo struct { + Type string + OpN int + Size int +} + +func (op *op) info() (info OpInfo) { + if int(op.typ) < len(opTypes) { + info.Type = opTypes[op.typ] + } else { + info.Type = fmt.Sprintf("unknown-type-%d", op.typ) + } + info.OpN = op.opN + info.Size = op.size() + return info +} + // apply executes the operation against a bitmap. func (op *op) apply(b *Bitmap) (changed bool) { switch op.typ { diff --git a/roaring/roaring_test.go b/roaring/roaring_test.go index 55395ea73..7dda0f5c6 100644 --- a/roaring/roaring_test.go +++ b/roaring/roaring_test.go @@ -1733,7 +1733,7 @@ type benchmarkSampleData struct { var sampleData benchmarkSampleData func isAllType(b *roaring.Bitmap, typ string) bool { - bi := b.Info() + bi := b.Info(true) for _, c := range bi.Containers { if c.Type != typ { return false diff --git a/roaring/unmarshal_binary.go b/roaring/unmarshal_binary.go index c9071b1ca..aeab0ef2f 100644 --- a/roaring/unmarshal_binary.go +++ b/roaring/unmarshal_binary.go @@ -1,4 +1,4 @@ -// Copyright 2017 Pilosa Corp. +// Copyright 2019 Pilosa Corp. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -15,232 +15,231 @@ package roaring import ( - "encoding/binary" - "fmt" + "errors" + "io" "unsafe" - - "github.com/pkg/errors" ) -// 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 { +// UnmarshalBinary reads Pilosa's format, or upstream roaring (mostly; +// it may not handle some edge cases), and decodes them into the given +// bitmap, replacing the existing contents. +func (b *Bitmap) UnmarshalBinary(data []byte) (err error) { if data == nil { - // Nothing to unmarshal - return nil - } - statsHit("Bitmap/UnmarshalBinary") - // reset ops/opN since we're reading new data. - b.ops = 0 - b.opN = 0 - fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2])) - if fileMagic == MagicNumber { // if pilosa roaring - return errors.Wrap(b.unmarshalPilosaRoaring(data), "unmarshaling as pilosa roaring") + return errors.New("no roaring bitmap provided") } + var itr roaringIterator + var itrKey uint64 + var itrCType byte + var itrN int + var itrLen int + var itrPointer *uint16 + var itrErr error - keyN, containerTyper, header, pos, haveRuns, err := readOfficialHeader(data) + itr, err = newRoaringIterator(data) if err != nil { - return errors.Wrap(err, "reading roaring header") + return err } - // Only the Pilosa roaring format has flags. The official Roaring format - // hasn't got space in its header for flags. - b.Flags = 0 - - b.Containers.ResetN(int(keyN)) - // Descriptive header section: Read container keys and cardinalities. - for i, buf := uint(0), data[header:]; i < uint(keyN); i, buf = i+1, buf[4:] { - card := int(binary.LittleEndian.Uint16(buf[2:4])) + 1 - b.Containers.PutContainerValues( - uint64(binary.LittleEndian.Uint16(buf[0:2])), - containerTyper(i, card), /// container type voodo with isRunBitmap - card, - true) + if itr == nil { + return errors.New("failed to create roaring iterator, but don't know why") } - // Read container offsets and attach data. - if haveRuns { - err := readWithRuns(b, data, pos, keyN) - if err != nil { - return errors.Wrap(err, "reading offsets from official roaring format") + b.Containers.Reset() + + itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next() + for itrErr == nil { + newC := &Container{ + typeID: itrCType, + n: int32(itrN), + len: int32(itrLen), + cap: int32(itrLen), + pointer: itrPointer, + flags: flagMapped, } - } else { - err := readOffsets(b, data, pos, keyN) - if err != nil { - return errors.Wrap(err, "reading official roaring format") + if !b.preferMapping { + newC.unmapOrClone() } + b.Containers.Put(itrKey, newC) + itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next() } - return nil -} - -func readOffsets(b *Bitmap, data []byte, pos int, keyN uint32) error { - - citer, _ := b.Containers.Iterator(0) - for i, buf := 0, data[pos:]; i < int(keyN); i, buf = i+1, buf[4:] { - // Verify the offset is fully formed - if len(buf) < 4 { - return fmt.Errorf("insufficient data for offsets: len=%d", len(buf)) - } - offset := binary.LittleEndian.Uint32(buf[0:4]) - // Verify the offset is within the bounds of the input data. - if int(offset) >= len(data) { - return fmt.Errorf("offset out of bounds: off=%d, len=%d", offset, len(data)) - } - - // Map byte slice directly to the container data. - citer.Next() - k, c := citer.Value() - if !c.Mapped() { - fmt.Printf("inexplicable: container %d (%d/%d) doesn't think it's mapped. fixing that.\n", - k, i, keyN) - c.setMapped(true) - } - switch c.typ() { - case containerArray: - c.setArray((*[0xFFFFFFF]uint16)(unsafe.Pointer(&data[offset]))[:c.N():c.N()]) - case containerBitmap: - c.setBitmap((*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[offset]))[:bitmapN:bitmapN]) - default: - return fmt.Errorf("unsupported container type %d", c.typ()) - } - } - return nil -} - -func readWithRuns(b *Bitmap, data []byte, pos int, keyN uint32) error { - if len(data) < pos+runCountHeaderSize { - return fmt.Errorf("insufficient data for offsets(run): len=%d", len(data)) - } - citer, _ := b.Containers.Iterator(0) - for i := 0; i < int(keyN); i++ { - citer.Next() - k, c := citer.Value() - if !c.Mapped() { - fmt.Printf("inexplicable: container %d (%d/%d) doesn't think it's mapped. fixing that.\n", - k, i, keyN) - c.setMapped(true) - } - switch c.typ() { - case containerRun: - runCount := binary.LittleEndian.Uint16(data[pos : pos+runCountHeaderSize]) - c.setRuns((*[0xFFFFFFF]interval16)(unsafe.Pointer(&data[pos+runCountHeaderSize]))[:runCount:runCount]) - runs := c.runs() - - for o := range runs { // must convert from start:length to start:end :( - runs[o].last = runs[o].start + runs[o].last - } - pos += int((runCount * interval16Size) + runCountHeaderSize) - case containerArray: - c.setArray((*[0xFFFFFFF]uint16)(unsafe.Pointer(&data[pos]))[:c.N():c.N()]) - pos += int(c.N() * 2) - case containerBitmap: - c.setBitmap((*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[pos]))[:bitmapN:bitmapN]) - pos += bitmapN * 8 - } - } - return nil -} - -func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error { - if len(data) < headerBaseSize { - return errors.New("data too small") - } - - // Verify the first two bytes are a valid MagicNumber, and second two bytes match current storageVersion. - fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2])) - fileVersion := uint32(data[2]) - b.Flags = data[3] - if fileMagic != MagicNumber { - return fmt.Errorf("invalid roaring file, magic number %v is incorrect", fileMagic) - } - - if fileVersion != storageVersion { - return fmt.Errorf("wrong roaring version, file is v%d, server requires v%d", fileVersion, storageVersion) - } - - // Read key count in bytes sizeof(cookie)+sizeof(flag):(sizeof(cookie)+sizeof(uint32)). - keyN := binary.LittleEndian.Uint32(data[3+1 : 8]) - if int64(len(data)) < headerBaseSize+int64(keyN)*12 { - return fmt.Errorf("insufficient data for header + offsets: key-cardinality not provided for %d containers", keyN) - } - - headerSize := headerBaseSize - b.Containers.ResetN(int(keyN)) - // Descriptive header section: Read container keys and cardinalities. - for i, buf := 0, data[headerSize:]; i < int(keyN); i, buf = i+1, buf[12:] { - b.Containers.PutContainerValues( - binary.LittleEndian.Uint64(buf[0:8]), - byte(binary.LittleEndian.Uint16(buf[8:10])), - int(binary.LittleEndian.Uint16(buf[10:12]))+1, - true) - } - opsOffset := int64(headerSize) + int64(keyN)*12 - - // Read container offsets and attach data. - citer, _ := b.Containers.Iterator(0) - // if you have enough containers that the *headers alone* exceed 4GB, we - // need to start with a higher cycle offset. - cycleOffset := opsOffset &^ ((1 << 32) - 1) - prevOffset32 := uint32(opsOffset) - for i, buf := 0, data[opsOffset:]; i < int(keyN); i, buf = i+1, buf[4:] { - offset32 := binary.LittleEndian.Uint32(buf[0:4]) - if offset32 < prevOffset32 { - cycleOffset += (1 << 32) - } - prevOffset32 = offset32 - offset := int64(offset32) + cycleOffset - // Verify the offset is within the bounds of the input data. - if offset >= int64(len(data)) { - return fmt.Errorf("offset out of bounds: off=%d, len=%d", offset, len(data)) - } - - // Map byte slice directly to the container data. - citer.Next() - k, c := citer.Value() - - // this shouldn't happen, since we don't normally store nils. - if c == nil { - continue - } - if !c.Mapped() { - fmt.Printf("inexplicable: container %d (%d/%d) doesn't think it's mapped. fixing that.\n", - k, i, keyN) - c.setMapped(true) - } - switch c.typ() { - case containerRun: - runCount := binary.LittleEndian.Uint16(data[offset : offset+runCountHeaderSize]) - c.setRuns((*[0xFFFFFFF]interval16)(unsafe.Pointer(&data[offset+runCountHeaderSize]))[:runCount:runCount]) - opsOffset = offset + runCountHeaderSize + int64(len(c.runs()))*interval16Size - case containerArray: - c.setArray((*[0xFFFFFFF]uint16)(unsafe.Pointer(&data[offset]))[:c.N():c.N()]) - opsOffset = offset + int64(len(c.array()))*2 // sizeof(uint32) - case containerBitmap: - c.setBitmap((*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[offset]))[:bitmapN:bitmapN]) - opsOffset = offset + int64(len(c.bitmap()))*8 // sizeof(uint64) - } + // note: if we get a non-EOF err, it's possible that we made SOME + // changes but didn't log them. I don't have a good solution to this. + if itrErr != io.EOF { + return itrErr } // Read ops log until the end of the file. - buf := data[opsOffset:] - + b.ops = 0 + b.opN = 0 + buf, lastValidOffset := itr.Remaining() for { // Exit when there are no more ops to parse. if len(buf) == 0 { break } + // Unmarshal the op and apply it. var opr op if err := opr.UnmarshalBinary(buf); err != nil { - return newFileShouldBeTruncatedError(err, int64(opsOffset)) + return newFileShouldBeTruncatedError(err, int64(lastValidOffset)) } + opr.apply(b) + // Increase the op count. b.ops++ b.opN += opr.count() - opsOffset += int64(opr.size()) - // Move the buffer forward. - buf = data[opsOffset:] - } + // Move the buffer forward. + opSize := opr.size() + buf = buf[opSize:] + lastValidOffset += int64(opSize) + } return nil } + +// InspectBinary reads a roaring bitmap, plus a possible ops log, +// and reports back on the contents, including distinguishing between +// the original ops log and the post-ops-log contents. +func InspectBinary(data []byte, mapped bool, info *BitmapInfo) (b *Bitmap, mappedAny bool, err error) { + b = NewFileBitmap() + b.PreferMapping(mapped) + if data == nil { + return b, mappedAny, errors.New("no roaring bitmap provided") + } + var itr roaringIterator + var itrKey uint64 + var itrCType byte + var itrN int + var itrLen int + var itrPointer *uint16 + var itrErr error + + itr, err = newRoaringIterator(data) + if err != nil { + return b, mappedAny, err + } + if itr == nil { + return b, mappedAny, errors.New("failed to create roaring iterator, but don't know why") + } + keys := itr.Len() + info.Containers = make([]ContainerInfo, 0, keys) + + itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next() + for itrErr == nil { + newC := &Container{ + typeID: itrCType, + n: int32(itrN), + len: int32(itrLen), + cap: int32(itrLen), + pointer: itrPointer, + flags: flagMapped, + } + if !mapped { + newC.unmapOrClone() + } + newC.flags |= flagPristine + if newC.flags&flagMapped != 0 { + mappedAny = true + } + var size int + b.Containers.Put(itrKey, newC) + switch itrCType { + case containerArray: + size = int(newC.n) * 2 + case containerBitmap: + size = 8192 + case containerRun: + size = itrLen*interval16Size + runCountHeaderSize + } + info.Containers = append(info.Containers, ContainerInfo{ + N: newC.n, + Mapped: newC.flags&flagMapped != 0, + Type: containerTypeNames[itrCType], + Alloc: size, + Pointer: uintptr(unsafe.Pointer(newC.pointer)), + Key: itrKey, + Flags: newC.flags.String(), + }) + info.ContainerCount++ + info.BitCount += uint64(newC.n) + itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next() + } + // note: if we get a non-EOF err, it's possible that we made SOME + // changes but didn't log them. I don't have a good solution to this. + if itrErr != io.EOF { + return b, mappedAny, itrErr + } + // stash pointer ranges + info.From = uintptr(unsafe.Pointer(&data[0])) + info.To = info.From + uintptr(len(data)) + + // Read ops log until the end of the file. + b.ops = 0 + b.opN = 0 + buf, lastValidOffset := itr.Remaining() + // if there's no ops log, we're done and can just return the + // info so far. + if len(buf) == 0 { + return b, mappedAny, err + } + for { + // Exit when there are no more ops to parse. + if len(buf) == 0 { + break + } + + // Unmarshal the op and apply it. + var opr op + if err = opr.UnmarshalBinary(buf); err != nil { + // we break out here, but we continue on to + // return the bitmap as-is, along with data about + // it, and the error. this lets us share the + // "is anything mapped" check with that code. + break + } + opr.apply(b) + + // Increase the op count. + if info != nil { + info.Ops++ + info.OpN += opr.count() + info.OpDetails = append(info.OpDetails, opr.info()) + } + // Move the buffer forward. + opSize := opr.size() + buf = buf[opSize:] + lastValidOffset += int64(opSize) + } + citer, _ := b.Containers.Iterator(0) + // it's possible the ops log unmapped every mapped container, so we recheck. + mappedAny = false + if info == nil { + for citer.Next() { + _, c := citer.Value() + if c.Mapped() { + mappedAny = true + break + } + } + return b, mappedAny, err + } + // now we want to compute the actual container and bit counts after + // ops, and create a report of just the containers which got changed. + info.ContainerCount = 0 + info.BitCount = 0 + for citer.Next() { + k, c := citer.Value() + if c.Mapped() { + mappedAny = true + } + info.ContainerCount++ + info.BitCount += uint64(c.N()) + if c.flags&flagPristine != 0 { + continue + } + ci := c.info() + ci.Key = k + info.OpContainers = append(info.OpContainers, ci) + } + return b, mappedAny, err +} diff --git a/server.go b/server.go index 131a6971c..0afd2fa43 100644 --- a/server.go +++ b/server.go @@ -66,9 +66,10 @@ type Server struct { // nolint: maligned extensions []*ext.ExtensionInfo // External - systemInfo SystemInfo - gcNotifier GCNotifier - logger logger.Logger + systemInfo SystemInfo + gcNotifier GCNotifier + logger logger.Logger + snapshotQueue SnapshotQueue nodeID string uri URI @@ -533,6 +534,9 @@ func (s *Server) UpAndDown() error { func (s *Server) Open() error { s.logger.Printf("open server") + // Start background monitoring. + s.snapshotQueue = newSnapshotQueue(10, 2, s.logger) + // Log startup err := s.holder.logStartup() if err != nil { @@ -560,6 +564,9 @@ func (s *Server) Open() error { if err := s.holder.Open(); err != nil { return errors.Wrap(err, "opening Holder") } + // bring up the background tasks for the holder. + s.holder.SnapshotQueue = s.snapshotQueue + s.holder.Activate() if err := s.cluster.setNodeState(nodeStateReady); err != nil { return errors.Wrap(err, "setting nodeState") } @@ -571,7 +578,6 @@ func (s *Server) Open() error { // buffered channel. s.cluster.listenForJoins() - // Start background monitoring. s.wg.Add(3) go func() { defer s.wg.Done(); s.monitorAntiEntropy() }() go func() { defer s.wg.Done(); s.monitorRuntime() }() @@ -596,6 +602,11 @@ func (s *Server) Close() error { if s.holder != nil { errh = s.holder.Close() } + if s.snapshotQueue != nil { + s.holder.SnapshotQueue = nil + s.snapshotQueue.Stop() + s.snapshotQueue = nil + } // prefer to return holder error over cluster // error. This order is somewhat arbitrary. It would be better if we had // some way to combine all the errors, but probably not important enough to diff --git a/snapshotqueue.go b/snapshotqueue.go index 373e5c538..1bdcdb7a9 100644 --- a/snapshotqueue.go +++ b/snapshotqueue.go @@ -15,7 +15,10 @@ package pilosa import ( + "context" "fmt" + "io" + "math/bits" "os" "sync" "sync/atomic" @@ -39,12 +42,22 @@ import ( // Await, Enqueue, and Immediate should be called only with the fragment lock // held. // -// ScanHolder spawns a new goroutine. You don't need to use `go` on it. -type snapshotQueue interface { +// If you create a queue, it should get stopped at some point. The +// atomicSnapshotQueue implementation used as defaultSnapshotQueue has +// a Start function which will tell you whether it actually started a +// queue. This logic exists because in a normal server case, you probably +// want the queue to be shut down as part of server shutdown, but if you're +// running cluster tests, you probably want to start and shop the queue as +// part of the test, not stop it when any server terminates. +// +// It's less likely to be desireable to start/stop individual queues, +// because fragments use the defaultSnapshotQueue anyway. This design +// needs revisiting. +type SnapshotQueue interface { Immediate(*fragment) error Enqueue(*fragment) Await(*fragment) error - ScanHolder(*Holder) + ScanHolder(*Holder, chan struct{}) Stop() } @@ -53,7 +66,8 @@ type snapshotQueue interface { type queuelessSnapshotQueue struct{} func (q *queuelessSnapshotQueue) Enqueue(f *fragment) { - _ = f.snapshot() + // We don't actually try to enqueue the snapshot; it breaks things + // if a snapshot gets caused during a transaction. } func (q *queuelessSnapshotQueue) Await(f *fragment) error { @@ -64,20 +78,27 @@ func (q *queuelessSnapshotQueue) Immediate(f *fragment) error { return f.snapshot() } -func (q *queuelessSnapshotQueue) ScanHolder(h *Holder) { +func (q *queuelessSnapshotQueue) ScanHolder(h *Holder, done chan struct{}) { } func (q *queuelessSnapshotQueue) Stop() { } -// defaultSnapshotQueue is the fallback to use if none is available, -// and currently uses queueless -- it runs all snapshots immediately. -var defaultSnapshotQueue *queuelessSnapshotQueue +var defaultSnapshotQueue = &queuelessSnapshotQueue{} // newSnapshotQueue makes a new snapshot queue, of depth N, with // w worker threads. -func newSnapshotQueue(n int, w int, l logger.Logger) snapshotQueue { - sq := prioritySnapshotQueue{normal: make(chan snapshotRequest, n), urgent: make(chan snapshotRequest), background: make(chan snapshotRequest), done: make(chan struct{}), logger: l} +func newSnapshotQueue(n int, w int, l logger.Logger) SnapshotQueue { + ctx, cancel := context.WithCancel(context.Background()) + sq := prioritySnapshotQueue{ + normal: make(chan snapshotRequest, n), + urgent: make(chan snapshotRequest), + background: make(chan snapshotRequest), + ctx: ctx, + cancel: cancel, + maxOpN: 10000, + logger: l, + } if sq.logger == nil { sq.logger = logger.NewStandardLogger(os.Stderr) } @@ -105,50 +126,52 @@ type prioritySnapshotQueue struct { urgent chan snapshotRequest normal chan snapshotRequest background chan snapshotRequest - done chan struct{} + ctx context.Context + cancel context.CancelFunc mu sync.RWMutex scanWG, workerWG sync.WaitGroup + maxOpN int + observedOpN [16]uint32 stats struct { - enqueued uint64 - skipped uint64 + enqueued uint32 + skipped uint32 } } func (sq *prioritySnapshotQueue) spawnWorkers(w int) { sq.mu.Lock() defer sq.mu.Unlock() - if sq.done == nil { - sq.logger.Printf("prioritySnapshotQueue worker: no done channel, already done?") + if sq.ctx.Err() != nil { + sq.logger.Printf("prioritySnapshotQueue worker: already done") return } sq.workerWG.Add(w) for i := 0; i < w; i++ { - go sq.worker(sq.urgent, sq.normal, sq.background, sq.done) + go sq.worker(sq.ctx, sq.urgent, sq.normal, sq.background) } } -func (sq *prioritySnapshotQueue) worker(urgent, normal, background chan snapshotRequest, done chan struct{}) { - // We don't want a race condition on these. If they're non-nil when - // we get them, they should get closed at some point. If done is - // already nil, we shouldn't do anything. +func (sq *prioritySnapshotQueue) worker(ctx context.Context, urgent, normal, background chan snapshotRequest) { defer sq.workerWG.Done() + done := ctx.Done() ok := true var req snapshotRequest for ok { req.frag = nil - select { + case _, ok = <-done: case req, ok = <-urgent: default: select { + case _, ok = <-done: case req, ok = <-urgent: case req, ok = <-normal: default: select { + case _, ok = <-done: case req, ok = <-urgent: case req, ok = <-normal: case req, ok = <-background: - case _, ok = <-done: } } } @@ -182,17 +205,18 @@ func (sq *prioritySnapshotQueue) process(req snapshotRequest) { func (sq *prioritySnapshotQueue) Stop() { sq.mu.Lock() defer sq.mu.Unlock() - close(sq.done) + sq.cancel() // scanners need to be done before we close the other channels. sq.scanWG.Wait() - sq.done = nil close(sq.normal) sq.normal = nil close(sq.urgent) sq.urgent = nil close(sq.background) sq.background = nil - if sq.stats.skipped > 0 || sq.stats.enqueued > 1 { + enqueued := atomic.LoadUint32(&sq.stats.enqueued) + skipped := atomic.LoadUint32(&sq.stats.skipped) + if skipped > 0 || enqueued > 1 { sq.logger.Printf("snapshot queue: enqueued %d, skipped %d\n", sq.stats.enqueued, sq.stats.skipped) } } @@ -203,6 +227,7 @@ func (sq *prioritySnapshotQueue) Enqueue(f *fragment) { if f.snapshotPending { return } + sq.observeOpN(uint32(f.opN)) sq.mu.RLock() defer sq.mu.RUnlock() if sq.normal == nil { @@ -217,10 +242,10 @@ func (sq *prioritySnapshotQueue) Enqueue(f *fragment) { // try to enqueue snapshot select { case sq.normal <- snapshotRequest{frag: f, when: time.Now()}: - atomic.AddUint64(&sq.stats.enqueued, 1) + atomic.AddUint32(&sq.stats.enqueued, 1) return default: - atomic.AddUint64(&sq.stats.skipped, 1) + atomic.AddUint32(&sq.stats.skipped, 1) f.snapshotPending = false return } @@ -230,6 +255,11 @@ func (sq *prioritySnapshotQueue) Enqueue(f *fragment) { // held. Await waits on a condition variable inside f, associated with the // fragment's lock, so this does not conflict with the lock being used for // snapshots. +// +// Note that workers don't stop just because the queue's been stopped; only +// the background scanner is stopped. So an Await shouldn't block forever +// even if the queue gets shut down. If you're reading this, possibly that +// analysis is incorrect. func (sq *prioritySnapshotQueue) Await(f *fragment) (err error) { for f.snapshotPending { f.snapshotCond.Wait() @@ -252,6 +282,7 @@ func (sq *prioritySnapshotQueue) Immediate(f *fragment) error { return errors.New("requested immediate snapshot after snapshot queue was closed") } f.snapshotPending = true + sq.observeOpN(uint32(f.opN)) req := snapshotRequest{frag: f, when: time.Now()} // if the fragment was already in the work queue, it's *possible* // that the only available worker just picked it off the queue, and @@ -268,141 +299,202 @@ func (sq *prioritySnapshotQueue) Immediate(f *fragment) error { return sq.Await(f) } -// needsSnapshot determines whether a fragment probably wants snapshotting. -// Specifically, it looks for fragments not already marked to receive -// snapshots, but which have a high enough opN to justify a snapshot. This -// is only used from the background scan. -func (sq *prioritySnapshotQueue) needsSnapshot(f *fragment) bool { - if f == nil { - return false - } - f.mu.Lock() - defer f.mu.Unlock() - if f.snapshotPending { - return false - } - if f.opN > f.MaxOpN { - return true - } - return false -} - // ScanHolder spawns a goroutine which iterates through the holder's // indexes/fields/views/fragments, looking for fragments which have OpN // high enough to justify a snapshot but don't seem to have one pending. // It then dumps these in the low priority background queue. -func (sq *prioritySnapshotQueue) ScanHolder(h *Holder) { +func (sq *prioritySnapshotQueue) ScanHolder(h *Holder, done chan struct{}) { sq.mu.Lock() sq.scanWG.Add(1) - go sq.scanHolderWorker(h, sq.background, sq.done) + go sq.scanHolderWorker(h, sq.background, done) sq.mu.Unlock() } +// observeOpN reports that a given value of opN was "observed", meaning, +// we encountered a fragment which had that value. This happens for every +// enqueue/immediate, including enqueue attempts which fail to actually +// enter the queue, and it also happens for fragments noticed by the background +// scan but which don't have high enough opN to trigger a snapshot. +func (sq *prioritySnapshotQueue) observeOpN(n uint32) { + // aka "log2(n) + 1", or 0 for n==0 + pow2 := 32 - bits.LeadingZeros32(n) + // 15 == 16384. Our usual fragment maxOpN is 10k, so most fragments + // should end up in the 8k-16k bucket, rather than the 16k+ bucket, + // unless we've got a lot of ingests with large batches going on, + // in which case the 16k bucket will win. + if pow2 > 15 { + pow2 = 15 + } + // store in inverse order so the lowest slot in the array is the + // highest cardinality + atomic.AddUint32(&sq.observedOpN[15-pow2], 1) +} + +// computeMaxOpN tries to pick a reasonable new maxOpN for the background +// scan to use. On a quiet system, we want to gradually lower opN, picking +// the fragments with the highest opN values first, because those offer the +// largest benefit. So, whenever we check a fragment in the background, if we +// *don't* snapshot it, we'll "observe" its OpN value, and then we pick a +// value which picks up at least 1/4 of them. +// +// If there's ingest activity, the Immediate and Enqueue operations will +// "observe" the OpN of fragments submitted to them. This can drive OpN back +// up, if those fragments frequently have very high opN values, which reflects +// the fact that we have enough of that activity that we don't need the +// background scanner adding more. +// +// If we have enough ingest activity that the background scanner never actually +// gets to submit work, we'll rarely get here, because the background scanner +// will block until there's no snapshots pending for the normal workload. +// When we do, we'll probably pick a MaxOpN which is dominated by the ingest +// workload's opN values. So for instance, if everything coming in from the +// ingest workload has 10k or more items, because that's the default fragment +// maxOpN, that will probably set the background snapshot queue value to 8k. +func (sq *prioritySnapshotQueue) computeMaxOpN() { + sq.logger.Debugf("observedOpN by power of 2: %d\n", sq.observedOpN[:]) + total := uint32(0) + for i := range sq.observedOpN { + total += atomic.LoadUint32(&sq.observedOpN[i]) + } + target := (total / 4) + 1 + subTotal := uint32(0) + for i := range sq.observedOpN { + v := atomic.LoadUint32(&sq.observedOpN[i]) + subTotal += v + if subTotal >= target { + prevMaxOpN := sq.maxOpN + sq.maxOpN = (1 << (15 - uint(i))) / 2 + if sq.maxOpN > 0 { + sq.maxOpN-- + } + if prevMaxOpN != sq.maxOpN { + sq.logger.Printf("background scan: %d/%d fragments considered have opN %d or higher\n", + subTotal, total, sq.maxOpN) + } + break + } + } + // It's conceptually possible that we'll miss a couple of observations + // here but that's not really important. This is all pretty approximate. + for i := range sq.observedOpN { + atomic.StoreUint32(&sq.observedOpN[i], 0) + } +} + +// prioritySnapshotQueueScanner is the data type that implements HolderOperator +// and represents a single scan of a holder, with a given maxOpN. +type prioritySnapshotQueueScanner struct { + HolderFilterAll + HolderProcessNone + sq *prioritySnapshotQueue + holder *Holder + queue chan snapshotRequest + ctx context.Context + maxOpN int + seen, hits, counter int +} + +func (s *prioritySnapshotQueueScanner) ProcessFragment(f *fragment) error { + if f == nil { + return nil + } + s.seen++ + // we can't defer this reasonably, because otherwise we'll keep + // the fragment locked forever if we end up trying to send it + // to the queue, but the workers are busy on other fragments. + f.mu.Lock() + open := f.open + snapshotPending, opN := f.snapshotPending, f.opN + f.mu.Unlock() + + // a pending snapshot is one that is either in the normal or + // immediate queue, or is trying to get into the normal queue + // and about to fail, but either way, it already got observed + // there, so we don't need to observe it here. A closed fragment + // doesn't matter to us -- it should be a transient state that + // happens during a shutdown, or shouldn't happen, but we don't + // care about it. + if snapshotPending || !open { + return nil + } + if opN <= s.maxOpN { + // observe the value but don't do a snapshot + s.sq.observeOpN(uint32(opN)) + s.counter++ + if s.counter == 1000 { + select { + case <-time.After(1 * time.Second): + case <-s.ctx.Done(): + return io.EOF + } + s.counter = 0 + } + return nil + } + // we don't observe values when we decide to trigger a snapshot, + // because those values will be changing anyway. we could also + // observe them as zero, but that's also sort of wrong. + s.hits++ + select { + case s.queue <- snapshotRequest{frag: f, when: time.Now()}: + s.sq.logger.Debugf("found fragment needing snapshot: %s\n", f.path) + case <-s.ctx.Done(): + return io.EOF + } + return nil + +} + +func contextMergedWithStructChan(ctx context.Context, ch chan struct{}) (context.Context, context.CancelFunc) { + canCancel, cancel := context.WithCancel(ctx) + go func() { + select { + case <-ctx.Done(): + cancel() + case <-ch: + cancel() + case <-canCancel.Done(): + // don't need to cancel, but do need to exit this + // function + } + }() + return canCancel, cancel +} + // scanHolderWorker is a background task that scans a holder looking for // fragments which need snapshots taken. It's the cleanup task for snapshots // that would have been requested by Enqueue, but the queue was full. func (sq *prioritySnapshotQueue) scanHolderWorker(h *Holder, background chan snapshotRequest, done chan struct{}) { defer sq.scanWG.Done() - var indexNames, fieldNames, viewNames []string - var fragNums []uint64 + ctx, cancel := contextMergedWithStructChan(sq.ctx, done) + defer cancel() + scanner := &prioritySnapshotQueueScanner{ + sq: sq, + holder: h, + queue: background, + ctx: sq.ctx, + maxOpN: sq.maxOpN, + } for { - // To avoid abusing things, cap activity rate; every time we finish - // the holder, or every couple hundred fragments considered, we - // pause for a bit. - counter := 0 - hits := 0 - h.mu.Lock() - indexNames = indexNames[:0] - for indexName := range h.indexes { - indexNames = append(indexNames, indexName) + err := h.Process(ctx, scanner) + if err != nil { + return } - h.mu.Unlock() - for _, indexName := range indexNames { - h.mu.Lock() - index := h.indexes[indexName] - h.mu.Unlock() - if index == nil { - continue - } - fieldNames = fieldNames[:0] - index.mu.Lock() - for fieldName := range index.fields { - fieldNames = append(fieldNames, fieldName) - } - index.mu.Unlock() - for _, fieldName := range fieldNames { - index.mu.Lock() - field := index.fields[fieldName] - index.mu.Unlock() - if field == nil { - continue - } - viewNames = viewNames[:0] - field.mu.Lock() - for viewName := range field.viewMap { - viewNames = append(viewNames, viewName) - } - field.mu.Unlock() - for _, viewName := range viewNames { - field.mu.Lock() - view := field.viewMap[viewName] - field.mu.Unlock() - if view == nil { - continue - } - fragNums := fragNums[:0] - view.mu.Lock() - for fragNum := range view.fragments { - fragNums = append(fragNums, fragNum) - } - view.mu.Unlock() - for _, fragNum := range fragNums { - view.mu.Lock() - frag := view.fragments[fragNum] - view.mu.Unlock() - if sq.needsSnapshot(frag) { - hits++ - select { - case background <- snapshotRequest{frag: frag, when: time.Now()}: - sq.logger.Debugf("found fragment needing snapshot: %s\n", frag.path) - case <-done: - return - } - } else { - // Count fragments examined *without* finding anything that - // needed a snapshot. When we find things that need snapshots, - // the time it takes the workers to respond to us is enough - // of a delay to keep us from eating every CPU. So, if a lot - // of things need snapshots, and the workers aren't doing - // anything else, ScanHolder will mostly keep them saturated. - // If they're busy, we'll block forever in the write to the - // background queue. If there's nothing that needs snapshots, - // we pause frequently for a second or so at a time. - counter++ - if counter == 100 { - select { - case <-time.After(1 * time.Second): - case <-done: - return - } - counter = 0 - } - } - } - } - } - } - if hits > 0 { - sq.logger.Printf("background scan: %d fragments needed snapshots\n", hits) - hits = 0 + + if scanner.hits > 0 { + sq.logger.Printf("background scan: %d/%d fragments needed snapshots\n", scanner.hits, scanner.seen) + scanner.hits = 0 } else { sq.logger.Debugf("background scan: no fragments needed snapshots, waiting\n") // No reason to be active if we're not finding anything. select { case <-time.After(60 * time.Second): - case <-done: + case <-ctx.Done(): return } } + scanner.seen = 0 + sq.computeMaxOpN() + scanner.maxOpN = sq.maxOpN } } diff --git a/test/field.go b/test/field.go index b95e4f88b..f56672834 100644 --- a/test/field.go +++ b/test/field.go @@ -33,7 +33,7 @@ func newField(opts pilosa.FieldOption) *Field { if err != nil { panic(err) } - field, err := pilosa.NewField(path, "i", "f", opts) + field, err := pilosa.NewField(pilosa.NewHolder(pilosa.DefaultPartitionN), path, "i", "f", opts) if err != nil { panic(err) } @@ -63,7 +63,7 @@ func (f *Field) reopen() error { } path, index, name := f.Path(), f.Index(), f.Name() - f.Field, err = pilosa.NewField(path, index, name, pilosa.OptFieldTypeDefault()) + f.Field, err = pilosa.NewField(pilosa.NewHolder(pilosa.DefaultPartitionN), path, index, name, pilosa.OptFieldTypeDefault()) if err != nil { return err } diff --git a/test/index.go b/test/index.go index bf65c6ae2..b90702ae2 100644 --- a/test/index.go +++ b/test/index.go @@ -32,7 +32,7 @@ func newIndex() *Index { if err != nil { panic(err) } - index, err := pilosa.NewIndex(path, "i", pilosa.DefaultPartitionN) + index, err := pilosa.NewIndex(pilosa.NewHolder(pilosa.DefaultPartitionN), path, "i") if err != nil { panic(err) } @@ -62,7 +62,7 @@ func (i *Index) Reopen() error { } path, name := i.Path(), i.Name() - i.Index, err = pilosa.NewIndex(path, name, pilosa.DefaultPartitionN) + i.Index, err = pilosa.NewIndex(pilosa.NewHolder(pilosa.DefaultPartitionN), path, name) if err != nil { return err } diff --git a/view.go b/view.go index 478725e1e..af73b7fe2 100644 --- a/view.go +++ b/view.go @@ -26,7 +26,6 @@ import ( "sync/atomic" "time" - "github.com/pilosa/pilosa/v2/logger" "github.com/pilosa/pilosa/v2/pql" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/stats" @@ -49,6 +48,8 @@ type view struct { field string name string + holder *Holder + fieldType string cacheType string cacheSize uint32 @@ -56,24 +57,24 @@ type view struct { // Fragments by shard. fragments map[uint64]*fragment - broadcaster broadcaster - stats stats.StatsClient - rowAttrStore AttrStore - logger logger.Logger - snapshotQueue snapshotQueue + broadcaster broadcaster + stats stats.StatsClient + rowAttrStore AttrStore knownShards *roaring.Bitmap knownShardsCopied uint32 } // newView returns a new instance of View. -func newView(path, index, field, name string, fieldOptions FieldOptions) *view { +func newView(holder *Holder, path, index, field, name string, fieldOptions FieldOptions) *view { return &view{ path: path, index: index, field: field, name: name, + holder: holder, + fieldType: fieldOptions.Type, cacheType: fieldOptions.CacheType, cacheSize: fieldOptions.CacheSize, @@ -82,7 +83,6 @@ func newView(path, index, field, name string, fieldOptions FieldOptions) *view { broadcaster: NopBroadcaster, stats: stats.NopStatsClient, - logger: logger.NopLogger, knownShards: roaring.NewSliceBitmap(), } } @@ -133,14 +133,14 @@ func (v *view) open() error { if err := func() error { // Ensure the view's path exists. - v.logger.Debugf("ensure view path exists: %s", v.path) + v.holder.Logger.Debugf("ensure view path exists: %s", v.path) if err := os.MkdirAll(v.path, 0777); err != nil { return errors.Wrap(err, "creating view directory") } else if err := os.MkdirAll(filepath.Join(v.path, "fragments"), 0777); err != nil { return errors.Wrap(err, "creating fragments directory") } - v.logger.Debugf("open fragments for index/field/view: %s/%s/%s", v.index, v.field, v.name) + v.holder.Logger.Debugf("open fragments for index/field/view: %s/%s/%s", v.index, v.field, v.name) if err := v.openFragments(); err != nil { return errors.Wrap(err, "opening fragments") } @@ -151,7 +151,7 @@ func (v *view) open() error { return err } - v.logger.Debugf("successfully opened index/field/view: %s/%s/%s", v.index, v.field, v.name) + v.holder.Logger.Debugf("successfully opened index/field/view: %s/%s/%s", v.index, v.field, v.name) return nil } @@ -190,12 +190,12 @@ fileLoop: // Parse filename into integer. shard, err := strconv.ParseUint(filepath.Base(fi.Name()), 10, 64) if err != nil { - v.logger.Debugf("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", v.index, v.field, v.name, fi.Name()) + v.holder.Logger.Debugf("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", v.index, v.field, v.name, fi.Name()) continue } workQueue <- struct{}{} - v.logger.Debugf("open index/field/view/fragment: %s/%s/%s/%d", v.index, v.field, v.name, shard) + v.holder.Logger.Debugf("open index/field/view/fragment: %s/%s/%s/%d", v.index, v.field, v.name, shard) eg.Go(func() error { defer func() { <-workQueue @@ -205,7 +205,7 @@ fileLoop: return fmt.Errorf("open fragment: shard=%d, err=%s", frag.shard, err) } frag.RowAttrStore = v.rowAttrStore - v.logger.Debugf("add index/field/view/fragment to view.fragments: %s/%s/%s/%d", v.index, v.field, v.name, shard) + v.holder.Logger.Debugf("add index/field/view/fragment to view.fragments: %s/%s/%s/%d", v.index, v.field, v.name, shard) mu.Lock() v.fragments[frag.shard] = frag v.addKnownShard(frag.shard) @@ -342,7 +342,7 @@ func (v *view) notifyIfNewShard(shard uint64) { // Broadcast a message that a new max shard was just created. err := v.broadcaster.SendSync(msg) if err != nil { - v.logger.Printf("broadcasting create shard: %v", err) + v.holder.Logger.Printf("broadcasting create shard: %v", err) } close(broadcastChan) }() @@ -352,19 +352,15 @@ func (v *view) notifyIfNewShard(shard uint64) { select { case <-broadcastChan: case <-time.After(50 * time.Millisecond): - v.logger.Debugf("broadcasting create shard took >50ms") + v.holder.Logger.Debugf("broadcasting create shard took >50ms") } } func (v *view) newFragment(path string, shard uint64) *fragment { - frag := newFragment(path, v.index, v.field, v.name, shard, v.flags()) + frag := newFragment(v.holder, path, v.index, v.field, v.name, shard, v.flags()) frag.CacheType = v.cacheType frag.CacheSize = v.cacheSize - frag.Logger = v.logger frag.stats = v.stats - if v.snapshotQueue != nil { - frag.snapshotQueue = v.snapshotQueue - } if v.fieldType == FieldTypeMutex { frag.mutexVector = newRowsVector(frag) } else if v.fieldType == FieldTypeBool { @@ -382,7 +378,7 @@ func (v *view) deleteFragment(shard uint64) error { return ErrFragmentNotFound } - v.logger.Printf("delete fragment: (%s/%s/%s) %d", v.index, v.field, v.name, shard) + v.holder.Logger.Printf("delete fragment: (%s/%s/%s) %d", v.index, v.field, v.name, shard) // Close data files before deletion. if err := fragment.Close(); err != nil { @@ -396,7 +392,7 @@ func (v *view) deleteFragment(shard uint64) error { // Delete fragment cache file. if err := os.Remove(fragment.cachePath()); err != nil { - v.logger.Printf("no cache file to delete for shard %d", shard) + v.holder.Logger.Printf("no cache file to delete for shard %d", shard) } delete(v.fragments, shard) diff --git a/view_internal_test.go b/view_internal_test.go index bb50003d2..bbb1a20f4 100644 --- a/view_internal_test.go +++ b/view_internal_test.go @@ -34,7 +34,7 @@ func mustOpenView(index, field, name string) *view { CacheSize: DefaultCacheSize, } - v := newView(path, index, field, name, fo) + v := newView(NewHolder(DefaultPartitionN), path, index, field, name, fo) if err := v.open(); err != nil { panic(err) }