stub out the service.Executor process

core/http.go

@@ -66,7 +66,13 @@ func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) {
 	cluster := self.service.Cluster
 	database := cluster.GetOrCreateDatabase("main")
 	pql := string(body)
-	query.Execute(database, pql)
+	query_plan := query.QueryPlanForPQL(database, pql)
+
+	results_ch := make(chan *query.QueryResults)
+	self.service.Executor.NewJob(query_plan, results_ch)
+	results := <-results_ch
+	spew.Dump(results)
+	close(results_ch)
 }
 
 func (self *WebService) HandleStats(w http.ResponseWriter, r *http.Request) {

core/service.go

@@ -25,6 +25,7 @@ type Service struct {
 	ProcessMap     *ProcessMap
 	Transport      interfaces.Transporter
 	Dispatch       interfaces.Dispatcher
+	Executor       interfaces.Executorer
 	WebService     *WebService
 	process_id     *uuid.UUID
 	Index          *index.FragmentContainer
@@ -79,6 +80,7 @@ func (service *Service) Run() {
 	go service.WebService.Run()
 	go service.Transport.Run()
 	go service.Dispatch.Run()
+	go service.Executor.Run()
 
 	sigterm, sighup := service.GetSignals()
 	for {

cruncher/cruncher.go

@@ -3,6 +3,7 @@ package cruncher
 import (
 	"pilosa/core"
 	"pilosa/dispatch"
+	"pilosa/executor"
 	"pilosa/index"
 	"pilosa/transport"
 
@@ -25,6 +26,7 @@ func NewCruncher() *Cruncher {
 	fragment_container := index.NewFragmentContainer()
 	cruncher := Cruncher{service, make(chan bool), fragment_container}
 	cruncher.Transport = transport.NewTcpTransport(service)
-	cruncher.Dispatch = dispatch.NewCruncherDispatch(service)
+	cruncher.Dispatch = dispatch.NewDispatch(service)
+	cruncher.Executor = executor.NewExecutor(service)
 	return &cruncher
 }

dispatch/dispatch.go

@@ -7,20 +7,20 @@ import (
 	"github.com/davecgh/go-spew/spew"
 )
 
-type CruncherDispatch struct {
+type Dispatch struct {
 	service *core.Service
 }
 
-func (self *CruncherDispatch) Init() error {
+func (self *Dispatch) Init() error {
 	log.Println("Starting Dispatcher")
 	return nil
 }
 
-func (self *CruncherDispatch) Close() {
+func (self *Dispatch) Close() {
 	log.Println("Shutting down Dispatcher")
 }
 
-func (self *CruncherDispatch) Run() {
+func (self *Dispatch) Run() {
 	log.Println("Dispatch Run...")
 	for {
 		message := self.service.Transport.Receive()
@@ -118,6 +118,6 @@ func (self *CruncherDispatch) Run() {
 	}
 }
 
-func NewCruncherDispatch(service *core.Service) *CruncherDispatch {
-	return &CruncherDispatch{service}
+func NewDispatch(service *core.Service) *Dispatch {
+	return &Dispatch{service}
 }

executor/executor.go

@@ -0,0 +1,63 @@
+package executor
+
+import (
+	"log"
+	"pilosa/core"
+	"pilosa/query"
+
+	"github.com/davecgh/go-spew/spew"
+)
+
+type Job struct {
+	query_plan *query.QueryPlan
+	results_ch chan *query.QueryResults
+}
+
+type Executor struct {
+	service *core.Service
+	inbox   chan *Job
+}
+
+func (self *Executor) Init() error {
+	log.Println("Starting Executor")
+	return nil
+}
+
+func (self *Executor) Close() {
+	log.Println("Shutting down Executor")
+}
+
+func (self *Executor) NewJob(qp *query.QueryPlan, results chan *query.QueryResults) {
+	j := Job{qp, results}
+	self.inbox <- &j
+}
+
+//func (self *Executor) ExecuteJob(qp *query.QueryPlan) {
+func (self *Executor) executeJob(job *Job) {
+	log.Println("Executor: ExecuteJob")
+	//spew.Dump(job)
+	query_plan := job.query_plan
+	for _, qs := range *query_plan {
+		spew.Dump(qs)
+	}
+	/*
+		spew.Dump("sleep...")
+		time.Sleep(4 * 1e9)
+	*/
+	// TODO: send the query steps out & wait for responses
+	qr := new(query.QueryResults)
+	qr.Data = 999
+	job.results_ch <- qr
+}
+
+func (self *Executor) Run() {
+	log.Println("Executor Run...")
+	for {
+		job := <-self.inbox
+		go self.executeJob(job)
+	}
+}
+
+func NewExecutor(service *core.Service) *Executor {
+	return &Executor{service, make(chan *Job)}
+}

interfaces/core.go

@@ -2,6 +2,7 @@ package interfaces
 
 import (
 	"pilosa/db"
+	"pilosa/query"
 )
 
 type Transporter interface {
@@ -16,3 +17,10 @@ type Dispatcher interface {
 	Close()
 	Run()
 }
+
+type Executorer interface {
+	Init() error
+	Close()
+	Run()
+	NewJob(*query.QueryPlan, chan *query.QueryResults)
+}

query/query.go

@@ -9,6 +9,10 @@ import (
 
 type QueryInput interface{}
 
+type QueryResults struct {
+	Data interface{}
+}
+
 type Query struct {
 	Operation string
 	Inputs    []QueryInput //"strconv"
@@ -17,7 +21,7 @@ type Query struct {
 	Profile_id int
 }
 
-func Execute(database *db.Database, pql string) {
+func QueryPlanForPQL(database *db.Database, pql string) *QueryPlan {
 	//spew.Dump("EXECUTE")
 	//spew.Dump(pql)
 	tokens := Lex(pql)
@@ -40,4 +44,5 @@ func Execute(database *db.Database, pql string) {
 	destination := db.Process{}
 	query_plan := query_planner.Plan(query, id, &destination)
 	spew.Dump(query_plan)
+	return query_plan
 }