diff --git a/index/bitmap_test.go b/index/bitmap_test.go index 4b592cf16..ae308ed94 100644 --- a/index/bitmap_test.go +++ b/index/bitmap_test.go @@ -2,6 +2,7 @@ package index import ( // "fmt" + "math/rand" "testing" // "time" @@ -117,6 +118,21 @@ func TestBitmaps(t *testing.T) { } */ +func benchmark(b *testing.B, size int, fill int) { + x := make(map[uint64]IBitmap) + for i := uint64(0); i < uint64(size); i++ { + x[i] = CreateRBBitmap() + } + for i := 0; i < b.N; i++ { + bid := rand.Int() % size + SetBit(x[uint64(bid)], uint64(i%fill)) + } +} +func BenchmarkSetBitL2(b *testing.B) { + benchmark(b, 50000, 1024*64) +} + +/* func BenchmarkSetBit(b *testing.B) { // run the Fib function b.N times a := CreateRBBitmap() @@ -124,3 +140,4 @@ func BenchmarkSetBit(b *testing.B) { SetBit(a, uint64(n)) } } +*/ diff --git a/index/brand.go b/index/brand.go index 061f96fbc..c777f0c23 100644 --- a/index/brand.go +++ b/index/brand.go @@ -88,7 +88,10 @@ func (self *Brand) Get(bitmap_id uint64) IBitmap { } func (self *Brand) GetFilter(bitmap_id, filter uint64) IBitmap { - b, _ := self.storage.Fetch(bitmap_id, self.db, self.frame, self.slice) + b, old_filter := self.storage.Fetch(bitmap_id, self.db, self.frame, self.slice) + if filter == 0 { + filter = old_filter + } self.cache_it(b, bitmap_id, filter) return b } @@ -124,10 +127,12 @@ func (self *Brand) SetBit(bitmap_id uint64, bit_pos uint64, filter uint64) bool change, chunk, address := SetBit(bm, bit_pos) if change { val := chunk.Value.Block[address.BlockIndex] - self.storage.BeginBatch() + /*self.storage.BeginBatch() self.storage.StoreBlock(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), val) self.storage.StoreBlock(bitmap_id, self.db, self.frame, self.slice, filter, COUNTERMASK, 0, bm.Count()) self.storage.EndBatch() + */ + self.storage.StoreBit(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), val, bm.Count()) self.rank_count++ } return change diff --git a/index/brand_test.go b/index/brand_test.go new file mode 100644 index 000000000..b039fbff3 --- /dev/null +++ b/index/brand_test.go @@ -0,0 +1,165 @@ +package index + +import ( + // "fmt" + "math/rand" + "testing" + // "time" + + // . "github.com/smartystreets/goconvey/convey" +) + +var ( + size int + membrand *Brand + cassbrand *Brand +) + +func init() { + println("GO") + size = 1000 + + membrand = NewBrand("db", "frame", 0, NewMemoryStorage(), size, size, 0) + for i := uint64(0); i < uint64(size); i++ { + membrand.SetBit(i, 0, 1) + } + cassbrand = NewBrand("db", "frame", 0, NewCassStorage(), size, size, 0) + for i := uint64(0); i < uint64(size); i++ { + cassbrand.SetBit(i, 0, 1) + } +} + +/* +func TestBitmaps(t *testing.T) { + Convey("function BitCount should equal method bm.Count()", t, func() { + bm := CreateRBBitmap() + for i := uint64(0); i < uint64(4096); i++ { + SetBit(bm, i) + } + bc1 := BitCount(bm) + bc2 := bm.Count() + So(bc1, ShouldEqual, bc2) + So(bc1, ShouldEqual, 4096) + }) + Convey("function Difference 1 and not 0 => true ", t, func() { + bm1 := CreateRBBitmap() + bm2 := CreateRBBitmap() + SetBit(bm1, 1) + //SetBit(bm2,2) + all := Difference(bm1, bm2) + res := BitCount(all) + + So(1, ShouldEqual, res) + }) + + Convey("function Difference 1 and not 0 => true ", t, func() { + bm1 := CreateRBBitmap() + bm2 := CreateRBBitmap() + SetBit(bm1, 1) + SetBit(bm1, 2) + SetBit(bm1, 3) + SetBit(bm1, 4) + SetBit(bm2, 3) + //SetBit(bm2,2) + all := Difference(bm1, bm2) + res := BitCount(all) + + So(3, ShouldEqual, res) + }) + + Convey("UNION even + odd equal 4096 ", t, func() { + even := CreateRBBitmap() + for i := uint64(0); i < uint64(4096); i += 2 { + SetBit(even, i) + } + + odd := CreateRBBitmap() + for i := uint64(1); i < uint64(4096); i += 2 { + SetBit(odd, i) + } + all := Union(even, odd) + total_bits := BitCount(all) + + So(total_bits, ShouldEqual, 4096) + }) + + Convey("Intersection even - odd equal 0 ", t, func() { + even := CreateRBBitmap() + for i := uint64(0); i < uint64(4096); i += 2 { + SetBit(even, i) + } + + odd := CreateRBBitmap() + for i := uint64(1); i < uint64(4096); i += 2 { + SetBit(odd, i) + } + all := Intersection(even, odd) + total_bits := BitCount(all) + + So(total_bits, ShouldEqual, 0) + + }) + + Convey("Bitcount< 1s ", t, func() { + all := CreateRBBitmap() + for i := uint64(0); i < uint64(65536); i++ { + SetBit(all, i) + } + start := time.Now() + BitCount(all) + So(start, ShouldHappenWithin, time.Duration(1)*time.Millisecond, time.Now()) + }) + + Convey("Compressed ", t, func() { + all := CreateRBBitmap() + for i := uint64(0); i < uint64(4096); i++ { + SetBit(all, i) + } + cs := all.ToCompressString() + fmt.Println(cs) + bm := CreateRBBitmap() + bm.FromCompressString(cs) + So(BitCount(all), ShouldEqual, BitCount(bm)) + }) + + Convey("AndCount ", t, func() { + a := CreateRBBitmap() + for i := uint64(0); i < uint64(4096); i++ { + SetBit(a, i) + } + b := CreateRBBitmap() + for i := uint64(0); i < uint64(8192); i++ { + SetBit(b, i) + } + c1 := IntersectionCount(a, b) + c := Intersection(a, b) + So(c1, ShouldEqual, BitCount(c)) + }) + +} +*/ +func benchmark_(b *testing.B, size int, fill int, brand *Brand) { + + println(b.N) + for i := 0; i < b.N; i++ { + bid := rand.Int() % size + profile := uint64(i % fill) + brand.SetBit(uint64(bid), profile, 1) + } +} +func BenchmarkBrandMemSetBitL2(b *testing.B) { + benchmark_(b, size, 1024*64, membrand) +} +func BenchmarkBrandCasSetBitL2(b *testing.B) { + benchmark_(b, size, 1024*64, cassbrand) +} + +/* +func BenchmarkSetBit(b *testing.B) { + // run the Fib function b.N times + a := CreateRBBitmap() + for n := 0; n < b.N; n++ { + SetBit(a, uint64(n)) + } +} +*/ diff --git a/index/general.go b/index/general.go index bab2e1609..3dbf7d727 100644 --- a/index/general.go +++ b/index/general.go @@ -57,10 +57,11 @@ func (self *General) SetBit(bitmap_id uint64, bit_pos uint64, filter uint64) boo change, chunk, address := SetBit(bm, bit_pos) if change { val := chunk.Value.Block[address.BlockIndex] - self.storage.BeginBatch() - self.storage.StoreBlock(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), val) - self.storage.StoreBlock(bitmap_id, self.db, self.frame, self.slice, filter, COUNTERMASK, 0, bm.Count()) - self.storage.EndBatch() + //self.storage.BeginBatch() + //self.storage.StoreBlock(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), val) + //self.storage.StoreBlock(bitmap_id, self.db, self.frame, self.slice, filter, COUNTERMASK, 0, bm.Count()) + //self.storage.EndBatch() + self.storage.StoreBit(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), val, bm.Count()) } return change diff --git a/index/storage.go b/index/storage.go index 50d0b38e0..ad0f67ab2 100644 --- a/index/storage.go +++ b/index/storage.go @@ -4,6 +4,7 @@ type Storage interface { Fetch(bitmap_id uint64, db string, frame string, slice int) (IBitmap, uint64) Store(id uint64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error StoreBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, block uint64) error + StoreBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) BeginBatch() EndBatch() FlushBatch() diff --git a/index/storage_cass.go b/index/storage_cass.go index 1cab65783..28e19f642 100644 --- a/index/storage_cass.go +++ b/index/storage_cass.go @@ -7,6 +7,7 @@ import ( "pilosa/config" "pilosa/util" + //"sync/atomic" "time" "github.com/gocql/gocql" @@ -20,6 +21,7 @@ type CassandraStorage struct { batch_counter int cass_time_window_secs float64 cass_flush_size int + cass_queue CassQueue } var cluster *gocql.ClusterConfig @@ -58,6 +60,8 @@ func NewCassStorage() Storage { obj.batch_counter = 0 obj.cass_time_window_secs = float64(config.GetIntDefault("cassandra_time_window_secs", 5)) obj.cass_flush_size = config.GetIntDefault("cassandra_max_size_batch", 15) + obj.cass_queue = NewCassQueue() + go obj.asyncStore() return obj } @@ -109,7 +113,7 @@ func (self *CassandraStorage) BeginBatch() { } func (self *CassandraStorage) runBatch(batch *gocql.Batch) { if batch != nil { - self.db.ExecuteBatch(batch) + go self.db.ExecuteBatch(batch) } } func (self *CassandraStorage) FlushBatch() { @@ -172,3 +176,52 @@ func (self *CassandraStorage) StoreBlock(bid uint64, db string, frame string, sl util.SendTimer("cassandra_storage_StoreBlock", delta.Nanoseconds()) return nil } + +type CassRecord struct { + bitmap_id uint64 + db string + frame string + slice int + filter uint64 + chunk uint64 + block_index int32 + val uint64 + count uint64 +} + +func (self *CassandraStorage) asyncStore() { + for { + rec, term := self.cass_queue.Pop() + if term { + break + } + self.BeginBatch() + self.StoreBlock(rec.bitmap_id, rec.db, rec.frame, rec.slice, rec.filter, rec.chunk, rec.block_index, rec.val) + self.StoreBlock(rec.bitmap_id, rec.db, rec.frame, rec.slice, rec.filter, COUNTERMASK, 0, rec.count) + self.EndBatch() + } +} + +func (self *CassandraStorage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) { + rec := CassRecord{bid, db, frame, slice, filter, bchunk, block_index, bblock, count} + self.cass_queue.Push(rec) +} + +type CassQueue struct { + size int64 + buffer chan CassRecord +} + +func NewCassQueue() CassQueue { + return CassQueue{0, make(chan CassRecord, 2048)} +} + +func (self *CassQueue) Push(rec CassRecord) { + // atomic.AddInt64(&self.size, 1) + self.buffer <- rec +} +func (self *CassQueue) Pop() (CassRecord, bool) { + ret := <-self.buffer + // atomic.AddInt64(&self.size, -1) + return ret, false +} diff --git a/index/storage_leveldb.go b/index/storage_leveldb.go index 191513c81..1e0077d3e 100644 --- a/index/storage_leveldb.go +++ b/index/storage_leveldb.go @@ -175,3 +175,11 @@ func (self *LevelDBStorage) Close() { self.FlushBatch() self.db.Close() } + +func (self *LevelDBStorage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) { + self.BeginBatch() + self.StoreBlock(bid, db, frame, slice, filter, bchunk, block_index, bblock) + self.StoreBlock(bid, db, frame, slice, filter, COUNTERMASK, 0, count) + self.EndBatch() + +} diff --git a/index/storage_mem.go b/index/storage_mem.go index 7a00b9483..d06d1575d 100644 --- a/index/storage_mem.go +++ b/index/storage_mem.go @@ -46,3 +46,5 @@ func (c *MemoryStorage) StoreBlock(bitmap_id uint64, db string, frame string, sl return nil } +func (self *MemoryStorage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) { +}