Performance improvement

client/worker_event_processing.js

@@ -1,11 +1,10 @@
 let height;
 let width;
-let wsURL;
+let eventURL;
 let portrait;
 let draw; 
 let latestX;
 let latestY;
-// Constants for the maximum values from the WebSocket messages
 const MAX_X_VALUE = 15725;
 const MAX_Y_VALUE = 20966;
 
@@ -16,7 +15,7 @@ onmessage = (event) => {
 		case 'init':
 			height = event.data.height;
 			width = event.data.width;
-			wsURL = event.data.wsURL;
+			eventURL = event.data.eventURL;
 			portrait = event.data.portrait;
 			initiateEventsListener();
 			break;
@@ -33,10 +32,9 @@ onmessage = (event) => {
 
 
 async function initiateEventsListener() {
-	const RETRY_DELAY_MS = 3000; // Delay before retrying the connection (in milliseconds)
-	ws = new WebSocket(wsURL);
+	const eventSource = new EventSource(eventURL);
 	draw = true;
-	ws.onmessage = (event) => {
+	eventSource.onmessage = (event) => {
 		const message = JSON.parse(event.data);
 		if (message.Type === 3) {
 			if (message.Code === 24) {
@@ -70,21 +68,21 @@ async function initiateEventsListener() {
 		}
 	}
 
-	ws.onerror = () => {
+	eventSource.onerror = () => {
 		postMessage({
 			type: 'error',
 			message: "websocket error",
 		});
-		console.error('WebSocket error occurred. Attempting to reconnect...');
+		console.error('EventStrean error occurred. Attempting to reconnect...');
 		//setTimeout(connectWebSocket, 3000); // Reconnect after 3 seconds
 	};
 
-	ws.onclose = () => {
+	eventSource.onclose = () => {
 		postMessage({
 			type: 'error',
 			message: 'closed connection'
 		});
-		console.log('WebSocket connection closed. Attempting to reconnect...');
+		console.log('EventStream connection closed. Attempting to reconnect...');
 		//setTimeout(connectWebSocket, 3000); // Reconnect after 3 seconds
 	};
 }

client/worker_gesture_processing.js

@@ -1,4 +1,3 @@
-let wsURL;
 // Constants for the maximum values from the WebSocket messages
 const SWIPE_DISTANCE = 200;
 
@@ -7,7 +6,6 @@ onmessage = (event) => {
 
 	switch (data.type) {
 		case 'init':
-			wsURL = event.data.wsURL;
 			fetchStream();
 			break;
 		case 'terminate':

client/worker_stream_processing.js

@@ -79,13 +79,19 @@ async function initiateStream() {
 								case 30: // red
 									imageData[offset+3] = 0;
 									break;
-								case 10: // red
+								case 6: // red
 									imageData[offset] = 255;
 									imageData[offset+1] = 0;
 									imageData[offset+2] = 0;
 									imageData[offset+3] = 255;
 									break;
-								case 18: // blue
+								case 8: // red
+									imageData[offset] = 255;
+									imageData[offset+1] = 0;
+									imageData[offset+2] = 0;
+									imageData[offset+3] = 255;
+									break;
+								case 12: // blue
 									imageData[offset] = 0;
 									imageData[offset+1] = 0;
 									imageData[offset+2] = 255;

client/workersHandling.js

@@ -32,20 +32,17 @@ streamWorker.onmessage = (event) => {
 
 
 // Determine the WebSocket protocol based on the current window protocol
-const wsProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
-const wsURL = `${wsProtocol}//${window.location.host}/events`;
-const wsGestureURL = `${wsProtocol}//${window.location.host}/gestures`;
+const eventURL = `/events`;
 // Send the OffscreenCanvas to the worker for initialization
 eventWorker.postMessage({ 
 	type: 'init', 
 	width: width, 
 	height: height, 
 	portrait: portrait,
-	wsURL: wsURL
+	eventURL: eventURL
 });
 gestureWorker.postMessage({ 
 	type: 'init', 
-	wsURL: wsGestureURL
 });
 
 gestureWorker.onmessage = (event) => {

go.mod

@@ -3,15 +3,12 @@ module github.com/owulveryck/goMarkableStream
 go 1.20
 
 require (
-	github.com/gobwas/ws v1.3.1
 	github.com/kelseyhightower/envconfig v1.4.0
 	golang.ngrok.com/ngrok v1.4.1
 )
 
 require (
 	github.com/go-stack/stack v1.8.1 // indirect
-	github.com/gobwas/httphead v0.1.0 // indirect
-	github.com/gobwas/pool v0.2.1 // indirect
 	github.com/inconshreveable/log15 v3.0.0-testing.3+incompatible // indirect
 	github.com/inconshreveable/log15/v3 v3.0.0-testing.5 // indirect
 	github.com/jpillora/backoff v1.0.0 // indirect

go.sum

@@ -1,12 +1,6 @@
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/go-stack/stack v1.8.1 h1:ntEHSVwIt7PNXNpgPmVfMrNhLtgjlmnZha2kOpuRiDw=
 github.com/go-stack/stack v1.8.1/go.mod h1:dcoOX6HbPZSZptuspn9bctJ+N/CnF5gGygcUP3XYfe4=
-github.com/gobwas/httphead v0.1.0 h1:exrUm0f4YX0L7EBwZHuCF4GDp8aJfVeBrlLQrs6NqWU=
-github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM=
-github.com/gobwas/pool v0.2.1 h1:xfeeEhW7pwmX8nuLVlqbzVc7udMDrwetjEv+TZIz1og=
-github.com/gobwas/pool v0.2.1/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw=
-github.com/gobwas/ws v1.3.1 h1:Qi34dfLMWJbiKaNbDVzM9x27nZBjmkaW6i4+Ku+pGVU=
-github.com/gobwas/ws v1.3.1/go.mod h1:hRKAFb8wOxFROYNsT1bqfWnhX+b5MFeJM9r2ZSwg/KY=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
@@ -35,7 +29,6 @@ golang.org/x/crypto v0.9.0 h1:LF6fAI+IutBocDJ2OT0Q1g8plpYljMZ4+lty+dsqw3g=
 golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.8.0 h1:EBmGv8NaZBZTWvrbjNoL6HVt+IVy3QDQpJs7VRIw3tU=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.8.0 h1:n5xxQn2i3PC0yLAbjTpNT85q/Kgzcr2gIoX9OrJUols=

internal/eventhttphandler/pen_handler.go

@@ -2,12 +2,9 @@ package eventhttphandler
 
 import (
 	"encoding/json"
-	"log"
+	"fmt"
 	"net/http"
 
-	"github.com/gobwas/ws"
-	"github.com/gobwas/ws/wsutil"
-
 	"github.com/owulveryck/goMarkableStream/internal/events"
 	"github.com/owulveryck/goMarkableStream/internal/pubsub"
 )
@@ -26,15 +23,14 @@ type EventHandler struct {
 
 // ServeHTTP implements http.Handler
 func (h *EventHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	conn, _, _, err := ws.UpgradeHTTP(r, w)
-	if err != nil {
-		http.Error(w, "cannot upgrade connection "+err.Error(), http.StatusInternalServerError)
-		return
-	}
 	eventC := h.inputEventBus.Subscribe("eventListener")
 	defer func() {
 		h.inputEventBus.Unsubscribe(eventC)
 	}()
+	// Set necessary headers to indicate a stream
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
 
 	for {
 		select {
@@ -53,12 +49,10 @@ func (h *EventHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				http.Error(w, "cannot send json encode the message "+err.Error(), http.StatusInternalServerError)
 				return
 			}
-			// Send the JSON message to the WebSocket client
-			err = wsutil.WriteServerText(conn, jsonMessage)
-			if err != nil {
-				log.Println(err)
-				return
-			}
+			// Send the event
+			fmt.Fprintf(w, "data: %s\n\n", jsonMessage)
+			w.(http.Flusher).Flush() // Ensure client receives the message immediately
+
 		}
 	}
 }

internal/rle/rle.go

@@ -10,7 +10,13 @@ import (
 
 var encodedPool = sync.Pool{
 	New: func() interface{} {
-		return make([]uint8, 0, remarkable.ScreenHeight*remarkable.ScreenWidth)
+		return new(bytes.Buffer)
+	},
+}
+
+var bufferPool = sync.Pool{
+	New: func() any {
+		return make([]byte, 0, remarkable.ScreenHeight*remarkable.ScreenWidth*2)
 	},
 }
 
@@ -38,26 +44,25 @@ func (rlewriter *RLE) Write(data []byte) (int, error) {
 	if length == 0 {
 		return 0, nil
 	}
-	encoded := encodedPool.Get().([]uint8) // Borrow a slice from the pool
-	defer encodedPool.Put(encoded)
+	buf := bufferPool.Get().([]uint8)
+	defer bufferPool.Put(buf)
 
 	current := data[0]
-	count := 0
+	count := uint8(0)
 
-	for _, datum := range data {
+	for i := 0; i < remarkable.ScreenWidth*remarkable.ScreenHeight; i++ {
+		datum := data[i*2]
 		if count < 254 && datum == current {
 			count++
 		} else {
-			encoded = append(encoded, uint8(count))
-			encoded = append(encoded, uint8(current))
+			buf = append(buf, count)
+			buf = append(buf, current)
 			current = datum
 			count = 1
 		}
 	}
+	buf = append(buf, count)
+	buf = append(buf, current)
 
-	encoded = append(encoded, uint8(count))
-	encoded = append(encoded, uint8(current))
-
-	n, err := io.Copy(rlewriter.sub, bytes.NewBuffer(encoded))
-	return int(n), err
+	return rlewriter.sub.Write(buf)
 }

internal/stream/bench_new

@@ -0,0 +1,16 @@
+goos: linux
+goarch: amd64
+pkg: github.com/owulveryck/goMarkableStream/internal/stream
+cpu: 11th Gen Intel(R) Core(TM) i3-1115G4 @ 3.00GHz
+BenchmarkFetchAndSend-4   	      64	  17426490 ns/op	 5398064 B/op	       0 allocs/op
+BenchmarkFetchAndSend-4   	      75	  19568566 ns/op	 9248058 B/op	       0 allocs/op
+BenchmarkFetchAndSend-4   	      76	  17417175 ns/op	 9015990 B/op	       0 allocs/op
+BenchmarkFetchAndSend-4   	      74	  17624161 ns/op	 9401254 B/op	       0 allocs/op
+BenchmarkFetchAndSend-4   	      75	  16933614 ns/op	 9248051 B/op	       0 allocs/op
+BenchmarkFetchAndSend-4   	      73	  18026694 ns/op	 9530045 B/op	       0 allocs/op
+BenchmarkFetchAndSend-4   	      75	  17434884 ns/op	 9136203 B/op	       0 allocs/op
+BenchmarkFetchAndSend-4   	      75	  17827753 ns/op	 9164056 B/op	       0 allocs/op
+BenchmarkFetchAndSend-4   	      70	  18889237 ns/op	 9908626 B/op	       0 allocs/op
+BenchmarkFetchAndSend-4   	      70	  18005032 ns/op	 9908626 B/op	       0 allocs/op
+PASS
+ok  	github.com/owulveryck/goMarkableStream/internal/stream	15.307s

internal/stream/bench_new2

@@ -0,0 +1,16 @@
+goos: linux
+goarch: amd64
+pkg: github.com/owulveryck/goMarkableStream/internal/stream
+cpu: 11th Gen Intel(R) Core(TM) i3-1115G4 @ 3.00GHz
+BenchmarkFetchAndSend-4   	      79	  14389482 ns/op	11893927 B/op	       2 allocs/op
+BenchmarkFetchAndSend-4   	     100	  15035708 ns/op	10108666 B/op	       2 allocs/op
+BenchmarkFetchAndSend-4   	     100	  12881031 ns/op	10087776 B/op	       2 allocs/op
+BenchmarkFetchAndSend-4   	     100	  13603009 ns/op	10087776 B/op	       2 allocs/op
+BenchmarkFetchAndSend-4   	     100	  13045376 ns/op	10114073 B/op	       2 allocs/op
+BenchmarkFetchAndSend-4   	     100	  13512878 ns/op	10087776 B/op	       2 allocs/op
+BenchmarkFetchAndSend-4   	     100	  12939981 ns/op	10087776 B/op	       2 allocs/op
+BenchmarkFetchAndSend-4   	     100	  13988847 ns/op	10087776 B/op	       2 allocs/op
+BenchmarkFetchAndSend-4   	     100	  12817359 ns/op	10140369 B/op	       2 allocs/op
+BenchmarkFetchAndSend-4   	     100	  13959212 ns/op	10087776 B/op	       2 allocs/op
+PASS
+ok  	github.com/owulveryck/goMarkableStream/internal/stream	15.604s

internal/stream/bench_new3

@@ -0,0 +1,16 @@
+goos: linux
+goarch: amd64
+pkg: github.com/owulveryck/goMarkableStream/internal/stream
+cpu: 11th Gen Intel(R) Core(TM) i3-1115G4 @ 3.00GHz
+BenchmarkFetchAndSend-4   	      86	  14446580 ns/op	 7992410 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	     100	  12809843 ns/op	 6820878 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	      97	  12796436 ns/op	 7053368 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	     100	  12317484 ns/op	 6820878 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	     100	  13181974 ns/op	 6820878 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	      97	  12577923 ns/op	 7031832 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	      98	  13408105 ns/op	 6960079 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	     100	  13360119 ns/op	 6926064 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	     100	  12104609 ns/op	 6820878 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	     100	  12607977 ns/op	 6820878 B/op	       1 allocs/op
+PASS
+ok  	github.com/owulveryck/goMarkableStream/internal/stream	14.923s

internal/stream/bench_new4

@@ -0,0 +1,16 @@
+goos: linux
+goarch: amd64
+pkg: github.com/owulveryck/goMarkableStream/internal/stream
+cpu: 11th Gen Intel(R) Core(TM) i3-1115G4 @ 3.00GHz
+BenchmarkFetchAndSend-4   	      96	  13647226 ns/op	 7159864 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	     100	  13476577 ns/op	 6873471 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	      98	  12466472 ns/op	 7121077 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	     100	  12856791 ns/op	 6873471 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	     100	  12318264 ns/op	 6873471 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	     100	  12955263 ns/op	 6873471 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	      93	  12505268 ns/op	 7526392 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	      96	  12427089 ns/op	 7159864 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	      98	  13074079 ns/op	 7013745 B/op	       1 allocs/op
+BenchmarkFetchAndSend-4   	     100	  12493149 ns/op	 6873471 B/op	       1 allocs/op
+PASS
+ok  	github.com/owulveryck/goMarkableStream/internal/stream	15.636s

internal/stream/bench_old

@@ -0,0 +1,7 @@
+goos: linux
+goarch: amd64
+pkg: github.com/owulveryck/goMarkableStream/internal/stream
+cpu: 11th Gen Intel(R) Core(TM) i3-1115G4 @ 3.00GHz
+BenchmarkFetchAndSend-4   	     337	  21152659 ns/op	11381657 B/op	       4 allocs/op
+PASS
+ok  	github.com/owulveryck/goMarkableStream/internal/stream	14.407s

internal/stream/benchfetchandsend_test.go

@@ -0,0 +1,29 @@
+package stream
+
+import (
+	"testing"
+
+	"github.com/owulveryck/goMarkableStream/internal/rle"
+)
+
+func BenchmarkFetchAndSend(b *testing.B) {
+	// Setup: Create a large enough mockReaderAt to test performance
+	width, height := 2872, 2404                  // Example size; adjust based on your needs
+	mockReader := NewMockReaderAt(width, height) // Using the mock from the previous example
+
+	handler := StreamHandler{
+		file:        mockReader,
+		pointerAddr: 0,
+	}
+
+	mockWriter := NewMockResponseWriter()
+
+	rleWriter := rle.NewRLE(mockWriter)
+
+	data := make([]byte, width*height) // Adjust based on your payload size
+
+	b.ResetTimer() // Start timing here
+	for i := 0; i < b.N; i++ {
+		handler.fetchAndSend(rleWriter, data)
+	}
+}

internal/stream/benchresult

@@ -0,0 +1,21 @@
+goos: linux
+goarch: amd64
+pkg: github.com/owulveryck/goMarkableStream/internal/stream
+cpu: 11th Gen Intel(R) Core(TM) i3-1115G4 @ 3.00GHz
+               │ /tmp/bench_old │            bench_new            │
+               │     sec/op     │    sec/op     vs base           │
+FetchAndSend-4     21.15m ± ∞ ¹   20.66m ± ∞ ¹  ~ (p=1.000 n=1) ²
+¹ need >= 6 samples for confidence interval at level 0.95
+² need >= 4 samples to detect a difference at alpha level 0.05
+
+               │ /tmp/bench_old │            bench_new             │
+               │      B/op      │     B/op       vs base           │
+FetchAndSend-4   10.854Mi ± ∞ ¹   8.746Mi ± ∞ ¹  ~ (p=1.000 n=1) ²
+¹ need >= 6 samples for confidence interval at level 0.95
+² need >= 4 samples to detect a difference at alpha level 0.05
+
+               │ /tmp/bench_old │           bench_new            │
+               │   allocs/op    │  allocs/op   vs base           │
+FetchAndSend-4      4.000 ± ∞ ¹   0.000 ± ∞ ¹  ~ (p=1.000 n=1) ²
+¹ need >= 6 samples for confidence interval at level 0.95
+² need >= 4 samples to detect a difference at alpha level 0.05