clear_vits v1

core/http.go

@@ -167,6 +167,7 @@ func (self *WebService) Run() {
 	} else {
 		mux.HandleFunc("/set_bits", self.HandleSetBit)
 	}
+	mux.HandleFunc("/clear_bits", self.HandleClearBit)
 	//mux.HandleFunc("/set_bits", self.HandleSetBit)
 	mux.HandleFunc("/flush", NewFlusher(flusher))
 	s := &http.Server{
@@ -408,7 +409,14 @@ func init() {
 	gob.Register(SBResult{})
 }
 
+func (self *WebService) HandleClearBit(w http.ResponseWriter, r *http.Request) {
+	self.HandleBit(w, r, false)
+}
 func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) {
+	self.HandleBit(w, r, true)
+}
+
+func (self *WebService) HandleBit(w http.ResponseWriter, r *http.Request, ToSet bool) {
 	if r.Method != "POST" {
 		http.Error(w, "Only POST allowed", http.StatusMethodNotAllowed)
 		return
@@ -467,8 +475,10 @@ func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) {
 		frag, err := database.GetFragmentFromProfile(frame, profile_id)
 		if err != nil {
 			//no fragment
-			self.service.TopologyMapper.MakeFragments(dbs, db.GetSlice(profile_id))
-			time.Sleep(2 * time.Second)
+			if ToSet {
+				self.service.TopologyMapper.MakeFragments(dbs, db.GetSlice(profile_id))
+				time.Sleep(2 * time.Second)
+			}
 			break
 		}
 		isLocal := util.Equal(frag.GetProcessId(), self.service.Id)
@@ -476,31 +486,35 @@ func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) {
 
 			if isLocal {
 				// The Local Route
-				result, _ = self.service.Index.SetBit(frag.GetId(), bitmap_id, profile_id, filter)
+				if ToSet {
+					result, _ = self.service.Index.SetBit(frag.GetId(), bitmap_id, profile_id, filter)
+				} else {
+					result, _ = self.service.Index.ClearBit(frag.GetId(), bitmap_id, profile_id)
+				}
 				bundle := SBResult{bitmap_id, frame, filter, profile_id, result}
 				results = append(results, bundle)
 			} else {
 
-				remoteSetBit.Add(frag, bitmap_id, profile_id, filter, frame)
+				remoteSetBit.Add(frag, bitmap_id, profile_id, filter, frame, ToSet)
 			}
 
 			//result, err := self.service.Executor.RunPQL(db, pql)
 			//pql := fmt.Sprintf("set(%d, %s, %d, %d)", bitmap_id, frame, filter, profile_id)
 
 			if err != nil {
-				log.Println("Error running set_bit", dbs, frame, profile_id)
+				log.Println("Error running set_bit", dbs, frame, profile_id, ToSet)
 				http.Error(w, err.Error(), http.StatusInternalServerError)
 				return
 			}
-		}
 
+		}
 	}
 	remoteSetBit.Request()
 	results = remoteSetBit.MergeResults(results)
 	encoder := json.NewEncoder(w)
 	err = encoder.Encode(results)
 	if err != nil {
-		log.Println("JSON SetBit ERROR:", err)
+		log.Println("JSON SetBit ERROR:", err, ToSet)
 		//log.Println("Error encoding set_bit", spew.Sdump(results))
 		//http.Error(w, "Error econding set_bit", http.StatusInternalServerError)
 		return

core/remotebits.go

@@ -32,6 +32,7 @@ type BitmapRequestItem struct {
 	Profile_id  uint64
 	Filter      uint64
 	Frame       string
+	SetUnset    bool
 }
 
 func NewRemoteSetBit(s *Service) *RemoteSetBit {
@@ -94,12 +95,12 @@ func (self *RemoteSetBit) MergeResults(local_results []SBResult) []SBResult {
 
 }
 
-func (self *RemoteSetBit) Add(frag *db.Fragment, bitmap_id, profile_id, filter uint64, frame string) {
+func (self *RemoteSetBit) Add(frag *db.Fragment, bitmap_id, profile_id, filter uint64, frame string, SetUnset bool) {
 	x, found := self.cluster[frag.GetProcessId()]
 	if !found {
 		x = make([]BitmapRequestItem, 0)
 	}
-	x = append(x, BitmapRequestItem{frag.GetId(), bitmap_id, profile_id, filter, frame})
+	x = append(x, BitmapRequestItem{frag.GetId(), bitmap_id, profile_id, filter, frame, SetUnset})
 	self.cluster[frag.GetProcessId()] = x
 
 }

dispatch/dispatch.go

@@ -35,8 +35,14 @@ func (self *Dispatch) Run() {
 			self.service.Transport.Send(&response, data.Source)
 		case core.BitsRequest:
 			var results []core.SBResult
+			result := false
+
 			for _, v := range data.Bits {
-				result, _ := self.service.Index.SetBit(v.Fragment_id, v.Bitmap_id, v.Profile_id, uint64(v.Filter))
+				if v.SetUnset {
+					result, _ = self.service.Index.SetBit(v.Fragment_id, v.Bitmap_id, v.Profile_id, uint64(v.Filter))
+				} else {
+					result, _ = self.service.Index.ClearBit(v.Fragment_id, v.Bitmap_id, v.Profile_id)
+				}
 				//jbundle := core.SBResult{v.Bitmap_id, ''v.Frame, v.Filter, v.Profile_id, result}
 				bundle := core.SBResult{v.Bitmap_id, v.Frame, v.Filter, v.Profile_id, result}
 				results = append(results, bundle)

index/timeframe_test.go

@@ -13,11 +13,12 @@ import (
 func getTime(id uint64, s string) {
 	const shortForm = "2006-01-02 15:04"
 	t1, _ := time.Parse(shortForm, s)
+	log.Println("BEFORE")
 
 	for i, v := range GetTimeIds(uint64(id), t1, YMDH) {
 		log.Println(i, v, s)
 	}
-	log.Println()
+	log.Println("AFTER")
 }
 func TestDemo(t *testing.T) {
 	Convey("Test ID", t, func() {

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)}
+}