Connection precache

cmd/restreamer/restreamer.go

@@ -159,7 +159,7 @@ func main() {
 
 			auth := auth.NewAuthenticator(streamdef.Authentication, config.UserList)
 
-			streamer := streaming.NewStreamer(streamdef.Serve, config.OutputBuffer, controller, auth)
+			streamer := streaming.NewStreamer(streamdef.Serve, config.OutputBuffer, config.CacheSize, controller, auth)
 			streamer.SetCollector(reg)
 			streamer.SetNotifier(queue)
 

configuration/config.go

@@ -137,6 +137,9 @@ type Configuration struct {
 	Resources []Resource `json:"resources"`
 	// Notifications defines event callbacks.
 	Notifications []Notification `json:"notifications"`
+	// CacheSize is the number of packets to precache to boost client stream
+	// startup time.
+	CacheSize uint `json:"cachesize"`
 }
 
 // DefaultConfiguration creates and returns a configuration object

streaming/streamer.go

@@ -161,6 +161,8 @@ type Streamer struct {
 	auth auth.Authenticator
 	// promCounter allows enabling/disabling Prometheus packet metrics.
 	promCounter bool
+	// cacheSize is the number of bytes to keep in precache
+	cacheSize int
 }
 
 // ConnectionBroker represents a policy handler for new connections.
@@ -178,9 +180,10 @@ type ConnectionBroker interface {
 // NewStreamer creates a new packet streamer.
 // queue is an input packet queue.
 // qsize is the length of each connection's queue (in packets).
+// cachesize is the size of the precache buffer, in number of packets
 // broker handles policy enforcement
 // stats is a statistics collector object.
-func NewStreamer(name string, qsize uint, broker ConnectionBroker, auth auth.Authenticator) *Streamer {
+func NewStreamer(name string, cachesize uint, qsize uint, broker ConnectionBroker, auth auth.Authenticator) *Streamer {
 	streamer := &Streamer{
 		name:      name,
 		broker:    broker,
@@ -189,6 +192,7 @@ func NewStreamer(name string, qsize uint, broker ConnectionBroker, auth auth.Aut
 		stats:     &metrics.DummyCollector{},
 		request:   make(chan *ConnectionRequest),
 		auth:      auth,
+		cacheSize: int(cachesize) * protocol.MpegTsPacketSize,
 	}
 	// start the command eater
 	go streamer.eatCommands()
@@ -248,9 +252,10 @@ func (streamer *Streamer) eatCommands() {
 // This routine will block; you should run it asynchronously like this:
 //
 // queue := make(chan protocol.MpegTsPacket, inputQueueSize)
-// go func() {
-//   log.Fatal(streamer.Stream(queue))
-// }
+//
+//	go func() {
+//	  log.Fatal(streamer.Stream(queue))
+//	}
 //
 // or simply:
 //
@@ -271,6 +276,9 @@ func (streamer *Streamer) Stream(queue <-chan protocol.MpegTsPacket) error {
 		Command: streamerCommandStart,
 	}
 
+	// prepare the precache buffer
+	precache := util.CreateSlidingWindow(streamer.cacheSize)
+
 	logger.Logkv(
 		"event", eventStreamerStart,
 		"message", "Starting streaming",
@@ -286,7 +294,10 @@ func (streamer *Streamer) Stream(queue <-chan protocol.MpegTsPacket) error {
 				//log.Printf("Got packet (length %d):\n%s\n", len(packet), hex.Dump(packet))
 				//log.Printf("Got packet (length %d)\n", len(packet))
 
+				precache.Put(packet)
+
 				for conn := range pool {
+
 					select {
 					case conn.Queue <- packet:
 						// packet distributed, done
@@ -338,6 +349,9 @@ func (streamer *Streamer) Stream(queue <-chan protocol.MpegTsPacket) error {
 					)
 					pool[request.Connection] = true
 					request.Ok = true
+					// write precached data
+					// TODO maybe don't write this directly, use the queue?
+					request.Connection.writer.Write(precache.Get())
 				} else {
 					logger.Logkv(
 						"event", eventStreamerError,

util/window.go

@@ -0,0 +1,58 @@
+/* Copyright (c) 2018 Gregor Riepl
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package util
+
+import (
+	"sync"
+)
+
+// SlidingWindow implements a buffer that continuously overwrites old data.
+// A single fetch function that copies the whole window is provided.
+//
+// Note that the implementation uses a variable buffer and may not execute
+// in deterministic time.
+//
+// All operations are thread-safe.
+type SlidingWindow struct {
+	window  []byte
+	lock    sync.RWMutex
+}
+
+// SlidingWindow creates a sliding window buffer with a fixed size.
+// Note that the buffer is pre-filled with 0s.
+func CreateSlidingWindow(size int) *SlidingWindow {
+	return &SlidingWindow{
+		window: make([]byte, size),
+	}
+}
+
+// Put copies the contents of data into the sliding window buffer.
+// If data is longer than the buffer, the head will be cut off until it fits.
+func (w *SlidingWindow) Put(data []byte) {
+	w.lock.Lock()
+	defer w.lock.Unlock()
+	w.window = append(w.window, data...)[len(data):]
+}
+
+// Get returns the contents of the sliding window buffer.
+// No copying is performed, the return value is simply a slice of th e buffer.
+// Take care not to modify the contents.
+func (w *SlidingWindow) Get() []byte {
+	w.lock.RLock()
+	defer w.lock.RUnlock()
+	return w.window
+}

util/window_test.go

@@ -0,0 +1,79 @@
+/* Copyright (c) 2018 Gregor Riepl
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package util
+
+import (
+	"testing"
+	"bytes"
+	"encoding/hex"
+)
+
+func TestSlidingWindow01(t *testing.T) {
+	w := CreateSlidingWindow(10)
+	c := []byte{0,1,2,3}
+	w.Put(c)
+	r := w.Get()
+	x := []byte{0,0,0,0,0,0,0,1,2,3}
+	if bytes.Compare(r, x) != 0 {
+		t.Errorf("t01: smaller-than-capacity buffer did not compare to padded value:\n%s", hex.Dump(r))
+	}
+}
+
+func TestSlidingWindow02(t *testing.T) {
+	w := CreateSlidingWindow(4)
+	c := []byte{0,1,2,3}
+	w.Put(c)
+	r := w.Get()
+	x := []byte{0,1,2,3}
+	if bytes.Compare(r, x) != 0 {
+		t.Errorf("t01: at-capacity buffer did not compare to the same value:\n%s", hex.Dump(r))
+	}
+}
+
+func TestSlidingWindow03(t *testing.T) {
+	w := CreateSlidingWindow(4)
+	c := []byte{0,1,2,3,4,5}
+	w.Put(c)
+	r := w.Get()
+	x := []byte{2,3,4,5}
+	if bytes.Compare(r, x) != 0 {
+		t.Errorf("t01: over-capacity buffer did not compare to tail:\n%s", hex.Dump(r))
+	}
+}
+
+func BenchmarkSlidingWindowPutSingle100(b *testing.B) {
+	w := CreateSlidingWindow(100)
+	for n := 0; n < b.N; n++ {
+		w.Put([]byte{0xaa})
+	}
+}
+
+func BenchmarkSlidingWindowPutMany100(b *testing.B) {
+	w := CreateSlidingWindow(100)
+	buf := bytes.Repeat([]byte{0xaa}, 50)
+	for n := 0; n < b.N; n++ {
+		w.Put(buf)
+	}
+}
+
+func BenchmarkSlidingWindowPutMany100000(b *testing.B) {
+	w := CreateSlidingWindow(1000000)
+	buf := bytes.Repeat([]byte{0xaa}, 5000)
+	for n := 0; n < b.N; n++ {
+		w.Put(buf)
+	}
+}