diff --git a/index/brand.go b/index/brand.go index 4b562d21b..d1d86568c 100644 --- a/index/brand.go +++ b/index/brand.go @@ -394,20 +394,10 @@ func (self *Brand) Load(requestChan chan Command, f *Fragment) { //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 4bdc315b4..6449be3fc 100644 --- a/index/fragment_container.go +++ b/index/fragment_container.go @@ -75,9 +75,6 @@ func (self *FragmentContainer) LoadBitmap(frag_id util.SUUID, bitmap_id uint64, func (self *FragmentContainer) GetFragment(frag_id util.SUUID) (*Fragment, bool) { c, v := self.fragments[frag_id] - if v { - c.inc() - } return c, v } @@ -287,9 +284,10 @@ func (self *FragmentContainer) AddFragment(db string, frame string, slice int, i if !ok { log.Println("ADD FRAGMENT", frame, db, slice, util.SUUID_to_Hex(id)) f := NewFragment(id, db, slice, frame) + loader := make(chan Command) self.fragments[id] = f - go f.ServeFragment() - go f.Load() + go f.ServeFragment(loader) + go f.Load(loader) } } @@ -453,108 +451,35 @@ func (self *Fragment) Persist() { log.Println("Error saving:", err) } } -func (self *Fragment) Load() { - self.impl.Load(self.requestChan, self) +func (self *Fragment) Load(loadChan chan Command) { + self.impl.Load(loadChan, self) } -func (self *Fragment) inc() { - self.queue_size++ +func (self *Fragment) processCommand(req Command) { + self.mesg_count++ + start := time.Now() + answer := req.Execute(self) + delta := time.Since(start) + self.mesg_count += 1 + self.mesg_time += delta + req.ResponseChannel() <- Result{answer, delta} } -func (self *Fragment) dec() { - self.queue_size-- - if self.queue_size < 0 { - self.queue_size = 0 - } -} -func (self *Fragment) ServeFragment() { +func (self *Fragment) ServeFragment(loadChan chan Command) { for { select { case req := <-self.requestChan: - self.mesg_count++ - start := time.Now() - answer := req.Execute(self) - - if req.QueryType() != "LoadRequest" { - self.dec() + self.processCommand(req) + default: + select { + case req := <-self.requestChan: + self.processCommand(req) + case req := <-loadChan: + self.processCommand(req) + case wg := <-self.exit: + log.Println("Fragment Shutdown") + self.Persist() + wg.Done() } - delta := time.Since(start) - self.mesg_count += 1 - self.mesg_time += delta - req.ResponseChannel() <- Result{answer, delta} - - case wg := <-self.exit: - log.Println("Fragment Shutdown") - self.Persist() - wg.Done() } } } - -/* -type RequestJSON struct { - Request string - Fragment string - Args json.RawMessage -} -func (a *FragmentContainer) ServeHTTP(w http.ResponseWriter, r *http.Request) { - handler(w, r, a.fragments) -} - -func (a *FragmentContainer) RunServer(porti int, closeChannel chan bool, started chan bool) { - http.Handle("/", a) - port := fmt.Sprintf(":%d", porti) - - s := &http.Server{ - Addr: port, - Handler: nil, - ReadTimeout: 10 * time.Second, - WriteTimeout: 10 * time.Second, - MaxHeaderBytes: 1 << 20, - } - - l, e := net.Listen("tcp", port) - if e != nil { - log.Panicf(e.Error()) - } - go s.Serve(l) - started <- true - select { - case <-closeChannel: - log.Printf("Server thread exit") - l.Close() - // Shutdown() - return - break - } -} - -func handler(w http.ResponseWriter, r *http.Request, fragments map[string]*Fragment) { - if r.Method == "POST" { - var f RequestJSON - - bin, _ := ioutil.ReadAll(r.Body) - err := json.Unmarshal(bin, &f) - - if err != nil { - w.Header().Set("Content-Type", "application/json") - fmt.Fprintf(w, fmt.Sprintf(`{ "error":"%s"}`, err)) - - } - decoder := json.NewDecoder(bytes.NewReader(f.Args)) - request := BuildCommandFactory(&f, decoder) - w.Header().Set("Content-Type", "application/json") - if request != nil { - output := `{"Error":"Invalid Fragment"}` - fc, found := fragments[f.Fragment] //f.FragmentIndex