diff --git a/index/bitmap.go b/index/bitmap.go index 1a41fda8b..8a32f715b 100644 --- a/index/bitmap.go +++ b/index/bitmap.go @@ -5,8 +5,9 @@ package index import ( "bytes" "encoding/gob" - "github.com/yasushi-saito/rbtree" "log" + + "github.com/yasushi-saito/rbtree" ) const ( @@ -46,10 +47,6 @@ func (set *IntSet) Size() int { // -var ( - errors map[error]int -) - /* ** native version turned out to be slower func popcount(i uint64)uint64{ val:= C.__builtin_popcountll(C.ulonglong(i)) @@ -95,6 +92,13 @@ func BlockArray_invert(a *BlockArray) BlockArray { return o } +func BlockArray_copy(a *BlockArray) BlockArray { + var o = BlockArray{} + for i, _ := range a.Block { + o.Block[i] = a.Block[i] + } + return o +} func BlockArray_intersection(a *BlockArray, b *BlockArray) BlockArray { var o = BlockArray{} for i, _ := range a.Block { @@ -126,6 +130,23 @@ func Compare(a uint64, b uint64) int { } return 0 } +func Clone(a_bm IBitmap) IBitmap { + + var a = a_bm.Min() + output := CreateRBBitmap() + for { + if a.Limit() { + break + } + var a_node = a.Item() + var o = BlockArray_copy(&a_node.Value) + var o_node = &Chunk{a_node.Key, o} + output.AddChunk(o_node) + a = a.Next() + } + return output +} + func Intersection(a_bm IBitmap, b_bm IBitmap) IBitmap { var a = a_bm.Min() var b = b_bm.Min() diff --git a/index/commands.go b/index/commands.go index e5d775d40..297656e7b 100644 --- a/index/commands.go +++ b/index/commands.go @@ -1,9 +1,6 @@ package index -import ( - "encoding/json" - "fmt" -) +import "time" type Rank struct { Key, Count uint64 @@ -15,76 +12,79 @@ func (p RankList) Swap(i, j int) { p[i], p[j] = p[j], p[i] } func (p RankList) Len() int { return len(p) } func (p RankList) Less(i, j int) bool { return p[i].Count > p[j].Count } +type Result struct { + answer Calculation + exec_time time.Duration +} + type Responder struct { - result chan string + result chan Result query_type string } func NewResponder(query_type string) *Responder { - return &Responder{make(chan string), query_type} + return &Responder{make(chan Result), query_type} } func (cmd *Responder) QueryType() string { return cmd.query_type } -func (cmd *Responder) Response() string { +func (cmd *Responder) Response() Result { return <-cmd.result } -func (cmd *Responder) ResponseChannel() chan string { +func (cmd *Responder) ResponseChannel() chan Result { return cmd.result } +type Calculation interface{} + type Command interface { - Execute(*Fragment) string + Execute(*Fragment) Calculation GetResponder() *Responder } - -func BuildCommandFactory(req *RequestJSON, decoder *json.Decoder) Command { - var result Command - - switch req.Request { - default: - result = &CmdUnknown{NewResponder("UnknownCommand"), req.Request} - case "UnionCount": - result = NewUnion(decoder) - case "IntersectCount": - result = NewIntersect(decoder) - case "SetBit": - result = NewSetBit(decoder) - } - return result +type CmdGet struct { + meta *Responder + bitmap_id uint64 } -type CmdUnknown struct { - meta *Responder - response string +func NewGet(bitmap_id uint64) *CmdGet { + return &CmdGet{NewResponder("Get"), bitmap_id} } -func (cmd *CmdUnknown) Execute(f *Fragment) string { - return fmt.Sprintf(`{ "Unknown Command":"%s" }`, cmd.response) -} - -func (cmd *CmdUnknown) GetResponder() *Responder { +func (cmd *CmdGet) GetResponder() *Responder { return cmd.meta } +func (cmd *CmdGet) Execute(f *Fragment) Calculation { + return f.NewHandle(cmd.bitmap_id) +} + +type CmdCount struct { + meta *Responder + bitmap BitmapHandle +} + +func NewCount(bitmap_handle BitmapHandle) *CmdCount { + return &CmdCount{NewResponder("Count"), bitmap_handle} +} + +func (cmd *CmdCount) GetResponder() *Responder { + return cmd.meta +} +func (cmd *CmdCount) Execute(f *Fragment) Calculation { + bm, _ := f.getBitmap(cmd.bitmap) + return BitCount(bm) +} type CmdUnion struct { meta *Responder - bitmap_ids []uint64 -} -type Args struct { - Bitmaps []uint64 + bitmap_ids []BitmapHandle } -func NewUnion(decoder *json.Decoder) *CmdUnion { - var f Args - decoder.Decode(&f) - result := &CmdUnion{NewResponder("UnionCount"), f.Bitmaps} +func NewUnion(bitmaps []BitmapHandle) *CmdUnion { + result := &CmdUnion{NewResponder("Union"), bitmaps} return result } -func (cmd *CmdUnion) Execute(f *Fragment) string { - bm := f.impl.Union(cmd.bitmap_ids) - result := BitCount(bm) - return fmt.Sprintf(`{ "value":%d }`, result) +func (cmd *CmdUnion) Execute(f *Fragment) Calculation { + return f.union(cmd.bitmap_ids) } func (cmd *CmdUnion) GetResponder() *Responder { return cmd.meta @@ -92,20 +92,14 @@ func (cmd *CmdUnion) GetResponder() *Responder { type CmdIntersect struct { meta *Responder - bitmaps []uint64 + bitmaps []BitmapHandle } -func NewIntersect(decoder *json.Decoder) *CmdIntersect { - var f Args - decoder.Decode(&f) - - result := &CmdIntersect{NewResponder("IntersectCount"), f.Bitmaps} - return result +func NewIntersect(bh []BitmapHandle) *CmdIntersect { + return &CmdIntersect{NewResponder("Intersect"), bh} } -func (cmd *CmdIntersect) Execute(f *Fragment) string { - bm := f.impl.Intersect(cmd.bitmaps) - result := BitCount(bm) - return fmt.Sprintf(`{ "value":%d }`, result) +func (cmd *CmdIntersect) Execute(f *Fragment) Calculation { + return f.intersect(cmd.bitmaps) } func (cmd *CmdIntersect) GetResponder() *Responder { return cmd.meta @@ -115,29 +109,19 @@ type BitArgs struct { Bitmap_id uint64 Bit_pos uint64 } - type CmdSetBit struct { - meta *Responder - - id uint64 + meta *Responder + bitmap BitmapHandle bit_pos uint64 } -func NewSetBit(decoder *json.Decoder) *CmdSetBit { - var f BitArgs - decoder.Decode(&f) - result := &CmdSetBit{NewResponder("SetBit"), f.Bitmap_id, f.Bit_pos} +func NewSetBit(bitmap BitmapHandle, bit_pos uint64) *CmdSetBit { + result := &CmdSetBit{NewResponder("SetBit"), bitmap, bit_pos} return result } -func (cmd *CmdSetBit) Execute(f *Fragment) string { - bitmap := f.impl.Get(cmd.id) - val := SetBit(bitmap, cmd.bit_pos) - m := 0 - if val { - m = 1 - } - result := BitCount(bitmap) - return fmt.Sprintf(`{ "value":%d , "changed":%d}`, result, m) +func (cmd *CmdSetBit) Execute(f *Fragment) Calculation { + bitmap, _ := f.getBitmap(cmd.bitmap) + return SetBit(bitmap, cmd.bit_pos) } func (cmd *CmdSetBit) GetResponder() *Responder { return cmd.meta diff --git a/index/general.go b/index/general.go index fa26a5945..582b248de 100644 --- a/index/general.go +++ b/index/general.go @@ -21,32 +21,6 @@ func NewGeneral(db string, slice int, s Storage) *General { } -func (f *General) Union(bitmaps []uint64) IBitmap { - result := NewBitmap() - for i, id := range bitmaps { - bm := f.Get(id) - if i == 0 { - result = bm - } else { - result = Union(result, bm) - } - } - return result -} - -func (f *General) Intersect(bitmaps []uint64) IBitmap { - result := NewBitmap() - for i, id := range bitmaps { - bm := f.Get(id) - if i == 0 { - result = bm - } else { - result = Intersection(result, bm) - } - } - return result -} - func (f *General) Get(bitmap_id uint64) IBitmap { bm, ok := f.bitmap_cache.Get(bitmap_id) if ok { diff --git a/index/server.go b/index/server.go index a6f99c470..2e934e755 100644 --- a/index/server.go +++ b/index/server.go @@ -1,35 +1,179 @@ package index import ( - "bytes" - "encoding/json" - "fmt" - "io/ioutil" - "net" - "net/http" - //"sort" - "log" + "errors" "time" + + "github.com/golang/groupcache/lru" + "github.com/nu7hatch/gouuid" ) type FragmentContainer struct { - fragments map[string]*Fragment + fragments map[*uuid.UUID]*Fragment } func NewFragmentContainer() *FragmentContainer { - return &FragmentContainer{make(map[string]*Fragment)} + return &FragmentContainer{make(map[*uuid.UUID]*Fragment)} } +type BitmapHandle uint64 + +func (a *FragmentContainer) GetFragment(frag_guid *uuid.UUID) (*Fragment, bool) { + //lock + c, v := a.fragments[frag_guid] + return c, v +} + +func (a *FragmentContainer) Intersect(frag_guid *uuid.UUID, bh []BitmapHandle) (BitmapHandle, error) { + if fragment, found := a.GetFragment(frag_guid); found { + request := NewIntersect(bh) + fragment.requestChan <- request + return request.GetResponder().Response().answer.(BitmapHandle), nil + } + return 0, errors.New("Invalid Bitmap Handle") +} +func (a *FragmentContainer) Union(frag_guid *uuid.UUID, bh []BitmapHandle) (BitmapHandle, error) { + if fragment, found := a.GetFragment(frag_guid); found { + request := NewUnion(bh) + fragment.requestChan <- request + return request.GetResponder().Response().answer.(BitmapHandle), nil + } + return 0, errors.New("Invalid Bitmap Handle") +} + +func (a *FragmentContainer) Get(frag_guid *uuid.UUID, bitmap_id uint64) (BitmapHandle, error) { + if fragment, found := a.GetFragment(frag_guid); found { + request := NewGet(bitmap_id) + fragment.requestChan <- request + return request.GetResponder().Response().answer.(BitmapHandle), nil + } + return 0, errors.New("Invalid Bitmap Handle") +} +func (a *FragmentContainer) Count(frag_guid *uuid.UUID, bitmap BitmapHandle) (uint64, error) { + if fragment, found := a.GetFragment(frag_guid); found { + request := NewCount(bitmap) + fragment.requestChan <- request + return request.GetResponder().Response().answer.(uint64), nil + } + return 0, errors.New("Invalid Bitmap Handle") +} + +func (a *FragmentContainer) SetBit(frag_guid *uuid.UUID, bitmap BitmapHandle, pos uint64) (bool, error) { + if fragment, found := a.GetFragment(frag_guid); found { + request := NewSetBit(bitmap, pos) + fragment.requestChan <- request + return request.GetResponder().Response().answer.(bool), nil + } + return false, errors.New("Invalid Bitmap Handle") +} + +func (a *FragmentContainer) AddFragment(frame string, db string, slice int, guid *uuid.UUID) { + f := NewFragment(guid, db, slice, frame) + a.fragments[guid] = f + go f.ServeFragment() +} + +type Pilosa interface { + Get(id uint64) IBitmap +} + +type Fragment struct { + requestChan chan Command + fragment_uuid *uuid.UUID + impl Pilosa + counter uint64 + slice int + cache *lru.Cache +} + +func NewFragment(guid *uuid.UUID, db string, slice int, frame string) *Fragment { + f := new(Fragment) + f.requestChan = make(chan Command, 64) + f.fragment_uuid = guid + f.cache = lru.New(10000) + f.impl = NewGeneral(db, slice, NewMemoryStorage()) + f.slice = slice + return f +} + +func (f *Fragment) getBitmap(bitmap BitmapHandle) (IBitmap, bool) { + bm, ok := f.cache.Get(bitmap) + return bm.(IBitmap), ok +} + +func (f *Fragment) NewHandle(bitmap_id uint64) BitmapHandle { + bm := f.impl.Get(bitmap_id) + return f.AllocHandle(bm) + //given a bitmap_id return a newly allocated handle +} +func (f *Fragment) AllocHandle(bm IBitmap) 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.counter % 1024 + f.counter += 1 + return BitmapHandle(id) +} + +func (f *Fragment) union(bitmaps []BitmapHandle) BitmapHandle { + result := NewBitmap() + for i, id := range bitmaps { + bm, _ := f.getBitmap(id) + if i == 0 { + result = bm + } else { + result = Union(result, bm) + } + } + return f.AllocHandle(result) +} +func (f *Fragment) intersect(bitmaps []BitmapHandle) BitmapHandle { + var result IBitmap + for i, id := range bitmaps { + bm, _ := f.getBitmap(id) + if i == 0 { + result = Clone(bm) + } else { + result = Intersection(result, bm) + } + } + return f.AllocHandle(result) +} + +func (f *Fragment) ServeFragment() { + for { + req := <-f.requestChan + start := time.Now() + responder := req.GetResponder() + answer := req.Execute(f) + delta := time.Since(start) + /* + var buffer bytes.Buffer + buffer.WriteString(`{ "results":`) + buffer.WriteString(answer) + buffer.WriteString(fmt.Sprintf(`,"query type": "%s"`, responder.QueryType())) + buffer.WriteString(fmt.Sprintf(`, "elapsed": "%s"}`, delta)) + */ + responder.ResponseChannel() <- Result{answer, delta} + } +} + +/* +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) AddFragment(frame string, db string, slice int, frag_guid string) { - f := &Fragment{make(chan Command), frag_guid, NewGeneral(db, slice, NewMemoryStorage())} - a.fragments[frag_guid] = f - go f.ServeFragment() -} - func (a *FragmentContainer) RunServer(porti int, closeChannel chan bool, started chan bool) { http.Handle("/", a) port := fmt.Sprintf(":%d", porti) @@ -58,40 +202,6 @@ func (a *FragmentContainer) RunServer(porti int, closeChannel chan bool, started } } -type Pilosa interface { - Union([]uint64) IBitmap - Intersect([]uint64) IBitmap - Get(id uint64) IBitmap -} - -type RequestJSON struct { - Request string - Fragment string - Args json.RawMessage -} -type Fragment struct { - requestChan chan Command - FragmentGuid string - impl Pilosa -} - -func (f *Fragment) ServeFragment() { - for { - req := <-f.requestChan - start := time.Now() - answer := `""` - responder := req.GetResponder() - answer = req.Execute(f) - delta := time.Since(start) - var buffer bytes.Buffer - buffer.WriteString(`{ "results":`) - buffer.WriteString(answer) - buffer.WriteString(fmt.Sprintf(`,"query type": "%s"`, responder.QueryType())) - buffer.WriteString(fmt.Sprintf(`, "elapsed": "%s"}`, delta)) - responder.ResponseChannel() <- buffer.String() - } -} - func handler(w http.ResponseWriter, r *http.Request, fragments map[string]*Fragment) { if r.Method == "POST" { var f RequestJSON @@ -121,3 +231,4 @@ func handler(w http.ResponseWriter, r *http.Request, fragments map[string]*Fragm } } } +*/ diff --git a/index/server_test.go b/index/server_test.go index 2220b69db..216045f2b 100644 --- a/index/server_test.go +++ b/index/server_test.go @@ -1,92 +1,50 @@ package index import ( - "encoding/json" - . "github.com/smartystreets/goconvey/convey" - "net/http" - "net/http/httptest" "testing" + // "io/ioutil" // "time" - "fmt" - "log" - "strings" + "github.com/nu7hatch/gouuid" + . "github.com/smartystreets/goconvey/convey" ) -func simple(id1, id2 int) string { - return fmt.Sprintf(`{ - "Request": "UnionCount", - "Fragment": "AAA-BBB-CCC", - "Args": { - "Bitmaps":[%d,%d] - } - }`, id1, id2) -} -func set_bit(id, pos int) string { - return fmt.Sprintf(`{ - "Request": "SetBit", - "Fragment": "AAA-BBB-CCC", - "Args": { - "Bitmap_id":%d, - "Bit_pos": %d - } - }`, id, pos) -} - -func sendRequest(msg string, dummy *FragmentContainer) (int, []byte) { - - r, err := http.NewRequest("POST", "http://api/foo", strings.NewReader(msg)) - if err != nil { - log.Fatal(err) - } - - w := httptest.NewRecorder() - - dummy.ServeHTTP(w, r) - //return w.Code, w.Body.String() - return w.Code, []byte(w.Body.String()) - -} -func getResult(key string, s []byte) interface{} { - var f interface{} - err := json.Unmarshal(s, &f) - if err != nil { - - log.Println(err) - return nil - } - m := f.(map[string]interface{}) - x := m["results"] - o := x.(map[string]interface{}) - return o[key] -} - func TestServer(t *testing.T) { - dummy := &FragmentContainer{make(map[string]*Fragment)} - dummy.AddFragment("general", "25", 0, "AAA-BBB-CCC") - var ( - c int - s []byte - ) - Convey("Set Bit 1 1", t, func() { - c, s = sendRequest(set_bit(1, 1), dummy) - So(c, ShouldEqual, 200) - v := getResult("value", s) - So(v, ShouldEqual, 1) + id, _ := uuid.NewV4() + dummy := NewFragmentContainer() + dummy.AddFragment("general", "25", 0, id) + + Convey("Get ", t, func() { + bh, _ := dummy.Get(id, 1234) + So(bh, ShouldNotEqual, 0) }) - Convey("Set Bit 2 2", t, func() { - c, s = sendRequest(set_bit(2, 2), dummy) - So(c, ShouldEqual, 200) - v := getResult("value", s) - So(v, ShouldEqual, 1) + Convey("SetBit/Count 1 1", t, func() { + bh, _ := dummy.Get(id, 1234) + changed, _ := dummy.SetBit(id, bh, 1) + So(changed, ShouldEqual, true) + changed, _ = dummy.SetBit(id, bh, 1) + So(changed, ShouldEqual, false) + num, _ := dummy.Count(id, bh) + So(num, ShouldEqual, 1) }) - Convey("Union", t, func() { - c, s = sendRequest(simple(1, 2), dummy) - So(c, ShouldEqual, 200) - v := getResult("value", s) - So(v, ShouldEqual, 2) + Convey("Union/Intersect", t, func() { + bh1, _ := dummy.Get(id, 1234) + // dummy.SetBit(id, bh1, 1) + + bh2, _ := dummy.Get(id, 4321) + dummy.SetBit(id, bh2, 2) + + handles := []BitmapHandle{bh1, bh2} + result, _ := dummy.Union(id, handles) + + num, _ := dummy.Count(id, result) + So(num, ShouldEqual, 2) + result, _ = dummy.Intersect(id, handles) + + num, _ = dummy.Count(id, result) + So(num, ShouldEqual, 0) }) } diff --git a/index/storage_mem.go b/index/storage_mem.go index d95dc8d2b..03473b6c2 100644 --- a/index/storage_mem.go +++ b/index/storage_mem.go @@ -2,16 +2,14 @@ package index // #cgo CFLAGS:-mpopcnt -import ( - // "log" - "fmt" -) +import "fmt" type MemoryStorage struct { db map[string]*Bitmap } func NewMemoryStorage() Storage { + // log.Println("Hello") obj := new(MemoryStorage) obj.db = make(map[string]*Bitmap) @@ -19,6 +17,8 @@ func NewMemoryStorage() Storage { } func (c *MemoryStorage) Fetch(bitmap_id uint64, db string, slice int) IBitmap { + // log.Println("hello") + key := fmt.Sprintf("%d:%s:%d", bitmap_id, db, slice) bitmap, found := c.db[key] if !found { diff --git a/util/id.go b/util/id.go new file mode 100644 index 000000000..e6db0eff4 --- /dev/null +++ b/util/id.go @@ -0,0 +1,39 @@ +package util + +import ( + "bytes" + "encoding/binary" + "encoding/hex" + "math/rand" + "time" +) + +var ( + counter = uint64(0) +) + +func init() { + rand.Seed(time.Now().UTC().UnixNano()) +} + +type SUUID uint64 + +func Id() SUUID { + millis := uint64(time.Now().UTC().UnixNano()) + id := millis << (64 - 41) + id |= uint64(rand.Intn(128)) << (64 - 41 - 13) + id |= counter % 1024 + counter += 1 + 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()) +} +func Hex_to_SUUID(str string) SUUID { + b, _ := hex.DecodeString(str) + num := binary.BigEndian.Uint64(b) + return SUUID(num) +} diff --git a/util/util_test.go b/util/util_test.go new file mode 100644 index 000000000..982dd2c17 --- /dev/null +++ b/util/util_test.go @@ -0,0 +1,76 @@ +package util + +import ( + "testing" + + "github.com/nu7hatch/gouuid" + . "github.com/smartystreets/goconvey/convey" +) + +/* +var ( + array [1000000]int + muid = make(map[SUUID]int) + muuid = make(map[*uuid.UUID]int) + r int +) + +func init() { + for i, _ := range array { + muid[Id()] = i + id, _ := uuid.NewV4() + muuid[id] = i + } + +} +*/ +func TestId(t *testing.T) { + Convey("Basic Usage", t, func() { + bc1 := Id() + println(SUUID_to_Hex(bc1)) + println(SUUID_to_Hex(bc1)) + bc2 := Id() + println(SUUID_to_Hex(bc2)) + So(bc1, ShouldNotEqual, bc2) + }) + Convey("Hex Encoded Usage", t, func() { + b1 := Id() + s := SUUID_to_Hex(b1) + b2 := Hex_to_SUUID(s) + So(b1, ShouldEqual, b2) + }) + +} + +func BenchmarkId(b *testing.B) { + // run the Fib function b.N times + for n := 0; n < b.N; n++ { + Id() + } +} + +func BenchmarkUUID(b *testing.B) { + // run the Fib function b.N times + for n := 0; n < b.N; n++ { + uuid.NewV4() + } +} + +/* +func BenchmarkLookupId(b *testing.B) { + x := Id() + for i := 0; i < b.N; i++ { + if a, found := muid[x]; found { + muid[x] = a + 1 + } + } +} +func BenchmarkLookupUUID(b *testing.B) { + x, _ := uuid.NewV4() + for i := 0; i < b.N; i++ { + if a, found := muuid[x]; found { + muuid[x] = a + 1 + } + } +} +*/