diff --git a/server.go b/server.go index 78cbeea..ed7bc4a 100644 --- a/server.go +++ b/server.go @@ -39,6 +39,13 @@ type eventBatch struct { events <-chan Event } +// maxEventsPerFlush caps how many batch events the handler encodes between flushes. +// The value is large enough that the per-flush cost is fully amortized for big +// replays, while keeping a deterministic bound on how long the handler can stay away +// from its main select loop (which watches for connection close and MaxConnTime) +// while a batch is streaming. +const maxEventsPerFlush = 512 + // Server manages any number of event-publishing channels and allows subscribers to consume them. // To use it within an HTTP server, create a handler for each channel with Handler(). type Server struct { @@ -137,7 +144,7 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { flusher.Flush() enc := NewEncoder(w, useGzip) - writeEventOrComment := func(ec eventOrComment) bool { + encodeEventOrComment := func(ec eventOrComment) bool { if err := enc.Encode(ec); err != nil { srv.unsubs <- sub if srv.Logger != nil { @@ -145,6 +152,13 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { } return false // if this happens, we'll end the handler early because something's clearly broken } + return true + } + + writeEventOrComment := func(ec eventOrComment) bool { + if !encodeEventOrComment(ec) { + return false + } flusher.Flush() return true } @@ -265,7 +279,15 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { if !ok { // end of batch readBatchCh = nil readMainCh = eventCh - } else if !writeEventOrComment(ev) { + continue + } + encodeOK, batchDone := encodeAvailableBatchEvents(ev, readBatchCh, encodeEventOrComment) + if batchDone { + readBatchCh = nil + readMainCh = eventCh + } + flusher.Flush() + if !encodeOK { break ReadLoop } } @@ -412,6 +434,43 @@ func (srv *Server) markServerClosed() { srv.isClosed = true } +// Encodes ev plus any further events that are immediately available from batch, +// without flushing in between. Batches can contain many events that are already +// available (for example, a replay of a large data set), and flushing the connection +// after every one of them costs an HTTP chunk write per event; the caller flushes +// once after this returns. Encoding stops when the channel goes momentarily quiet, +// the batch ends, or maxEventsPerFlush events have been written. Memory stays +// bounded either way -- the ResponseWriter's buffer writes through to the connection +// as it fills, and a slow client blocks the encode -- but the cap guarantees the +// handler returns to its select loop regularly, so connection shutdown and +// MaxConnTime are still honored while a long batch is streaming. +// +// encodeOK is false if an encode failed (the handler should exit). batchDone is true +// if the batch channel was closed. +func encodeAvailableBatchEvents( + ev eventOrComment, + batch <-chan Event, + encode func(eventOrComment) bool, +) (encodeOK, batchDone bool) { + for encoded := 1; ; encoded++ { + if !encode(ev) { + return false, false + } + if encoded >= maxEventsPerFlush { + return true, false // budget spent; do not pull another event before flushing + } + select { + case next, more := <-batch: + if !more { + return true, true + } + ev = next + default: + return true, false + } + } +} + // Attempts to send an event or comment to the subscription's channel. // // We do not want to block the main Server goroutine, so this is a non-blocking send. If it fails, diff --git a/server_replay_benchmark_test.go b/server_replay_benchmark_test.go new file mode 100644 index 0000000..27ec91f --- /dev/null +++ b/server_replay_benchmark_test.go @@ -0,0 +1,128 @@ +package eventsource + +import ( + "bufio" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +type prerenderedEvent struct { + name string + data string +} + +func (e prerenderedEvent) Id() string { return "" } //nolint:revive +func (e prerenderedEvent) Event() string { return e.name } +func (e prerenderedEvent) Data() string { return e.data } + +// batchRepository replays a fixed set of events through a buffered channel, so every +// event is immediately available to the handler. This models a server-side proxy +// replaying a large pre-rendered data set to a newly connected client. +type batchRepository struct { + events []Event +} + +func (r batchRepository) Replay(channel, id string) chan Event { + out := make(chan Event, len(r.events)) + for _, e := range r.events { + out <- e + } + close(out) + return out +} + +// unbufferedRepository replays the same events through an unbuffered channel fed by a +// goroutine, one rendezvous per event. This models callers (like ld-relay today) that +// stream replay events instead of preloading them. +type unbufferedRepository struct { + events []Event +} + +func (r unbufferedRepository) Replay(channel, id string) chan Event { + out := make(chan Event) + go func() { + defer close(out) + for _, e := range r.events { + out <- e + } + }() + return out +} + +func benchmarkReplayBatch(b *testing.B, numEvents, dataSize int) { + benchmarkReplay(b, numEvents, dataSize, func(events []Event) Repository { + return batchRepository{events: events} + }) +} + +func benchmarkReplayUnbuffered(b *testing.B, numEvents, dataSize int) { + benchmarkReplay(b, numEvents, dataSize, func(events []Event) Repository { + return unbufferedRepository{events: events} + }) +} + +func benchmarkReplay(b *testing.B, numEvents, dataSize int, makeRepo func([]Event) Repository) { + events := make([]Event, 0, numEvents) + data := strings.Repeat("x", dataSize) + for i := 0; i < numEvents; i++ { + events = append(events, prerenderedEvent{name: "put-object", data: data}) + } + + server := NewServer() + server.ReplayAll = true + defer server.Close() + server.Register("test", makeRepo(events)) + + httpServer := httptest.NewServer(server.Handler("test")) + defer httpServer.Close() + + b.SetBytes(int64(numEvents * dataSize)) + b.ResetTimer() + + for i := 0; i < b.N; i++ { + resp, err := http.Get(httpServer.URL) + if err != nil { + b.Fatal(err) + } + scanner := bufio.NewScanner(resp.Body) + count := 0 + for scanner.Scan() { + if strings.HasPrefix(scanner.Text(), "data: ") { + count++ + if count == numEvents { + break + } + } + } + _ = resp.Body.Close() + if count != numEvents { + b.Fatalf("expected %d events, got %d", numEvents, count) + } + } +} + +func BenchmarkReplayBatch(b *testing.B) { + for _, bc := range []struct{ numEvents, dataSize int }{ + {100, 300}, + {1000, 300}, + {5000, 300}, + {5000, 2000}, + } { + b.Run(fmt.Sprintf("%devents_%dB", bc.numEvents, bc.dataSize), func(b *testing.B) { + benchmarkReplayBatch(b, bc.numEvents, bc.dataSize) + }) + } +} + +func BenchmarkReplayUnbuffered(b *testing.B) { + for _, bc := range []struct{ numEvents, dataSize int }{ + {5000, 300}, + } { + b.Run(fmt.Sprintf("%devents_%dB", bc.numEvents, bc.dataSize), func(b *testing.B) { + benchmarkReplayUnbuffered(b, bc.numEvents, bc.dataSize) + }) + } +}