scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-12 07:41:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

make cassandra writes async

Browse source
This commit is contained in:
Todd Gruben 2015-01-15 20:12:14 +00:00
parent f32ca7c3d4
commit 79f5c58fbb
8 changed files with 259 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
index/bitmap_test.go
View file

@ -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))
}
}
*/

9
index/brand.go
View file

@ -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

165
index/brand_test.go Normal file
View file

@ -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))
}
}
*/

9
index/general.go
View file

@ -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

1
index/storage.go
View file

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

55
index/storage_cass.go
View file

@ -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
}

8
index/storage_leveldb.go
View file

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

2
index/storage_mem.go
View file

@ -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) {
}