diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index a7f0a3ddf..262a74f27 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -12,9 +12,13 @@ import ( "os" "strconv" "strings" + "syscall" + "text/tabwriter" "time" + "unsafe" "github.com/umbel/pilosa" + "github.com/umbel/pilosa/roaring" ) var ( @@ -78,6 +82,7 @@ The commands are: import imports data from a CSV file backup backs up a frame to an archive file restore restores a frame from an archive file + inspect inspects fragment data files bench benchmarks operations Use the "-h" flag with any command for more information. @@ -108,6 +113,8 @@ func (m *Main) ParseFlags(args []string) error { m.Cmd = NewBackupCommand(m.Stdin, m.Stdout, m.Stderr) case "restore": m.Cmd = NewRestoreCommand(m.Stdin, m.Stdout, m.Stderr) + case "inspect": + m.Cmd = NewInspectCommand(m.Stdin, m.Stdout, m.Stderr) case "bench": m.Cmd = NewBenchCommand(m.Stdin, m.Stdout, m.Stderr) default: @@ -515,6 +522,116 @@ func (cmd *RestoreCommand) Run() error { return nil } +// InspectCommand represents a command for inspecting fragment data files. +type InspectCommand struct { + // Path to data file + Path string + + // Standard input/output + Stdin io.Reader + Stdout io.Writer + Stderr io.Writer +} + +// NewInspectCommand returns a new instance of InspectCommand. +func NewInspectCommand(stdin io.Reader, stdout, stderr io.Writer) *InspectCommand { + return &InspectCommand{ + Stdin: stdin, + Stdout: stdout, + Stderr: stderr, + } +} + +// ParseFlags parses command line flags from args. +func (cmd *InspectCommand) 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") + } else if fs.NArg() > 1 { + return errors.New("only one path allowed") + } + cmd.Path = fs.Arg(0) + + return nil +} + +// Usage returns the usage message to be printed. +func (cmd *InspectCommand) Usage() string { + return strings.TrimSpace(` +usage: pilosactl inspect PATH + +Inspects a data file and provides stats. + +`) +} + +// Run executes the main program execution. +func (cmd *InspectCommand) Run() error { + // Open file handle. + f, err := os.Open(cmd.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. + t := time.Now() + fmt.Fprintf(cmd.Stderr, "unmarshaling bitmap...") + bm := roaring.NewBitmap() + buf := (*[0x7FFFFFFF]byte)(unsafe.Pointer(&data[0]))[:fi.Size()] + if err := bm.UnmarshalBinary(buf); err != nil { + return err + } + fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t)) + + // Retrieve stats. + t = time.Now() + fmt.Fprintf(cmd.Stderr, "calculating stats...") + info := bm.Info() + fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t)) + + // 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.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])), + ) + } + tw.Flush() + + return nil +} + // BenchCommand represents a command for benchmarking database operations. type BenchCommand struct { // Destination host and port. diff --git a/fragment.go b/fragment.go index 7be7993e1..08c304494 100644 --- a/fragment.go +++ b/fragment.go @@ -47,9 +47,6 @@ const ( ) const ( - // DefaultCacheFlushInterval is the default value for Fragment.CacheFlushInterval. - DefaultCacheFlushInterval = 1 * time.Minute - // DefaultFragmentMaxOpN is the default value for Fragment.MaxOpN. DefaultFragmentMaxOpN = 1000 ) @@ -76,13 +73,6 @@ type Fragment struct { // Cached checksums for each block. checksums map[int][]byte - // Close management - wg sync.WaitGroup - closing chan struct{} - - // The interval at which the cached bitmap ids are persisted to disk. - CacheFlushInterval time.Duration - // Number of operations performed before performing a snapshot. // This limits the size of fragments on the heap and flushes them to disk // so that they can be mmapped and heap utilization can be kept low. @@ -99,15 +89,13 @@ type Fragment struct { // NewFragment returns a new instance of Fragment. func NewFragment(path, db, frame string, slice uint64) *Fragment { return &Fragment{ - path: path, - db: db, - frame: frame, - slice: slice, - closing: make(chan struct{}, 0), + path: path, + db: db, + frame: frame, + slice: slice, - LogOutput: os.Stderr, - CacheFlushInterval: DefaultCacheFlushInterval, - MaxOpN: DefaultFragmentMaxOpN, + LogOutput: os.Stderr, + MaxOpN: DefaultFragmentMaxOpN, } } @@ -149,10 +137,6 @@ func (f *Fragment) Open() error { // Clear checksums. f.checksums = make(map[int][]byte) - // Periodically flush cache. - f.wg.Add(1) - go func() { defer f.wg.Done(); f.monitorCacheFlush() }() - return nil }(); err != nil { f.close() @@ -250,7 +234,7 @@ func (f *Fragment) openCache() error { // Read in all bitmaps by ID. // This will cause them to be added to the cache. for _, bitmapID := range pb.GetBitmapIDs() { - f.bitmap(bitmapID) + bm := f.bitmap(bitmapID) } return nil @@ -264,12 +248,6 @@ func (f *Fragment) Close() error { } func (f *Fragment) close() error { - // Notify goroutines of closing and wait for completion. - close(f.closing) - f.mu.Unlock() - f.wg.Wait() - f.mu.Lock() - // 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) @@ -964,24 +942,6 @@ func (f *Fragment) snapshot() error { return nil } -// monitorCacheFlush periodically flushes the cache to disk. -// This is run in a goroutine. -func (f *Fragment) monitorCacheFlush() { - ticker := time.NewTicker(f.CacheFlushInterval) - defer ticker.Stop() - - for { - select { - case <-f.closing: - return - case <-ticker.C: - if err := f.FlushCache(); err != nil { - f.logger().Printf("error flushing cache: err=%s, path=%s", err, f.CachePath()) - } - } - } -} - // FlushCache writes the cache data to disk. func (f *Fragment) FlushCache() error { f.mu.Lock() diff --git a/frame.go b/frame.go index d15e2e5d1..bd16c172f 100644 --- a/frame.go +++ b/frame.go @@ -161,6 +161,18 @@ func (f *Frame) Fragment(slice uint64) *Fragment { func (f *Frame) fragment(slice uint64) *Fragment { return f.fragments[slice] } +// Fragments returns a list of all fragments in the frame. +func (f *Frame) Fragments() []*Fragment { + f.mu.Lock() + defer f.mu.Unlock() + + other := make([]*Fragment, 0, len(f.fragments)) + for _, fragment := range f.fragments { + other = append(other, fragment) + } + return other +} + // CreateFragmentIfNotExists returns a fragment in the frame by slice. func (f *Frame) CreateFragmentIfNotExists(slice uint64) (*Fragment, error) { f.mu.Lock() diff --git a/index.go b/index.go index 0fe535768..cb9f033cd 100644 --- a/index.go +++ b/index.go @@ -3,12 +3,18 @@ package pilosa import ( "errors" "fmt" + "io" + "log" "os" "path/filepath" "sort" "sync" + "time" ) +// DefaultCacheFlushInterval is the default value for Fragment.CacheFlushInterval. +const DefaultCacheFlushInterval = 1 * time.Minute + // Index represents a container for fragments. type Index struct { mu sync.Mutex @@ -17,8 +23,17 @@ type Index struct { // Databases by name. dbs map[string]*DB + // Close management + wg sync.WaitGroup + closing chan struct{} + // Data directory path. Path string + + // The interval at which the cached bitmap ids are persisted to disk. + CacheFlushInterval time.Duration + + LogOutput io.Writer } // NewIndex returns a new instance of Index. @@ -26,6 +41,11 @@ func NewIndex() *Index { return &Index{ dbs: make(map[string]*DB), remoteMax: 0, + closing: make(chan struct{}, 0), + + CacheFlushInterval: DefaultCacheFlushInterval, + + LogOutput: os.Stderr, } } @@ -52,17 +72,28 @@ func (i *Index) Open() error { continue } + i.logger().Printf("opening database: %s", filepath.Base(fi.Name())) + db := NewDB(i.DBPath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) if err := db.Open(); err != nil { return fmt.Errorf("open db: name=%s, err=%s", db.Name(), err) } i.dbs[db.Name()] = db } + + // Periodically flush cache. + i.wg.Add(1) + go func() { defer i.wg.Done(); i.monitorCacheFlush() }() + return nil } // Close closes all open fragments. func (i *Index) Close() error { + // Notify goroutines of closing and wait for completion. + close(i.closing) + i.wg.Wait() + for _, db := range i.dbs { db.Close() } @@ -196,6 +227,42 @@ func (i *Index) SetMax(newmax uint64) { i.remoteMax = newmax } +// monitorCacheFlush periodically flushes all fragment caches sequentially. +// This is run in a goroutine. +func (i *Index) monitorCacheFlush() { + ticker := time.NewTicker(i.CacheFlushInterval) + defer ticker.Stop() + + for { + select { + case <-i.closing: + return + case <-ticker.C: + i.flushCaches() + } + } +} + +func (i *Index) flushCaches() { + for _, db := range i.DBs() { + for _, frame := range db.Frames() { + for _, fragment := range frame.Fragments() { + select { + case <-i.closing: + return + default: + } + + if err := fragment.FlushCache(); err != nil { + i.logger().Printf("error flushing cache: err=%s, path=%s", err, fragment.CachePath()) + } + } + } + } +} + +func (i *Index) logger() *log.Logger { return log.New(i.LogOutput, "", log.LstdFlags) } + // IndexSyncer is an active anti-entropy tool that compares the local index // with a remote index based on block checksums and resolves differences. type IndexSyncer struct { diff --git a/index_test.go b/index_test.go index 1c5cdc208..19859af93 100644 --- a/index_test.go +++ b/index_test.go @@ -1,6 +1,7 @@ package pilosa_test import ( + "bytes" "io/ioutil" "os" "reflect" @@ -118,6 +119,7 @@ func TestIndexSyncer_SyncIndex(t *testing.T) { // Index is a test wrapper for pilosa.Index. type Index struct { *pilosa.Index + LogOutput bytes.Buffer } // NewIndex returns a new instance of Index with a temporary path. @@ -129,6 +131,8 @@ func NewIndex() *Index { i := &Index{Index: pilosa.NewIndex()} i.Path = path + i.Index.LogOutput = &i.LogOutput + return i } diff --git a/roaring/roaring.go b/roaring/roaring.go index 2e3c16c76..f9857d13b 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -409,6 +409,9 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error { } op.apply(b) + // Increase the op count. + b.opN++ + // Move the buffer forward. buf = buf[op.size():] } @@ -437,6 +440,28 @@ func (b *Bitmap) Iterator() *Iterator { return itr } +// Info returns stats for the bitmap. +func (b *Bitmap) Info() BitmapInfo { + info := BitmapInfo{ + OpN: b.opN, + Containers: make([]ContainerInfo, len(b.containers)), + } + + for i, c := range b.containers { + ci := c.info() + ci.Key = b.keys[i] + info.Containers[i] = ci + } + + return info +} + +// BitmapInfo represents a point-in-time snapshot of bitmap stats. +type BitmapInfo struct { + OpN int + Containers []ContainerInfo +} + // Iterator represents an iterator over a Bitmap. type Iterator struct { bitmap *Bitmap @@ -836,6 +861,38 @@ func (c *container) size() int { return len(c.bitmap) * 8 } +// info returns the current stats about the container. +func (c *container) info() ContainerInfo { + info := ContainerInfo{N: c.n} + + if c.isArray() { + info.Type = "array" + info.Alloc = len(c.array) * 4 + } else { + info.Type = "bitmap" + info.Alloc = len(c.bitmap) * 8 + } + + if c.mapped { + if c.isArray() { + info.Pointer = unsafe.Pointer(&c.array[0]) + } else { + info.Pointer = unsafe.Pointer(&c.bitmap[0]) + } + } + + return info +} + +// ContainerInfo represents a point-in-time snapshot of container stats. +type ContainerInfo struct { + Key uint64 // container key + Type string // container type (array or bitmap) + N int // number of bits + Alloc int // memory used + Pointer unsafe.Pointer // offset within the mmap +} + func intersectionCount(a, b *container) uint64 { if a.isArray() { if b.isArray() {