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
}