diff --git a/index/brand.go b/index/brand.go index a0f64bbd1..4b562d21b 100644 --- a/index/brand.go +++ b/index/brand.go @@ -1,6 +1,7 @@ package index import ( + "math/rand" "time" _ "github.com/go-sql-driver/mysql" @@ -392,10 +393,21 @@ func (self *Brand) Load(requestChan chan Command, f *Fragment) { return //log.Println("Bad mojo") } + time.Sleep(time.Duration(rand.Intn(15)) * time.Second) //trying to avoid mass cassandra hit + batch_count := 0 for _, k := range keys { request := NewLoadRequest(k) + if f.queue_size > 0 { + time.Sleep(1 * time.Second) + } + if batch_count > 200 { + time.Sleep(1 * time.Second) + batch_count = 0 + } requestChan <- request request.Response() + batch_count++ + //so execute will decrement...GetFragment Increments..since this doesn't call GetFragment it doesn't increment } } diff --git a/index/fragment_container.go b/index/fragment_container.go index 8c6112178..5c1e0ae22 100644 --- a/index/fragment_container.go +++ b/index/fragment_container.go @@ -74,10 +74,10 @@ func (self *FragmentContainer) LoadBitmap(frag_id util.SUUID, bitmap_id uint64, } func (self *FragmentContainer) GetFragment(frag_id util.SUUID) (*Fragment, bool) { - //lock c, v := self.fragments[frag_id] - //log.Println(self.fragments) - //log.Println(c) + if v { + c.inc() + } return c, v } @@ -316,6 +316,7 @@ type Fragment struct { mesg_count uint64 mesg_time time.Duration exit chan *sync.WaitGroup + queue_size int } func getStorage(db string, slice int, frame string, fid util.SUUID) Storage { @@ -349,6 +350,7 @@ func NewFragment(frag_id util.SUUID, db string, slice int, frame string) *Fragme f.impl = impl //NewGeneral(db, slice, NewMemoryStorage()) f.slice = slice f.exit = make(chan *sync.WaitGroup) + f.queue_size = 0 return f } @@ -454,6 +456,15 @@ func (self *Fragment) Load() { self.impl.Load(self.requestChan, self) } +func (self *Fragment) inc() { + self.queue_size++ +} +func (self *Fragment) dec() { + self.queue_size-- + if self.queue_size < 0 { + self.queue_size = 0 + } +} func (self *Fragment) ServeFragment() { for { select { @@ -461,6 +472,10 @@ func (self *Fragment) ServeFragment() { self.mesg_count++ start := time.Now() answer := req.Execute(self) + + if req.QueryType() != "LoadRequest" { + self.dec() + } delta := time.Since(start) self.mesg_count += 1 self.mesg_time += delta diff --git a/index/storage_cass.go b/index/storage_cass.go index 9d4e37f95..fb318ce3c 100644 --- a/index/storage_cass.go +++ b/index/storage_cass.go @@ -30,7 +30,8 @@ func init() { cluster = gocql.NewCluster(hosts...) cluster.Keyspace = keyspace cluster.Consistency = gocql.One - cluster.Timeout = 3 * time.Second + cluster.Timeout = 5 * time.Second + cluster.RetryPolicy = gocql.RetryPolicy{NumRetries: 10} } func BuildSchema() {