removed pql from setbit path

core/http.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"compress/gzip"
 	"encoding/base64"
+	"encoding/gob"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
@@ -403,6 +404,10 @@ type SBResult struct {
 	Result     interface{}
 }
 
+func init() {
+	gob.Register(SBResult{})
+}
+
 func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) {
 	if r.Method != "POST" {
 		http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed)
@@ -424,6 +429,8 @@ func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "Request To large", http.StatusBadRequest)
 		return
 	}
+	result := false
+	remoteSetBit := NewRemoteSetBit(self.service)
 	for _, obj := range args {
 		if obj["profile_id"] == nil {
 			http.Error(w, "Missing Profile", http.StatusBadRequest)
@@ -451,8 +458,6 @@ func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) {
 		}
 		t = float64(obj["filter"].(float64))
 		filter := uint64(t)
-		result := false
-		remoteSetBit := NewRemoteSetBit(self.service)
 		for bitmap_id := range bitmaps(frame, obj) {
 			start := time.Now()
 
@@ -467,6 +472,8 @@ func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) {
 			if util.Equal(frag.GetProcessId(), self.service.Id) {
 				// The Local Route
 				result, _ = self.service.Index.SetBit(frag.GetId(), bitmap_id, profile_id, filter)
+				bundle := SBResult{bitmap_id, frame, filter, profile_id, result}
+				results = append(results, bundle)
 			} else {
 
 				remoteSetBit.Add(frag, bitmap_id, profile_id, filter, frame)
@@ -482,14 +489,11 @@ func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) {
 				http.Error(w, err.Error(), http.StatusInternalServerError)
 				return
 			}
-			bundle := SBResult{bitmap_id, frame, filter, profile_id, result}
-			results = append(results, bundle)
 		}
 
-		remoteSetBit.Request()
-		remoteSetBit.MergeResults(results)
-
 	}
