closer (#132)

algorithm.go

@@ -65,6 +65,7 @@ type TxProfile struct {
 	ReadsQueueSize          int
 	PoolBufferSize          int
 	RxPortalPacingThreshold float64
+	CloseCheckMs            int
 }
 
 func DefaultTxProfile() *TxProfile {
@@ -77,5 +78,6 @@ func DefaultTxProfile() *TxProfile {
 		ReadsQueueSize:          1024,
 		PoolBufferSize:          64 * 1024,
 		RxPortalPacingThreshold: 0.5,
+		CloseCheckMs:            500,
 	}
 }

closer.go

@@ -52,6 +52,57 @@ func (c *closer) timeout() {
 }
 
 func (c *closer) run() {
+	logrus.Info("started")
+	defer logrus.Info("exited")
+
+closeWait:
+	for {
+		select {
+		case rxCloseSeq, ok := <-c.rxCloseSeqIn:
+			if !ok {
+				logrus.Info("!rx close seq")
+				break closeWait
+			}
+			c.rxCloseSeq = rxCloseSeq
+			c.lastEvent = time.Now()
+			logrus.Infof("got rx close seq [%d]", rxCloseSeq)
+			if c.txCloseSeq == notClosed {
+				if err := c.txp.sendClose(c.seq); err != nil {
+					logrus.Errorf("error sending close (%v)", err)
+				}
+			}
+			if c.readyToClose() {
+				break closeWait
+			}
+
+		case txCloseSeq, ok := <-c.txCloseSeqIn:
+			if !ok {
+				logrus.Infof("!tx close seq")
+				break closeWait
+			}
+			c.txCloseSeq = txCloseSeq
+			c.lastEvent = time.Now()
+			logrus.Infof("got tx close seq [%d]", txCloseSeq)
+			if c.readyToClose() {
+				break closeWait
+			}
+
+		case <-time.After(time.Duration(c.txp.alg.Profile().CloseCheckMs) * time.Millisecond):
+			if c.readyToClose() {
+				break closeWait
+			}
+		}
+	}
+	logrus.Info("ready to close")
+
+	c.txp.close()
+	c.rxp.Close()
+
+	if c.closeHook != nil {
+		c.closeHook()
+	}
+
+	logrus.Info("close complete")
 }
 
 func (c *closer) readyToClose() bool {

txportal.go

@@ -24,6 +24,7 @@ type TxPortal struct {
 	alg       TxAlgorithm
 	monitor   *TxMonitor
 	closer    *closer
+	closeSent bool
 	closed    bool
 	pool      *Pool
 }
@@ -127,6 +128,28 @@ func (txp *TxPortal) ack(acks []Ack) error {
 	return nil
 }
 
+func (txp *TxPortal) sendClose(seq *util.Sequence) error {
+	txp.lock.Lock()
+	defer txp.lock.Unlock()
+
+	if !txp.closeSent {
+		wm, err := newClose(seq.Next(), txp.pool)
+		if err != nil {
+			return errors.Wrap(err, "close")
+		}
+		txp.tree.Put(wm.Seq, wm)
+		txp.monitor.add(wm)
+
+		if err := writeWireMessage(wm, txp.transport); err != nil {
+			return errors.Wrap(err, "tx close")
+		}
+		txp.closer.txCloseSeqIn <- wm.Seq
+		txp.closeSent = true
+	}
+
+	return nil
+}
+
 func (txp *TxPortal) close() {
 	txp.closed = true
 	txp.monitor.close()