diff --git a/commands/pilosa-loader/loader.go b/commands/pilosa-loader/loader.go index a5b1c799a..68bd873fe 100644 --- a/commands/pilosa-loader/loader.go +++ b/commands/pilosa-loader/loader.go @@ -9,13 +9,22 @@ import ( "net/http" "net/url" "os" + "pilosa/db" + "pilosa/index" + "strconv" "strings" + + "github.com/davecgh/go-spew/spew" ) -func post(database, base_url, id, pid, fragment_type string) { +func post(database, base_url, id, compressed_string, fragment_type string, slice int) { values := make(url.Values) values.Set("db", database) - values.Set("pql", fmt.Sprintf("set(%s,%s,%s)", id, fragment_type, pid)) + values.Set("id", id) + values.Set("frame", fragment_type) + values.Set("slice", fmt.Sprintf("%d", slice)) + values.Set("bitmap", compressed_string) + spew.Dump(values) r, err := http.PostForm(base_url, values) if err != nil { log.Printf("error posting stat to stathat: %s", err) @@ -38,10 +47,31 @@ func Load(database, url, fullpath string, fragment_type string) error { recs := strings.Split(line, "|") id := recs[0] log.Println(id) - for _, profile_id := range recs[1:] { - post(database, url, id, profile_id, fragment_type) + bm := index.NewBitmap() + slice := -1 + last_slice := -1 + for _, pid := range recs[1:] { + profile_id, _ := strconv.ParseUint(strings.TrimSpace(pid), 10, 64) + + if slice < 0 { + slice = db.GetSlice(profile_id) + last_slice = slice + } else { + slice = db.GetSlice(profile_id) + } + if last_slice != slice { + post(database, url, id, bm.ToCompressString(), fragment_type, last_slice) + bm = index.NewBitmap() + last_slice = slice + } + + index.SetBit(bm, profile_id) } + if slice != -1 { + post(database, url, id, bm.ToCompressString(), fragment_type, slice) + } + line, e = Readln(r) } return nil @@ -68,7 +98,7 @@ func Readln(r *bufio.Reader) (string, error) { func main() { flag.Parse() - full_url := fmt.Sprintf("http://%s/query", *host_port) + full_url := fmt.Sprintf("http://%s/bulk", *host_port) //fun(*file) Load(*database, full_url, *file, *fragment) } diff --git a/core/http.go b/core/http.go index 351510c46..f1c90983a 100644 --- a/core/http.go +++ b/core/http.go @@ -34,6 +34,7 @@ func (self *WebService) Run() { mux.HandleFunc("/test", self.HandleTest) mux.HandleFunc("/version", self.HandleVersion) mux.HandleFunc("/ping", self.HandlePing) + mux.HandleFunc("/batch", self.HandleBatch) s := &http.Server{ Addr: ":" + port_string, Handler: mux, @@ -55,6 +56,66 @@ func (self *WebService) HandleMessage(w http.ResponseWriter, r *http.Request) { //service.Inbox <- &message } +func (self *WebService) HandleBatch(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" { + http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed) + return + } + /* values.Set("db", database) + values.Set("id", id) + values.Set("frame", fragment_type) + values.Set("slice", fmt.Sprintf("%d", slice)) + values.Set("bitmap", compressed_string) + */ + + err := r.ParseForm() + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + database_name := r.Form.Get("db") + if database_name == "" { + http.Error(w, "Provide a database (db)", http.StatusNotFound) + return + } + + bitmap_id_string := r.Form.Get("id") + if bitmap_id_string == "" { + http.Error(w, "Provide a bitmap id (id)", http.StatusNotFound) + return + } + bitmap_id, _ := strconv.ParseUint(bitmap_id_string, 10, 64) + + slice_string := r.Form.Get("slice") + if bitmap_id_string == "" { + http.Error(w, "Provide a slice (slice)", http.StatusNotFound) + return + } + slice, _ := strconv.ParseInt(slice_string, 10, 32) + + frame := r.Form.Get("frame") + if bitmap_id_string == "" { + http.Error(w, "Provide a frame (frame)", http.StatusNotFound) + return + } + + compressed_bitmap := r.Form.Get("bitmap") + if bitmap_id_string == "" { + http.Error(w, "Provide a compressed base64 bitmap (bitmap)", http.StatusNotFound) + return + } + + results := self.service.Batch(database_name, frame, compressed_bitmap, bitmap_id, int(slice)) + + encoder := json.NewEncoder(w) + err = encoder.Encode(results) + if err != nil { + log.Fatal("Error encoding results") + } + +} + func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) { if r.Method != "POST" { http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed) diff --git a/db/topology.go b/db/topology.go index 32ed4ab86..6c33c83fb 100644 --- a/db/topology.go +++ b/db/topology.go @@ -390,13 +390,16 @@ func (d *Database) GetOrCreateFragment(frame *Frame, slice *Slice, fragment_id u func (f *Fragment) SetProcess(process *Process) { f.process = process } +func GetSlice(profile_id uint64) int { + return int(profile_id / SLICE_WIDTH) +} /////////////////////////////////////////////////////////////////////////////////////////////// // Get a slice from a database + func (d *Database) GetSliceForProfile(profile_id uint64) (*Slice, error) { - slice_id := profile_id / SLICE_WIDTH - return d.getSlice(int(slice_id)) + return d.getSlice(GetSlice(profile_id)) } type Bitmap struct { diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index 7a3facd27..f0b47ef9a 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -27,6 +27,10 @@ func (self *Dispatch) Run() { for { message := self.service.Transport.Receive() switch data := message.Data.(type) { + case core.BatchRequest: + response := db.Message{Data: core.BatchResponse{Id: data.Id}} + self.service.Index.LoadBitmap(data.Fragment_id, data.Bitmap_id, data.Compressed_bitmap) + self.service.Transport.Send(&response, data.Source) case core.PingRequest: pong := db.Message{Data: core.PongRequest{Id: data.Id}} self.service.Transport.Send(&pong, data.Source) diff --git a/index/bitmap.go b/index/bitmap.go index 74381e001..563144c47 100644 --- a/index/bitmap.go +++ b/index/bitmap.go @@ -4,7 +4,10 @@ package index import ( "bytes" + "compress/gzip" + "encoding/base64" "encoding/gob" + "io/ioutil" "log" "github.com/yasushi-saito/rbtree" @@ -354,6 +357,8 @@ type IBitmap interface { BuildFromBits(bits []uint64) ToBytes() []byte FromBytes([]byte) + ToCompressString() string + FromCompressString(string) } func NewRB() *rbtree.Tree { @@ -363,8 +368,29 @@ func NewRB() *rbtree.Tree { func CreateRBBitmap() IBitmap { return &Bitmap{nodes: NewRB(), bcount: 0} } -func (b *Bitmap) AddChunk(a *Chunk) { - b.nodes.Insert(a) + +func (self *Bitmap) FromCompressString(str string) { + compressed_data, err := base64.StdEncoding.DecodeString(str) + if err != nil { + log.Println(err) + return + } + reader, _ := gzip.NewReader(bytes.NewReader(compressed_data)) + data, _ := ioutil.ReadAll(reader) + self.FromBytes(data) +} + +func (self *Bitmap) ToCompressString() string { + var b bytes.Buffer + w := gzip.NewWriter(&b) + w.Write(self.ToBytes()) + w.Flush() + w.Close() + return base64.StdEncoding.EncodeToString(b.Bytes()) +} + +func (self *Bitmap) AddChunk(a *Chunk) { + self.nodes.Insert(a) } func (b *Bitmap) Min() ChunkIterator { return &RBNodeIterator{b.nodes.Min()} diff --git a/index/brand.go b/index/brand.go index dd78e5ee0..7211bcdae 100644 --- a/index/brand.go +++ b/index/brand.go @@ -125,6 +125,12 @@ func packagePairs(r RankList) []Pair { } return res } +func (self *Brand) Store(bitmap_id uint64, bm IBitmap) { + //oldbm:=self.Get(bitmap_id) + //nbm = Union(oldbm, bm) + self.storage.Store(int64(bitmap_id), self.db, self.slice, bm.(*Bitmap)) + self.cache_it(bm, bitmap_id) +} func (self *Brand) TopN(src_bitmap IBitmap, n int) []Pair { breakout := 500 diff --git a/index/commands.go b/index/commands.go index 86312a37f..4a2889186 100644 --- a/index/commands.go +++ b/index/commands.go @@ -202,3 +202,19 @@ func NewClear() *CmdClear { func (self *CmdClear) Execute(f *Fragment) Calculation { return f.impl.Clear() } + +type CmdLoader struct { + *Responder + bitmap_id uint64 + compressed_bitmap string +} + +func NewLoader(bitmap_id uint64, compressed_bitmap string) *CmdLoader { + return &CmdLoader{NewResponder("Loader"), bitmap_id, compressed_bitmap} +} +func (self *CmdLoader) Execute(f *Fragment) Calculation { + nbm := NewBitmap() + nbm.FromCompressString(self.compressed_bitmap) + f.impl.Store(self.bitmap_id, nbm) + return "ok" +} diff --git a/index/fragment_container.go b/index/fragment_container.go index 3f4fee0d4..5d09b4714 100644 --- a/index/fragment_container.go +++ b/index/fragment_container.go @@ -25,6 +25,13 @@ func init() { gob.Register(vh) } +func (self *FragmentContainer) LoadBitmap(frag_id SUUID, bitmap_id uint64, compressed_bitmap string) { + if fragment, found := self.GetFragment(frag_id); found { + loader := NewLoader(bitmap_id, compressed_bitmap) + fragment.requestChan <- loader + } +} + func (self *FragmentContainer) GetFragment(frag_id SUUID) (*Fragment, bool) { //lock c, v := self.fragments[frag_id] @@ -144,6 +151,7 @@ type Pilosa interface { SetBit(id uint64, bit_pos uint64) bool TopN(b IBitmap, n int) []Pair Clear() bool + Store(bitmap_id uint64, bm IBitmap) } type Fragment struct { diff --git a/index/general.go b/index/general.go index 88156a97a..bdc8bc0a6 100644 --- a/index/general.go +++ b/index/general.go @@ -48,3 +48,9 @@ func (self *General) TopN(b IBitmap, n int) []Pair { return nil } +func (self *General) Store(bitmap_id uint64, bm IBitmap) { + //oldbm:=self.Get(bitmap_id) + //nbm = Union(oldbm, bm) + self.storage.Store(int64(bitmap_id), self.db, self.slice, bm.(*Bitmap)) + self.bitmap_cache.Add(bitmap_id, bm) +} diff --git a/util/util_test.go b/util/util_test.go index 246ac0116..12e1e496d 100644 --- a/util/util_test.go +++ b/util/util_test.go @@ -46,6 +46,13 @@ func TestId(t *testing.T) { b2 := Hex_to_SUUID(s) So(b1, ShouldEqual, b2) }) + Convey("Gen 10", t, func() { + for i := 0; i < 10; i++ { + bc1 := Id() + println(SUUID_to_Hex(bc1)) + } + So(1, ShouldEqual, 1) + }) }