diff --git a/commands/pilosa-cruncher/cruncher.go b/commands/pilosa-cruncher/cruncher.go index a94dd842f..0efe20441 100644 --- a/commands/pilosa-cruncher/cruncher.go +++ b/commands/pilosa-cruncher/cruncher.go @@ -1,7 +1,7 @@ package main import ( - "pilosa/cruncher" + "pilosa/core" "pilosa/db" "flag" "log" @@ -26,6 +26,6 @@ func main() { log.Fatal("Location not valid:", httpLoc) } - cruncher := cruncher.NewCruncher(tcp, http) - cruncher.Run() + service := core.NewService(tcp, http) + service.Run() } diff --git a/core/cruncher.go b/core/cruncher.go new file mode 100644 index 000000000..1e2af8684 --- /dev/null +++ b/core/cruncher.go @@ -0,0 +1,13 @@ +package core + +import ( + "github.com/davecgh/go-spew/spew" +) + + +type Cruncher struct { +} + +func (cruncher *Cruncher) Run() { + spew.Dump("Cruncher.Run") +} diff --git a/core/cruncher_test.go b/core/cruncher_test.go new file mode 100644 index 000000000..9d0b42df3 --- /dev/null +++ b/core/cruncher_test.go @@ -0,0 +1,30 @@ +package core + +import ( + "testing" + //"github.com/nu7hatch/gouuid" + "github.com/davecgh/go-spew/spew" + . "github.com/smartystreets/goconvey/convey" +) + +func TestCruncher(t *testing.T) { + Convey("Basic Cruncher Tests", t, func() { + spew.Dump("cruncher test") + + /* + cluster := NewCluster() + database := cluster.GetOrCreateDatabase("main") + + frame := database.GetOrCreateFrame("general") + slice := database.GetOrCreateSlice(0) + + fragment_id, _ := uuid.ParseHex("6a9aea17-2915-4eb4-858f-a8d7d4dc0a1e") + spew.Dump(fragment_id) + database.GetOrCreateFragment(frame, slice, fragment_id) + + spew.Dump(database) + spew.Dump("DONE") + */ + + }) +} diff --git a/core/etcd.go b/core/etcd.go index bb19b91d4..054ffe96a 100644 --- a/core/etcd.go +++ b/core/etcd.go @@ -1,88 +1,122 @@ package core import ( + "github.com/davecgh/go-spew/spew" "github.com/coreos/go-etcd/etcd" - "log" - "strings" - "time" "encoding/gob" "pilosa/db" + "log" + "strings" + "github.com/nu7hatch/gouuid" + "strconv" + "errors" ) func (service *Service) SetupEtcd() { gob.Register(db.Location{}) service.Etcd = etcd.NewClient(nil) - service.NodeMapMutex.Lock() - defer service.NodeMapMutex.Unlock() - service.NodeMap = db.NodeMap{} +} - nodes, err := service.Etcd.Get("nodes", false) +func flatten(node *etcd.Node) []*etcd.Node { + nodes := make([]*etcd.Node, 0) + nodes = append(nodes, node) + for _, node := range node.Nodes { + nodes = append(nodes, flatten(&node)...) + } + return nodes +} + +func handlenode(node *etcd.Node, namespace string, cluster *db.Cluster) error { + key := node.Key[len(namespace)+1:] + bits := strings.Split(key, "/") + var database *db.Database + var frame *db.Frame + var fragment *db.Fragment + var fragment_uuid *uuid.UUID + var slice *db.Slice + var process_uuid *uuid.UUID + var process *db.Process + var err error + + if len(bits) <= 1 || bits[0] != "db" { + return nil + } + if len(bits) > 1 { + database = cluster.GetOrCreateDatabase(bits[1]) + } + if len(bits) > 2 { + if bits[2] != "frame" { + return errors.New("no frame") + } + } + if len(bits) > 3 { + frame = database.GetOrCreateFrame(bits[3]) + } + if len(bits) > 4 { + if bits[4] != "slice" { + return errors.New("no slice") + } + } + if len(bits) > 5 { + slice_int, err := strconv.Atoi(bits[5]) + if err != nil { + return err + } + slice = database.GetOrCreateSlice(slice_int) + } + if len(bits) > 6 { + if bits[6] != "fragment" { + return errors.New("no fragment") + } + } + if len(bits) > 7 { + fragment_uuid, err = uuid.ParseHex(bits[7]) + if err != nil { + return err + } + fragment = database.GetOrCreateFragment(frame, slice, fragment_uuid) + } + + if len(bits) > 8 { + if bits[8] != "process" { + return errors.New("no process") + } + process_uuid, err = uuid.ParseHex(node.Value) + if err != nil { + return err + } + process = db.NewProcess(process_uuid) + fragment.SetProcess(process) + } + return err +} + +func (service *Service) MetaWatcher() { + namespace := "/pilosa/0" + log.Println(namespace + "/db") + cluster := db.NewCluster() + resp, err := service.Etcd.Get(namespace + "/db", false, true) if err != nil { log.Fatal(err) } - for _, node := range nodes.Kvs { - nodestring := strings.Split(node.Key, "/")[2] - location, err := db.NewLocation(nodestring) + for _, node := range flatten(resp.Node) { + err := handlenode(node, namespace, cluster) if err != nil { - log.Fatal(err) + spew.Dump(node) + log.Println(err) } - routerlocation, err := db.NewLocation(node.Value) - if err != nil { - log.Fatal(err) - } - service.NodeMap[*location] = *routerlocation } - log.Println(service.NodeMap) -} - -func (service *Service) WatchEtcd() { - var receiver = make(chan *etcd.Response) - var stop chan bool - go func () { - _, err := service.Etcd.Watch("nodes/", 0, receiver, stop) - if err != nil { - log.Fatal(err) + receiver := make(chan *etcd.Response) + stop := make(chan bool) + go func() { + _, _ = service.Etcd.Watch(namespace + "/db", 0, true, receiver, stop) + }() + go func() { + for resp = range receiver { + switch resp.Action { + case "set": + handlenode(resp.Node, namespace, cluster) + } } }() - - exit, done := service.GetExitChannels() - - for { - select { - case response := <-receiver: - switch response.Action { - case "SET": - nodestring := strings.Split(response.Key, "/")[2] - node, err := db.NewLocation(nodestring) - if err != nil { - log.Fatal(err) - } - router, err := db.NewLocation(response.Value) - if err != nil { - log.Fatal(err) - } - service.NodeMapMutex.Lock() - service.NodeMap[*node] = *router - service.NodeMapMutex.Unlock() - case "DELETE": - nodestring := strings.Split(response.Key, "/")[2] - node, err := db.NewLocation(nodestring) - if err != nil { - log.Fatal(err) - } - service.NodeMapMutex.Lock() - delete(service.NodeMap, *node) - service.NodeMapMutex.Unlock() - default: - log.Println("unhandled etcd message", response) - } - //log.Println(response.Action, response.Key, response.Value) - log.Println(service.NodeMap) - case <-exit: - log.Println("cleaning up watchetcd service thing.") - time.Sleep(time.Second/2) - log.Println("done!") - done <- 1 - } - } } diff --git a/core/service.go b/core/service.go index c8914255e..90b62b3de 100644 --- a/core/service.go +++ b/core/service.go @@ -50,6 +50,7 @@ type Service struct { ConnectionRegisterChannel chan *PersistentConnection Stats *Stats //Cluster query.Cluster + Cruncher *Cruncher } func NewService(tcp, http *db.Location) *Service { @@ -59,6 +60,7 @@ func NewService(tcp, http *db.Location) *Service { service.Outbox = make(chan *db.Envelope) service.Inbox = make(chan *db.Message) service.Stats = new(Stats) + service.Cruncher = new(Cruncher) return service } @@ -290,3 +292,43 @@ func (service *Service) NewListener() chan *db.Message { ch := make(chan *db.Message) return ch } + + +//////////////////////////////////////////////// + + +func (service *Service) Run() { + log.Println("Running service...") + service.SetupEtcd() + //go r.SyncEtcd() + //go service.WatchEtcd() + //go service.HandleConnections() + //service.SetupNetwork() + //go service.Serve() + //go service.HandleInbox() + //go service.ServeHTTP() + go service.MetaWatcher() + go service.Cruncher.Run() + + sigterm, sighup := service.GetSignals() + for { + select { + case <- sighup: + log.Println("SIGHUP! Reloading configuration...") + // TODO: reload configuration + case <- sigterm: + log.Println("SIGTERM! Cleaning up...") + service.Stop() + return + } + } +} + +func (service *Service) HandleInbox() { + for { + select { + case message := <-service.Inbox: + log.Println("process", message) + } + } +} diff --git a/cruncher/cruncher.go b/cruncher/cruncher.go deleted file mode 100644 index 309a22a48..000000000 --- a/cruncher/cruncher.go +++ /dev/null @@ -1,55 +0,0 @@ -package cruncher - -import ( - "pilosa/core" - "pilosa/db" - "log" -) - -type Cruncher struct { - core.Service -} - -func (c *Cruncher) Init() { - log.Println("Initializing cruncher...") -} - -func (c *Cruncher) Run() { - log.Println("Running cruncher...") - c.SetupEtcd() - //go r.SyncEtcd() - go c.WatchEtcd() - go c.HandleConnections() - c.SetupNetwork() - go c.Serve() - go c.HandleInbox() - go c.ServeHTTP() - - sigterm, sighup := c.GetSignals() - for { - select { - case <- sighup: - log.Println("SIGHUP! Reloading configuration...") - // TODO: reload configuration - case <- sigterm: - log.Println("SIGTERM! Cleaning up...") - c.Stop() - return - } - } -} - -func NewCruncher(tcp, http *db.Location) *Cruncher { - service := core.NewService(tcp, http) - cruncher := Cruncher{*service} - return &cruncher -} - -func (c *Cruncher) HandleInbox() { - for { - select { - case message := <-c.Inbox: - log.Println("process", message) - } - } -} diff --git a/db/constants.go b/db/constants.go new file mode 100644 index 000000000..d2fbdd0eb --- /dev/null +++ b/db/constants.go @@ -0,0 +1,3 @@ +package db + +const SLICE_WIDTH = 65536 diff --git a/db/topology.go b/db/topology.go index 64eba14bc..b4332bc7d 100644 --- a/db/topology.go +++ b/db/topology.go @@ -2,20 +2,34 @@ package db import ( "github.com/stathat/consistent" + //"github.com/davecgh/go-spew/spew" + "github.com/nu7hatch/gouuid" "log" "fmt" "errors" "strings" "strconv" + "sync" ) var FrameDoesNotExistError = errors.New("Frame does not exist.") +var SliceDoesNotExistError = errors.New("Slice does not exist.") +var FragmentDoesNotExistError = errors.New("Fragment does not exist.") +var FrameSliceIntersectDoesNotExistError = errors.New("FrameSliceIntersect does not exist.") type Location struct { Ip string Port int } +type Process struct { + id *uuid.UUID +} + +func NewProcess(id *uuid.UUID) *Process { + return &Process{id} +} + // Create a Location struct given a string in form "0.0.0.0:0" func NewLocation(location_string string) (*Location, error) { splitstring := strings.Split(location_string, ":") @@ -37,98 +51,298 @@ func (location *Location) ToString() string { // Map of node location to their router type NodeMap map[Location]Location -// A fragment is a collection of bitmaps within a slice. The fragment contains a reference to the responsible node for that fragment. The node is in the form ip:port -type Fragment struct { - Node string -} -// A slice is the vertical combination of every fragment. It contains the hashring used to delegate bitmaps to fragments -type Slice struct { - Fragments []Fragment - Hashring *consistent.Consistent -} -// A frame is a collection of slices in a given category (brands, demographics, etc), specific to a database -type Frame struct { - Name string - Slices []*Slice -} -// Add a slice to a frame with given Node addresses -func (f *Frame) AddSlice(addrs ...string) *Slice { - slice := Slice{} - slice.Hashring = consistent.New() - slice.Hashring.NumberOfReplicas = 200 - sliceIndex := len(f.Slices) - for index, addr := range addrs { - slice.Fragments = append(slice.Fragments, Fragment{addr}) - slice.Hashring.Add(fmt.Sprintf("%d %d", sliceIndex, index)) - } - f.Slices = append(f.Slices, &slice) - return &slice -} +/////////// CLUSTERS //////////////////////////////////////////////////////////////////// // Represents the entire cluster, and a reference to the Node this instance is running on type Cluster struct { - Databases map[string]*Database - Self string + databases map[string]*Database + mutex sync.Mutex } -// Add a database to a cluster -func (c *Cluster) AddDatabase(name string) *Database { - database := Database{Name: name} - if c.Databases == nil { - c.Databases = make(map[string]*Database) - } - c.Databases[name] = &database - return &database +func NewCluster() *Cluster { + cluster := Cluster{} + cluster.databases = make(map[string]*Database) + return &cluster } + +/////////// DATABASES //////////////////////////////////////////////////////////////////// + // A database is a collection of all the frames within a given profile space type Database struct { Name string - Frames []*Frame + frames []*Frame + slices []*Slice + frame_slice_intersects []*FrameSliceIntersect + mutex sync.Mutex +} + +// Add a database to a cluster +func (c *Cluster) addDatabase(name string) *Database { + database := Database{Name: name} + if c.databases == nil { + c.databases = make(map[string]*Database) + } + c.databases[name] = &database + return &database +} + +func (c *Cluster) getDatabase(name string) (*Database, error) { + value, ok := c.databases[name] + if !ok { + return nil, errors.New("The database does not exist!") + } else { + return value, nil + } +} + +func (c *Cluster) GetOrCreateDatabase(name string) *Database { + c.mutex.Lock() + defer c.mutex.Unlock() + database, err := c.getDatabase(name) + if err == nil { + return database + } + return c.addDatabase(name) } // Count the number of slices in a database func (d *Database) NumSlices() (int, error) { - if len(d.Frames) < 1 { + if len(d.slices) < 1 { return 0, errors.New("Database is empty") } - return len(d.Frames[0].Slices), nil + return len(d.slices), nil } -// Add a frame to a database -func (d *Database) AddFrame(name string) *Frame { - frame := Frame{Name: name} - d.Frames = append(d.Frames, &frame) - return &frame +///////// FRAMES //////////////////////////////////////////////////////////////////// + +// A frame is a collection of slices in a given category +// (brands, demographics, etc), specific to a database +type Frame struct { + name string } // Get a frame from a database -func (d *Database) GetFrame(name string) (*Frame, error) { - for _, frame := range d.Frames { - if frame.Name == name { +func (d *Database) getFrame(name string) (*Frame, error) { + for _, frame := range d.frames { + if frame.name == name { return frame, nil } } return nil, FrameDoesNotExistError } -// For debugging, prints cluster information -func (c *Cluster) Describe() { - for _, database := range c.Databases { - log.Println("frames", database.Frames) - for _, frame := range database.Frames { - log.Println(frame.Name, database.Name) - for _, slice := range frame.Slices { - log.Println(" ", slice) - } - } - } +// Add a frame to a database +func (d *Database) addFrame(name string) *Frame { + frame := Frame{name: name} + d.frames = append(d.frames, &frame) + // add intersections + for _, slice := range d.slices { + d.AddFrameSliceIntersect(&frame, slice) + } + return &frame } -type Bitmap struct { - FrameType string - Id int +func (d *Database) GetOrCreateFrame(name string) *Frame { + d.mutex.Lock() + defer d.mutex.Unlock() + frame, err := d.getFrame(name) + if err == nil { + return frame + } + return d.addFrame(name) +} + + +///////// SLICES ///////////////////////////////////////////////////////////////////////// + +// A slice is the vertical combination of every fragment. +type Slice struct { + id int +} + +// Get a slice from a database +func (d *Database) getSlice(slice_id int) (*Slice, error) { + for _, slice := range d.slices { + if slice.id == slice_id { + return slice, nil + } + } + return nil, SliceDoesNotExistError +} + +// Add a slice to a database +func (d *Database) addSlice(slice_id int) *Slice { + slice := Slice{id: slice_id} + d.slices = append(d.slices, &slice) + // add intersections + for _, frame := range d.frames { + d.AddFrameSliceIntersect(frame, &slice) + } + return &slice +} + +func (d *Database) GetOrCreateSlice(slice_id int) *Slice { + d.mutex.Lock() + defer d.mutex.Unlock() + slice, err := d.getSlice(slice_id) + if err == nil { + return slice + } + return d.addSlice(slice_id) +} + + +///////// FRAME-SLICE INTERSECT ////////////////////////////////////////////////////////////// + +type FrameSliceIntersect struct { + frame *Frame + slice *Slice + fragments []*Fragment + hashring *consistent.Consistent +} + +func (d *Database) AddFrameSliceIntersect(frame *Frame, slice *Slice) *FrameSliceIntersect { + frameslice := FrameSliceIntersect{frame: frame, slice: slice} + d.frame_slice_intersects = append(d.frame_slice_intersects, &frameslice) + frameslice.hashring = consistent.New() + frameslice.hashring.NumberOfReplicas = 16 + return &frameslice +} + +func (d *Database) GetFrameSliceIntersect(frame *Frame, slice *Slice) (*FrameSliceIntersect, error) { + for _, frameslice := range d.frame_slice_intersects { + if frameslice.frame == frame && frameslice.slice == slice { + return frameslice, nil + } + } + return nil, FrameSliceIntersectDoesNotExistError +} + +func (fsi *FrameSliceIntersect) GetFragment(fragment_id *uuid.UUID) (*Fragment, error) { + for _, fragment := range fsi.fragments { + if fragment.id == fragment_id { + return fragment, nil + } + } + return nil, FragmentDoesNotExistError +} + +func (fsi *FrameSliceIntersect) AddFragment(fragment *Fragment) { + fsi.fragments = append(fsi.fragments, fragment) + fsi.hashring.Add(fragment.id.String()) +} + + + +///////// FRAGMENTS //////////////////////////////////////////////////////////////////////// + +// A fragment is a collection of bitmaps within a slice. The fragment contains a reference to the responsible node for that fragment. The node is in the form ip:port +type Fragment struct { + id *uuid.UUID + process *Process +} + +// rename this one +func (d *Database) OldGetFragment(bitmap Bitmap, profile_id int) (*Fragment, error) { + d.mutex.Lock() + defer d.mutex.Unlock() + slice, _ := d.GetSliceForProfile(profile_id) + frame, _ := d.getFrame(bitmap.FrameType) + fsi, err := d.GetFrameSliceIntersect(frame, slice) + frag_id_s, err := fsi.hashring.Get(fmt.Sprintf("%d", bitmap.Id)) + frag_id, err := uuid.ParseHex(frag_id_s) + if err != nil { + log.Fatal(err) + } + return fsi.GetFragment(frag_id) +} + + +/* +// NOT IMPLEMENTED +// this would loop through all frame_slice_intersect[], then all fragmments to find a match +func (d *Database) GetFragmentById(fragment_id *uuid.UUID) *Fragment { +} +*/ +func (d *Database) getFragment(frame *Frame, slice *Slice, fragment_id *uuid.UUID) (*Fragment, error) { + fsi, err := d.GetFrameSliceIntersect(frame, slice) + if err != nil { + log.Fatal(err) + } + return fsi.GetFragment(fragment_id) +} + +func (d *Database) addFragment(frame *Frame, slice *Slice, fragment_id *uuid.UUID) *Fragment { + fsi, err := d.GetFrameSliceIntersect(frame, slice) + if err != nil { + log.Fatal(err) + } + fragment := Fragment{id: fragment_id} + fsi.AddFragment(&fragment) + return &fragment +} + +/* +func (d *Database) AllocateFragment(frame *Frame, slice *Slice) *Fragment { + // from ETCD, randomly get a process that has available_fragments > 0 + // atomically decrement available_fragments (as long as it's not 0) + // if it IS 0, try until we find a process with available capacity + + * + process, err := GetAvailableProcess() + if err != nil { + log.Fatal(err) + } + * + process_id, _ := uuid.NewV4() + process := NewProcess(process_id) + return nil + //return d.AddFragment(&frame, &slice, process) +} +*/ + +/* +func (d *Database) AddFragmentByProcess(frame *Frame, slice *Slice, process *Process) *Fragment { + d.mutex.Lock() + defer d.mutex.Unlock() + frameslice, _ := d.GetFrameSliceIntersect(frame, slice) + fragment_id, _ := uuid.NewV4() + fragment := Fragment{id: fragment_id, process: process} + frameslice.fragments = append(frameslice.fragments, &fragment) + frameslice.hashring.Add(fragment.id.String()) + return &fragment +} +*/ + +func (d *Database) GetOrCreateFragment(frame *Frame, slice *Slice, fragment_id *uuid.UUID) *Fragment { + d.mutex.Lock() + defer d.mutex.Unlock() + fragment, err := d.getFragment(frame, slice, fragment_id) + if err == nil { + return fragment + } + return d.addFragment(frame, slice, fragment_id) +} + +func (f *Fragment) SetProcess(process *Process) { + f.process = process +} + +/////////////////////////////////////////////////////////////////////////////////////////////// + + +// Get a slice from a database +func (d *Database) GetSliceForProfile(profile_id int) (*Slice, error) { + slice_id := profile_id / SLICE_WIDTH + return d.getSlice(slice_id) +} + + +type Bitmap struct { + Id int + FrameType string } diff --git a/db/topology_test.go b/db/topology_test.go new file mode 100644 index 000000000..0666ef0e7 --- /dev/null +++ b/db/topology_test.go @@ -0,0 +1,29 @@ +package db + +import ( + "testing" + "log" + "github.com/nu7hatch/gouuid" + "github.com/davecgh/go-spew/spew" + . "github.com/smartystreets/goconvey/convey" +) + +func TestTopology(t *testing.T) { + Convey("Basic DB structures", t, func() { + log.Println("topology test") + + cluster := NewCluster() + database := cluster.GetOrCreateDatabase("main") + + frame := database.GetOrCreateFrame("general") + slice := database.GetOrCreateSlice(0) + + fragment_id, _ := uuid.ParseHex("6a9aea17-2915-4eb4-858f-a8d7d4dc0a1e") + spew.Dump(fragment_id) + database.GetOrCreateFragment(frame, slice, fragment_id) + + spew.Dump(database) + spew.Dump("DONE") + + }) +} diff --git a/deps.json b/deps.json index 8df70badd..fc18b8318 100644 --- a/deps.json +++ b/deps.json @@ -6,7 +6,7 @@ }, "etcd": { "repo": "github.com/coreos/go-etcd/etcd", - "version": "8a4461a676eb65fb74f10da1f8198cc9f67da366", + "version": "8a4461a", "type": "git" }, "goconvey": { diff --git a/query/lexer.go b/query/lexer.go new file mode 100644 index 000000000..68762b6ac --- /dev/null +++ b/query/lexer.go @@ -0,0 +1,186 @@ +package query + +import ( + "errors" + "strings" + "log" + "unicode" + "unicode/utf8" + //"github.com/davecgh/go-spew/spew" +) + +const ( + TYPE_FUNC = iota + TYPE_LP = iota + TYPE_RP = iota + TYPE_ID = iota + TYPE_COMMA = iota +) + +type Token struct { + Text string + Type int +} + +type statefn func(lexer *Lexer) statefn + +type Lexer struct { + text string // the string being scanned. + pos int // current position in the input. + width int // width of last rune read from input. + start int // start position of this item. + state int // current state of lexer NEEDED??? + ch chan Token // channel of scanned items (Tokens). +} + + +func (lexer *Lexer) emit(typ int) { + lexer.ch <- Token{lexer.text[lexer.start:lexer.pos], typ} + lexer.start = lexer.pos +} + +func (lexer *Lexer) acceptUntil(chars string) error { + for { + if strings.HasPrefix(lexer.text[lexer.pos:], chars) { + return nil + } + // if we receive a reserved character that we are not expecting, throw a parse error + lexer.pos += 1 + if lexer.pos > len(lexer.text) { + return errors.New("Parse error, expecting " + string(chars)) + } + } +} + +func (lexer *Lexer) acceptRun(valid string) { + for strings.IndexRune(valid, lexer.next()) >= 0 { + } + lexer.backup() +} + +// next returns the next rune in the input. +func (lexer *Lexer) next() (runey rune) { + if lexer.pos >= len(lexer.text) { + lexer.width = 0 + return 0 + } + runey, lexer.width = utf8.DecodeRuneInString(lexer.text[lexer.pos:]) + lexer.pos += lexer.width + return runey +} + +// ignore skips over the pending input before this point. +func (lexer *Lexer) ignore() { + lexer.start = lexer.pos +} + +// backup steps back one rune. +// Can be called only once per call of next. +func (lexer *Lexer) backup() { + lexer.pos -= lexer.width +} + +// peek returns but does not consume +// the next rune in the input. +func (lexer *Lexer) peek() rune { + for { + next_rune := lexer.next() + // ignore spaces + if next_rune != rune(' ') { + lexer.backup() + return next_rune + } + lexer.ignore() + } +} + +func stateFunc(lexer *Lexer) statefn { + err := lexer.acceptUntil("(") + if err != nil { + log.Fatal(err) + } + lexer.emit(TYPE_FUNC) + return stateLP +} + +func stateLP(lexer *Lexer) statefn { + lexer.pos += 1 + lexer.emit(TYPE_LP) + // handle multiple LPs + if lexer.peek() == rune('(') { + return stateLP + } + return stateArgs +} + +func stateArgs(lexer *Lexer) statefn { + if unicode.IsNumber(lexer.peek()) { + return stateID + } else { + return stateFunc + } +} + +func stateID(lexer *Lexer) statefn { + + digits := "0123456789" + lexer.acceptRun(digits) + lexer.emit(TYPE_ID) + + // if next is comma + peeked := lexer.peek() + if peeked == rune(',') { + return stateComma + } else if peeked == rune(')') { + return stateRP + } else { + return stateID + } +} + +func stateRP(lexer *Lexer) statefn { + lexer.pos += 1 + lexer.emit(TYPE_RP) + + peeked := lexer.peek() + if peeked == rune(',') { + return stateComma + } else if peeked == rune(')') { + return stateRP + } else { + return stateEOF + } +} + +func stateComma(lexer *Lexer) statefn { + lexer.pos += 1 + lexer.emit(TYPE_COMMA) + return stateArgs +} + +func stateEOF(lexer *Lexer) statefn { + close(lexer.ch) + return nil +} + +func (lexer *Lexer) Lex() []Token{ + tokens := make([]Token, 0) + state := stateFunc + go func () { + for { + state = state(lexer) + if state == nil { + return + } + } + }() + for t := range lexer.ch { + tokens = append(tokens, t) + } + return tokens +} + +func Lex(input string) []Token { + lexer := Lexer{input, 0, 0, 0, TYPE_FUNC, make(chan Token)} + return lexer.Lex() +} diff --git a/query/lexer_test.go b/query/lexer_test.go new file mode 100644 index 000000000..f0c632384 --- /dev/null +++ b/query/lexer_test.go @@ -0,0 +1,104 @@ +package query + +import ( + "testing" + . "github.com/smartystreets/goconvey/convey" +) + +func TestLexer(t *testing.T) { + Convey("Basic lexical analysis", t, func() { + tokens := Lex("get(10)") + So(len(tokens), ShouldEqual, 4) + So(tokens[0].Text, ShouldEqual, "get") + So(tokens[0].Type, ShouldEqual, TYPE_FUNC) + So(tokens[1].Text, ShouldEqual, "(") + So(tokens[1].Type, ShouldEqual, TYPE_LP) + So(tokens[2].Text, ShouldEqual, "10") + So(tokens[2].Type, ShouldEqual, TYPE_ID) + So(tokens[3].Text, ShouldEqual, ")") + So(tokens[3].Type, ShouldEqual, TYPE_RP) + + tokens2 := Lex("intersect(get(10), get(11), get(12))") + So(len(tokens2), ShouldEqual, 17) + So(tokens2[0].Text, ShouldEqual, "intersect") + So(tokens2[0].Type, ShouldEqual, TYPE_FUNC) + So(tokens2[1].Text, ShouldEqual, "(") + So(tokens2[1].Type, ShouldEqual, TYPE_LP) + So(tokens2[2].Text, ShouldEqual, "get") + So(tokens2[2].Type, ShouldEqual, TYPE_FUNC) + So(tokens2[3].Text, ShouldEqual, "(") + So(tokens2[3].Type, ShouldEqual, TYPE_LP) + So(tokens2[4].Text, ShouldEqual, "10") + So(tokens2[4].Type, ShouldEqual, TYPE_ID) + So(tokens2[5].Text, ShouldEqual, ")") + So(tokens2[5].Type, ShouldEqual, TYPE_RP) + So(tokens2[6].Text, ShouldEqual, ",") + So(tokens2[6].Type, ShouldEqual, TYPE_COMMA) + So(tokens2[7].Text, ShouldEqual, "get") + So(tokens2[7].Type, ShouldEqual, TYPE_FUNC) + So(tokens2[8].Text, ShouldEqual, "(") + So(tokens2[8].Type, ShouldEqual, TYPE_LP) + So(tokens2[9].Text, ShouldEqual, "11") + So(tokens2[9].Type, ShouldEqual, TYPE_ID) + So(tokens2[10].Text, ShouldEqual, ")") + So(tokens2[10].Type, ShouldEqual, TYPE_RP) + So(tokens2[11].Text, ShouldEqual, ",") + So(tokens2[11].Type, ShouldEqual, TYPE_COMMA) + So(tokens2[12].Text, ShouldEqual, "get") + So(tokens2[12].Type, ShouldEqual, TYPE_FUNC) + So(tokens2[13].Text, ShouldEqual, "(") + So(tokens2[13].Type, ShouldEqual, TYPE_LP) + So(tokens2[14].Text, ShouldEqual, "12") + So(tokens2[14].Type, ShouldEqual, TYPE_ID) + So(tokens2[15].Text, ShouldEqual, ")") + So(tokens2[15].Type, ShouldEqual, TYPE_RP) + So(tokens2[16].Text, ShouldEqual, ")") + So(tokens2[16].Type, ShouldEqual, TYPE_RP) + + tokens3 := Lex("intersect(get(10), get(11), concat(12,14))") + So(len(tokens3), ShouldEqual, 19) + So(tokens3[0].Text, ShouldEqual, "intersect") + So(tokens3[0].Type, ShouldEqual, TYPE_FUNC) + So(tokens3[1].Text, ShouldEqual, "(") + So(tokens3[1].Type, ShouldEqual, TYPE_LP) + So(tokens3[2].Text, ShouldEqual, "get") + So(tokens3[2].Type, ShouldEqual, TYPE_FUNC) + So(tokens3[3].Text, ShouldEqual, "(") + So(tokens3[3].Type, ShouldEqual, TYPE_LP) + So(tokens3[4].Text, ShouldEqual, "10") + So(tokens3[4].Type, ShouldEqual, TYPE_ID) + So(tokens3[5].Text, ShouldEqual, ")") + So(tokens3[5].Type, ShouldEqual, TYPE_RP) + So(tokens3[6].Text, ShouldEqual, ",") + So(tokens3[6].Type, ShouldEqual, TYPE_COMMA) + So(tokens3[7].Text, ShouldEqual, "get") + So(tokens3[7].Type, ShouldEqual, TYPE_FUNC) + So(tokens3[8].Text, ShouldEqual, "(") + So(tokens3[8].Type, ShouldEqual, TYPE_LP) + So(tokens3[9].Text, ShouldEqual, "11") + So(tokens3[9].Type, ShouldEqual, TYPE_ID) + So(tokens3[10].Text, ShouldEqual, ")") + So(tokens3[10].Type, ShouldEqual, TYPE_RP) + So(tokens3[11].Text, ShouldEqual, ",") + So(tokens3[11].Type, ShouldEqual, TYPE_COMMA) + So(tokens3[12].Text, ShouldEqual, "concat") + So(tokens3[12].Type, ShouldEqual, TYPE_FUNC) + So(tokens3[13].Text, ShouldEqual, "(") + So(tokens3[13].Type, ShouldEqual, TYPE_LP) + So(tokens3[14].Text, ShouldEqual, "12") + So(tokens3[14].Type, ShouldEqual, TYPE_ID) + So(tokens3[15].Text, ShouldEqual, ",") + So(tokens3[15].Type, ShouldEqual, TYPE_COMMA) + So(tokens3[16].Text, ShouldEqual, "14") + So(tokens3[16].Type, ShouldEqual, TYPE_ID) + So(tokens3[17].Text, ShouldEqual, ")") + So(tokens3[17].Type, ShouldEqual, TYPE_RP) + So(tokens3[18].Text, ShouldEqual, ")") + So(tokens3[18].Type, ShouldEqual, TYPE_RP) + + tokens4 := Lex("concat(1,2,345,890)") + So(len(tokens4), ShouldEqual, 10) + So(tokens4[6].Text, ShouldEqual, "345") + So(tokens4[6].Type, ShouldEqual, TYPE_ID) + }) +} diff --git a/query/parser.go b/query/parser.go index 130b48bba..c94b8cc26 100644 --- a/query/parser.go +++ b/query/parser.go @@ -1,105 +1,12 @@ package query import ( - "encoding/json" + "encoding/json" "errors" - "log" "pilosa/db" - "github.com/davecgh/go-spew/spew" + //"github.com/davecgh/go-spew/spew" ) -const ( - TYPE_FUNC = iota - TYPE_LP = iota - TYPE_RP = iota - TYPE_ID = iota -) - -type Token struct { - Text string - Type int -} - -type statefn func(lexer *Lexer) statefn - -type Lexer struct { - text string - pos int - start int - state int - ch chan Token -} - -func (lexer *Lexer) emit(typ int) { - lexer.ch <- Token{lexer.text[lexer.start:lexer.pos], typ} - lexer.start = lexer.pos -} - -func (lexer *Lexer) accept(char uint8) error { - for { - if lexer.text[lexer.pos] == char { - return nil - } - lexer.pos += 1 - if lexer.pos > len(lexer.text) { - return errors.New("Parse error, expecting " + string(char)) - } - } -} - -func stateFunc(lexer *Lexer) statefn { - err := lexer.accept('(') - if err != nil { - log.Fatal(err) - } - lexer.emit(TYPE_FUNC) - return stateLP -} - -func stateLP(lexer *Lexer) statefn { - lexer.pos += 1 - lexer.emit(TYPE_LP) - return stateID -} - -func stateID(lexer *Lexer) statefn { - err := lexer.accept(')') - if err != nil { - log.Fatal(err) - } - lexer.emit(TYPE_ID) - return stateRP -} - -func stateRP(lexer *Lexer) statefn { - lexer.pos += 1 - lexer.emit(TYPE_RP) - close(lexer.ch) - return nil -} - -func (lexer *Lexer) Lex() []Token{ - tokens := make([]Token, 0) - state := stateFunc - go func () { - for { - state = state(lexer) - if state == nil { - return - } - } - }() - for t := range lexer.ch { - spew.Dump(t) - tokens = append(tokens, t) - } - return tokens -} - -func Lex(input string) []Token { - lexer := Lexer{input, 0, 0, TYPE_FUNC, make(chan Token)} - return lexer.Lex() -} var InvalidQueryError = errors.New("Invalid query format.") @@ -141,7 +48,7 @@ func (q *QueryParser) Walk(data interface{}) (*Query, error) { return nil, InvalidQueryError } id_int := int(id) - query.Inputs = []QueryInput{db.Bitmap{frame, id_int}} + query.Inputs = []QueryInput{db.Bitmap{id_int, frame}} } return query, nil diff --git a/query/parser_test.go b/query/parser_test.go deleted file mode 100644 index 772aab41c..000000000 --- a/query/parser_test.go +++ /dev/null @@ -1,21 +0,0 @@ -package query - -import ( - "testing" - . "github.com/smartystreets/goconvey/convey" -) - -func TestParser(t *testing.T) { - Convey("Basic parsing", t, func() { - tokens := Lex("get(10)") - So(len(tokens), ShouldEqual, 4) - So(tokens[0].Text, ShouldEqual, "get") - So(tokens[0].Type, ShouldEqual, TYPE_FUNC) - So(tokens[1].Text, ShouldEqual, "(") - So(tokens[1].Type, ShouldEqual, TYPE_LP) - So(tokens[2].Text, ShouldEqual, "10") - So(tokens[2].Type, ShouldEqual, TYPE_ID) - So(tokens[3].Text, ShouldEqual, ")") - So(tokens[3].Type, ShouldEqual, TYPE_RP) - }) -} diff --git a/query/planner.go b/query/planner.go index b70261468..8551e0438 100644 --- a/query/planner.go +++ b/query/planner.go @@ -2,7 +2,7 @@ package query import ( "github.com/nu7hatch/gouuid" - "strconv" + //"strconv" "fmt" "math/rand" "pilosa/db" @@ -69,6 +69,7 @@ type GetQueryTree struct { // Uses consistent hashing function to select node containing data for GET operation func (qt *GetQueryTree) getLocation(d *db.Database) string { + /* frame, err := d.GetFrame(qt.bitmap.FrameType) if err != nil { panic(err) @@ -82,6 +83,8 @@ func (qt *GetQueryTree) getLocation(d *db.Database) string { fragment := slice.Fragments[fragIndex] return fmt.Sprintf(fragment.Node) + */ + return "Nothing yet" } // Builds QueryTree object from Query. Pass slice=-1 to perform operation on all slices diff --git a/query/planner_test.go b/query/planner_test.go index db6e6a623..f63aa4160 100644 --- a/query/planner_test.go +++ b/query/planner_test.go @@ -3,8 +3,11 @@ package query import ( "testing" "log" + . "github.com/smartystreets/goconvey/convey" ) func TestQueryPlanner(t *testing.T) { - log.Println("query planner test") + Convey("Basic query plan", t, func() { + log.Println("query planner test") + }) } diff --git a/router/router.go b/router/router.go deleted file mode 100644 index 5dc700c97..000000000 --- a/router/router.go +++ /dev/null @@ -1,94 +0,0 @@ -package router - -import ( - "log" - //"time" - //"strings" - "pilosa/core" - "pilosa/db" -) - -type Router struct { - core.Service -} - -func (r *Router) Init() { - log.Println("Initializing router...") -} - -func (r *Router) Run() { - log.Println("Running router...") - r.SetupEtcd() - //go r.SyncEtcd() - go r.WatchEtcd() - go r.HandleConnections() - go r.HandleInbox() - go r.ServeHTTP() - - //go func() { - // for { - // r.SendMessage(&core.Message{"ping", core.Location{"127.0.0.1", 1200}, core.Location{"127.0.0.1", 1300}}) - // time.Sleep(2*time.Second) - // //log.Println(r.GetRouterLocation(core.Location{"127.0.0.1", 1200})) - // } - //}() - - sigterm, sighup := r.GetSignals() - for { - select { - case <- sighup: - log.Println("SIGHUP! Reloading configuration...") - // TODO: reload configuration - case <- sigterm: - log.Println("SIGTERM! Cleaning up...") - r.Stop() - return - } - } -} - -func (r *Router) HandleInbox() { - for { - select { - case message := <-r.Inbox: - log.Println("process", message) - } - } -} - -func (r *Router) SetupEtcd() { - r.Service.SetupEtcd() - //routers, err := r.Etcd.Get("topology") - //if err != nil { - // log.Fatal(err) - //} - //for _, router := range routers { - // routerstring := strings.Split(router.Key, "/")[2] - // routerlocation, err := core.NewLocation(routerstring) - // if err != nil { - // log.Fatal(err) - // } - // nodes, err := r.Etcd.Get("topology/" + routerstring) - // if err != nil { - // log.Fatal(err) - // } - // for _, node := range nodes { - // nodestring := strings.Split(node.Key, "/")[3] - // nodelocation, err := core.NewLocation(nodestring) - // if err != nil { - // log.Fatal(err) - // } - // r.NodeMap[*nodelocation] = *routerlocation - // } - //} -} - -func (r *Router) HandleMessage(m *db.Message) { - log.Println(m) -} - -func NewRouter(tcp, http *db.Location) *Router { - service := core.NewService(tcp, http) - router := Router{*service} - return &router -}