merge travis

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()
 }

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")
+}

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")
+        */
+
+    })
+}

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
-		}
-	}
 }

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)
+        }
+    }
+}

cruncher/cruncher.go

@@ -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)
-		}
-	}
-}

db/constants.go

@@ -0,0 +1,3 @@
+package db
+
+const SLICE_WIDTH = 65536

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
 }

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")
+
+    })
+}

deps.json

@@ -6,7 +6,7 @@
     },
     "etcd": {
         "repo": "github.com/coreos/go-etcd/etcd",
-        "version": "8a4461a676eb65fb74f10da1f8198cc9f67da366",
+        "version": "8a4461a",
         "type": "git"
     },
     "goconvey": {

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()
+}

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)
+    })
+}

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

query/parser_test.go

@@ -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)
-        })
-}

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

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")
+    })
 }

router/router.go

@@ -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
-}