Separate network-level and application-level measurements (#23)

* Separate network-level and application-level measurements.

This commit adds application-level measurements and namespaces to better separate
the two kind of measurements in the Measurement object.

It is a breaking change and will need a schema update.

* Make application-level byte counters atomic.

* sync.Once and atomic.Int64 do not need to be pointers.

Author: robertodauria

internal/measurer/measurer.go

@@ -96,10 +96,12 @@ func (m *throughput1Measurer) measure(ctx context.Context) {
 	case <-ctx.Done():
 		// NOTHING
 	case m.dstChan <- model.Measurement{
-		ElapsedTime:   time.Since(m.startTime).Microseconds(),
-		BytesSent:     int64(totalWritten) - m.bytesWrittenAtStart,
-		BytesReceived: int64(totalRead) - m.bytesReadAtStart,
-		BBRInfo:       &bbrInfo,
+		ElapsedTime: time.Since(m.startTime).Microseconds(),
+		Network: model.ByteCounters{
+			BytesSent:     int64(totalWritten) - m.bytesWrittenAtStart,
+			BytesReceived: int64(totalRead) - m.bytesReadAtStart,
+		},
+		BBRInfo: &bbrInfo,
 		TCPInfo: &model.TCPInfo{
 			LinuxTCPInfo: tcpInfo,
 			ElapsedTime:  time.Since(m.connInfo.AcceptTime()).Microseconds(),

internal/measurer/measurer_test.go

@@ -35,7 +35,7 @@ func TestNdt8Measurer_Start(t *testing.T) {
 	select {
 	case m := <-mchan:
 		fmt.Println("received measurement")
-		if m.BytesSent != 4 {
+		if m.Network.BytesSent != 4 {
 			t.Errorf("invalid byte counter value")
 		}
 	case <-time.After(1 * time.Second):

pkg/throughput1/model/measurement.go

@@ -24,13 +24,10 @@ type WireMeasurement struct {
 // The Measurement struct contains measurement results. This structure is
 // meant to be serialised as JSON and sent as a textual message.
 type Measurement struct {
-	// BytesSent is the number of bytes sent at the application level by the
-	// party sending this Measurement.
-	BytesSent int64 `json:",omitempty"`
-
-	// BytesReceived is the number of bytes received at the application level
-	// by the party sending this Measurement.
-	BytesReceived int64 `json:",omitempty"`
+	// Application contains the application-level BytesSent/Received pair.
+	Application ByteCounters
+	// Network contains the network-level BytesSent/Received pair.
+	Network ByteCounters
 
 	// ElapsedTime is the time elapsed since the start of the measurement
 	// according to the party sending this Measurement.
@@ -47,6 +44,14 @@ type Measurement struct {
 	TCPInfo *TCPInfo `json:",omitempty"`
 }
 
+type ByteCounters struct {
+	// BytesSent is the number of bytes sent.
+	BytesSent int64 `json:",omitempty"`
+
+	// BytesReceived is the number of bytes received.
+	BytesReceived int64 `json:",omitempty"`
+}
+
 // TCPInfo is an extension to Linux's TCPInfo struct that includes the time
 // elapsed since the connection was accepted.
 type TCPInfo struct {

pkg/throughput1/protocol.go

@@ -10,6 +10,7 @@ import (
 	"net"
 	"net/http"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/gorilla/websocket"
@@ -42,7 +43,10 @@ type Protocol struct {
 	connInfo netx.ConnInfo
 	rnd      *rand.Rand
 	measurer Measurer
-	once     *sync.Once
+	once     sync.Once
+
+	applicationBytesReceived atomic.Int64
+	applicationBytesSent     atomic.Int64
 }
 
 // New returns a new Protocol with the specified connection and every other
@@ -54,7 +58,6 @@ func New(conn *websocket.Conn) *Protocol {
 		// Seed randomness source with the current time.
 		rnd:      rand.New(rand.NewSource(time.Now().UnixMilli())),
 		measurer: &DefaultMeasurer{},
-		once:     &sync.Once{},
 	}
 }
 
@@ -138,7 +141,8 @@ func (p *Protocol) senderReceiverLoop(ctx context.Context,
 }
 
 // receiver reads from the connection until NextReader fails. It returns
-// the measurements received over the provided channel.
+// the measurements received over the provided channel and updates the sent and
+// received byte counters as needed.
 func (p *Protocol) receiver(ctx context.Context,
 	results chan<- model.WireMeasurement, errCh chan<- error) {
 	for {
@@ -147,12 +151,22 @@ func (p *Protocol) receiver(ctx context.Context,
 			errCh <- err
 			return
 		}
+		if kind == websocket.BinaryMessage {
+			// Binary messages are discarded after reading their size.
+			size, err := io.Copy(io.Discard, reader)
+			if err != nil {
+				errCh <- err
+				return
+			}
+			p.applicationBytesReceived.Add(size)
+		}
 		if kind == websocket.TextMessage {
 			data, err := io.ReadAll(reader)
 			if err != nil {
 				errCh <- err
 				return
 			}
+			p.applicationBytesReceived.Add(int64(len(data)))
 			var m model.WireMeasurement
 			if err := json.Unmarshal(data, &m); err != nil {
 				errCh <- err
@@ -177,12 +191,27 @@ func (p *Protocol) sendCounterflow(ctx context.Context,
 				wm = p.createWireMeasurement(ctx)
 			})
 			wm.Measurement = m
-			err := p.conn.WriteJSON(wm)
+			wm.Application = model.ByteCounters{
+				BytesSent:     p.applicationBytesSent.Load(),
+				BytesReceived: p.applicationBytesReceived.Load(),
+			}
+			// Encode as JSON separately so we can read the message size before
+			// sending.
+			jsonwm, err := json.Marshal(wm)
+			if err != nil {
+				log.Printf("failed to encode measurement (ctx: %p, err: %v)",
+					ctx, err)
+				errCh <- err
+				return
+			}
+			err = p.conn.WriteMessage(websocket.TextMessage, jsonwm)
 			if err != nil {
 				log.Printf("failed to write measurement JSON (ctx: %p, err: %v)", ctx, err)
 				errCh <- err
 				return
 			}
+			p.applicationBytesSent.Add(int64(len(jsonwm)))
+
 			// This send is non-blocking in case there is no one to read the
 			// Measurement message and the channel's buffer is full.
 			select {
@@ -195,7 +224,6 @@ func (p *Protocol) sendCounterflow(ctx context.Context,
 
 func (p *Protocol) sender(ctx context.Context, measurerCh <-chan model.Measurement,
 	results chan<- model.WireMeasurement, errCh chan<- error) {
-	ci := netx.ToConnInfo(p.conn.UnderlyingConn())
 	size := spec.MinMessageSize
 	message, err := p.makePreparedMessage(size)
 	if err != nil {
@@ -219,12 +247,27 @@ func (p *Protocol) sender(ctx context.Context, measurerCh <-chan model.Measureme
 				wm = p.createWireMeasurement(ctx)
 			})
 			wm.Measurement = m
-			err = p.conn.WriteJSON(wm)
+			wm.Application = model.ByteCounters{
+				BytesReceived: p.applicationBytesReceived.Load(),
+				BytesSent:     p.applicationBytesSent.Load(),
+			}
+			// Encode as JSON separately so we can read the message size before
+			// sending.
+			jsonwm, err := json.Marshal(wm)
+			if err != nil {
+				log.Printf("failed to encode measurement (ctx: %p, err: %v)",
+					ctx, err)
+				errCh <- err
+				return
+			}
+			err = p.conn.WriteMessage(websocket.TextMessage, jsonwm)
 			if err != nil {
 				log.Printf("failed to write measurement JSON (ctx: %p, err: %v)", ctx, err)
 				errCh <- err
 				return
 			}
+			p.applicationBytesSent.Add(int64(len(jsonwm)))
+
 			// This send is non-blocking in case there is no one to read the
 			// Measurement message and the channel's buffer is full.
 			select {
@@ -238,14 +281,14 @@ func (p *Protocol) sender(ctx context.Context, measurerCh <-chan model.Measureme
 				errCh <- err
 				return
 			}
+			p.applicationBytesSent.Add(int64(size))
 
 			// Determine whether it's time to scale the message size.
 			if size >= spec.MaxScaledMessageSize {
 				continue
 			}
 
-			_, w := ci.ByteCounters()
-			if uint64(size) > w/spec.ScalingFraction {
+			if size > int(p.applicationBytesSent.Load())/spec.ScalingFraction {
 				continue
 			}
 
@@ -264,11 +307,15 @@ func (p *Protocol) sender(ctx context.Context, measurerCh <-chan model.Measureme
 func (p *Protocol) close(ctx context.Context) {
 	msg := websocket.FormatCloseMessage(
 		websocket.CloseNormalClosure, "Done sending")
+
 	err := p.conn.WriteControl(websocket.CloseMessage, msg, time.Now().Add(time.Second))
 	if err != nil {
 		log.Printf("WriteControl failed (ctx: %p, err: %v)", ctx, err)
 		return
 	}
+	// The closing message is part of the measurement and added to bytesSent.
+	p.applicationBytesSent.Add(int64(len(msg)))
+
 	log.Printf("Close message sent (ctx: %p)", ctx)
 }
 

pkg/throughput1/protocol_test.go

@@ -121,8 +121,10 @@ func TestProtocol_Download(t *testing.T) {
 		case <-context.Background().Done():
 			return
 		case m := <-senderCh:
-			fmt.Printf("senderCh BytesReceived: %d, BytesSent: %d\n", m.BytesReceived, m.BytesSent)
-			fmt.Printf("senderCh Goodput: %f Mb/s\n", float64(m.BytesReceived)/float64(time.Since(start).Microseconds())*8)
+			fmt.Printf("senderCh Network.BytesReceived: %d, Network.BytesSent: %d\n",
+				m.Network.BytesReceived, m.Network.BytesSent)
+			fmt.Printf("senderCh Network throughput: %f Mb/s\n",
+				float64(m.Network.BytesReceived)/float64(time.Since(start).Microseconds())*8)
 		case <-receiverCh:
 
 		case err := <-errCh: