dilithium.TxPortal.tx. TxAlgorithm tweaks. (#132)

algorithm.go

@@ -6,11 +6,16 @@ import "time"
 // instance.
 //
 type TxAlgorithm interface {
-	Ready(int)
 	Tx(int)
 	Success(int)
 	DuplicateAck()
 	Retransmission(int)
+
+	// ProbeRTT will return bool when the transmitter is due to probe round trip time. It will also record that true
+	// response and will not return true again until the algorithm wants another RTT probe.
+	//
+	ProbeRTT() bool
+
 	UpdateRTT(rttMs int)
 	RetxMs() int
 	RxPortalSize() int
@@ -20,6 +25,7 @@ type TxAlgorithm interface {
 // TxProfile defines all of the configurable values that are requested by a flow control algorithm.
 //
 type TxProfile struct {
+	MaxSegmentSize    int
 	RetxBatchMs       int
 	SendKeepalive     bool
 	ConnectionTimeout time.Duration

txportal.go

@@ -3,7 +3,11 @@ package dilithium
 import (
 	"github.com/emirpasic/gods/trees/btree"
 	"github.com/emirpasic/gods/utils"
+	"github.com/openziti/dilithium/util"
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	"io"
+	"math"
 	"sync"
 	"time"
 )
@@ -44,6 +48,49 @@ func (txp *TxPortal) start() {
 	}
 }
 
+func (txp *TxPortal) tx(p []byte, seq *util.Sequence) (n int, err error) {
+	txp.lock.Lock()
+	defer txp.lock.Unlock()
+
+	if txp.closed {
+		return -1, io.EOF
+	}
+
+	remaining := len(p)
+	n = 0
+	for remaining > 0 {
+		segmentSize := int(math.Min(float64(remaining), float64(txp.alg.Profile().MaxSegmentSize)))
+
+		var rtt *uint16
+		if txp.alg.ProbeRTT() {
+			now := time.Now()
+			rtt = new(uint16)
+			*rtt = uint16(now.UnixNano() / int64(time.Millisecond))
+			segmentSize -= 2
+		}
+
+		txp.alg.Tx(segmentSize)
+
+		wm, err := newData(seq.Next(), rtt, p[n:n+segmentSize], txp.pool)
+		if err != nil {
+			return 0, errors.Wrap(err, "new data")
+		}
+		txp.tree.Put(wm.Seq, wm)
+
+		if err := writeWireMessage(wm, txp.transport); err != nil {
+			return 0, errors.Wrap(err, "tx")
+		}
+		txp.lastTx = time.Now()
+
+		txp.monitor.add(wm)
+
+		n += segmentSize
+		remaining -= segmentSize
+	}
+
+	return n, nil
+}
+
 func (txp *TxPortal) keepaliveSender() {
 	logrus.Info("started")
 	defer logrus.Info("exited")

westworld.go

@@ -11,9 +11,6 @@ func NewWestworldAlgorithm(pf *WestworldProfile, txPortal *TxPortal) TxAlgorithm
 	return &WestworldAlgorithm{pf, txPortal}
 }
 
-func (self *WestworldAlgorithm) Ready(size int) {
-}
-
 func (self *WestworldAlgorithm) Tx(size int) {
 }
 
@@ -26,6 +23,10 @@ func (self *WestworldAlgorithm) DuplicateAck() {
 func (self *WestworldAlgorithm) Retransmission(size int) {
 }
 
+func (self *WestworldAlgorithm) ProbeRTT() bool {
+	return false
+}
+
 func (self *WestworldAlgorithm) UpdateRTT(rttMs int) {
 }