From 7048aa519b8394e6a4aef53154b5e497efb4df53 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Thu, 3 Sep 2015 15:50:08 -0600 Subject: [PATCH 1/2] clean up ids --- cmd/pilosa/main.go | 2 +- core/batch.go | 2 +- core/etcd.go | 6 +-- core/ping.go | 2 +- core/remotebits.go | 2 +- db/topology.go | 6 +-- executor/executor.go | 6 +-- fragment_container.go | 6 +-- id.go => pilosa.go | 72 ++++++++++++++++-------------------- id_test.go => pilosa_test.go | 28 +++++++------- query/parser.go | 2 +- query/planner.go | 16 ++++---- query/query.go | 2 +- storage/leveldb/leveldb.go | 2 +- 14 files changed, 73 insertions(+), 81 deletions(-) rename id.go => pilosa.go (57%) rename id_test.go => pilosa_test.go (50%) diff --git a/cmd/pilosa/main.go b/cmd/pilosa/main.go index a575c45d2..5addecc66 100644 --- a/cmd/pilosa/main.go +++ b/cmd/pilosa/main.go @@ -68,7 +68,7 @@ func (m *Main) Run(args ...string) error { // Generate an ID if one is not specified in the config. id := config.ID if id == nil { - *id = pilosa.RandomUUID() + *id = pilosa.NewGUID() } // Set up profiling. diff --git a/core/batch.go b/core/batch.go index a140607a8..647d0e59a 100644 --- a/core/batch.go +++ b/core/batch.go @@ -58,7 +58,7 @@ func (b *Batcher) Batch(database_name, frame, compressed_bitmap string, bitmap_i fragment, err := database.GetFragmentForBitmap(oslice, &db.Bitmap{Id: bitmap_id, FrameType: frame, Filter: filter}) if err == nil { - id := pilosa.RandomUUID() + id := pilosa.NewGUID() batch := db.Message{Data: BatchRequest{Id: &id, Source: &b.ID, Fragment_id: fragment.GetId(), Bitmap_id: bitmap_id, Compressed_bitmap: compressed_bitmap}} dest_id := fragment.GetProcess().Id() b.Transport.Send(&batch, &dest_id) diff --git a/core/etcd.go b/core/etcd.go index 436bd95ae..d2be6c597 100644 --- a/core/etcd.go +++ b/core/etcd.go @@ -193,7 +193,7 @@ func (self *TopologyMapper) AllocateFragment(process_guid, db, frame string, sli //to create the node, just write off the items to etcd and the watch should spawn //be nice if something would notify perhaps queue //so i need db, frame, slice , fragment_id - fuid := pilosa.SUUID_to_Hex(pilosa.Id()) + fuid := pilosa.NewSUUID().String() fragment_key := fmt.Sprintf("%s/db/%s/frame/%s/slice/%d/fragment/%s/process", self.namespace, db, frame, slice_int, fuid) // need to check value to see how many we have left log.Warn("ALLOC:", process_guid, len(process_guid)) @@ -268,7 +268,7 @@ func (self *TopologyMapper) handlenode(node *etcd.Node) error { } } if len(bits) > 7 { - fragment_id = pilosa.Hex_to_SUUID(bits[7]) + fragment_id = pilosa.ParseSUUID(bits[7]) fragment = database.GetOrCreateFragment(frame, slice, fragment_id) } @@ -284,7 +284,7 @@ func (self *TopologyMapper) handlenode(node *etcd.Node) error { process = db.NewProcess(&process_uuid) fragment.SetProcess(process) - if pilosa.Equal(&self.ID, &process_uuid) { + if self.ID.Equals(&process_uuid) { self.Index.AddFragment(bits[1], bits[3], slice_int, fragment_id) } diff --git a/core/ping.go b/core/ping.go index 9c35d015c..0d281580f 100644 --- a/core/ping.go +++ b/core/ping.go @@ -48,7 +48,7 @@ func NewPinger(id pilosa.GUID) *Pinger { } func (self *Pinger) Ping(process_id *pilosa.GUID) (*time.Duration, error) { - id := pilosa.RandomUUID() + id := pilosa.NewGUID() ping := db.Message{Data: PingRequest{Id: &id, Source: &self.ID}} start := time.Now() self.Transport.Send(&ping, process_id) diff --git a/core/remotebits.go b/core/remotebits.go index 4283befc7..47649b7e6 100644 --- a/core/remotebits.go +++ b/core/remotebits.go @@ -55,7 +55,7 @@ func (self *RemoteSetBit) Request() { self.requests = make([]remote_task, 0) source_process, _ := self.ProcessMap.GetProcess(&self.ID) for process, request := range self.cluster { - random_id := pilosa.RandomUUID() + random_id := pilosa.NewGUID() msg := new(db.Message) msg.Data = BitsRequest{ Bits: request, diff --git a/db/topology.go b/db/topology.go index a15b2a6eb..29e8667fd 100644 --- a/db/topology.go +++ b/db/topology.go @@ -323,7 +323,7 @@ func (self *FrameSliceIntersect) GetFragment(fragment_id pilosa.SUUID) (*Fragmen func (self *FrameSliceIntersect) AddFragment(fragment *Fragment) { self.fragments = append(self.fragments, fragment) - self.hashring.Add(pilosa.SUUID_to_Hex(fragment.id)) + self.hashring.Add(fragment.id.String()) } ///////// FRAGMENTS @@ -373,7 +373,7 @@ func (d *Database) GetFragmentForBitmap(slice *Slice, bitmap *Bitmap) (*Fragment log.Warn(err) return nil, err } - frag_id := pilosa.Hex_to_SUUID(frag_id_s) + frag_id := pilosa.ParseSUUID(frag_id_s) return fsi.GetFragment(frag_id) } @@ -391,7 +391,7 @@ func (d *Database) GetFragmentForFrameSlice(frame *Frame, slice *Slice) (*Fragme log.Warn(err) return nil, err } - frag_id := pilosa.Hex_to_SUUID(frag_id_s) + frag_id := pilosa.ParseSUUID(frag_id_s) return fsi.GetFragment(frag_id) } diff --git a/executor/executor.go b/executor/executor.go index ec7a48f90..4f16d6d10 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -444,7 +444,7 @@ func (self *Executor) GetQueryStepHandler(msg *db.Message) { bh, err := self.Index.Get(qs.Location.FragmentId, qs.Bitmap.Id) if err != nil { spew.Dump(err) - log.Error("GetQueryStepHandler1", pilosa.SUUID_to_Hex(qs.Location.FragmentId), qs.Bitmap.Id) + log.Error("GetQueryStepHandler1", qs.Location.FragmentId.String(), qs.Bitmap.Id) log.Error("GetQueryStepHandler2", err) } @@ -455,7 +455,7 @@ func (self *Executor) GetQueryStepHandler(msg *db.Message) { bm, err := self.Index.GetBytes(qs.Location.FragmentId, bh) if err != nil { spew.Dump(err) - log.Error("GetQueryStepHandlerr3", pilosa.SUUID_to_Hex(qs.Location.FragmentId), qs.Bitmap.Id) + log.Error("GetQueryStepHandlerr3", qs.Location.FragmentId.String(), qs.Bitmap.Id) log.Error("GetQueryStepHandler4", err) } result = bm @@ -709,7 +709,7 @@ func newtask(p pilosa.GUID) *Task { result := new(Task) result.processid = p result.f = make(map[pilosa.SUUID]pilosa.FillArgs) - result.hold_id = pilosa.RandomUUID() + result.hold_id = pilosa.NewGUID() return result } diff --git a/fragment_container.go b/fragment_container.go index e51310d45..c09645593 100644 --- a/fragment_container.go +++ b/fragment_container.go @@ -173,7 +173,7 @@ func (self *FragmentContainer) TopN(frag_id SUUID, bh BitmapHandle, n int, categ statsd.SendTimer("fragmant_container_TopN", result.exec_time.Nanoseconds()) return result.answer.([]Pair), nil } - return nil, errors.New(fmt.Sprintf("Fragment not found:%s", SUUID_to_Hex(frag_id))) + return nil, errors.New(fmt.Sprintf("Fragment not found:%s", frag_id.String())) } func (self *FragmentContainer) TopNAll(frag_id SUUID, n int, categories []uint64) ([]Pair, error) { @@ -185,7 +185,7 @@ func (self *FragmentContainer) TopNAll(frag_id SUUID, n int, categories []uint64 statsd.SendTimer("fragmant_container_TopNAll", result.exec_time.Nanoseconds()) return result.answer.([]Pair), nil } - return nil, errors.New(fmt.Sprintf("Fragment not found:%s", SUUID_to_Hex(frag_id))) + return nil, errors.New(fmt.Sprintf("Fragment not found:%s", frag_id.String())) } func (self *FragmentContainer) TopFillBatch(args []FillArgs) ([]Pair, error) { @@ -319,7 +319,7 @@ func (self *FragmentContainer) AddFragment(db string, frame string, slice int, i _, ok := self.fragments[id] if !ok { // dumpHandlesToLog() - log.Warn("ADD FRAGMENT", frame, db, slice, SUUID_to_Hex(id)) + log.Warn("ADD FRAGMENT", frame, db, slice, id.String()) f := NewFragment(id, db, slice, frame) loader := make(chan Command) self.fragments[id] = f diff --git a/id.go b/pilosa.go similarity index 57% rename from id.go rename to pilosa.go index e747d2186..a480b3946 100644 --- a/id.go +++ b/pilosa.go @@ -1,36 +1,27 @@ package pilosa import ( - "bytes" "encoding/binary" "encoding/hex" "fmt" "math/rand" - "os" "strings" "time" - log "github.com/cihub/seelog" "github.com/gocql/gocql" ) -var ( - counter = uint64(0) - Random *os.File -) +var counter = uint64(0) func init() { rand.Seed(time.Now().UTC().UnixNano()) - f, err := os.Open("/dev/urandom") - if err != nil { - log.Warn(err) - } - Random = f } +// SUUID represents a sequential UUID. type SUUID uint64 -func Id() SUUID { +// SUUID returns a new SUUID. +func NewSUUID() SUUID { millis := uint64(time.Now().UTC().UnixNano()) id := millis << (64 - 41) id |= uint64(rand.Intn(128)) << (64 - 41 - 13) @@ -39,26 +30,24 @@ func Id() SUUID { return SUUID(id) } -func SUUID_to_Hex(a SUUID) string { - buf := new(bytes.Buffer) - binary.Write(buf, binary.BigEndian, a) - return hex.EncodeToString(buf.Bytes()) +// String returns a string representation of id. +func (id SUUID) String() string { + var buf [8]byte + binary.BigEndian.PutUint64(buf[:], uint64(id)) + return hex.EncodeToString(buf[:]) } -func Hex_to_SUUID(str string) SUUID { - l := len(str) - var m string - if l < 16 { - m = strings.Repeat("0", 16-l) + str - } else { - m = str +// ParseSUUID parses s into an SUUID. +func ParseSUUID(s string) SUUID { + if n := len(s); n < 16 { + s = strings.Repeat("0", 16-n) + s } - b, _ := hex.DecodeString(m) - num := binary.BigEndian.Uint64(b) - return SUUID(num) + b, _ := hex.DecodeString(s) + return SUUID(binary.BigEndian.Uint64(b)) } +// GUID represents a globally unique identifier. type GUID [16]byte // UnmarshalText parses a text value into a GUID. @@ -73,11 +62,11 @@ func (id *GUID) UnmarshalText(text []byte) error { return nil } -func (self GUID) String() string { +func (id GUID) String() string { var offsets = [...]int{0, 2, 4, 6, 9, 11, 14, 16, 19, 21, 24, 26, 28, 30, 32, 34} const hexString = "0123456789abcdef" r := make([]byte, 36) - for i, b := range self { + for i, b := range id { r[offsets[i]] = hexString[b>>4] r[offsets[i]+1] = hexString[b&0xF] } @@ -89,26 +78,29 @@ func (self GUID) String() string { } -func Equal(a, b *GUID) bool { - for i, v := range a { - if v != b[i] { +// Equals returns true if id equals other. +func (id *GUID) Equals(other *GUID) bool { + for i, v := range id { + if v != other[i] { return false } } return true } -func RandomUUID() GUID { +// NewGUID returns a random GUID. +func NewGUID() GUID { uid, _ := gocql.RandomUUID() - var r GUID - copy(r[:], uid[:]) - return r + var id GUID + copy(id[:], uid[:]) + return id } -func ParseGUID(input string) (GUID, error) { +// ParseGUID parses s into a GUID. +func ParseGUID(s string) (GUID, error) { var u GUID j := 0 - for _, r := range input { + for _, r := range s { switch { case r == '-' && j&1 == 0: continue @@ -119,12 +111,12 @@ func ParseGUID(input string) (GUID, error) { case r >= 'A' && r <= 'F' && j < 32: u[j/2] |= byte(r-'A'+10) << uint(4-j&1*4) default: - return GUID{}, fmt.Errorf("invalid GUID %q", input) + return GUID{}, fmt.Errorf("invalid GUID %q", s) } j += 1 } if j != 32 { - return GUID{}, fmt.Errorf("invalid GUID %q", input) + return GUID{}, fmt.Errorf("invalid GUID %q", s) } return u, nil } diff --git a/id_test.go b/pilosa_test.go similarity index 50% rename from id_test.go rename to pilosa_test.go index f4c047ca3..f1cb30bde 100644 --- a/id_test.go +++ b/pilosa_test.go @@ -8,44 +8,44 @@ import ( ) // Ensure id can be parsed from string. -func TestId_Small(t *testing.T) { - if v := pilosa.Hex_to_SUUID("1"); v != 1 { +func TestSUUID_Small(t *testing.T) { + if v := pilosa.ParseSUUID("1"); v != 1 { t.Fatalf("unexpected SUUID: %v", v) } } // Ensure generated IDs are unique. -func TestId_Unique(t *testing.T) { - a, b := pilosa.Id(), pilosa.Id() +func TestSUUID_Unique(t *testing.T) { + a, b := pilosa.NewSUUID(), pilosa.NewSUUID() if a == b { t.Fatalf("ids should be unique: %v != %v", a, b) } } // Ensure ids can be converted to and from hex. -func TestId_Hex(t *testing.T) { - a := pilosa.Id() - b := pilosa.Hex_to_SUUID(pilosa.SUUID_to_Hex(a)) +func TestSUUID_Hex(t *testing.T) { + a := pilosa.NewSUUID() + b := pilosa.ParseSUUID(a.String()) if a != b { t.Fatalf("ids not equal: %v != %v", a, b) } } // Ensure ids can be generated in sequence. -func TestId_Multiple(t *testing.T) { +func TestSUUID_Multiple(t *testing.T) { for i := 0; i < 10; i++ { - println(pilosa.SUUID_to_Hex(pilosa.Id())) + println(pilosa.NewSUUID().String()) } } -// Ensure a random UUID can be converted to a string. -func TestRandomUUID_String(t *testing.T) { - fmt.Println(pilosa.RandomUUID().String()) +// Ensure a random GUID can be converted to a string. +func TestGUID_String(t *testing.T) { + fmt.Println(pilosa.NewGUID().String()) } -func BenchmarkId(b *testing.B) { +func BenchmarkSUUID(b *testing.B) { // run the Fib function b.N times for n := 0; n < b.N; n++ { - pilosa.Id() + pilosa.NewSUUID() } } diff --git a/query/parser.go b/query/parser.go index 63a68487f..d15fec54a 100644 --- a/query/parser.go +++ b/query/parser.go @@ -52,7 +52,7 @@ func (self *QueryParser) Parse() (query *Query, err error) { }() var token *Token - id := pilosa.RandomUUID() + id := pilosa.NewGUID() query = &Query{Id: &id, Subqueries: make([]Query, 0), Args: make(map[string]interface{})} token = self.next() diff --git a/query/planner.go b/query/planner.go index 0f7d962de..bc7de4391 100644 --- a/query/planner.go +++ b/query/planner.go @@ -75,7 +75,7 @@ func (self *BaseQueryStep) GetLocation() *db.Location { func (self *BaseQueryStep) LocIsDest() bool { log.Trace("BaseQueryStep.LocIsDest") - if pilosa.Equal(self.Location.ProcessId, self.Destination.ProcessId) && + if self.Location.ProcessId.Equals(self.Destination.ProcessId) && self.Location.FragmentId == self.Destination.FragmentId { log.Trace("BaseQueryStep.LocIsDest Return true") return true @@ -616,7 +616,7 @@ func (self *QueryPlanner) flatten(qt QueryTree, id *pilosa.GUID, location *db.Lo } step := CatQueryStep{&BaseQueryStep{id, "cat", loc, location}, inputs, cat.N} for index, subq := range cat.subqueries { - sub_id := pilosa.RandomUUID() + sub_id := pilosa.NewGUID() step.Inputs[index] = &sub_id subq_steps, err := self.flatten(subq, &sub_id, loc) if err != nil { @@ -634,7 +634,7 @@ func (self *QueryPlanner) flatten(qt QueryTree, id *pilosa.GUID, location *db.Lo } step := StashQueryStep{&BaseQueryStep{id, "stash", loc, location}, inputs, stash.N} for index, subq := range stash.subqueries { - sub_id := pilosa.RandomUUID() + sub_id := pilosa.NewGUID() step.Inputs[index] = &sub_id subq_steps, err := self.flatten(subq, &sub_id, loc) if err != nil { @@ -651,7 +651,7 @@ func (self *QueryPlanner) flatten(qt QueryTree, id *pilosa.GUID, location *db.Lo } step := UnionQueryStep{&BaseQueryStep{id, "union", loc, location}, inputs} for index, subq := range union.subqueries { - sub_id := pilosa.RandomUUID() + sub_id := pilosa.NewGUID() step.Inputs[index] = &sub_id subq_steps, err := self.flatten(subq, &sub_id, loc) if err != nil { @@ -668,7 +668,7 @@ func (self *QueryPlanner) flatten(qt QueryTree, id *pilosa.GUID, location *db.Lo } step := IntersectQueryStep{&BaseQueryStep{id, "intersect", loc, location}, inputs} for index, subq := range intersect.subqueries { - sub_id := pilosa.RandomUUID() + sub_id := pilosa.NewGUID() step.Inputs[index] = &sub_id subq_steps, err := self.flatten(subq, &sub_id, loc) if err != nil { @@ -685,7 +685,7 @@ func (self *QueryPlanner) flatten(qt QueryTree, id *pilosa.GUID, location *db.Lo } step := DifferenceQueryStep{&BaseQueryStep{id, "difference", loc, location}, inputs} for index, subq := range difference.subqueries { - sub_id := pilosa.RandomUUID() + sub_id := pilosa.NewGUID() step.Inputs[index] = &sub_id subq_steps, err := self.flatten(subq, &sub_id, loc) if err != nil { @@ -727,7 +727,7 @@ func (self *QueryPlanner) flatten(qt QueryTree, id *pilosa.GUID, location *db.Lo plan := QueryPlan{step} return &plan, nil } else if cnt, ok := qt.(*CountQueryTree); ok { - sub_id := pilosa.RandomUUID() + sub_id := pilosa.NewGUID() loc, err := cnt.getLocation(self.Database) if err != nil { return nil, err @@ -740,7 +740,7 @@ func (self *QueryPlanner) flatten(qt QueryTree, id *pilosa.GUID, location *db.Lo plan = append(plan, *subq_steps...) plan = append(plan, step) } else if topn, ok := qt.(*TopNQueryTree); ok { - sub_id := pilosa.RandomUUID() + sub_id := pilosa.NewGUID() loc, err := topn.getLocation(self.Database) if err != nil { return nil, err diff --git a/query/query.go b/query/query.go index f012d4d27..b50c904e8 100644 --- a/query/query.go +++ b/query/query.go @@ -68,7 +68,7 @@ func QueryPlanForTokens(database *db.Database, tokens []Token, destination *db.L func QueryPlanForQuery(database *db.Database, query *Query, destination *db.Location) (*QueryPlan, error) { log.Trace("QueryPlanForQuery", database, query, destination) query_planner := QueryPlanner{Database: database, Query: query} - id := pilosa.RandomUUID() + id := pilosa.NewGUID() query_plan, err := query_planner.Plan(query, &id, destination) if err != nil { return nil, err diff --git a/storage/leveldb/leveldb.go b/storage/leveldb/leveldb.go index 4af2920cd..5a78afca6 100644 --- a/storage/leveldb/leveldb.go +++ b/storage/leveldb/leveldb.go @@ -40,7 +40,7 @@ func NewStorage(opt pilosa.StorageOptions) *Storage { opt.DB, strconv.Itoa(opt.Slice), opt.Frame, - pilosa.SUUID_to_Hex(opt.FragmentID), + opt.FragmentID.String(), ) return &Storage{path: path} From 26bc85d5657da424e8fb0a9005bfcfe263416551 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Thu, 10 Sep 2015 17:26:32 -0600 Subject: [PATCH 2/2] replace command queue with mutex This commit simplifies Fragment by replacing the request/response channels with a `sync.Mutex` to restrict access. This also simplifies the `FragmentContainer` code which can largely be removed now since most of the methods are simply wrappers around `GetFragment()` and `Fragment`. --- brand.go | 24 +- commands.go | 372 ------------------------- fragment.go | 213 ++++++++++++++ fragment_container.go | 632 +++++++++++++++--------------------------- general.go | 15 +- 5 files changed, 459 insertions(+), 797 deletions(-) delete mode 100644 commands.go create mode 100644 fragment.go diff --git a/brand.go b/brand.go index 33cf34d54..c3c39ebe8 100644 --- a/brand.go +++ b/brand.go @@ -14,11 +14,7 @@ import ( var FragmentBase string -var globalLock *sync.Mutex - -func init() { - globalLock = new(sync.Mutex) -} +var globalLock sync.Mutex type Pair struct { Key, Count uint64 @@ -199,10 +195,15 @@ func (b *Brand) Stats() interface{} { "skip": b.skip} return stats } -func (b *Brand) Store(bitmap_id uint64, bm *Bitmap, filter uint64) { - b.storage.Store(bitmap_id, b.db, b.frame, b.slice, filter, bm) + +func (b *Brand) Store(bitmap_id uint64, bm *Bitmap, filter uint64) error { + if err := b.storage.Store(bitmap_id, b.db, b.frame, b.slice, filter, bm); err != nil { + return err + } b.cache_it(bm, bitmap_id, filter) + return nil } + func (b *Brand) checkRank() { if len(b.rankings) < 50 { b.Rank() @@ -393,7 +394,7 @@ func (b *Brand) Persist() error { return encoder.Encode(results) } -func (b *Brand) Load(requestChan chan Command, f *Fragment) { +func (b *Brand) Load(f *Fragment) { log.Warn("Brand Load") time.Sleep(time.Duration(rand.Intn(32)) * time.Second) //trying to avoid mass cassandra hit r, err := openFile(b.getFileName()) @@ -406,15 +407,14 @@ func (b *Brand) Load(requestChan chan Command, f *Fragment) { if err := dec.Decode(&keys); err != nil { return } + globalLock.Lock() defer globalLock.Unlock() + // probaly need to get a etcd lock too someday for _, k := range keys { - request := NewLoadRequest(k) - requestChan <- request - request.Response() + b.Get(k) time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond) //trying to avoid mass cassandra hit - } } diff --git a/commands.go b/commands.go deleted file mode 100644 index 73c723d3e..000000000 --- a/commands.go +++ /dev/null @@ -1,372 +0,0 @@ -package pilosa - -import ( - "bytes" - "compress/gzip" - "encoding/base64" - "io/ioutil" - "time" - - log "github.com/cihub/seelog" -) - -type Result struct { - answer Calculation - exec_time time.Duration -} - -type Responder struct { - result chan Result - queryType string -} - -func NewResponder(queryType string) *Responder { - return &Responder{ - result: make(chan Result), - queryType: queryType, - } -} - -func (r *Responder) QueryType() string { return r.queryType } -func (r *Responder) Response() Result { return <-r.result } -func (r *Responder) ResponseChannel() chan Result { return r.result } - -type Calculation interface{} - -type Command interface { - Execute(*Fragment) Calculation - QueryType() string - Response() Result - ResponseChannel() chan Result -} - -type CmdGet struct { - *Responder - bitmap_id uint64 -} - -func NewGet(bitmap_id uint64) *CmdGet { - return &CmdGet{NewResponder("Get"), bitmap_id} -} - -func (self *CmdGet) Execute(f *Fragment) Calculation { - return f.NewHandle(self.bitmap_id) -} - -type CmdCount struct { - *Responder - bitmap BitmapHandle -} - -func NewCount(bitmap_handle BitmapHandle) *CmdCount { - return &CmdCount{NewResponder("Count"), bitmap_handle} -} - -func (self *CmdCount) Execute(f *Fragment) Calculation { - bm, ok := f.getBitmap(self.bitmap) - if ok == false { - return uint64(0) - } - return bm.BitCount() -} - -type CmdUnion struct { - *Responder - bitmap_ids []BitmapHandle -} - -func NewUnion(bitmaps []BitmapHandle) *CmdUnion { - result := &CmdUnion{NewResponder("Union"), bitmaps} - return result -} -func (self *CmdUnion) Execute(f *Fragment) Calculation { - return f.union(self.bitmap_ids) -} - -type CmdDifference struct { - *Responder - bitmap_ids []BitmapHandle -} - -func NewDifference(bitmaps []BitmapHandle) *CmdDifference { - result := &CmdDifference{NewResponder("Difference"), bitmaps} - return result -} -func (self *CmdDifference) Execute(f *Fragment) Calculation { - return f.difference(self.bitmap_ids) -} - -type CmdIntersect struct { - *Responder - bitmaps []BitmapHandle -} - -func NewIntersect(bh []BitmapHandle) *CmdIntersect { - return &CmdIntersect{NewResponder("Intersect"), bh} -} -func (self *CmdIntersect) Execute(f *Fragment) Calculation { - return f.intersect(self.bitmaps) -} - -type BitArgs struct { - Bitmap_id uint64 - Bit_pos uint64 -} -type CmdSetBit struct { - *Responder - bitmap_id uint64 - bit_pos uint64 - filter uint64 -} - -func NewSetBit(bitmap_id uint64, bit_pos uint64, filter uint64) *CmdSetBit { - result := &CmdSetBit{NewResponder("SetBit"), bitmap_id, bit_pos, filter} - return result -} -func (self *CmdSetBit) Execute(f *Fragment) Calculation { - return f.impl.SetBit(self.bitmap_id, self.bit_pos, self.filter) -} - -type CmdGetBytes struct { - *Responder - bitmap BitmapHandle -} - -func NewGetBytes(bh BitmapHandle) *CmdGetBytes { - return &CmdGetBytes{NewResponder("GetBytes"), bh} -} - -func (self *CmdGetBytes) Execute(f *Fragment) Calculation { - bm, ok := f.getBitmap(self.bitmap) - //*Compress it - if !ok { - bm = NewBitmap() - log.Warn("cache miss") - } - var b bytes.Buffer - w := gzip.NewWriter(&b) - w.Write(bm.ToBytes()) - w.Flush() - w.Close() - - return b.Bytes() -} - -type CmdFromBytes struct { - *Responder - compressed_bytes []byte -} - -func NewFromBytes(bytes []byte) *CmdFromBytes { - return &CmdFromBytes{NewResponder("FromBytes"), bytes} -} - -func (self *CmdFromBytes) Execute(f *Fragment) Calculation { - reader, _ := gzip.NewReader(bytes.NewReader(self.compressed_bytes)) - b, _ := ioutil.ReadAll(reader) - - result := NewBitmap() - result.FromBytes(b) - return f.AllocHandle(result) -} - -type CmdEmpty struct { - *Responder -} - -func NewEmpty() *CmdEmpty { - return &CmdEmpty{NewResponder("Empty")} -} - -func (self *CmdEmpty) Execute(f *Fragment) Calculation { - result := NewBitmap() - return f.AllocHandle(result) -} - -type CmdGetList struct { - *Responder - bitmap_ids []uint64 -} - -func NewGetList(bitmap_ids []uint64) *CmdGetList { - return &CmdGetList{NewResponder("GetList"), bitmap_ids} -} - -func (self *CmdGetList) Execute(f *Fragment) Calculation { - ret := make([]BitmapHandle, len(self.bitmap_ids)) - for i, v := range self.bitmap_ids { - ret[i] = f.NewHandle(v) - } - - return ret -} - -type CmdTopN struct { - *Responder - bitmap BitmapHandle - n int - categories []uint64 -} - -func NewTopN(b BitmapHandle, n int, categories []uint64) *CmdTopN { - return &CmdTopN{NewResponder("TopN"), b, n, categories} -} -func (self *CmdTopN) Execute(f *Fragment) Calculation { - return f.TopN(self.bitmap, self.n, self.categories) -} - -type CmdClear struct { - *Responder -} - -func NewClear() *CmdClear { - return &CmdClear{NewResponder("Clear")} -} -func (self *CmdClear) Execute(f *Fragment) Calculation { - return f.impl.Clear() -} - -type CmdLoader struct { - *Responder - bitmap_id uint64 - compressed_bitmap string - filter uint64 -} - -func NewLoader(bitmap_id uint64, compressed_bitmap string, filter uint64) *CmdLoader { - return &CmdLoader{NewResponder("Loader"), bitmap_id, compressed_bitmap, filter} -} -func (self *CmdLoader) Execute(f *Fragment) Calculation { - buf, err := base64.StdEncoding.DecodeString(self.compressed_bitmap) - if err != nil { - log.Warn(err) - return "ok" - } - reader, _ := gzip.NewReader(bytes.NewReader(buf)) - data, _ := ioutil.ReadAll(reader) - - bm := NewBitmap() - bm.FromBytes(data) - f.impl.Store(self.bitmap_id, bm, self.filter) - return "ok" -} - -type CmdStats struct { - *Responder -} - -func NewStats() *CmdStats { - return &CmdStats{NewResponder("Stats")} -} -func (self *CmdStats) Execute(f *Fragment) Calculation { - return f.impl.Stats() -} - -type CmdLoadRequest struct { - *Responder - bitmap_id uint64 -} - -func NewLoadRequest(bitmap_id uint64) *CmdLoadRequest { - result := &CmdLoadRequest{NewResponder("LoadRequest"), bitmap_id} - return result -} -func (self *CmdLoadRequest) Execute(f *Fragment) Calculation { - f.impl.Get(self.bitmap_id) - return 0 -} - -type CmdRange struct { - *Responder - bitmap_id uint64 - start_time time.Time - end_time time.Time -} - -func NewRange(bitmap_id uint64, start, end time.Time) *CmdRange { - return &CmdRange{NewResponder("Range"), bitmap_id, start, end} -} - -func (self *CmdRange) Execute(f *Fragment) Calculation { - return f.build_time_range_bitmap(self.bitmap_id, self.start_time, self.end_time) -} - -type CmdMask struct { - *Responder - start, end uint64 -} - -func NewMask(start, end uint64) *CmdMask { - return &CmdMask{NewResponder("Mask"), start, end} -} - -func (self *CmdMask) Execute(f *Fragment) Calculation { - result := NewBitmap() - for i := self.start; i < self.end; i++ { - result.SetBit(i) - } - return f.AllocHandle(result) -} - -type CmdTopFill struct { - *Responder - args FillArgs -} - -func NewTopFill(a FillArgs) *CmdTopFill { - return &CmdTopFill{NewResponder("TopFill"), a} -} - -func (self *CmdTopFill) Execute(f *Fragment) Calculation { - result := make([]Pair, 0) - for _, v := range self.args.Bitmaps { - if f.exists(v) { - a := f.NewHandle(v) - if self.args.Handle == 0 { - // return just the count - bm, ok := f.getBitmap(a) - if ok && bm.Count() > 0 { - result = append(result, Pair{v, bm.Count()}) - } - } else { - res := f.intersect([]BitmapHandle{self.args.Handle, a}) - bm, ok := f.getBitmap(res) - if ok { - bc := bm.BitCount() - if bc > 0 { - result = append(result, Pair{v, bc}) - } - } - } - } - } - return result -} - -type CmdTopNAll struct { - *Responder - n int - categories []uint64 -} - -func NewTopNAll(n int, categories []uint64) *CmdTopNAll { - return &CmdTopNAll{NewResponder("TopNAll"), n, categories} -} - -func (self *CmdTopNAll) Execute(f *Fragment) Calculation { - return f.TopNAll(self.n, self.categories) -} - -type CmdClearBit struct { - *Responder - bitmap_id uint64 - bit_pos uint64 -} - -func NewClearBit(bitmap_id uint64, bit_pos uint64) *CmdClearBit { - result := &CmdClearBit{NewResponder("ClearBit"), bitmap_id, bit_pos} - return result -} -func (self *CmdClearBit) Execute(f *Fragment) Calculation { - return f.impl.ClearBit(self.bitmap_id, self.bit_pos) -} diff --git a/fragment.go b/fragment.go new file mode 100644 index 000000000..a8c2898a3 --- /dev/null +++ b/fragment.go @@ -0,0 +1,213 @@ +package pilosa + +import ( + "strings" + "sync" + "time" + + log "github.com/cihub/seelog" + "github.com/golang/groupcache/lru" +) + +type Fragment struct { + mu sync.Mutex + id SUUID + slice int + impl Pilosa + cache *lru.Cache + + // Provide an autoincrementing index for bitmap handles. + seq uint64 + + // Stats for how many messages have been processed. + stats FragmentStats +} + +func NewFragment(id SUUID, db string, slice int, frame string) *Fragment { + storage := NewStorage(Backend, StorageOptions{ + DB: db, + Slice: slice, + Frame: frame, + FragmentID: id, + LevelDBPath: LevelDBPath, + }) + + var impl Pilosa + if strings.HasSuffix(frame, ".n") { + impl = NewBrand(db, frame, slice, storage, 50000, 45000, 100) + } else { + impl = NewGeneral(db, frame, slice, storage) + } + + return &Fragment{ + id: id, + cache: lru.New(50000), + impl: impl, + slice: slice, + } +} + +func (f *Fragment) Bitmap(bh BitmapHandle) (*Bitmap, bool) { + f.mu.Lock() + defer f.mu.Unlock() + return f.bitmap(bh) +} + +func (f *Fragment) bitmap(bh BitmapHandle) (*Bitmap, bool) { + bm, ok := f.cache.Get(bh) + if ok && bm != nil { + return bm.(*Bitmap), ok + } + return NewBitmap(), false //cache fail +} + +func (f *Fragment) exists(bitmapID uint64) bool { + f.mu.Lock() + defer f.mu.Unlock() + return f.impl.Exists(bitmapID) +} + +func (f *Fragment) TopNAll(n int, categories []uint64) []Pair { + f.mu.Lock() + defer f.mu.Unlock() + return f.impl.TopNAll(n, categories) +} + +func (f *Fragment) TopN(bitmap BitmapHandle, n int, categories []uint64) []Pair { + f.mu.Lock() + defer f.mu.Unlock() + + bm, ok := f.cache.Get(bitmap) + if ok { + return f.impl.TopN(bm.(*Bitmap), n, categories) + } + return nil +} + +func (f *Fragment) NewHandle(bitmapID uint64) BitmapHandle { + f.mu.Lock() + defer f.mu.Unlock() + return f.allocHandle(f.impl.Get(bitmapID)) +} + +func (f *Fragment) AllocHandle(bm *Bitmap) BitmapHandle { + f.mu.Lock() + defer f.mu.Unlock() + return f.allocHandle(bm) +} + +func (f *Fragment) allocHandle(bm *Bitmap) BitmapHandle { + handle := f.nextHandle() + f.cache.Add(handle, bm) + return handle +} + +func (f *Fragment) nextHandle() BitmapHandle { + millis := uint64(time.Now().UTC().UnixNano()) + id := millis << (64 - 41) + id |= uint64(f.slice) << (64 - 41 - 13) + id |= f.seq % 1024 + f.seq += 1 + return BitmapHandle(id) +} + +func (f *Fragment) Union(bitmaps []BitmapHandle) BitmapHandle { + f.mu.Lock() + defer f.mu.Unlock() + + result := NewBitmap() + for i, id := range bitmaps { + bm, _ := f.bitmap(id) + if i == 0 { + result = bm + } else { + result = result.Union(bm) + } + } + return f.allocHandle(result) +} + +func (f *Fragment) build_time_range_bitmap(bitmapID uint64, start, end time.Time) BitmapHandle { + f.mu.Lock() + defer f.mu.Unlock() + + result := NewBitmap() + for i, bid := range GetRange(start, end, bitmapID) { + bm := f.impl.Get(bid) + if i == 0 { + result = bm + } else { + result = result.Union(bm) + } + } + return f.AllocHandle(result) +} + +func (f *Fragment) Intersect(bitmaps []BitmapHandle) BitmapHandle { + f.mu.Lock() + defer f.mu.Unlock() + + var result *Bitmap + for i, id := range bitmaps { + bm, _ := f.bitmap(id) + if i == 0 { + result = bm.Clone() + } else { + result = result.Intersection(bm) + } + } + return f.allocHandle(result) +} + +func (f *Fragment) Difference(bitmaps []BitmapHandle) BitmapHandle { + f.mu.Lock() + defer f.mu.Unlock() + + result := NewBitmap() + for i, id := range bitmaps { + bm, _ := f.bitmap(id) + if i == 0 { + result = bm + } else { + result = result.Difference(bm) + } + } + return f.allocHandle(result) +} + +func (f *Fragment) Persist() { + f.mu.Lock() + defer f.mu.Unlock() + + err := f.impl.Persist() + if err != nil { + log.Warn("Error saving:", err) + } +} + +func (f *Fragment) Load() { + f.mu.Lock() + defer f.mu.Unlock() + f.impl.Load(f) +} + +// FragmentStats represents in-memory stats for a single fragment. +type FragmentStats struct { + // Messages processed by the fragment + ProcessN uint64 + ProcessTime time.Duration +} + +type Pilosa interface { + Get(id uint64) *Bitmap + SetBit(id uint64, bit_pos uint64, filter uint64) bool + ClearBit(id uint64, bit_pos uint64) bool + TopN(b *Bitmap, n int, categories []uint64) []Pair + TopNAll(n int, categories []uint64) []Pair + Clear() bool + Store(bitmapID uint64, bm *Bitmap, filter uint64) error + Stats() interface{} + Persist() error + Load(fragment *Fragment) + Exists(id uint64) bool +} diff --git a/fragment_container.go b/fragment_container.go index c09645593..6ad4d7f9f 100644 --- a/fragment_container.go +++ b/fragment_container.go @@ -1,17 +1,19 @@ package pilosa import ( + "bytes" + "compress/gzip" + "encoding/base64" "encoding/gob" "errors" "fmt" - "strings" + "io/ioutil" "sync" "syscall" "time" log "github.com/cihub/seelog" - "github.com/golang/groupcache/lru" - "github.com/umbel/pilosa/statsd" + // "github.com/umbel/pilosa/statsd" ) // DefaultBackend is the default data storage layer. @@ -21,179 +23,160 @@ var Backend = DefaultBackend var LevelDBPath string +var ( + ErrFragmentNotFound = errors.New("fragment not found") +) + +func init() { + gob.Register(BitmapHandle(0)) + gob.Register([]Pair{}) +} + type FragmentContainer struct { + mu sync.Mutex fragments map[SUUID]*Fragment - mutex *sync.Mutex } func NewFragmentContainer() *FragmentContainer { - f := new(FragmentContainer) - f.fragments = make(map[SUUID]*Fragment) - f.mutex = &sync.Mutex{} - return f + return &FragmentContainer{ + fragments: make(map[SUUID]*Fragment), + } } type BitmapHandle uint64 + type FillArgs struct { - Frag_id SUUID - Handle BitmapHandle - Bitmaps []uint64 + FragmentID SUUID + Handle BitmapHandle + Bitmaps []uint64 } -func init() { - var vh BitmapHandle - gob.Register(vh) - var lp []Pair - gob.Register(lp) +func (fc *FragmentContainer) Fragment(fragmentID SUUID) (*Fragment, bool) { + fc.mu.Lock() + f, ok := fc.fragments[fragmentID] + fc.mu.Unlock() + return f, ok } -func (self *FragmentContainer) Shutdown() { - log.Warn("Container Shutdown Started") - var wg sync.WaitGroup - wg.Add(len(self.fragments)) - - for _, v := range self.fragments { - v.exit <- &wg +func (fc *FragmentContainer) Get(fragmentID SUUID, bitmapID uint64) (BitmapHandle, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return 0, ErrFragmentNotFound } - wg.Wait() - log.Warn("Container Shutdown Complete") + return f.NewHandle(bitmapID), nil } -func (self *FragmentContainer) LoadBitmap(frag_id SUUID, bitmap_id uint64, compressed_bitmap string, filter uint64) { - log.Trace("LoadBitmap") - if fragment, found := self.GetFragment(frag_id); found { - request := NewLoader(bitmap_id, compressed_bitmap, filter) - fragment.requestChan <- request - request.Response() - return +func (fc *FragmentContainer) LoadBitmap(fragmentID SUUID, bitmapID uint64, data string, filter uint64) error { + f, ok := fc.Fragment(fragmentID) + if !ok { + return ErrFragmentNotFound } + + // Decode from base64 encoding. + buf, err := base64.StdEncoding.DecodeString(data) + if err != nil { + return err + } + + // Decompress data. + reader, err := gzip.NewReader(bytes.NewReader(buf)) + if err != nil { + return err + } + b, err := ioutil.ReadAll(reader) + if err != nil { + return err + } + + // Build bitmap from data. + bm := NewBitmap() + bm.FromBytes(b) + + // Write bitmap to the underlying store. + return f.impl.Store(bitmapID, bm, filter) } -func (self *FragmentContainer) GetFragment(frag_id SUUID) (*Fragment, bool) { - log.Trace("index.GetFragment") - self.mutex.Lock() - c, v := self.fragments[frag_id] - self.mutex.Unlock() - return c, v +func (fc *FragmentContainer) Stats(fragmentID SUUID) interface{} { + f, ok := fc.Fragment(fragmentID) + if !ok { + return nil + } + return f.impl.Stats() } -func (self *FragmentContainer) Stats(frag_id SUUID) interface{} { - if fragment, found := self.GetFragment(frag_id); found { - request := NewStats() - fragment.requestChan <- request - return request.Response().answer +func (fc *FragmentContainer) Empty(fragmentID SUUID) (BitmapHandle, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return 0, ErrFragmentNotFound } - return nil -} -func (self *FragmentContainer) Empty(frag_id SUUID) (BitmapHandle, error) { - log.Trace("index.Empty") - if fragment, found := self.GetFragment(frag_id); found { - request := NewEmpty() - fragment.requestChan <- request - return request.Response().answer.(BitmapHandle), nil - } - return 0, errors.New("Invalid Bitmap Handle Empty") + return f.AllocHandle(NewBitmap()), nil } -func (self *FragmentContainer) Intersect(frag_id SUUID, bh []BitmapHandle) (BitmapHandle, error) { - log.Trace("index.Intersect") - if fragment, found := self.GetFragment(frag_id); found { - request := NewIntersect(bh) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_Intersect", result.exec_time.Nanoseconds()) - return result.answer.(BitmapHandle), nil +func (fc *FragmentContainer) Intersect(fragmentID SUUID, bh []BitmapHandle) (BitmapHandle, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return 0, ErrFragmentNotFound } - return 0, errors.New("Invalid Bitmap Handle Intersect") -} -func (self *FragmentContainer) Union(frag_id SUUID, bh []BitmapHandle) (BitmapHandle, error) { - log.Trace("index.Union") - if fragment, found := self.GetFragment(frag_id); found { - request := NewUnion(bh) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_Union", result.exec_time.Nanoseconds()) - return result.answer.(BitmapHandle), nil - } - return 0, errors.New("Invalid Bitmap Handle Union") + return f.Intersect(bh), nil } -func (self *FragmentContainer) Difference(frag_id SUUID, bh []BitmapHandle) (BitmapHandle, error) { - log.Trace("index.Difference") - if fragment, found := self.GetFragment(frag_id); found { - request := NewDifference(bh) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_Difference", result.exec_time.Nanoseconds()) - return result.answer.(BitmapHandle), nil +func (fc *FragmentContainer) Union(fragmentID SUUID, bh []BitmapHandle) (BitmapHandle, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return 0, ErrFragmentNotFound } - return 0, errors.New("Invalid Bitmap Handle Diff") + return f.Union(bh), nil } -func (self *FragmentContainer) Get(frag_id SUUID, bitmap_id uint64) (BitmapHandle, error) { - log.Trace("index.Get") - if fragment, found := self.GetFragment(frag_id); found { - request := NewGet(bitmap_id) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_Get", result.exec_time.Nanoseconds()) - return result.answer.(BitmapHandle), nil +func (fc *FragmentContainer) Difference(fragmentID SUUID, bh []BitmapHandle) (BitmapHandle, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return 0, ErrFragmentNotFound } - return 0, errors.New("Invalid Bitmap Handle Get") + return f.Difference(bh), nil } -func (self *FragmentContainer) Mask(frag_id SUUID, start, end uint64) (BitmapHandle, error) { - if fragment, found := self.GetFragment(frag_id); found { - request := NewMask(start, end) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_Mask", result.exec_time.Nanoseconds()) - return result.answer.(BitmapHandle), nil +func (fc *FragmentContainer) Mask(fragmentID SUUID, start, end uint64) (BitmapHandle, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return 0, ErrFragmentNotFound } - return 0, errors.New("Invalid Bitmap Handle") + + bm := NewBitmap() + for i := start; i < end; i++ { + bm.SetBit(i) + } + return f.AllocHandle(bm), nil } -func (self *FragmentContainer) Range(frag_id SUUID, bitmap_id uint64, start, end time.Time) (BitmapHandle, error) { - if fragment, found := self.GetFragment(frag_id); found { - request := NewRange(bitmap_id, start, end) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_Range", result.exec_time.Nanoseconds()) - return result.answer.(BitmapHandle), nil +func (fc *FragmentContainer) Range(fragmentID SUUID, bitmapID uint64, start, end time.Time) (BitmapHandle, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return 0, ErrFragmentNotFound } - return 0, errors.New("Invalid Bitmap Handle") + return f.build_time_range_bitmap(bitmapID, start, end), nil } -func (self *FragmentContainer) TopN(frag_id SUUID, bh BitmapHandle, n int, categories []uint64) ([]Pair, error) { - log.Trace("index.TopN") - if fragment, found := self.GetFragment(frag_id); found { - request := NewTopN(bh, n, categories) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_TopN", result.exec_time.Nanoseconds()) - return result.answer.([]Pair), nil +func (fc *FragmentContainer) TopN(fragmentID SUUID, bh BitmapHandle, n int, categories []uint64) ([]Pair, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return nil, ErrFragmentNotFound } - return nil, errors.New(fmt.Sprintf("Fragment not found:%s", frag_id.String())) + return f.TopN(bh, n, categories), nil } -func (self *FragmentContainer) TopNAll(frag_id SUUID, n int, categories []uint64) ([]Pair, error) { - log.Trace("index.TopNAll") - if fragment, found := self.GetFragment(frag_id); found { - request := NewTopNAll(n, categories) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_TopNAll", result.exec_time.Nanoseconds()) - return result.answer.([]Pair), nil +func (fc *FragmentContainer) TopNAll(fragmentID SUUID, n int, categories []uint64) ([]Pair, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return nil, ErrFragmentNotFound } - return nil, errors.New(fmt.Sprintf("Fragment not found:%s", frag_id.String())) + return f.TopNAll(n, categories), nil } -func (self *FragmentContainer) TopFillBatch(args []FillArgs) ([]Pair, error) { - log.Trace("index.TopFillBatch") - //should probaly make this concurrent but then all hell breaks loose +func (fc *FragmentContainer) TopFillBatch(args []FillArgs) ([]Pair, error) { results := make(map[uint64]uint64) for _, v := range args { - items, _ := self.TopFillFragment(v) + items, _ := fc.TopFillFragment(v) if len(args) == 1 { return items, nil } @@ -210,100 +193,143 @@ func (self *FragmentContainer) TopFillBatch(args []FillArgs) ([]Pair, error) { return ret_val, nil } -func (self *FragmentContainer) TopFillFragment(arg FillArgs) ([]Pair, error) { - log.Trace("index.TopFillFragment") - if fragment, found := self.GetFragment(arg.Frag_id); found { - request := NewTopFill(arg) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_TopFillFragment", result.exec_time.Nanoseconds()) - return result.answer.([]Pair), nil +func (fc *FragmentContainer) TopFillFragment(args FillArgs) ([]Pair, error) { + f, ok := fc.Fragment(args.FragmentID) + if !ok { + return nil, ErrFragmentNotFound } - return nil, errors.New("Invalid Bitmap Handle") + + result := make([]Pair, 0) + for _, v := range args.Bitmaps { + if !f.exists(v) { + continue + } + + a := f.NewHandle(v) + if args.Handle == 0 { + // Return just the count + if bm, ok := f.Bitmap(a); ok && bm.Count() > 0 { + result = append(result, Pair{v, bm.Count()}) + } + continue + } + + res := f.Intersect([]BitmapHandle{args.Handle, a}) + if bm, ok := f.Bitmap(res); ok { + bc := bm.BitCount() + if bc > 0 { + result = append(result, Pair{v, bc}) + } + } + } + return result, nil } -func (self *FragmentContainer) GetList(frag_id SUUID, bitmap_id []uint64) ([]BitmapHandle, error) { - log.Trace("index.GetList") - if fragment, found := self.GetFragment(frag_id); found { - request := NewGetList(bitmap_id) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_GetList", result.exec_time.Nanoseconds()) - return result.answer.([]BitmapHandle), nil +func (fc *FragmentContainer) GetList(fragmentID SUUID, bitmapIDs []uint64) ([]BitmapHandle, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return nil, ErrFragmentNotFound } - return nil, errors.New("Invalid Bitmap Handle GetList") + + a := make([]BitmapHandle, len(bitmapIDs)) + for i, v := range bitmapIDs { + a[i] = f.NewHandle(v) + } + return a, nil } -func (self *FragmentContainer) Count(frag_id SUUID, bitmap BitmapHandle) (uint64, error) { - log.Trace("index.Count") - if fragment, found := self.GetFragment(frag_id); found { - request := NewCount(bitmap) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_Count", result.exec_time.Nanoseconds()) - return result.answer.(uint64), nil +func (fc *FragmentContainer) Count(fragmentID SUUID, bh BitmapHandle) (uint64, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return 0, ErrFragmentNotFound } - return 0, errors.New("Invalid Bitmap Handle Count") + + bm, ok := f.Bitmap(bh) + if ok == false { + return 0, nil + } + return bm.BitCount(), nil } -func (self *FragmentContainer) GetBytes(frag_id SUUID, bh BitmapHandle) ([]byte, error) { - log.Trace("index.GetBytes") - if fragment, found := self.GetFragment(frag_id); found { - request := NewGetBytes(bh) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_GetBytes", result.exec_time.Nanoseconds()) - return result.answer.([]byte), nil +func (fc *FragmentContainer) GetBytes(fragmentID SUUID, bh BitmapHandle) ([]byte, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return nil, ErrFragmentNotFound } - return nil, errors.New("Invalid Bitmap Handle GetBytes") + + bm, ok := f.Bitmap(bh) + if !ok { + bm = NewBitmap() + log.Warn("cache miss") + } + + var b bytes.Buffer + w := gzip.NewWriter(&b) + if _, err := w.Write(bm.ToBytes()); err != nil { + return nil, err + } + if err := w.Flush(); err != nil { + return nil, err + } + if err := w.Close(); err != nil { + return nil, err + } + return b.Bytes(), nil } -func (self *FragmentContainer) FromBytes(frag_id SUUID, bytes []byte) (BitmapHandle, error) { - log.Trace("index.FromBytes") - if fragment, found := self.GetFragment(frag_id); found { - request := NewFromBytes(bytes) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_FromBytes", result.exec_time.Nanoseconds()) - return result.answer.(BitmapHandle), nil +func (fc *FragmentContainer) FromBytes(fragmentID SUUID, data []byte) (BitmapHandle, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return 0, ErrFragmentNotFound } - return 0, errors.New("Invalid Bitmap Handle FromBytes") + + r, _ := gzip.NewReader(bytes.NewReader(data)) + b, _ := ioutil.ReadAll(r) + + bm := NewBitmap() + bm.FromBytes(b) + return f.AllocHandle(bm), nil } -func (self *FragmentContainer) SetBit(frag_id SUUID, bitmap_id uint64, pos uint64, category uint64) (bool, error) { - log.Trace("SetBit", frag_id, bitmap_id, pos, category) - if fragment, found := self.GetFragment(frag_id); found { - request := NewSetBit(bitmap_id, pos, category) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_SetBit", result.exec_time.Nanoseconds()) - statsd.SendInc("fragmant_container_SetBit") - return result.answer.(bool), nil +func (fc *FragmentContainer) SetBit(fragmentID SUUID, bitmapID uint64, pos uint64, category uint64) (bool, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return false, ErrFragmentNotFound } - return false, errors.New("Invalid Bitmap Handle SetBit") + return f.impl.SetBit(bitmapID, pos, category), nil } -func (self *FragmentContainer) ClearBit(frag_id SUUID, bitmap_id uint64, pos uint64) (bool, error) { - log.Trace("ClearBit", frag_id, bitmap_id, pos) - if fragment, found := self.GetFragment(frag_id); found { - request := NewClearBit(bitmap_id, pos) - fragment.requestChan <- request - result := request.Response() - statsd.SendTimer("fragmant_container_ClearBit", result.exec_time.Nanoseconds()) - statsd.SendInc("fragmant_container_ClearBit") - return result.answer.(bool), nil +func (fc *FragmentContainer) ClearBit(fragmentID SUUID, bitmapID uint64, pos uint64) (bool, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return false, ErrFragmentNotFound } - return false, errors.New("Invalid Bitmap Handle ClearBit") + return f.impl.ClearBit(bitmapID, pos), nil } -func (self *FragmentContainer) Clear(frag_id SUUID) (bool, error) { - if fragment, found := self.GetFragment(frag_id); found { - request := NewClear() - fragment.requestChan <- request - return request.Response().answer.(bool), nil +func (fc *FragmentContainer) Clear(fragmentID SUUID) (bool, error) { + f, ok := fc.Fragment(fragmentID) + if !ok { + return false, ErrFragmentNotFound } - return false, errors.New("Invalid Fragment ID") + return f.impl.Clear(), nil } + +func (fc *FragmentContainer) AddFragment(db string, frame string, slice int, id SUUID) { + fc.mu.Lock() + defer fc.mu.Unlock() + + _, ok := fc.fragments[id] + if ok { + return + } + + log.Warn("ADD FRAGMENT", frame, db, slice, id.String()) + f := NewFragment(id, db, slice, frame) + fc.fragments[id] = f + // go f.Load(loader) +} + func dumpHandlesToLog() { var limit syscall.Rlimit if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &limit); err != nil { @@ -312,209 +338,3 @@ func dumpHandlesToLog() { mesg := fmt.Sprintf("%v file descriptors out of a maximum of %v available\n", limit.Cur, limit.Max) log.Warn(mesg) } - -func (self *FragmentContainer) AddFragment(db string, frame string, slice int, id SUUID) { - self.mutex.Lock() - defer self.mutex.Unlock() - _, ok := self.fragments[id] - if !ok { - // dumpHandlesToLog() - log.Warn("ADD FRAGMENT", frame, db, slice, id.String()) - f := NewFragment(id, db, slice, frame) - loader := make(chan Command) - self.fragments[id] = f - go f.ServeFragment(loader) - go f.Load(loader) - } - -} - -type Pilosa interface { - Get(id uint64) *Bitmap - SetBit(id uint64, bit_pos uint64, filter uint64) bool - ClearBit(id uint64, bit_pos uint64) bool - TopN(b *Bitmap, n int, categories []uint64) []Pair - TopNAll(n int, categories []uint64) []Pair - Clear() bool - Store(bitmap_id uint64, bm *Bitmap, filter uint64) - Stats() interface{} - Persist() error - Load(requestChan chan Command, fragment *Fragment) - Exists(id uint64) bool -} - -type Fragment struct { - requestChan chan Command - fragment_id SUUID - impl Pilosa - counter uint64 - slice int - cache *lru.Cache - mesg_count uint64 - mesg_time time.Duration - exit chan *sync.WaitGroup - queue_size int -} - -func NewFragment(frag_id SUUID, db string, slice int, frame string) *Fragment { - var impl Pilosa - log.Warn(fmt.Sprintf("XXXXXXXXXXXXXXXXXXXXXXXXXXX(%s)", frame)) - - storage := NewStorage(Backend, StorageOptions{ - DB: db, - Slice: slice, - Frame: frame, - FragmentID: frag_id, - LevelDBPath: LevelDBPath, - }) - - if strings.HasSuffix(frame, ".n") { - impl = NewBrand(db, frame, slice, storage, 50000, 45000, 100) - } else { - impl = NewGeneral(db, frame, slice, storage) - } - - f := new(Fragment) - f.requestChan = make(chan Command, 64) - f.fragment_id = frag_id - f.cache = lru.New(50000) - f.impl = impl //NewGeneral(db, slice, NewMemoryStorage()) - f.slice = slice - f.exit = make(chan *sync.WaitGroup) - f.queue_size = 0 - return f -} - -func (self *Fragment) getBitmap(bitmap BitmapHandle) (*Bitmap, bool) { - bm, ok := self.cache.Get(bitmap) - if ok && bm != nil { - return bm.(*Bitmap), ok - } - return NewBitmap(), false //cache fail but return ting em -} - -func (self *Fragment) exists(bitmap_id uint64) bool { - return self.impl.Exists(bitmap_id) -} -func (self *Fragment) TopNAll(n int, categories []uint64) []Pair { - return self.impl.TopNAll(n, categories) -} - -func (self *Fragment) TopN(bitmap BitmapHandle, n int, categories []uint64) []Pair { - - bm, ok := self.cache.Get(bitmap) - if ok { - return self.impl.TopN(bm.(*Bitmap), n, categories) - } - return nil -} - -func (self *Fragment) NewHandle(bitmap_id uint64) BitmapHandle { - bm := self.impl.Get(bitmap_id) - return self.AllocHandle(bm) - //given a bitmap_id return a newly allocated handle -} -func (self *Fragment) AllocHandle(bm *Bitmap) BitmapHandle { - handle := self.nextHandle() - self.cache.Add(handle, bm) - return handle -} - -func (self *Fragment) nextHandle() BitmapHandle { - millis := uint64(time.Now().UTC().UnixNano()) - id := millis << (64 - 41) - id |= uint64(self.slice) << (64 - 41 - 13) - id |= self.counter % 1024 - self.counter += 1 - return BitmapHandle(id) -} - -func (self *Fragment) union(bitmaps []BitmapHandle) BitmapHandle { - result := NewBitmap() - for i, id := range bitmaps { - bm, _ := self.getBitmap(id) - if i == 0 { - result = bm - } else { - result = result.Union(bm) - } - } - return self.AllocHandle(result) -} -func (self *Fragment) build_time_range_bitmap(bitmap_id uint64, start, end time.Time) BitmapHandle { - result := NewBitmap() - for i, bid := range GetRange(start, end, bitmap_id) { - bm := self.impl.Get(bid) - if i == 0 { - result = bm - } else { - result = result.Union(bm) - } - } - return self.AllocHandle(result) -} - -func (self *Fragment) intersect(bitmaps []BitmapHandle) BitmapHandle { - var result *Bitmap - for i, id := range bitmaps { - bm, _ := self.getBitmap(id) - if i == 0 { - result = bm.Clone() - } else { - result = result.Intersection(bm) - } - } - return self.AllocHandle(result) -} - -func (self *Fragment) difference(bitmaps []BitmapHandle) BitmapHandle { - result := NewBitmap() - for i, id := range bitmaps { - bm, _ := self.getBitmap(id) - if i == 0 { - result = bm - } else { - result = result.Difference(bm) - } - } - return self.AllocHandle(result) -} - -func (self *Fragment) Persist() { - err := self.impl.Persist() - if err != nil { - log.Warn("Error saving:", err) - } -} -func (self *Fragment) Load(loadChan chan Command) { - self.impl.Load(loadChan, self) -} - -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) ServeFragment(loadChan chan Command) { - for { - select { - case req := <-self.requestChan: - self.processCommand(req) - default: - select { - case req := <-self.requestChan: - self.processCommand(req) - case req := <-loadChan: - self.processCommand(req) - case wg := <-self.exit: - log.Warn("Fragment Shutdown") - self.Persist() - wg.Done() - } - } - } -} diff --git a/general.go b/general.go index e7f2f3eef..5c96d9672 100644 --- a/general.go +++ b/general.go @@ -63,11 +63,15 @@ func (self *General) TopN(b *Bitmap, n int, categories []uint64) []Pair { return empty } -func (self *General) Store(bitmap_id uint64, bm *Bitmap, filter uint64) { - self.storage.Store(bitmap_id, self.db, self.frame, self.slice, filter, bm) +func (self *General) Store(bitmap_id uint64, bm *Bitmap, filter uint64) error { + if err := self.storage.Store(bitmap_id, self.db, self.frame, self.slice, filter, bm); err != nil { + return err + } self.bitmap_cache.Add(bitmap_id, bm) self.keys[bitmap_id] = 0 + return nil } + func (self *General) OnEvicted(key lru.Key, value interface{}) { delete(self.keys, key.(uint64)) } @@ -108,7 +112,7 @@ func (self *General) Persist() error { return encoder.Encode(results) } -func (self *General) Load(requestChan chan Command, f *Fragment) { +func (self *General) Load(f *Fragment) { log.Warn("General Load") r, err := openFile(self.getFileName()) if err != nil { @@ -123,14 +127,11 @@ func (self *General) Load(requestChan chan Command, f *Fragment) { return } for _, k := range keys { - request := NewLoadRequest(k) - requestChan <- request - request.Response() + self.Get(k) } } func (self *General) TopNAll(n int, categories []uint64) []Pair { - results := make([]Pair, 0, 0) count := 0