+	remoteSetBit.Request()
+	results = remoteSetBit.MergeResults(results)
 	encoder := json.NewEncoder(w)
 	err = encoder.Encode(results)
 	if err != nil {

core/remotebits.go

@@ -1,6 +1,8 @@
 package core
 
 import (
+	//	"github.com/davecgh/go-spew/spew"
+	"encoding/gob"
 	"log"
 	"pilosa/db"
 	"pilosa/util"
@@ -12,6 +14,12 @@ type RemoteSetBit struct {
 	service  *Service
 }
 
+func init() {
+	gob.Register(BitmapRequestItem{})
+	gob.Register(BitsRequest{})
+	gob.Register(BitsResponse{})
+}
+
 type BitsRequest struct {
 	Bits            []BitmapRequestItem
 	ReturnProcessId util.GUID
@@ -34,7 +42,7 @@ func NewRemoteSetBit(s *Service) *RemoteSetBit {
 }
 
 func (self *RemoteSetBit) Request() {
-	self.requests = make([]util.GUID, len(self.cluster), len(self.cluster))
+	self.requests = make([]util.GUID, 0)
 	source_process, _ := self.service.GetProcess()
 	for process, request := range self.cluster {
 		random_id := util.RandomUUID()
@@ -79,7 +87,7 @@ func (self *RemoteSetBit) MergeResults(local_results []SBResult) []SBResult {
 func (self *RemoteSetBit) Add(frag *db.Fragment, bitmap_id, profile_id, filter uint64, frame string) {
 	x, found := self.cluster[frag.GetProcessId()]
 	if !found {
-		x = make([]BitmapRequestItem, 8)
+		x = make([]BitmapRequestItem, 0)
 	}
 	x = append(x, BitmapRequestItem{frag.GetId(), bitmap_id, profile_id, filter, frame})
 	self.cluster[frag.GetProcessId()] = x

dispatch/dispatch.go

@@ -51,7 +51,10 @@ func (self *Dispatch) Run() {
 			go self.service.Executor.NewJob(message)
 		case core.TopFill:
 			go self.service.TopFillHandler(message)
+		case core.BitsResponse:
+			self.service.Hold.Set(data.ResultId(), data.ResultData(), 30)
 		default:
+			println("Dispatch Unhandled")
 			spew.Dump(data)
 			log.Println("Unprocessed message", data)
 		}

transport/tcp.go

@@ -10,7 +10,6 @@ import (
 	"pilosa/core"
 	"pilosa/db"
 	. "pilosa/util"
-	"runtime/debug"
 	"sync"
 
 	"time"
@@ -19,121 +18,74 @@ import (
 )
 
 type connection struct {
-	transport *TcpTransport
-	inbox     chan *db.Message
-	outbox    chan *db.Message
-	conn      net.Conn
-	process   *GUID
-	id        int
-	terminate bool
-	exit      chan int
-}
-
-type newconnection struct {
-	id         *GUID
-	connection *connection
+	inbox      chan *db.Message
+	outbox     chan *db.Message
+	conn       net.Conn
+	encoder    *gob.Encoder
+	decoder    *gob.Decoder
+	transport  *TcpTransport
+	process_id *GUID
+	exit       chan int
 }
 
 func init() {
 	gob.Register(GUID{})
 }
 
-func newConnection(transport *TcpTransport, conn net.Conn, proc *GUID) *connection {
-	println("New Connection", proc)
+func newConnection(conn net.Conn, enc *gob.Encoder, dec *gob.Decoder, t *TcpTransport, g *GUID) *connection {
 	p := new(connection)
-	p.transport = transport
 	p.outbox = make(chan *db.Message, 2048)
 	p.inbox = make(chan *db.Message, 2048)
 	p.conn = conn
-	p.process = proc
+	p.encoder = enc
+	p.decoder = dec
+	p.transport = t
+	p.process_id = g
 	p.exit = make(chan int)
 	return p
 }
 
-func (self *connection) manage() {
-	for {
-		self.exit = make(chan int)
-		self.serviceConnection()
-		close(self.exit)
-		time.Sleep(2 * time.Second)
-		self.conn = nil
-		if self.terminate {
-			break
-		}
-
-	}
+func (self *connection) Close() {
+	close(self.outbox)
+	close(self.inbox)
 }
 
-func (self *connection) Shutdown() {
-	self.terminate = true
-	if self.conn != nil {
-		self.conn.Close()
-	}
-
-}
-func (self *connection) serviceConnection() {
-	var host_string string
-	if self.conn == nil {
-		println("Service Connection", self.process)
-		process, err := self.transport.service.ProcessMap.GetProcess(self.process)
-		if err != nil {
-			log.Println("transport/tcp: error getting process, retrying in 2 seconds... ", self.process, err)
-			return
-		}
-		host_string = fmt.Sprintf("%s:%d", process.Host(), process.PortTcp())
-		log.Println("Connecting:", host_string)
-		conn, err := net.Dial("tcp", host_string)
-		if err != nil {
-			log.Println("transport/tcp: error dialing: ", host_string, " Retrying in 2 seconds...")
-			return
-		}
-		self.conn = conn
-		go func() {
-			//register on server
-			self.outbox <- &db.Message{self.transport.service.Id}
-		}()
-	}
-	encoder := gob.NewEncoder(self.conn)
-	decoder := gob.NewDecoder(self.conn)
+func (self *connection) run() {
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
 		for {
 			var mess *db.Message
-			err := decoder.Decode(&mess)
+			err := self.decoder.Decode(&mess)
 			if err != nil {
-				log.Println("transport/tcp: error decoding message: ", host_string, err.Error())
+				log.Println("transport/tcp: error decoding message: ", err.Error())
+				time.Sleep(2 * time.Second)
 				self.exit <- 1
 				wg.Done()
 				return
+			} else {
+				self.inbox <- mess
 			}
-			self.inbox <- mess
 		}
 	}()
 	breakout := true
 	for breakout {
 		select {
 		case message := <-self.outbox:
-			err := encoder.Encode(message)
+			err := self.encoder.Encode(message)
 			if err != nil {
-				log.Println("Sending to ", host_string, err.Error())
+				log.Println("Failed to Send on Socket Transport")
 				breakout = false
 			}
 		case message := <-self.inbox:
-			identifier, ok := message.Data.(GUID)
-			if ok {
-				// message is connection registration; bypass inbox and register
-				self.process = &identifier
-				self.transport.reg <- &newconnection{&identifier, self}
-			} else {
-				self.transport.inbox <- message
-			}
+			self.transport.inbox <- message
 		case <-self.exit:
 			breakout = false
 		}
 	}
 	self.conn.Close()
 	wg.Wait()
+	self.transport.removeConnection(self)
 }
 
 type TcpTransport struct {
@@ -141,21 +93,20 @@ type TcpTransport struct {
 	port        int
 	inbox       chan *db.Message
 	outbox      chan db.Envelope
-	connections map[GUID]*connection
-	reg         chan *newconnection
+	connections map[*GUID]*connection
+	mutex       sync.Mutex
 	enc         *gob.Encoder
 	dec         *gob.Decoder
 }
 
 func NewTcpTransport(service *core.Service) *TcpTransport {
 	p := new(TcpTransport)
-	var network bytes.Buffer // Stand-in for a network connection
 	p.service = service
 	p.port = config.GetInt("port_tcp")
 	p.inbox = make(chan *db.Message, 2048)
 	p.outbox = make(chan db.Envelope, 2048)
-	p.connections = make(map[GUID]*connection)
-	p.reg = make(chan *newconnection)
+	p.connections = make(map[*GUID]*connection)
+	var network bytes.Buffer         // Stand-in for a network connection
 	p.enc = gob.NewEncoder(&network) // Will write to network.
 	p.dec = gob.NewDecoder(&network) // Will read from network.
 	return p
@@ -167,19 +118,13 @@ func (self *TcpTransport) Run() {
 	for {
 		select {
 		case env := <-self.outbox: //transport outbox
-			con, ok := self.connections[*(env.Host)]
-			if !ok {
-				con = newConnection(self, nil, env.Host)
-				go con.manage()
-				self.connections[*env.Host] = con
+			con, need := self.getConnection(env.Host)
+			if need {
+				con = self.connectRemotePeer(env.Host)
 			}
-			con.outbox <- env.Message
-		case nc := <-self.reg:
-			before, present := self.connections[*nc.id]
-			if present {
-				before.Shutdown()
+			if con != nil {
+				con.outbox <- env.Message
 			}
-			self.connections[*nc.id] = nc.connection
 		}
 	}
 }
@@ -197,13 +142,60 @@ func (self *TcpTransport) listen() {
 			time.Sleep(2 * time.Second)
 			continue
 		}
-		go self.manage(conn)
+		self.addPeer(conn)
 	}
 }
+func (self *TcpTransport) connectRemotePeer(remoteProcessId *GUID) *connection {
+	process, err := self.service.ProcessMap.GetProcess(remoteProcessId)
+	host_string := fmt.Sprintf("%s:%d", process.Host(), process.PortTcp())
+	conn, err := net.Dial("tcp", host_string)
+	if err != nil {
+		log.Println("transport/tcp: error dialing: ", host_string, " Retrying in 2 seconds...")
+		return nil
+	}
+	encoder := gob.NewEncoder(conn)
+	decoder := gob.NewDecoder(conn)
+	err = encoder.Encode(self.service.Id) //send registration
+	if err != nil {
+		log.Println("Error sending processid:", host_string)
+	}
+	acon := newConnection(conn, encoder, decoder, self, remoteProcessId)
+	self.addConnection(acon)
+	return acon
+
+}
+
+func (self *TcpTransport) addPeer(conn net.Conn) {
+	decoder := gob.NewDecoder(conn)
+	encoder := gob.NewEncoder(conn)
+	var processid GUID
+	err := decoder.Decode(&processid)
+	if err != nil {
+		log.Println("Failed to recieve remote processid")
+		return
+	}
+	acon := newConnection(conn, encoder, decoder, self, &processid)
+	self.addConnection(acon)
+}
+func (self *TcpTransport) getConnection(guid *GUID) (*connection, bool) {
+	self.mutex.Lock()
+	defer self.mutex.Unlock()
+	con, ok := self.connections[guid]
+	return con, !ok
+}
+
+func (self *TcpTransport) addConnection(c *connection) {
+	self.mutex.Lock()
+	defer self.mutex.Unlock()
+	self.connections[c.process_id] = c
+	go c.run()
+}
+func (self *TcpTransport) removeConnection(c *connection) {
+	self.mutex.Lock()
+	defer self.mutex.Unlock()
+	delete(self.connections, c.process_id)
+	c.Close()
 
-func (self *TcpTransport) manage(c net.Conn) {
-	con := newConnection(self, c, nil)
-	con.manage()
 }
 
 func (self *TcpTransport) Close() {
@@ -231,6 +223,7 @@ func (self *TcpTransport) Send(message *db.Message, host *GUID) {
 }
 
 func (self *TcpTransport) Receive() *db.Message {
+
 	message := <-self.inbox
 	notify.Post("inbox", message)
 	return message