From ccdcac5ebbcc67a0398e0a56d18af5d2e722617d Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Thu, 12 Dec 2013 05:29:21 -0600 Subject: [PATCH] refactored command interface; fmt code --- index/bitmap.go | 93 ++++++++++--------- index/bitmap_test.go | 116 +++++++++++------------ index/commands.go | 209 ++++++++++++++++++++---------------------- index/general.go | 57 ++++++------ index/server.go | 69 +++++++------- index/server_test.go | 153 +++++++++++++++---------------- index/storage_cass.go | 45 ++++----- index/storage_mem.go | 58 ++++++------ 8 files changed, 393 insertions(+), 407 deletions(-) diff --git a/index/bitmap.go b/index/bitmap.go index a236ddfa5..1a41fda8b 100644 --- a/index/bitmap.go +++ b/index/bitmap.go @@ -3,10 +3,10 @@ package index // #cgo CFLAGS:-mpopcnt import ( - "log" "bytes" "encoding/gob" "github.com/yasushi-saito/rbtree" + "log" ) const ( @@ -163,8 +163,8 @@ func Invert(a_bm IBitmap) IBitmap { return output } -func NewBitmap()IBitmap{ - return CreateRBBitmap() +func NewBitmap() IBitmap { + return CreateRBBitmap() } func Union(a_bm IBitmap, b_bm IBitmap) IBitmap { @@ -231,50 +231,50 @@ func AND_NOT(a_bm IBitmap, b_bm IBitmap) IBitmap { defer a.Close() defer b.Close() output := CreateRBBitmap() - var o_last_Key = uint64(0) - - if o_last_Key != 0{ - o_last_Key = uint64(0) - } + var o_last_Key = uint64(0) - for { - if a.Limit() && b.Limit() { - break - } else if a.Limit() { - break - } else if b.Limit() { - var a_node = a.Item() - var o_node = &Chunk{a_node.Key, a_node.Value} - output.AddChunk(o_node) - o_last_Key = o_node.Key - a = a.Next() - } else if a.Item().Key < b.Item().Key { - var a_node = a.Item() - var o_node = &Chunk{a_node.Key, a_node.Value} - output.AddChunk(o_node) - o_last_Key = o_node.Key - a = a.Next() - } else if a.Item().Key > b.Item().Key { - var b_node = b.Item() - o_last_Key = b_node.Key - b = b.Next() - } else if a.Item().Key == b.Item().Key { - var a_node = a.Item() - var b_node = BlockArray_invert(&b.Item().Value) //probably need to copy this out - var o = BlockArray_intersection(&a_node.Value, &b_node) - var o_node = &Chunk{a_node.Key, o} - //could not add if all zero - if o_node.Value.bitcount()>0{ - output.AddChunk(o_node) - } - o_last_Key = o_node.Key - a = a.Next() - b = b.Next() - } else { - log.Println("NEVER SHOULD BE HERE") - break - } - } + if o_last_Key != 0 { + o_last_Key = uint64(0) + } + + for { + if a.Limit() && b.Limit() { + break + } else if a.Limit() { + break + } else if b.Limit() { + var a_node = a.Item() + var o_node = &Chunk{a_node.Key, a_node.Value} + output.AddChunk(o_node) + o_last_Key = o_node.Key + a = a.Next() + } else if a.Item().Key < b.Item().Key { + var a_node = a.Item() + var o_node = &Chunk{a_node.Key, a_node.Value} + output.AddChunk(o_node) + o_last_Key = o_node.Key + a = a.Next() + } else if a.Item().Key > b.Item().Key { + var b_node = b.Item() + o_last_Key = b_node.Key + b = b.Next() + } else if a.Item().Key == b.Item().Key { + var a_node = a.Item() + var b_node = BlockArray_invert(&b.Item().Value) //probably need to copy this out + var o = BlockArray_intersection(&a_node.Value, &b_node) + var o_node = &Chunk{a_node.Key, o} + //could not add if all zero + if o_node.Value.bitcount() > 0 { + output.AddChunk(o_node) + } + o_last_Key = o_node.Key + a = a.Next() + b = b.Next() + } else { + log.Println("NEVER SHOULD BE HERE") + break + } + } return output } @@ -459,4 +459,3 @@ func BitCount(b IBitmap) uint64 { } return total } - diff --git a/index/bitmap_test.go b/index/bitmap_test.go index cdcc8f518..77f6c51d8 100644 --- a/index/bitmap_test.go +++ b/index/bitmap_test.go @@ -1,72 +1,72 @@ package index import ( - "testing" - . "github.com/smartystreets/goconvey/convey" - "time" - ) + . "github.com/smartystreets/goconvey/convey" + "testing" + "time" +) - func TestBitmaps(t *testing.T) { - Convey("function BitCount should equal method bm.Count()", t, func() { - bm:= CreateRBBitmap() - for i:=uint64(0); i true ", t, func() { - bm1:= CreateRBBitmap() - bm2:= CreateRBBitmap() - SetBit(bm1,1) - //SetBit(bm2,2) - all:= AND_NOT(bm1,bm2) - res :=BitCount(all) +func TestBitmaps(t *testing.T) { + Convey("function BitCount should equal method bm.Count()", t, func() { + bm := CreateRBBitmap() + for i := uint64(0); i < uint64(4096); i++ { + SetBit(bm, i) + } + bc1 := BitCount(bm) + bc2 := bm.Count() + So(bc1, ShouldEqual, bc2) + }) + Convey("function AND_NOT 1 and not 0 => true ", t, func() { + bm1 := CreateRBBitmap() + bm2 := CreateRBBitmap() + SetBit(bm1, 1) + //SetBit(bm2,2) + all := AND_NOT(bm1, bm2) + res := BitCount(all) - So(1, ShouldEqual, res) - }) + So(1, ShouldEqual, res) + }) - Convey("UNION even + odd equal 4096 ", t, func() { - even:= CreateRBBitmap() - for i:=uint64(0); i p[j].Count } +type Responder struct { + result chan string + query_type string +} + +func NewResponder(query_type string) *Responder { + return &Responder{make(chan string), query_type} +} +func (cmd *Responder) QueryType() string { + return cmd.query_type +} +func (cmd *Responder) Response() string { + return <-cmd.result +} +func (cmd *Responder) ResponseChannel() chan string { + return cmd.result +} type Command interface { - Execute(*Fragment)string - QueryType() string - Response() string - ResponseChannel() chan string + Execute(*Fragment) string + GetResponder() *Responder } +func BuildCommandFactory(req *RequestJSON, decoder *json.Decoder) Command { + var result Command -func BuildCommandFactory(req *RequestJSON,decoder *json.Decoder)Command{ - var result Command - - switch req.Request{ - default: - result=&CmdUnknown{make(chan string),req.Request} - case "UnionCount": - result= NewUnion(decoder) - case "IntersectCount": - result= NewIntersect(decoder) - case "SetBit": - result= NewSetBit(decoder) - } - return result - } - - -type CmdUnknown struct{ - result chan string - response string -} -func (cmd *CmdUnknown) Execute(f *Fragment)string { - return fmt.Sprintf(`{ "Unknown Command":"%s" }`,cmd.response) -} -func (cmd *CmdUnknown) QueryType()string { - return "UnknownCommand" -} -func (cmd *CmdUnknown) Response()string { - return <-cmd.result -} -func (cmd *CmdUnknown) ResponseChannel()chan string { - return cmd.result + 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 CmdUnion struct{ - result chan string - bitmap_ids []uint64 +type CmdUnknown struct { + meta *Responder + response string +} + +func (cmd *CmdUnknown) Execute(f *Fragment) string { + return fmt.Sprintf(`{ "Unknown Command":"%s" }`, cmd.response) +} + +func (cmd *CmdUnknown) GetResponder() *Responder { + return cmd.meta +} + +type CmdUnion struct { + meta *Responder + bitmap_ids []uint64 } type Args struct { - Bitmaps[] uint64 + Bitmaps []uint64 } - -func NewUnion(decoder *json.Decoder) *CmdUnion{ - var f Args - decoder.Decode(&f) - - result:= &CmdUnion{make(chan string),f.Bitmaps} - return result +func NewUnion(decoder *json.Decoder) *CmdUnion { + var f Args + decoder.Decode(&f) + result := &CmdUnion{NewResponder("UnionCount"), f.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) string { + bm := f.impl.Union(cmd.bitmap_ids) + result := BitCount(bm) + return fmt.Sprintf(`{ "value":%d }`, result) } -func (cmd *CmdUnion) QueryType()string { - return "UnionCount" -} -func (cmd *CmdUnion) Response()string { - return <-cmd.result -} -func (cmd *CmdUnion) ResponseChannel()chan string { - return cmd.result +func (cmd *CmdUnion) GetResponder() *Responder { + return cmd.meta } - -type CmdIntersect struct{ - result chan string - bitmaps []uint64 +type CmdIntersect struct { + meta *Responder + bitmaps []uint64 } -func NewIntersect(decoder *json.Decoder) *CmdIntersect{ - var f Args - decoder.Decode(&f) - - result:= &CmdIntersect{ make(chan string), f.Bitmaps } - return result +func NewIntersect(decoder *json.Decoder) *CmdIntersect { + var f Args + decoder.Decode(&f) + + result := &CmdIntersect{NewResponder("IntersectCount"), f.Bitmaps} + return result } -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) string { + bm := f.impl.Intersect(cmd.bitmaps) + result := BitCount(bm) + return fmt.Sprintf(`{ "value":%d }`, result) } -func (cmd *CmdIntersect) QueryType()string { - return "IntersectCount" -} -func (cmd *CmdIntersect) Response()string { - return <-cmd.result -} -func (cmd *CmdIntersect) ResponseChannel()chan string { - return cmd.result +func (cmd *CmdIntersect) GetResponder() *Responder { + return cmd.meta } type BitArgs struct { - Bitmap_id uint64 - Bit_pos uint64 + Bitmap_id uint64 + Bit_pos uint64 } +type CmdSetBit struct { + meta *Responder -type CmdSetBit struct{ - result chan string - id uint64 - bit_pos uint64 + id uint64 + bit_pos uint64 } -func NewSetBit(decoder *json.Decoder) *CmdSetBit{ - var f BitArgs - decoder.Decode(&f) - - result:= &CmdSetBit{make(chan string),f.Bitmap_id,f.Bit_pos} - return result +func NewSetBit(decoder *json.Decoder) *CmdSetBit { + var f BitArgs + decoder.Decode(&f) + result := &CmdSetBit{NewResponder("SetBit"), f.Bitmap_id, f.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) 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) QueryType()string { - return "SetBit" -} -func (cmd *CmdSetBit) Response()string { - return <-cmd.result -} -func (cmd *CmdSetBit) ResponseChannel()chan string { - return cmd.result +func (cmd *CmdSetBit) GetResponder() *Responder { + return cmd.meta } diff --git a/index/general.go b/index/general.go index b6e345b8d..fa26a5945 100644 --- a/index/general.go +++ b/index/general.go @@ -6,45 +6,45 @@ import ( type General struct { bitmap_cache *lru.Cache - db string - slice int - storage Storage + db string + slice int + storage Storage } func NewGeneral(db string, slice int, s Storage) *General { f := new(General) f.bitmap_cache = lru.New(10000) f.storage = s - f.slice = slice - f.db = db + f.slice = slice + f.db = db return f } -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) 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) 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 { @@ -52,8 +52,7 @@ func (f *General) Get(bitmap_id uint64) IBitmap { if ok { return bm.(*Bitmap) } - bm = f.storage.Fetch(bitmap_id, f.db,f.slice) + bm = f.storage.Fetch(bitmap_id, f.db, f.slice) f.bitmap_cache.Add(bitmap_id, bm) return bm.(*Bitmap) } - diff --git a/index/server.go b/index/server.go index 0923ee99f..a6f99c470 100644 --- a/index/server.go +++ b/index/server.go @@ -8,30 +8,31 @@ import ( "net" "net/http" //"sort" + "log" "time" - "log" ) type FragmentContainer struct { - fragments map[string] *Fragment + fragments map[string]*Fragment } -func NewFragmentContainer() *FragmentContainer{ - return &FragmentContainer{make( map[string]*Fragment)} + +func NewFragmentContainer() *FragmentContainer { + return &FragmentContainer{make(map[string]*Fragment)} } - + func (a *FragmentContainer) ServeHTTP(w http.ResponseWriter, r *http.Request) { - handler(w , r,a.fragments) + 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() + 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) { +func (a *FragmentContainer) RunServer(porti int, closeChannel chan bool, started chan bool) { http.Handle("/", a) - port := fmt.Sprintf(":%d",porti) + port := fmt.Sprintf(":%d", porti) s := &http.Server{ Addr: port, @@ -46,32 +47,32 @@ func (a *FragmentContainer) RunServer(porti int, closeChannel chan bool,started log.Panicf(e.Error()) } go s.Serve(l) - started<- true + started <- true select { case <-closeChannel: log.Printf("Server thread exit") l.Close() - // Shutdown() + // Shutdown() return break } } -type Pilosa interface{ - Union([]uint64) IBitmap - Intersect([] uint64) IBitmap - Get(id uint64 )IBitmap +type Pilosa interface { + Union([]uint64) IBitmap + Intersect([]uint64) IBitmap + Get(id uint64) IBitmap } type RequestJSON struct { - Request string - Fragment string - Args json.RawMessage + Request string + Fragment string + Args json.RawMessage } type Fragment struct { - requestChan chan Command - FragmentGuid string - impl Pilosa + requestChan chan Command + FragmentGuid string + impl Pilosa } func (f *Fragment) ServeFragment() { @@ -79,18 +80,19 @@ func (f *Fragment) ServeFragment() { 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"`, req.QueryType())) + buffer.WriteString(fmt.Sprintf(`,"query type": "%s"`, responder.QueryType())) buffer.WriteString(fmt.Sprintf(`, "elapsed": "%s"}`, delta)) - req.ResponseChannel() <- buffer.String() + responder.ResponseChannel() <- buffer.String() } } -func handler(w http.ResponseWriter, r *http.Request,fragments map[string]*Fragment) { +func handler(w http.ResponseWriter, r *http.Request, fragments map[string]*Fragment) { if r.Method == "POST" { var f RequestJSON @@ -106,17 +108,16 @@ func handler(w http.ResponseWriter, r *http.Request,fragments map[string]*Fragme 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