Merge pull request #102 from benbjohnson/serialize-cache-flush

Serialize cache flush

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.

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()

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()

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 {

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
 }
 

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() {