Merge pull request #84 from umbel/slow_import

Slow import optimization

client.go

@@ -176,12 +176,29 @@ func (c *Client) Import(db, frame string, slice uint64, bits []Bit) error {
 		return ErrFrameRequired
 	}
 
+	buf, err := MarshalImportPayload(db, frame, slice, bits)
+	if err != nil {
+		return fmt.Errorf("Error Creating Payload: %s", err)
+	}
+
 	// Retrieve a list of nodes that own the slice.
 	nodes, err := c.SliceNodes(slice)
+
 	if err != nil {
 		return fmt.Errorf("slice nodes: %s", err)
 	}
 
+	// Import to each node.
+	for _, node := range nodes {
+		if err := c.importNode(node, buf); err != nil {
+			return fmt.Errorf("import node: host=%s, err=%s", node.Host, err)
+		}
+	}
+
+	return nil
+}
+
+func MarshalImportPayload(db, frame string, slice uint64, bits []Bit) ([]byte, error) {
 	// Separate bitmap and profile IDs to reduce allocations.
 	bitmapIDs := Bits(bits).BitmapIDs()
 	profileIDs := Bits(bits).ProfileIDs()
@@ -195,17 +212,9 @@ func (c *Client) Import(db, frame string, slice uint64, bits []Bit) error {
 		ProfileIDs: profileIDs,
 	})
 	if err != nil {
-		return fmt.Errorf("marshal import request: %s", err)
+		return nil, fmt.Errorf("marshal import request: %s", err)
 	}
-
-	// Import to each node.
-	for _, node := range nodes {
-		if err := c.importNode(node, buf); err != nil {
-			return fmt.Errorf("import node: host=%s, err=%s", node.Host, err)
-		}
-	}
-
-	return nil
+	return buf, nil
 }
 
 // importNode sends a pre-marshaled import request to a node.

cmd/pilosa/main.go

@@ -8,7 +8,6 @@ import (
 	"math/rand"
 	"os"
 	"os/signal"
-	"os/user"
 	"path/filepath"
 	"strings"
 	"time"
@@ -149,13 +148,14 @@ func (m *Main) ParseFlags(args []string) error {
 	// Expand home directory.
 	prefix := "~" + string(filepath.Separator)
 	if strings.HasPrefix(m.Config.DataDir, prefix) {
-		u, err := user.Current()
-		if err != nil {
+		//	u, err := user.Current()
+		HomeDir := os.Getenv("HOME")
+		/*if err != nil {
 			return err
-		} else if u.HomeDir == "" {
+		} else*/if HomeDir == "" {
 			return errors.New("data directory not specified and no home dir available")
 		}
-		m.Config.DataDir = filepath.Join(u.HomeDir, strings.TrimPrefix(m.Config.DataDir, prefix))
+		m.Config.DataDir = filepath.Join(HomeDir, strings.TrimPrefix(m.Config.DataDir, prefix))
 	}
 
 	return nil

fragment.go

@@ -25,7 +25,7 @@ import (
 
 const (
 	// SliceWidth is the number of profile IDs in a slice.
-	SliceWidth = 65536
+	SliceWidth = 2097152
 
 	// SnapshotExt is the file extension used for an in-process snapshot.
 	SnapshotExt = ".snapshotting"
@@ -821,9 +821,11 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
 	// Process every bit.
 	// If an error occurs then reopen the storage.
 	if err := func() error {
+		lastID := uint64(0)
+		bmCounter := uint64(0)
+		var bitmap *Bitmap
 		for i := range bitmapIDs {
 			bitmapID, profileID := bitmapIDs[i], profileIDs[i]
-
 			// Determine the position of the bit in the storage.
 			pos, err := f.pos(bitmapID, profileID)
 			if err != nil {
@@ -835,11 +837,24 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
 				return err
 			}
 
+			// import optimization to avoid linear foreach calls
+			// slight risk of concurrent cache counter being off but
+			// no real danger
+			if i == 0 || bitmapID != lastID {
+				bitmap = f.bitmap(bitmapID)
+				if i != 0 {
+					f.cache.Add(lastID, bmCounter)
+				}
+				bmCounter = bitmap.Count()
+				lastID = bitmapID
+			}
+
 			// Invalidate block checksum.
 			delete(f.checksums, int(bitmapID/HashBlockSize))
-
+			if bitmap.SetBit(profileID) {
+				bmCounter += 1
+			}
 			// Update the cache.
-			f.bitmap(bitmapID).SetBit(profileID)
 		}
 		return nil
 	}(); err != nil {

roaring/assembly.go

@@ -0,0 +1,7 @@
+package roaring
+
+func hasAsm() bool
+
+func BSFQ(memory uint64) int
+
+func POPCNTQ(memory uint64) int

roaring/assembly_amd64.s

@@ -0,0 +1,21 @@
+#include "textflag.h"
+
+TEXT ·hasAsm(SB),4,$0
+	MOVQ $1, AX
+	CPUID
+	SHRQ $23, CX
+	ANDQ $1, CX
+	MOVB CX, ret+0(FP)
+	RET
+
+TEXT ·POPCNTQ(SB),NOSPLIT,$0-8
+	MOVQ x+0(FP), BP
+	POPCNTQ BP, BX
+	MOVQ BX, ret+8(FP)
+	RET
+
+TEXT ·BSFQ(SB),NOSPLIT,$0-8
+	MOVQ x+0(FP), BP
+	BSFQ BP, BX
+	MOVQ BX, ret+8(FP)
+	RET

roaring/assembly_test.go

@@ -0,0 +1,53 @@
+package roaring
+
+import "testing"
+
+func TestBSFQ(t *testing.T) {
+	result := BSFQ(2)
+	if result != 1 {
+		t.Fatalf("BSF INCORRECT: %d", result)
+	}
+}
+
+func TestBSFQ_CompareGo(t *testing.T) {
+	v := uint64(1)
+	for i := 0; i < 64; i++ {
+		if BSFQ(v) != trailingZeroN(v) {
+			t.Fatalf("BSF INCORRECT: %d %d", BSFQ(v), trailingZeroN(v))
+		}
+		if v == 0 {
+			v = 1
+		} else {
+			v *= 2
+		}
+	}
+	/*
+		if bsfq(0) != trailingZeroN(0) {
+			fmt.Println(bsfq(0))
+			t.Fatalf("BSF INCORRECT")
+		}
+	*/
+}
+func BenchmarkBSF(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		BSFQ(uint64(i))
+	}
+}
+
+func BenchmarkTrailingZeroN(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		trailingZeroN(uint64(i))
+	}
+}
+
+func BenchmarkPOPCNTQ(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		POPCNTQ(uint64(i))
+	}
+}
+
+func BenchmarkPopcount(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		popcount(uint64(i))
+	}
+}