mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 09:05:55 +00:00
Merge pull request #1901 from molecula/fb-1114-2-rip-inspect
remove inspect command
This commit is contained in:
Matthew Jaffee
GitHub
commit
fac5bdbfbf
11 changed files with 0 additions and 1147 deletions
4
api.go
4
api.go
|
|
@ -2307,10 +2307,6 @@ 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")
|
||||
|
|
|
|||
|
|
@ -1,43 +0,0 @@
|
|||
// Copyright 2021 Molecula Corp. All rights reserved.
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
"github.com/molecula/featurebase/v3/ctl"
|
||||
)
|
||||
|
||||
var inspector *ctl.InspectCommand
|
||||
|
||||
func newInspectCommand(stdin io.Reader, stdout, stderr io.Writer) *cobra.Command {
|
||||
inspector = ctl.NewInspectCommand(stdin, stdout, stderr)
|
||||
|
||||
inspectCmd := &cobra.Command{
|
||||
Use: "inspect",
|
||||
Short: "Get stats on a FeatureBase data file.",
|
||||
Long: `
|
||||
Inspects a data file and provides stats.
|
||||
`,
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
if len(args) == 0 {
|
||||
return fmt.Errorf("path required")
|
||||
} else if len(args) > 1 {
|
||||
return fmt.Errorf("only one path allowed")
|
||||
}
|
||||
inspector.Path = args[0]
|
||||
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
|
||||
}
|
||||
|
|
@ -1,29 +0,0 @@
|
|||
// Copyright 2021 Molecula Corp. All rights reserved.
|
||||
package cmd_test
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestInspectHelp(t *testing.T) {
|
||||
output, err := ExecNewRootCommand(t, "inspect", "--help")
|
||||
if !strings.Contains(output, "Usage:") ||
|
||||
!strings.Contains(output, "featurebase inspect") || err != nil {
|
||||
t.Fatalf("Command 'inspect --help' not working, err: '%v', output: '%s'", err, output)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInspectNoPath(t *testing.T) {
|
||||
output, err := ExecNewRootCommand(t, "inspect")
|
||||
if !strings.Contains(err.Error(), "path required") {
|
||||
t.Fatalf("Command 'inspect' without args should error but: err: '%v', output: '%v'", err, output)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInspectMultiPath(t *testing.T) {
|
||||
output, err := ExecNewRootCommand(t, "inspect", "one", "two")
|
||||
if !strings.Contains(err.Error(), "only one path") {
|
||||
t.Fatalf("Command 'inspect' without args should error but: err: '%v', output: '%v'", err, output)
|
||||
}
|
||||
}
|
||||
|
|
@ -57,7 +57,6 @@ at https://docs.molecula.cloud/.
|
|||
rc.AddCommand(newExportCommand(stdin, stdout, stderr))
|
||||
rc.AddCommand(newGenerateConfigCommand(stdin, stdout, stderr))
|
||||
rc.AddCommand(newImportCommand(stdin, stdout, stderr))
|
||||
rc.AddCommand(newInspectCommand(stdin, stdout, stderr))
|
||||
rc.AddCommand(newRBFCommand(stdin, stdout, stderr))
|
||||
rc.AddCommand(newServeCmd(stdin, stdout, stderr))
|
||||
rc.AddCommand(newHolderCmd(stdin, stdout, stderr))
|
||||
|
|
|
|||
394
ctl/inspect.go
394
ctl/inspect.go
|
|
@ -1,394 +0,0 @@
|
|||
// Copyright 2021 Molecula Corp. All rights reserved.
|
||||
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/molecula/featurebase/v3"
|
||||
"github.com/molecula/featurebase/v3/pb"
|
||||
"github.com/molecula/featurebase/v3/roaring"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
// InspectCommand represents a command for inspecting fragment data files.
|
||||
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
|
||||
}
|
||||
|
||||
// NewInspectCommand returns a new instance of InspectCommand.
|
||||
func NewInspectCommand(stdin io.Reader, stdout, stderr io.Writer) *InspectCommand {
|
||||
return &InspectCommand{
|
||||
CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
|
||||
}
|
||||
}
|
||||
|
||||
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(ctx context.Context) error {
|
||||
// Open file handle.
|
||||
f, err := os.Open(cmd.Path)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "opening file")
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
fi, err := f.Stat()
|
||||
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 pb.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(path, nil)
|
||||
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 {
|
||||
return errors.Wrap(err, "mmapping")
|
||||
}
|
||||
defer func() {
|
||||
err := syscall.Munmap(data)
|
||||
if err != nil {
|
||||
fmt.Fprintf(cmd.Stderr, "inspect command: munmap failed: %v", err)
|
||||
}
|
||||
}()
|
||||
mappedFrom := uintptr(unsafe.Pointer(&data[0]))
|
||||
mappedTo := mappedFrom + uintptr(len(data))
|
||||
// Attach the mmap file to the bitmap.
|
||||
t := time.Now()
|
||||
fmt.Fprintf(cmd.Stderr, "inspecting bitmap...")
|
||||
var info roaring.BitmapInfo
|
||||
bitmap, _, 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")
|
||||
}
|
||||
mappedIn, mappedOut, unmappedIn, errs, err := bitmap.SanityCheckMapping(mappedFrom, mappedTo)
|
||||
if err != nil {
|
||||
fmt.Fprintf(cmd.Stderr, "sanity check: %d mapped in, %d mapped out, %d unmapped in, %d errors\n",
|
||||
mappedIn, mappedOut, unmappedIn, errs)
|
||||
fmt.Fprintf(cmd.Stderr, "last error: %v\n", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
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, " 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.
|
||||
if !cmd.Quiet {
|
||||
if info.ContainerCount > 0 {
|
||||
cmd.PrintContainers(info, pC)
|
||||
}
|
||||
if info.Ops > 0 {
|
||||
cmd.PrintOps(info)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,48 +0,0 @@
|
|||
// Copyright 2021 Molecula Corp. All rights reserved.
|
||||
package ctl
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"io"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/molecula/featurebase/v3/testhook"
|
||||
)
|
||||
|
||||
func TestInspectCommand_Run(t *testing.T) {
|
||||
rder := []byte{}
|
||||
stdin := bytes.NewReader(rder)
|
||||
r, w, _ := os.Pipe()
|
||||
|
||||
cm := NewInspectCommand(stdin, w, w)
|
||||
file, err := testhook.TempFile(t, "inspectTest")
|
||||
if err != nil {
|
||||
t.Fatalf("Error creating tempfile: %s", err)
|
||||
}
|
||||
_, err = file.Write([]byte("12358267538963"))
|
||||
if err != nil {
|
||||
t.Fatalf("writing to tempfile: %v", err)
|
||||
}
|
||||
file.Close()
|
||||
cm.Path = file.Name()
|
||||
err = cm.Run(context.Background())
|
||||
expectedError := "inspecting: "
|
||||
if !strings.Contains(err.Error(), expectedError) {
|
||||
t.Fatalf("expected error '%s', got '%v'", expectedError, err)
|
||||
}
|
||||
|
||||
w.Close()
|
||||
var buf bytes.Buffer
|
||||
_, err = io.Copy(&buf, r)
|
||||
if err != nil {
|
||||
t.Fatalf("copying data: %v", err)
|
||||
}
|
||||
if !strings.Contains(buf.String(), "inspecting bitmap...") {
|
||||
t.Fatalf("Inspect doesn't work: %s", err)
|
||||
}
|
||||
|
||||
// Todo: need correct roaring file for happy path
|
||||
}
|
||||
12
fragment.go
12
fragment.go
|
|
@ -214,18 +214,6 @@ func (f *fragment) Index() *Index {
|
|||
return f.holder.Index(f.index())
|
||||
}
|
||||
|
||||
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()
|
||||
|
|
|
|||
28
handler.go
28
handler.go
|
|
@ -410,31 +410,3 @@ 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
|
||||
}
|
||||
}
|
||||
|
|
|
|||
437
holder.go
437
holder.go
|
|
@ -7,10 +7,8 @@ import (
|
|||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
|
@ -137,14 +135,6 @@ type Holder struct {
|
|||
// HolderOpts holds information about the holder which other things might want
|
||||
// to look up later while using the holder.
|
||||
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
|
||||
|
||||
// StorageBackend controls the tx/storage engine we instatiate. Set by
|
||||
// server.go OptServerStorageConfig
|
||||
StorageBackend string
|
||||
|
|
@ -295,280 +285,6 @@ func (h *Holder) IndexesPath() string {
|
|||
return filepath.Join(h.path, IndexesDir)
|
||||
}
|
||||
|
||||
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
|
||||
|
|
@ -773,7 +489,6 @@ func (h *Holder) processForeignIndexFields() error {
|
|||
|
||||
// Close closes all open fragments.
|
||||
func (h *Holder) Close() error {
|
||||
|
||||
if h == nil {
|
||||
return nil
|
||||
}
|
||||
|
|
@ -1951,130 +1666,6 @@ 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 fieldNames, viewNames []string
|
||||
var fragNums []uint64
|
||||
|
||||
indexes := h.Indexes()
|
||||
for _, idx := range indexes {
|
||||
if err = ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
if idx == nil {
|
||||
continue
|
||||
}
|
||||
indexName := idx.name
|
||||
process, recurse := op.CheckIndex(indexName)
|
||||
if !process && !recurse {
|
||||
continue
|
||||
}
|
||||
|
||||
if err = ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
if process {
|
||||
err = op.ProcessIndex(idx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if !recurse {
|
||||
continue
|
||||
}
|
||||
fieldNames = fieldNames[:0]
|
||||
idx.mu.Lock()
|
||||
for fieldName := range idx.fields {
|
||||
fieldNames = append(fieldNames, fieldName)
|
||||
}
|
||||
idx.mu.Unlock()
|
||||
for _, fieldName := range fieldNames {
|
||||
if err = ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
process, recurse := op.CheckField(idx.name, fieldName)
|
||||
if !process && !recurse {
|
||||
continue
|
||||
}
|
||||
idx.mu.Lock()
|
||||
field := idx.fields[fieldName]
|
||||
idx.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
|
||||
}
|
||||
|
||||
// used by Index.openFields(), enabling Tx / Txf by telling
|
||||
// the holder about its own indexes.
|
||||
func (h *Holder) addIndex(idx *Index) {
|
||||
|
|
@ -2095,34 +1686,6 @@ func (h *Holder) BeginTx(writable bool, idx *Index, shard uint64) (Tx, error) {
|
|||
return h.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard}), nil
|
||||
}
|
||||
|
||||
func (h *Holder) HasRoaringData() (has bool, err error) {
|
||||
idxs := h.Indexes()
|
||||
for _, idx := range idxs {
|
||||
paths, err := listFilesUnderDir(idx.path, false, "", true)
|
||||
if err != nil {
|
||||
return false, errors.Wrap(err, "HasRoaringData listFilesUnderDir")
|
||||
}
|
||||
index := idx.name
|
||||
|
||||
for _, relpath := range paths {
|
||||
field, view, shard, err := fragmentSpecFromRoaringPath(relpath)
|
||||
if err != nil {
|
||||
continue // ignore .meta paths
|
||||
}
|
||||
abspath := idx.path + sep + relpath
|
||||
|
||||
hasData, err := roaringFragmentHasData(abspath, index, field, view, shard)
|
||||
if err != nil {
|
||||
return false, errors.Wrap(err, "HasRoaringData roaringFragmentHasData")
|
||||
}
|
||||
if hasData {
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func decodeCreateIndexMessage(ser Serializer, b []byte) (*CreateIndexMessage, error) {
|
||||
var cim CreateIndexMessage
|
||||
if err := ser.Unmarshal(b, &cim); err != nil {
|
||||
|
|
|
|||
|
|
@ -2,82 +2,11 @@
|
|||
package pilosa
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/molecula/featurebase/v3/disco"
|
||||
"github.com/molecula/featurebase/v3/testhook"
|
||||
)
|
||||
|
||||
var _ = fmt.Printf
|
||||
|
||||
type testHolderOperator struct {
|
||||
indexSeen, indexProcessed int
|
||||
fieldSeen, fieldProcessed int
|
||||
viewSeen, viewProcessed int
|
||||
fragmentSeen, fragmentProcessed int
|
||||
waitHere chan struct{}
|
||||
}
|
||||
|
||||
func (t *testHolderOperator) CheckIndex(string) (bool, bool) {
|
||||
t.indexSeen++
|
||||
return true, true
|
||||
}
|
||||
|
||||
func (t *testHolderOperator) CheckField(string, string) (bool, bool) {
|
||||
t.fieldSeen++
|
||||
return true, true
|
||||
}
|
||||
|
||||
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(tb testing.TB, backend string) (*Holder, string, error) {
|
||||
path, err := testhook.TempDir(tb, "pilosa-")
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
cfg := mustHolderConfig()
|
||||
if backend != "" {
|
||||
cfg.StorageConfig.Backend = backend
|
||||
cfg.StorageConfig.FsyncEnabled = false
|
||||
}
|
||||
h := NewHolder(path, cfg)
|
||||
return h, path, h.Open()
|
||||
}
|
||||
|
||||
func testSetBit(t *testing.T, h *Holder, index, field string, rowID, columnID uint64) {
|
||||
|
||||
idx, err := h.CreateIndexIfNotExists(index, IndexOptions{})
|
||||
|
|
@ -95,85 +24,6 @@ func testSetBit(t *testing.T, h *Holder, index, field string, rowID, columnID ui
|
|||
}
|
||||
}
|
||||
|
||||
func TestHolderOperatorProcess(t *testing.T) {
|
||||
h, path, err := makeHolder(t, "")
|
||||
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(t, "")
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// mustHolderConfig sets up a default holder config for tests.
|
||||
func mustHolderConfig() *HolderConfig {
|
||||
cfg := DefaultHolderConfig()
|
||||
|
|
|
|||
|
|
@ -236,7 +236,6 @@ func TestClusterStuff(t *testing.T) {
|
|||
t.Fatalf("restore failed: %v", err)
|
||||
}
|
||||
|
||||
fmt.Println("pausing all featurebasen")
|
||||
if err = sendCmd("docker", "pause", container(t, "pilosa1")); err != nil {
|
||||
t.Fatalf("sending pause command: %v", err)
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue