mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-09 14:41:02 +00:00
refactored command interface; fmt code
This commit is contained in:
Todd Gruben
parent
648e44ca72
commit
ccdcac5ebb
8 changed files with 393 additions and 407 deletions
|
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<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)
|
||||
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<uint64(4096);i+=2{
|
||||
SetBit(even,i)
|
||||
}
|
||||
Convey("UNION even + odd equal 4096 ", t, func() {
|
||||
even := CreateRBBitmap()
|
||||
for i := uint64(0); i < uint64(4096); i += 2 {
|
||||
SetBit(even, i)
|
||||
}
|
||||
|
||||
odd:= CreateRBBitmap()
|
||||
for i:=uint64(1); i<uint64(4096);i+=2{
|
||||
SetBit(odd,i)
|
||||
}
|
||||
all:= Union(even,odd)
|
||||
total_bits:= BitCount(all)
|
||||
odd := CreateRBBitmap()
|
||||
for i := uint64(1); i < uint64(4096); i += 2 {
|
||||
SetBit(odd, i)
|
||||
}
|
||||
all := Union(even, odd)
|
||||
total_bits := BitCount(all)
|
||||
|
||||
So(total_bits, ShouldEqual, 4096)
|
||||
})
|
||||
So(total_bits, ShouldEqual, 4096)
|
||||
})
|
||||
|
||||
Convey("Intersection even - odd equal 0 ", t, func() {
|
||||
even:= CreateRBBitmap()
|
||||
for i:=uint64(0); i<uint64(4096);i+=2{
|
||||
SetBit(even,i)
|
||||
}
|
||||
Convey("Intersection even - odd equal 0 ", t, func() {
|
||||
even := CreateRBBitmap()
|
||||
for i := uint64(0); i < uint64(4096); i += 2 {
|
||||
SetBit(even, i)
|
||||
}
|
||||
|
||||
odd:= CreateRBBitmap()
|
||||
for i:=uint64(1); i<uint64(4096);i+=2{
|
||||
SetBit(odd,i)
|
||||
}
|
||||
all:= Intersection(even,odd)
|
||||
total_bits:= BitCount(all)
|
||||
odd := CreateRBBitmap()
|
||||
for i := uint64(1); i < uint64(4096); i += 2 {
|
||||
SetBit(odd, i)
|
||||
}
|
||||
all := Intersection(even, odd)
|
||||
total_bits := BitCount(all)
|
||||
|
||||
So(total_bits, ShouldEqual, 0)
|
||||
So(total_bits, ShouldEqual, 0)
|
||||
|
||||
})
|
||||
})
|
||||
|
||||
Convey("Bitcount< 1s ", t, func() {
|
||||
all:= CreateRBBitmap()
|
||||
for i:=uint64(0); i< uint64(65536);i++{
|
||||
SetBit(all,i)
|
||||
}
|
||||
start:= time.Now()
|
||||
BitCount(all)
|
||||
So(start, ShouldHappenWithin, time.Duration(1) * time.Millisecond, time.Now())
|
||||
})
|
||||
Convey("Bitcount< 1s ", t, func() {
|
||||
all := CreateRBBitmap()
|
||||
for i := uint64(0); i < uint64(65536); i++ {
|
||||
SetBit(all, i)
|
||||
}
|
||||
start := time.Now()
|
||||
BitCount(all)
|
||||
So(start, ShouldHappenWithin, time.Duration(1)*time.Millisecond, time.Now())
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ package index
|
|||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
type Rank struct {
|
||||
|
|
@ -15,143 +15,130 @@ 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 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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<len(fragments){
|
||||
if found{
|
||||
// fc := fragments[f.FragmentGuid]
|
||||
fc.requestChan <- request
|
||||
output = request.Response()
|
||||
}
|
||||
output := `{"Error":"Invalid Fragment"}`
|
||||
fc, found := fragments[f.Fragment] //f.FragmentIndex<len(fragments){
|
||||
if found {
|
||||
// fc := fragments[f.FragmentGuid]
|
||||
fc.requestChan <- request
|
||||
output = request.GetResponder().Response()
|
||||
}
|
||||
fmt.Fprintf(w, output)
|
||||
} else {
|
||||
fmt.Fprintf(w, "NoOp")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,93 +1,92 @@
|
|||
package index
|
||||
|
||||
import (
|
||||
"testing"
|
||||
. "github.com/smartystreets/goconvey/convey"
|
||||
"net/http"
|
||||
"encoding/json"
|
||||
"net/http/httptest"
|
||||
// "io/ioutil"
|
||||
// "time"
|
||||
"log"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
func simple(id1,id2 int)string {
|
||||
return fmt.Sprintf(`{
|
||||
. "github.com/smartystreets/goconvey/convey"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
// "io/ioutil"
|
||||
// "time"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
)
|
||||
|
||||
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(`{
|
||||
}`, 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)
|
||||
})
|
||||
|
||||
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("Union", t, func() {
|
||||
c,s=sendRequest(simple(1,2),dummy)
|
||||
So(c, ShouldEqual, 200)
|
||||
v:=getResult("value",s)
|
||||
So(v, ShouldEqual, 2)
|
||||
})
|
||||
}`, 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)
|
||||
})
|
||||
|
||||
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("Union", t, func() {
|
||||
c, s = sendRequest(simple(1, 2), dummy)
|
||||
So(c, ShouldEqual, 200)
|
||||
v := getResult("value", s)
|
||||
So(v, ShouldEqual, 2)
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,19 +7,20 @@ import (
|
|||
"tux21b.org/v1/gocql"
|
||||
)
|
||||
|
||||
type CassandraStorage struct{
|
||||
db *gocql.Session
|
||||
type CassandraStorage struct {
|
||||
db *gocql.Session
|
||||
}
|
||||
func BuildSchema(){
|
||||
/*
|
||||
"CREATE KEYSPACE IF NOT EXISTS hotbox WITH strategy_class = SimpleStrategy AND strategy_options:replication_factor = 1"
|
||||
"CREATE TABLE IF NOT EXISTS bitmap ( bitmap_id bigint, db varchar, slice int, ChunkKey bigint, BlockIndex int, block bigint, PRIMARY KEY ((bitmap_id, db, slice),ChunkKey,BlockIndex) )
|
||||
"
|
||||
*/
|
||||
|
||||
|
||||
func BuildSchema() {
|
||||
/*
|
||||
"CREATE KEYSPACE IF NOT EXISTS hotbox WITH strategy_class = SimpleStrategy AND strategy_options:replication_factor = 1"
|
||||
"CREATE TABLE IF NOT EXISTS bitmap ( bitmap_id bigint, db varchar, slice int, ChunkKey bigint, BlockIndex int, block bigint, PRIMARY KEY ((bitmap_id, db, slice),ChunkKey,BlockIndex) )
|
||||
"
|
||||
*/
|
||||
|
||||
}
|
||||
func NewCassStorage() Storage{
|
||||
obj := new(CassandraStorage)
|
||||
func NewCassStorage() Storage {
|
||||
obj := new(CassandraStorage)
|
||||
cluster := gocql.NewCluster("127.0.0.1")
|
||||
cluster.Keyspace = "hotbox"
|
||||
cluster.Consistency = gocql.Quorum
|
||||
|
|
@ -30,14 +31,14 @@ func NewCassStorage() Storage{
|
|||
log.Fatal(err)
|
||||
}
|
||||
|
||||
err = session.Query("USE hotbox").Exec()
|
||||
if err != nil {
|
||||
err = session.Query("USE hotbox").Exec()
|
||||
if err != nil {
|
||||
}
|
||||
obj.db = session
|
||||
return obj
|
||||
obj.db = session
|
||||
return obj
|
||||
}
|
||||
|
||||
func (c *CassandraStorage)Fetch( bitmap_id uint64, db string, slice int) IBitmap {
|
||||
func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, slice int) IBitmap {
|
||||
var dumb = COUNTERMASK
|
||||
last_key := int64(dumb)
|
||||
marker := int64(dumb)
|
||||
|
|
@ -49,7 +50,7 @@ func (c *CassandraStorage)Fetch( bitmap_id uint64, db string, slice int) IBitmap
|
|||
block_index uint32
|
||||
s8 uint8
|
||||
)
|
||||
log.Println("FETCHING ", bitmap_id, db,slice)
|
||||
log.Println("FETCHING ", bitmap_id, db, slice)
|
||||
|
||||
bitmap := CreateRBBitmap()
|
||||
iter := c.db.Query("SELECT Chunkkey,BlockIndex,block FROM bitmap WHERE bitmap_id=? AND db=? AND slice=? ", id, db, slice).Iter()
|
||||
|
|
@ -73,14 +74,14 @@ func (c *CassandraStorage)Fetch( bitmap_id uint64, db string, slice int) IBitmap
|
|||
return bitmap
|
||||
}
|
||||
|
||||
func (c *CassandraStorage) Store( id int64, db string, slice int, bitmap *Bitmap) error {
|
||||
func (c *CassandraStorage) Store(id int64, db string, slice int, bitmap *Bitmap) error {
|
||||
for i := bitmap.Min(); !i.Limit(); i = i.Next() {
|
||||
var chunk = i.Item()
|
||||
for idx, block := range chunk.Value.Block {
|
||||
block_index := int32(idx)
|
||||
iblock := int64(block)
|
||||
if iblock != 0 {
|
||||
c.StoreBlock(id, db,slice, int64(chunk.Key), block_index, iblock)
|
||||
c.StoreBlock(id, db, slice, int64(chunk.Key), block_index, iblock)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -89,13 +90,13 @@ func (c *CassandraStorage) Store( id int64, db string, slice int, bitmap *Bitmap
|
|||
var dumb = COUNTERMASK
|
||||
COUNTER_KEY := int64(dumb)
|
||||
|
||||
c.StoreBlock(id, db,slice, COUNTER_KEY, 0, cnt)
|
||||
c.StoreBlock(id, db, slice, COUNTER_KEY, 0, cnt)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *CassandraStorage)StoreBlock(id int64, db string, slice int, chunk int64, block_index int32, block int64) error {
|
||||
func (c *CassandraStorage) StoreBlock(id int64, db string, slice int, chunk int64, block_index int32, block int64) error {
|
||||
|
||||
if err := c.db.Query(`INSERT INTO bitmap (bitmap_id, db, slice , ChunkKey, BlockIndex,block) VALUES (?,?,?, ?,?,?);`, id, db, slice , chunk, block_index, block).Exec(); err != nil {
|
||||
if err := c.db.Query(`INSERT INTO bitmap (bitmap_id, db, slice , ChunkKey, BlockIndex,block) VALUES (?,?,?, ?,?,?);`, id, db, slice, chunk, block_index, block).Exec(); err != nil {
|
||||
log.Println(err)
|
||||
log.Println("INSERT ", id, chunk, block_index)
|
||||
return err
|
||||
|
|
|
|||
|
|
@ -3,45 +3,45 @@ package index
|
|||
// #cgo CFLAGS:-mpopcnt
|
||||
|
||||
import (
|
||||
// "log"
|
||||
"fmt"
|
||||
// "log"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
type MemoryStorage struct{
|
||||
db map[string]*Bitmap
|
||||
|
||||
}
|
||||
func NewMemoryStorage() Storage{
|
||||
obj := new(MemoryStorage)
|
||||
obj.db = make(map[string]*Bitmap)
|
||||
|
||||
return obj
|
||||
type MemoryStorage struct {
|
||||
db map[string]*Bitmap
|
||||
}
|
||||
|
||||
func (c *MemoryStorage)Fetch( bitmap_id uint64, db string, slice int) IBitmap {
|
||||
key := fmt.Sprintf("%d:%s:%d",bitmap_id,db,slice)
|
||||
bitmap,found := c.db[key]
|
||||
if !found{
|
||||
bitmap = CreateRBBitmap().(*Bitmap)
|
||||
c.db[key]=bitmap
|
||||
}
|
||||
func NewMemoryStorage() Storage {
|
||||
obj := new(MemoryStorage)
|
||||
obj.db = make(map[string]*Bitmap)
|
||||
|
||||
return obj
|
||||
}
|
||||
|
||||
func (c *MemoryStorage) Fetch(bitmap_id uint64, db string, slice int) IBitmap {
|
||||
key := fmt.Sprintf("%d:%s:%d", bitmap_id, db, slice)
|
||||
bitmap, found := c.db[key]
|
||||
if !found {
|
||||
bitmap = CreateRBBitmap().(*Bitmap)
|
||||
c.db[key] = bitmap
|
||||
}
|
||||
return bitmap
|
||||
}
|
||||
|
||||
func (c *MemoryStorage) Store( bitmap_id int64, db string, slice int, bitmap *Bitmap) error {
|
||||
key := fmt.Sprintf("%d:%s:%d",bitmap_id,db,slice)
|
||||
c.db[key]= bitmap
|
||||
func (c *MemoryStorage) Store(bitmap_id int64, db string, slice int, bitmap *Bitmap) error {
|
||||
key := fmt.Sprintf("%d:%s:%d", bitmap_id, db, slice)
|
||||
c.db[key] = bitmap
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *MemoryStorage)StoreBlock(bitmap_id int64, db string, slice int, chunk_key int64, block_index int32, block int64) error {
|
||||
bm := c.Fetch(uint64(bitmap_id),db,slice)
|
||||
node := GetChunk(bm,uint64(chunk_key))
|
||||
if node == nil{
|
||||
node = &Chunk{uint64(chunk_key), BlockArray{}}
|
||||
bm.AddChunk(node)
|
||||
}
|
||||
node.Value.Block[block_index]=uint64(block)
|
||||
func (c *MemoryStorage) StoreBlock(bitmap_id int64, db string, slice int, chunk_key int64, block_index int32, block int64) error {
|
||||
bm := c.Fetch(uint64(bitmap_id), db, slice)
|
||||
node := GetChunk(bm, uint64(chunk_key))
|
||||
if node == nil {
|
||||
node = &Chunk{uint64(chunk_key), BlockArray{}}
|
||||
bm.AddChunk(node)
|
||||
}
|
||||
node.Value.Block[block_index] = uint64(block)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue