added some connection cleanup code for tcp/transport and support for local connections

core/http.go

@@ -330,7 +330,7 @@ func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) {
 	encoder := json.NewEncoder(w)
 	err = encoder.Encode(results)
 	if err != nil {
-		http.Error(w, "Error encoding: "+err.Error(), http.StatusInternalServerError)
+		log.Println("Encode Error :", database_name, pql, err.Error())
 		return
 	}
 
@@ -459,7 +459,7 @@ func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) {
 			util.SendTimer("executor_setbit", delta.Nanoseconds())
 
 			if err != nil {
-				log.Println("Error running set_bit", pql)
+				log.Println("Error running set_bit", db, pql)
 				http.Error(w, err.Error(), http.StatusInternalServerError)
 				return
 			}

executor/executor.go

@@ -1,7 +1,6 @@
 package executor
 
 import (
-	"fmt"
 	"log"
 	"pilosa/config"
 	"pilosa/core"
@@ -53,7 +52,8 @@ func (self *Executor) NewJob(job *db.Message) {
 	//	case query.MaskQueryStep:
 	//		self.service.MaskQueryStepHandler(job)
 	default:
-		fmt.Println("unknown")
+		log.Println("unknown")
+		log.Println(spew.Sdump(job.Data))
 	}
 }
 

index/fragment_container.go

@@ -468,15 +468,6 @@ func (self *Fragment) processCommand(req Command) {
 	req.ResponseChannel() <- Result{answer, delta}
 }
 func (self *Fragment) ServeFragment(loadChan chan Command) {
-	ispanic := true
-	defer func() {
-		if r := recover(); r != nil {
-			if ispanic {
-				self.Persist()
-			}
-		}
-
-	}()
 	for {
 		select {
 		case req := <-self.requestChan:
@@ -490,7 +481,6 @@ func (self *Fragment) ServeFragment(loadChan chan Command) {
 			case wg := <-self.exit:
 				log.Println("Fragment Shutdown")
 				self.Persist()
-				ispanic = false
 				wg.Done()
 			}
 		}

transport/tcp.go

@@ -1,6 +1,7 @@
 package transport
 
 import (
+	"bytes"
 	"encoding/gob"
 	"fmt"
 	"log"
@@ -9,18 +10,23 @@ import (
 	"pilosa/core"
 	"pilosa/db"
 	. "pilosa/util"
+	"sync"
 
 	"time"
 
 	notify "github.com/bitly/go-notify"
+	"github.com/davecgh/go-spew/spew"
 )
 
 type connection struct {
 	transport *TcpTransport
 	inbox     chan *db.Message
 	outbox    chan *db.Message
-	conn      *net.Conn
+	conn      net.Conn
 	process   *GUID
+	id        int
+	terminate bool
+	exit      chan int
 }
 
 type newconnection struct {
@@ -32,70 +38,97 @@ func init() {
 	gob.Register(GUID{})
 }
 
+func newConnection(transport *TcpTransport, conn net.Conn, proc *GUID) *connection {
+	p := new(connection)
+	p.transport = transport
+	p.outbox = make(chan *db.Message, 100)
+	p.inbox = make(chan *db.Message, 100)
+	p.conn = conn
+	p.process = proc
+	p.exit = make(chan int)
+	return p
+}
+
 func (self *connection) manage() {
-BeginManageConnection:
 	for {
-		if self.conn == nil {
-			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)
-				time.Sleep(2 * time.Second)
-				continue
-			}
-			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...")
-				time.Sleep(2 * time.Second)
-				continue
-			}
-			self.conn = &conn
-			go func() {
-				self.outbox <- &db.Message{self.transport.service.Id}
-			}()
+		self.exit = make(chan int)
+		self.serviceConnection()
+		close(self.exit)
+		time.Sleep(2 * time.Second)
+		self.conn = nil
+		if self.terminate {
+			break
 		}
-		encoder := gob.NewEncoder(*self.conn)
-		decoder := gob.NewDecoder(*self.conn)
-		var exit = make(chan int)
+
+	}
+}
+
+func (self *connection) Shutdown() {
+	self.terminate = true
+	if self.conn != nil {
+		self.conn.Close()
+	}
+
+}
+func (self *connection) serviceConnection() {
+	if self.conn == nil {
+		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())
+		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() {
-			for {
-				var mess *db.Message
-				err := decoder.Decode(&mess)
-				if err != nil {
-					log.Println("transport/tcp: error decoding message: ", err.Error())
-					exit <- 1
-					return
-				}
-				self.inbox <- mess
-			}
+			//register on server
+			self.outbox <- &db.Message{self.transport.service.Id}
 		}()
+	}
+	encoder := gob.NewEncoder(self.conn)
+	decoder := gob.NewDecoder(self.conn)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
 		for {
-			select {
-			case message := <-self.outbox:
-				err := encoder.Encode(message)
-				if err != nil {
-					log.Println(err.Error())
-					return
-				}
-			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
-				}
-			case <-exit:
-				if self.process != nil {
-					self.conn = nil
-					continue BeginManageConnection
-				} else {
-					return
-				}
+			var mess *db.Message
+			err := decoder.Decode(&mess)
+			if err != nil {
+				log.Println("transport/tcp: error decoding message: ", err.Error())
+				self.exit <- 1
+				wg.Done()
+				return
 			}
+			self.inbox <- mess
+		}
+	}()
+	for {
+		select {
+		case message := <-self.outbox:
+			err := encoder.Encode(message)
+			if err != nil {
+				log.Println("Connection Outbox", err.Error())
+				break
+			}
+		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
+			}
+		case <-self.exit:
+			break
 		}
 	}
+	self.conn.Close()
+	wg.Wait()
 }
 
 type TcpTransport struct {
@@ -112,15 +145,19 @@ func (self *TcpTransport) Run() {
 	go self.listen()
 	for {
 		select {
-		case env := <-self.outbox:
+		case env := <-self.outbox: //transport outbox
 			con, ok := self.connections[*(env.Host)]
 			if !ok {
-				con = &connection{self, make(chan *db.Message, 100), make(chan *db.Message, 100), nil, env.Host}
+				con = newConnection(self, nil, env.Host)
 				go con.manage()
 				self.connections[*env.Host] = con
 			}
 			con.outbox <- env.Message
 		case nc := <-self.reg:
+			before, present := self.connections[*nc.id]
+			if present {
+				before.Shutdown()
+			}
 			self.connections[*nc.id] = nc.connection
 		}
 	}
@@ -139,23 +176,42 @@ func (self *TcpTransport) listen() {
 			time.Sleep(2 * time.Second)
 			continue
 		}
-		go self.manage(&conn)
+		go self.manage(conn)
 	}
 }
 
-func (self *TcpTransport) manage(conn *net.Conn) {
-	con := &connection{self, make(chan *db.Message, 1024), make(chan *db.Message, 1024), conn, nil}
+func (self *TcpTransport) manage(c net.Conn) {
+	con := newConnection(self, c, nil)
 	con.manage()
 }
 
 func (self *TcpTransport) Close() {
 	log.Println("Shutting down TCP transport")
 }
+func adjust(in *db.Message) *db.Message {
+	log.Println(spew.Sdump(in))
+	var network bytes.Buffer        // Stand-in for a network connection
+	enc := gob.NewEncoder(&network) // Will write to network.
+	dec := gob.NewDecoder(&network) // Will read from network.
+	err := enc.Encode(in)
+	var out db.Message
+	err = dec.Decode(&out)
+	if err != nil {
+		log.Println(err)
+	}
+	return &out
+
+}
 
 func (self *TcpTransport) Send(message *db.Message, host *GUID) {
-	envelope := db.Envelope{message, host}
-	notify.Post("outbox", &envelope)
-	self.outbox <- envelope
+	//I think I can avoid the outbox and go directly to the inbox if the transport process_id == GUID
+	if !Equal(host, self.service.Id) {
+		envelope := db.Envelope{message, host}
+		notify.Post("outbox", &envelope)
+		self.outbox <- envelope
+	} else {
+		self.inbox <- adjust(message)
+	}
 }
 
 func (self *TcpTransport) Receive() *db.Message {

util/id.go

@@ -63,6 +63,14 @@ func Hex_to_SUUID(str string) SUUID {
 
 type GUID [16]byte
 
+func Equal(a, b *GUID) bool {
+	for i, v := range a {
+		if v != b[i] {
+			return false
+		}
+	}
+	return true
+}
 func (self GUID) String() string {
 	var offsets = [...]int{0, 2, 4, 6, 9, 11, 14, 16, 19, 21, 24, 26, 28, 30, 32, 34}
 	const hexString = "0123456789abcdef"