From c1c54bca1f4a7b93a37ef2f4e0b4e94c51c7d246 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Wed, 28 Jan 2015 00:52:55 +0000 Subject: [PATCH 1/2] old transport --- transport/tcp.go | 240 +++++++++++++++++------------------------------ 1 file changed, 88 insertions(+), 152 deletions(-) diff --git a/transport/tcp.go b/transport/tcp.go index 1a70e4178..5bf113242 100644 --- a/transport/tcp.go +++ b/transport/tcp.go @@ -1,7 +1,6 @@ package transport import ( - "bytes" "encoding/gob" "fmt" "log" @@ -10,7 +9,6 @@ import ( "pilosa/core" "pilosa/db" . "pilosa/util" - "sync" "time" @@ -18,75 +16,86 @@ import ( ) type connection struct { - 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 + transport *TcpTransport + inbox chan *db.Message + outbox chan *db.Message + conn *net.Conn + process *GUID +} + +type newconnection struct { + id *GUID + connection *connection } func init() { gob.Register(GUID{}) } -func newConnection(conn net.Conn, enc *gob.Encoder, dec *gob.Decoder, t *TcpTransport, g *GUID) *connection { - log.Println("NewConnection") - p := new(connection) - p.outbox = make(chan *db.Message, 64) - p.inbox = make(chan *db.Message, 64) - p.conn = conn - p.encoder = enc - p.decoder = dec - p.transport = t - p.process_id = g - p.exit = make(chan int) - return p -} - -func (self *connection) Close() { - close(self.outbox) - close(self.inbox) -} - -func (self *connection) run() { - var wg sync.WaitGroup - wg.Add(1) - go func() { - for { - var mess *db.Message - err := self.decoder.Decode(&mess) +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 decoding message: ", err.Error()) + log.Println("transport/tcp: error getting process, retrying in 2 seconds... ", self.process, err) time.Sleep(2 * time.Second) - self.exit <- 1 - wg.Done() - return - } else { + 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} + }() + } + encoder := gob.NewEncoder(*self.conn) + decoder := gob.NewDecoder(*self.conn) + var exit = make(chan int) + 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 } - } - }() - breakout := true - for breakout { - select { - case message := <-self.outbox: - err := self.encoder.Encode(message) - if err != nil { - log.Println("Failed to Send on Socket Transport") - breakout = false + }() + 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 + } } - case message := <-self.inbox: - self.transport.inbox <- message - case <-self.exit: - breakout = false } } - self.conn.Close() - wg.Wait() - self.transport.removeConnection(self) } type TcpTransport struct { @@ -95,22 +104,7 @@ type TcpTransport struct { inbox chan *db.Message outbox chan db.Envelope connections map[GUID]*connection - mutex sync.Mutex - enc *gob.Encoder - dec *gob.Decoder -} - -func NewTcpTransport(service *core.Service) *TcpTransport { - p := new(TcpTransport) - p.service = service - p.port = config.GetInt("port_tcp") - p.inbox = make(chan *db.Message, 64) - p.outbox = make(chan db.Envelope, 64) - 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 + reg chan *newconnection } func (self *TcpTransport) Run() { @@ -118,14 +112,16 @@ func (self *TcpTransport) Run() { go self.listen() for { select { - case env := <-self.outbox: //transport outbox - con, need := self.getConnection(env.Host) - if need { - con = self.connectRemotePeer(env.Host) - } - if con != nil { - con.outbox <- env.Message + case env := <-self.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} + go con.manage() + self.connections[*env.Host] = con } + con.outbox <- env.Message + case nc := <-self.reg: + self.connections[*nc.id] = nc.connection } } } @@ -143,90 +139,26 @@ func (self *TcpTransport) listen() { time.Sleep(2 * time.Second) continue } - self.addPeer(conn) + go self.manage(&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...") - time.Sleep(2 * time.Second) - 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(conn *net.Conn) { + con := &connection{self, make(chan *db.Message, 1024), make(chan *db.Message, 1024), conn, nil} + con.manage() } func (self *TcpTransport) Close() { log.Println("Shutting down TCP transport") } -func (self *TcpTransport) adjust(in *db.Message) *db.Message { - err := self.enc.Encode(in) - var out db.Message - err = self.dec.Decode(&out) - if err != nil { - log.Println(err) - } - return &out - -} func (self *TcpTransport) Send(message *db.Message, host *GUID) { - if Equal(host, self.service.Id) { - self.inbox <- self.adjust(message) - } else { - envelope := db.Envelope{message, host} - notify.Post("outbox", &envelope) - self.outbox <- envelope - } + envelope := db.Envelope{message, host} + notify.Post("outbox", &envelope) + self.outbox <- envelope } func (self *TcpTransport) Receive() *db.Message { - message := <-self.inbox notify.Post("inbox", message) return message @@ -235,3 +167,7 @@ func (self *TcpTransport) Receive() *db.Message { func (self *TcpTransport) Push(message *db.Message) { self.inbox <- message } + +func NewTcpTransport(service *core.Service) *TcpTransport { + return &TcpTransport{service, config.GetInt("port_tcp"), make(chan *db.Message, 100), make(chan db.Envelope, 100), make(map[GUID]*connection), make(chan *newconnection)} +} From 0862a011328f4b33db3a29cf97585b8aef9f8016 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Wed, 4 Feb 2015 15:15:36 +0000 Subject: [PATCH 2/2] fix for waiting --- core/remotebits.go | 19 +++- transport/tcp.go | 240 ++++++++++++++++++++++++++++----------------- 2 files changed, 166 insertions(+), 93 deletions(-) diff --git a/core/remotebits.go b/core/remotebits.go index 6ac184430..5b3e130c3 100644 --- a/core/remotebits.go +++ b/core/remotebits.go @@ -9,7 +9,7 @@ import ( ) type RemoteSetBit struct { - requests []util.GUID + requests []remote_task cluster map[*util.GUID][]BitmapRequestItem service *Service } @@ -42,7 +42,7 @@ func NewRemoteSetBit(s *Service) *RemoteSetBit { } func (self *RemoteSetBit) Request() { - self.requests = make([]util.GUID, 0) + self.requests = make([]remote_task, 0) source_process, _ := self.service.GetProcess() for process, request := range self.cluster { random_id := util.RandomUUID() @@ -53,17 +53,26 @@ func (self *RemoteSetBit) Request() { QueryId: random_id, DestProcessId: *process, } - self.requests = append(self.requests, random_id) + wait := len(request) / 100 + if wait < 10 { + wait = 10 + } + self.requests = append(self.requests, remote_task{random_id, wait}) self.service.Transport.Send(msg, process) } } +type remote_task struct { + id util.GUID + wait_time int +} + func (self *RemoteSetBit) MergeResults(local_results []SBResult) []SBResult { answers := make(chan []SBResult) for _, task := range self.requests { - go func(id util.GUID) { - value, err := self.service.Hold.Get(&id, 10) //eiher need to be the frame process or the handler process? + go func(task remote_task) { + value, err := self.service.Hold.Get(&task.id, task.wait_time) //eiher need to be the frame process or the handler process? if value == nil { log.Println("Bad RemoteSetBit Result:", err) empty := make([]SBResult, 0, 0) diff --git a/transport/tcp.go b/transport/tcp.go index 5bf113242..1a70e4178 100644 --- a/transport/tcp.go +++ b/transport/tcp.go @@ -1,6 +1,7 @@ package transport import ( + "bytes" "encoding/gob" "fmt" "log" @@ -9,6 +10,7 @@ import ( "pilosa/core" "pilosa/db" . "pilosa/util" + "sync" "time" @@ -16,86 +18,75 @@ import ( ) type connection struct { - transport *TcpTransport - inbox chan *db.Message - outbox chan *db.Message - conn *net.Conn - process *GUID -} - -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 (self *connection) manage() { -BeginManageConnection: - for { - if self.conn == nil { - process, err := self.transport.service.ProcessMap.GetProcess(self.process) +func newConnection(conn net.Conn, enc *gob.Encoder, dec *gob.Decoder, t *TcpTransport, g *GUID) *connection { + log.Println("NewConnection") + p := new(connection) + p.outbox = make(chan *db.Message, 64) + p.inbox = make(chan *db.Message, 64) + p.conn = conn + p.encoder = enc + p.decoder = dec + p.transport = t + p.process_id = g + p.exit = make(chan int) + return p +} + +func (self *connection) Close() { + close(self.outbox) + close(self.inbox) +} + +func (self *connection) run() { + var wg sync.WaitGroup + wg.Add(1) + go func() { + for { + var mess *db.Message + err := self.decoder.Decode(&mess) if err != nil { - log.Println("transport/tcp: error getting process, retrying in 2 seconds... ", self.process, err) + log.Println("transport/tcp: error decoding message: ", err.Error()) 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} - }() - } - encoder := gob.NewEncoder(*self.conn) - decoder := gob.NewDecoder(*self.conn) - var exit = make(chan int) - 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.exit <- 1 + wg.Done() + return + } else { self.inbox <- mess } - }() - 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 - } + } + }() + breakout := true + for breakout { + select { + case message := <-self.outbox: + err := self.encoder.Encode(message) + if err != nil { + log.Println("Failed to Send on Socket Transport") + breakout = false } + case message := <-self.inbox: + self.transport.inbox <- message + case <-self.exit: + breakout = false } } + self.conn.Close() + wg.Wait() + self.transport.removeConnection(self) } type TcpTransport struct { @@ -104,7 +95,22 @@ type TcpTransport struct { inbox chan *db.Message outbox chan db.Envelope connections map[GUID]*connection - reg chan *newconnection + mutex sync.Mutex + enc *gob.Encoder + dec *gob.Decoder +} + +func NewTcpTransport(service *core.Service) *TcpTransport { + p := new(TcpTransport) + p.service = service + p.port = config.GetInt("port_tcp") + p.inbox = make(chan *db.Message, 64) + p.outbox = make(chan db.Envelope, 64) + 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 } func (self *TcpTransport) Run() { @@ -112,16 +118,14 @@ func (self *TcpTransport) Run() { go self.listen() for { select { - case env := <-self.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} - go con.manage() - self.connections[*env.Host] = con + case env := <-self.outbox: //transport outbox + con, need := self.getConnection(env.Host) + if need { + con = self.connectRemotePeer(env.Host) + } + if con != nil { + con.outbox <- env.Message } - con.outbox <- env.Message - case nc := <-self.reg: - self.connections[*nc.id] = nc.connection } } } @@ -139,26 +143,90 @@ 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...") + time.Sleep(2 * time.Second) + 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(conn *net.Conn) { - con := &connection{self, make(chan *db.Message, 1024), make(chan *db.Message, 1024), conn, nil} - con.manage() } func (self *TcpTransport) Close() { log.Println("Shutting down TCP transport") } +func (self *TcpTransport) adjust(in *db.Message) *db.Message { + err := self.enc.Encode(in) + var out db.Message + err = self.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 + if Equal(host, self.service.Id) { + self.inbox <- self.adjust(message) + } else { + envelope := db.Envelope{message, host} + notify.Post("outbox", &envelope) + self.outbox <- envelope + } } func (self *TcpTransport) Receive() *db.Message { + message := <-self.inbox notify.Post("inbox", message) return message @@ -167,7 +235,3 @@ func (self *TcpTransport) Receive() *db.Message { func (self *TcpTransport) Push(message *db.Message) { self.inbox <- message } - -func NewTcpTransport(service *core.Service) *TcpTransport { - return &TcpTransport{service, config.GetInt("port_tcp"), make(chan *db.Message, 100), make(chan db.Envelope, 100), make(map[GUID]*connection), make(chan *newconnection)} -}