Merge pull request #130 from benbjohnson/check

Add pilosactl check.

cmd/pilosactl/main.go

@@ -11,6 +11,7 @@ import (
 	"log"
 	"math/rand"
 	"os"
+	"path/filepath"
 	"sort"
 	"strconv"
 	"strings"
@@ -87,6 +88,7 @@ The commands are:
 	backup     backs up a frame to an archive file
 	restore    restores a frame from an archive file
 	inspect    inspects fragment data files
+	check      performs a consistency check of data files
 	bench      benchmarks operations
 
 Use the "-h" flag with any command for more information.
@@ -123,6 +125,8 @@ func (m *Main) ParseFlags(args []string) error {
 		m.Cmd = NewRestoreCommand(m.Stdin, m.Stdout, m.Stderr)
 	case "inspect":
 		m.Cmd = NewInspectCommand(m.Stdin, m.Stdout, m.Stderr)
+	case "check":
+		m.Cmd = NewCheckCommand(m.Stdin, m.Stdout, m.Stderr)
 	case "bench":
 		m.Cmd = NewBenchCommand(m.Stdin, m.Stdout, m.Stderr)
 	default:
@@ -818,7 +822,7 @@ func (cmd *InspectCommand) ParseFlags(args []string) error {
 // Usage returns the usage message to be printed.
 func (cmd *InspectCommand) Usage() string {
 	return strings.TrimSpace(`
-usage: pilosactl inspect PATH 
+usage: pilosactl inspect PATH
 
 Inspects a data file and provides stats.
 
@@ -885,6 +889,135 @@ func (cmd *InspectCommand) Run() error {
 	return nil
 }
 
+// CheckCommand represents a command for performing consistency checks on data files.
+type CheckCommand struct {
+	// Data file paths.
+	Paths []string
+
+	// Standard input/output
+	Stdin  io.Reader
+	Stdout io.Writer
+	Stderr io.Writer
+}
+
+// NewCheckCommand returns a new instance of CheckCommand.
+func NewCheckCommand(stdin io.Reader, stdout, stderr io.Writer) *CheckCommand {
+	return &CheckCommand{
+		Stdin:  stdin,
+		Stdout: stdout,
+		Stderr: stderr,
+	}
+}
+
+// ParseFlags parses command line flags from args.
+func (cmd *CheckCommand) ParseFlags(args []string) error {
+	fs := flag.NewFlagSet("pilosactl", flag.ContinueOnError)
+	fs.SetOutput(ioutil.Discard)
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+
+	// Parse path.
+	if fs.NArg() == 0 {
+		return errors.New("path required")
+	}
+	cmd.Paths = fs.Args()
+
+	return nil
+}
+
+// Usage returns the usage message to be printed.
+func (cmd *CheckCommand) Usage() string {
+	return strings.TrimSpace(`
+usage: pilosactl check PATHS...
+
+Performs a consistency check on data files.
+
+`)
+}
+
+// Run executes the main program execution.
+func (cmd *CheckCommand) Run() error {
+	for _, path := range cmd.Paths {
+		switch filepath.Ext(path) {
+		case "":
+			if err := cmd.checkBitmapFile(path); err != nil {
+				return err
+			}
+
+		case ".cache":
+			if err := cmd.checkCacheFile(path); err != nil {
+				return err
+			}
+
+		case ".snapshotting":
+			if err := cmd.checkSnapshotFile(path); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+// checkBitmapFile performs a consistency check on path for a roaring bitmap file.
+func (cmd *CheckCommand) checkBitmapFile(path string) error {
+	// Open file handle.
+	f, err := os.Open(path)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	fi, err := f.Stat()
+	if err != nil {
+		return err
+	}
+
+	// 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 err
+	}
+	defer syscall.Munmap(data)
+
+	// Attach the mmap file to the bitmap.
+	bm := roaring.NewBitmap()
+	if err := bm.UnmarshalBinary(data); err != nil {
+		return err
+	}
+
+	// Perform consistency check.
+	if err := bm.Check(); err != nil {
+		// Print returned errors.
+		switch err := err.(type) {
+		case roaring.ErrorList:
+			for i := range err {
+				fmt.Fprintf(cmd.Stdout, "%s: %s\n", path, err[i].Error())
+			}
+		default:
+			fmt.Fprintf(cmd.Stdout, "%s: %s\n", path, err.Error())
+		}
+	}
+
+	// Print success message if no errors were found.
+	fmt.Fprintf(cmd.Stdout, "%s: ok\n", path)
+
+	return nil
+}
+
+// checkCacheFile performs a consistency check on path for a cache file.
+func (cmd *CheckCommand) checkCacheFile(path string) error {
+	fmt.Fprintf(cmd.Stderr, "%s: ignoring cache file\n", path)
+	return nil
+}
+
+// checkSnapshotFile performs a consistency check on path for a snapshot file.
+func (cmd *CheckCommand) checkSnapshotFile(path string) error {
+	fmt.Fprintf(cmd.Stderr, "%s: ignoring snapshot file\n", path)
+	return nil
+}
+
 // BenchCommand represents a command for benchmarking database operations.
 type BenchCommand struct {
 	// Destination host and port.

roaring/roaring.go

@@ -278,6 +278,7 @@ func (b *Bitmap) container(key uint64) *container {
 	}
 	return b.containers[i]
 }
+
 func insertU64(original []uint64, position int, value uint64) []uint64 {
 	l := len(original)
 	target := original
@@ -574,6 +575,29 @@ func (b *Bitmap) Info() BitmapInfo {
 	return info
 }
 
+// Check performs a consistency check on the bitmap. Returns nil if consistent.
+func (b *Bitmap) Check() error {
+	var a ErrorList
+
+	// Check keys/containers match. Return immediately if this happens.
+	if len(b.keys) != len(b.containers) {
+		a.Append(fmt.Errorf("key/container count mismatch: %d != %d", len(b.keys), len(b.containers)))
+		return a
+	}
+
+	// Check each container.
+	for i, c := range b.containers {
+		if err := c.check(); err != nil {
+			a.AppendWithPrefix(err, fmt.Sprintf("%d/", b.keys[i]))
+		}
+	}
+
+	if len(a) == 0 {
+		return nil
+	}
+	return a
+}
+
 // BitmapInfo represents a point-in-time snapshot of bitmap stats.
 type BitmapInfo struct {
 	OpN        int
@@ -1050,6 +1074,26 @@ func (c *container) info() ContainerInfo {
 	return info
 }
 
+// check performs a consistency check on the container.
+func (c *container) check() error {
+	var a ErrorList
+
+	if c.n <= ArrayMaxSize {
+		if len(c.array) != c.n {
+			a.Append(fmt.Errorf("array count mismatch: count=%d, n=%d", len(c.array), c.n))
+		}
+	} else {
+		if n := c.bitmapCountRange(0, uint32(len(c.bitmap)*64)); n != c.n {
+			a.Append(fmt.Errorf("bitmap count mismatch: count=%d, n=%d", n, c.n))
+		}
+	}
+
+	if a == nil {
+		return nil
+	}
+	return a
+}
+
 // ContainerInfo represents a point-in-time snapshot of container stats.
 type ContainerInfo struct {
 	Key     uint64         // container key
@@ -1680,3 +1724,38 @@ func (itr *bufBitmapIterator) unread() {
 	}
 	itr.buf.full = true
 }
+
+// ErrorList represents a list of errors.
+type ErrorList []error
+
+func (a ErrorList) Error() string {
+	switch len(a) {
+	case 0:
+		return "no errors"
+	case 1:
+		return a[0].Error()
+	}
+	return fmt.Sprintf("%s (and %d more errors)", a[0], len(a)-1)
+}
+
+// Append appends an error to the list. If err is an ErrorList then all errors are appended.
+func (a *ErrorList) Append(err error) {
+	switch err := err.(type) {
+	case ErrorList:
+		*a = append(*a, err...)
+	default:
+		*a = append(*a, err)
+	}
+}
+
+// AppendWithPrefix appends an error to the list and includes a prefix.
+func (a *ErrorList) AppendWithPrefix(err error, prefix string) {
+	switch err := err.(type) {
+	case ErrorList:
+		for i := range err {
+			*a = append(*a, fmt.Errorf("%s%s", prefix, err[i]))
+		}
+	default:
+		*a = append(*a, fmt.Errorf("%s%s", prefix, err))
+	}
+}