From 3189f40da122348dbc9c1d5651156da64954ecd2 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Thu, 18 Dec 2014 21:46:47 +0000 Subject: [PATCH] added some connection cleanup code for tcp/transport and support for local connections --- core/http.go | 4 +- executor/executor.go | 4 +- index/fragment_container.go | 10 -- transport/tcp.go | 184 +++++++++++++++++++++++------------- util/id.go | 8 ++ 5 files changed, 132 insertions(+), 78 deletions(-) diff --git a/core/http.go b/core/http.go index 3563e8a08..b26c04c2e 100644 --- a/core/http.go +++ b/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 } diff --git a/executor/executor.go b/executor/executor.go index d6080971d..f99f0540b 100644 --- a/executor/executor.go +++ b/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)) } } diff --git a/index/fragment_container.go b/index/fragment_container.go index 4d0759ff1..0766703e3 100644 --- a/index/fragment_container.go +++ b/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() } } diff --git a/transport/tcp.go b/transport/tcp.go index 5bf113242..33f9030d9 100644 --- a/transport/tcp.go +++ b/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 { diff --git a/util/id.go b/util/id.go index a76908198..8a3bb41bf 100644 --- a/util/id.go +++ b/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"