scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-08 20:07:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fixed messaging layer bug

Browse source
This commit is contained in:
Todd Gruben 2015-01-26 19:27:00 +00:00
parent d8c58f57ad
commit e4416a91b8
3 changed files with 19 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
core/http.go
View file

@ -431,6 +431,8 @@ func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) {
}
result := false
remoteSetBit := NewRemoteSetBit(self.service)
var database *db.Database
olddbs := ""
for _, obj := range args {
if obj["profile_id"] == nil {
http.Error(w, "Missing Profile", http.StatusBadRequest)
@ -458,7 +460,10 @@ func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) {
}
t = float64(obj["filter"].(float64))
filter := uint64(t)
database := self.service.Cluster.GetOrCreateDatabase(dbs)
if dbs != olddbs {
database = self.service.Cluster.GetOrCreateDatabase(dbs)
olddbs = dbs
}
frag, err := database.GetFragmentFromProfile(frame, profile_id)
if err != nil {
//no fragment

1
core/remotebits.go
View file

@ -80,6 +80,7 @@ func (self *RemoteSetBit) MergeResults(local_results []SBResult) []SBResult {
local_results = append(local_results, item)
}
}
close(answers)
return local_results
}

21
transport/tcp.go
View file

@ -33,9 +33,10 @@ func init() {
}
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, 2048)
p.inbox = make(chan *db.Message, 2048)
p.outbox = make(chan *db.Message, 64)
p.inbox = make(chan *db.Message, 64)
p.conn = conn
p.encoder = enc
p.decoder = dec
@ -93,7 +94,7 @@ type TcpTransport struct {
port int
inbox chan *db.Message
outbox chan db.Envelope
connections map[*GUID]*connection
connections map[GUID]*connection
mutex sync.Mutex
enc *gob.Encoder
dec *gob.Decoder
@ -103,9 +104,9 @@ func NewTcpTransport(service *core.Service) *TcpTransport {
p := new(TcpTransport)
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.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.
@ -146,11 +147,13 @@ func (self *TcpTransport) listen() {
}
}
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)
@ -180,20 +183,20 @@ func (self *TcpTransport) addPeer(conn net.Conn) {
func (self *TcpTransport) getConnection(guid *GUID) (*connection, bool) {
self.mutex.Lock()
defer self.mutex.Unlock()
con, ok := self.connections[guid]
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
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)
delete(self.connections, *c.process_id)
c.Close()
}