diff --git a/client.go b/client.go index 3a0f221c0..42302bbaa 100644 --- a/client.go +++ b/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. diff --git a/cmd/pilosa/main.go b/cmd/pilosa/main.go index 648b5cb6d..fd26a5685 100644 --- a/cmd/pilosa/main.go +++ b/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 diff --git a/fragment.go b/fragment.go index f4dcf4559..9142ad573 100644 --- a/fragment.go +++ b/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 { diff --git a/roaring/assembly.go b/roaring/assembly.go new file mode 100644 index 000000000..16012be82 --- /dev/null +++ b/roaring/assembly.go @@ -0,0 +1,7 @@ +package roaring + +func hasAsm() bool + +func BSFQ(memory uint64) int + +func POPCNTQ(memory uint64) int diff --git a/roaring/assembly_amd64.s b/roaring/assembly_amd64.s new file mode 100644 index 000000000..a33ade6ab --- /dev/null +++ b/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 diff --git a/roaring/assembly_test.go b/roaring/assembly_test.go new file mode 100644 index 000000000..3a3aa2854 --- /dev/null +++ b/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)) + } +}