From 8b2398b8e554147991ce1cd1f26653b48ca21b77 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Wed, 4 Mar 2015 23:17:36 +0000 Subject: [PATCH] refactor delete bit --- core/service.go | 25 ++++++++----- index/brand.go | 3 +- index/general.go | 2 +- index/storage.go | 3 +- index/storage_cass.go | 81 ++++++++++++++-------------------------- index/storage_leveldb.go | 7 +++- index/storage_mem.go | 6 ++- query/planner.go | 8 ---- 8 files changed, 58 insertions(+), 77 deletions(-) diff --git a/core/service.go b/core/service.go index 836c7f7c0..2866033c7 100644 --- a/core/service.go +++ b/core/service.go @@ -57,19 +57,11 @@ func NewService() *Service { return service } -func (self *Service) PrepareLogging() { +func (self *Service) getProduction() string { base_path := config.GetString("log_path") if base_path == "" { base_path = "/tmp" } - /* - f, err := os.OpenFile(fmt.Sprintf("%s/%s.%s", base_path, self.name, self.Id), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666) - if err != nil { - log.Warn("error opening file: %v", err) - } - //defer f.Close() - log.SetOutput(f) - */ fname := fmt.Sprintf("%s/%s.%s", base_path, self.name, self.Id) // log_level := config.GetStringDefault("log_level", "info") @@ -80,7 +72,20 @@ func (self *Service) PrepareLogging() { `, log_level, fname) - logger, _ := log.LoggerFromConfigAsBytes([]byte(prod_config)) + return prod_config +} + +func (self *Service) getDev() string { + return ` + + + + + ` +} +func (self *Service) PrepareLogging() { + logger, _ := log.LoggerFromConfigAsBytes([]byte(self.getProduction())) + // logger, _ := log.LoggerFromConfigAsBytes([]byte(self.getDev())) log.ReplaceLogger(logger) } diff --git a/index/brand.go b/index/brand.go index ecd645783..0ac784909 100644 --- a/index/brand.go +++ b/index/brand.go @@ -428,6 +428,7 @@ func (self *Brand) Load(requestChan chan Command, f *Fragment) { } func (self *Brand) ClearBit(bitmap_id uint64, bit_pos uint64) bool { + log.Trace("ClearBit", bitmap_id, bit_pos) bm1, ok := self.bitmap_cache[bitmap_id] var bm IBitmap filter := uint64(0) @@ -450,7 +451,7 @@ func (self *Brand) ClearBit(bitmap_id uint64, bit_pos uint64) bool { self.storage.EndBatch() */ if val == 0 { - self.storage.RemoveBlock(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex)) + self.storage.RemoveBit(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), bm.Count()) } else { self.storage.StoreBit(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), val, bm.Count()) } diff --git a/index/general.go b/index/general.go index 0784733e0..24c92dde9 100644 --- a/index/general.go +++ b/index/general.go @@ -165,7 +165,7 @@ func (self *General) ClearBit(bitmap_id uint64, bit_pos uint64) bool { if change { val := chunk.Value.Block[address.BlockIndex] if val == 0 { - self.storage.RemoveBlock(bitmap_id, self.db, self.frame, self.slice, uint64(0), address.ChunkKey, int32(address.BlockIndex)) + self.storage.RemoveBit(bitmap_id, self.db, self.frame, self.slice, uint64(0), address.ChunkKey, int32(address.BlockIndex), bm.Count()) } else { self.storage.StoreBit(bitmap_id, self.db, self.frame, self.slice, uint64(0), address.ChunkKey, int32(address.BlockIndex), val, bm.Count()) } diff --git a/index/storage.go b/index/storage.go index 8be339c9d..c395db9f7 100644 --- a/index/storage.go +++ b/index/storage.go @@ -5,7 +5,8 @@ type Storage interface { 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, chunk uint64, block_index int32, block, count uint64) - RemoveBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32) error + RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64) + RemoveBlock(id uint64, db string, frame string, slice int, chunk uint64, block_index int32) BeginBatch() EndBatch() FlushBatch() diff --git a/index/storage_cass.go b/index/storage_cass.go index fbb34a58b..6e8eb4a16 100644 --- a/index/storage_cass.go +++ b/index/storage_cass.go @@ -22,7 +22,6 @@ type CassandraStorage struct { batch_counter int cass_time_window_secs float64 cass_flush_size int - cass_queue CassQueue } var cluster *gocql.ClusterConfig @@ -46,6 +45,7 @@ func BuildSchema() { */ } + func NewCassStorage() Storage { obj := new(CassandraStorage) // cluster.CQLVersion = "3.0.0" @@ -56,14 +56,12 @@ func NewCassStorage() Storage { obj.db = session obj.stmt = `INSERT INTO bitmap ( bitmap_id, db, frame, slice , filter, ChunkKey, BlockIndex, block) VALUES (?,?,?,?,?,?,?,?) USING timestamp ?;` - obj.dstmt = `DELETE BlockIndex,block FROM bitmap WHERE bitmap_id=? AND db=? AND frame=? AND slice=? AND filter=? AND ChunkKey=? AND BlockIndex=?;` + obj.dstmt = `DELETE FROM bitmap USING TIMESTAMP ? WHERE bitmap_id=? AND db=? AND frame=? AND slice=? AND chunkkey=? AND blockindex=?` obj.batch = nil obj.batch_time = time.Now() 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 } @@ -116,7 +114,10 @@ func (self *CassandraStorage) BeginBatch() { } func (self *CassandraStorage) runBatch(batch *gocql.Batch) { if batch != nil { - go self.db.ExecuteBatch(batch) + err := self.db.ExecuteBatch(batch) + if err != nil { + log.Warn("Batch ERROR", err) + } } } func (self *CassandraStorage) FlushBatch() { @@ -181,56 +182,32 @@ func (self *CassandraStorage) StoreBlock(bid uint64, db string, frame string, sl 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) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, val, count uint64) { + self.BeginBatch() + self.StoreBlock(bid, db, frame, slice, filter, chunk, block_index, val) + self.StoreBlock(bid, db, frame, slice, filter, COUNTERMASK, 0, count) + self.EndBatch() } -func (self *CassandraStorage) asyncStore() { - for { - rec, term := self.cass_queue.Pop() - if term { - break - } +func (self *CassandraStorage) RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64) { + log.Trace("RemoveBit", id, db, frame, slice, chunk, block_index) + self.BeginBatch() + self.RemoveBlock(id, db, frame, slice, chunk, block_index) + self.StoreBlock(id, db, frame, slice, filter, COUNTERMASK, 0, count) + self.EndBatch() +} + +func (self *CassandraStorage) RemoveBlock(bid uint64, db string, frame string, slice int, bchunk uint64, block_index int32) { + log.Trace("RemoveBBlock", bid, db, frame, slice, bchunk, block_index) + id := util.Uint64ToInt64(bid) //these fucntions ignore overflow + chunk := util.Uint64ToInt64(bchunk) + + if self.batch == nil { 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() } -} + start := time.Now() -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) -} - -func (self *CassandraStorage) RemoveBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32) error { - return nil -} - -type CassQueue struct { - size int64 - buffer chan CassRecord -} - -func NewCassQueue() CassQueue { - //return CassQueue{0, make(chan CassRecord, 4096)} - return CassQueue{0, make(chan CassRecord)} -} - -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 + self.batch.Query(self.dstmt, start.UnixNano(), id, db, frame, slice, chunk, block_index) + delta := time.Since(start) + util.SendTimer("cassandra_storage_DeleteBlock", delta.Nanoseconds()) } diff --git a/index/storage_leveldb.go b/index/storage_leveldb.go index 95980d29f..d08115803 100644 --- a/index/storage_leveldb.go +++ b/index/storage_leveldb.go @@ -170,8 +170,11 @@ func (self *LevelDBStorage) StoreBlock(id uint64, db string, frame string, slice util.SendTimer("leveldb_storage_StoreBlock", delta.Nanoseconds()) return nil } -func (self *LevelDBStorage) RemoveBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32) error { - return nil + +func (self *LevelDBStorage) RemoveBlock(id uint64, db string, frame string, slice int, chunk uint64, block_index int32) { +} + +func (self *LevelDBStorage) RemoveBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, count uint64) { } func (self *LevelDBStorage) Close() { diff --git a/index/storage_mem.go b/index/storage_mem.go index 398c6a146..c3d22f19c 100644 --- a/index/storage_mem.go +++ b/index/storage_mem.go @@ -48,6 +48,8 @@ func (c *MemoryStorage) StoreBlock(bitmap_id uint64, db string, frame string, sl } func (self *MemoryStorage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) { } -func (self *MemoryStorage) RemoveBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32) error { - return nil +func (self *MemoryStorage) RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64) { +} + +func (self *MemoryStorage) RemoveBlock(id uint64, db string, frame string, slice int, chunk uint64, block_index int32) { } diff --git a/query/planner.go b/query/planner.go index d81c9f425..8a783a1a0 100644 --- a/query/planner.go +++ b/query/planner.go @@ -464,26 +464,18 @@ func validateRange(Args map[string]interface{}) error { func (self *QueryPlanner) buildTree(query *Query, slice int) (QueryTree, error) { log.Trace("QueryPlanner.buildTree", query, slice) var tree QueryTree - spew.Dump(query) // handle SET operation regardless of the slice if query.Operation == "set" { - println("SET") - log.Warn("SET") - tree = &SetQueryTree{&db.Bitmap{query.Args["id"].(uint64), query.Args["frame"].(string), query.Args["filter"].(uint64)}, query.Args["profile_id"].(uint64)} return tree, nil } if query.Operation == "clear" { - - println("CLEAR") - log.Warn("CLEAR") tree = &ClearQueryTree{&db.Bitmap{query.Args["id"].(uint64), query.Args["frame"].(string), query.Args["filter"].(uint64)}, query.Args["profile_id"].(uint64)} return tree, nil } if query.Operation == "recall" { - tree = &RecallQueryTree{query.Args["stash"].(Stash)} return tree, nil }