diff --git a/core/http.go b/core/http.go index 47fbb9201..76cddd44c 100644 --- a/core/http.go +++ b/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 { diff --git a/core/remotebits.go b/core/remotebits.go index e0e9c0239..0463b2f3d 100644 --- a/core/remotebits.go +++ b/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 diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index 43bea66c9..8eddf415d 100644 --- a/dispatch/dispatch.go +++ b/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) } diff --git a/transport/tcp.go b/transport/tcp.go index 9a85739ad..7a142be46 100644 --- a/transport/tcp.go +++ b/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