Add benchmark for number of rooms on a user's account

bench_test.go

@@ -0,0 +1,97 @@
+package syncv3
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/matrix-org/sync-v3/sync2"
+	"github.com/matrix-org/sync-v3/sync3"
+	"github.com/matrix-org/sync-v3/testutils"
+)
+
+// The purpose of this benchmark is to ensure that sync v3 responds quickly regardless of how many
+// rooms the user has on their account. The first initial sync to fetch the rooms are ignored for
+// the purposes of the benchmark, only subsequent requests are taken into account. We expect this
+// value to grow slowly (searching more rows in a database, looping more elements in an array) but
+// this should not grow so quickly that it impacts performance or else something has gone terribly
+// wrong.
+func BenchmarkNumRooms(b *testing.B) {
+	testutils.Quiet = true
+	for _, numRooms := range []int{100, 200, 400, 800} {
+		n := numRooms
+		b.Run(fmt.Sprintf("num_rooms_%d", n), func(b *testing.B) {
+			benchNumV2Rooms(n, b)
+		})
+	}
+}
+
+func benchNumV2Rooms(numRooms int, b *testing.B) {
+	// setup code
+	boolFalse := false
+	pqString := testutils.PrepareDBConnectionString()
+	v2 := runTestV2Server(b)
+	v3 := runTestServer(b, v2, pqString)
+	defer v2.close()
+	defer v3.close()
+	allRooms := make([]roomEvents, numRooms)
+	for i := 0; i < len(allRooms); i++ {
+		ts := time.Now().Add(time.Duration(i) * time.Minute)
+		roomName := fmt.Sprintf("My Room %d", i)
+		allRooms[i] = roomEvents{
+			roomID: fmt.Sprintf("!benchNumV2Rooms_%d:localhost", i),
+			name:   roomName,
+			events: append(createRoomState(b, alice, ts), []json.RawMessage{
+				testutils.NewStateEvent(b, "m.room.name", "", alice, map[string]interface{}{"name": roomName}, testutils.WithTimestamp(ts.Add(3*time.Second))),
+				testutils.NewEvent(b, "m.room.message", alice, map[string]interface{}{"body": "A"}, testutils.WithTimestamp(ts.Add(4*time.Second))),
+				testutils.NewEvent(b, "m.room.message", alice, map[string]interface{}{"body": "B"}, testutils.WithTimestamp(ts.Add(5*time.Second))),
+				testutils.NewEvent(b, "m.room.message", alice, map[string]interface{}{"body": "C"}, testutils.WithTimestamp(ts.Add(6*time.Second))),
+			}...),
+		}
+	}
+	v2.addAccount(alice, aliceToken)
+	v2.queueResponse(alice, sync2.SyncResponse{
+		Rooms: sync2.SyncRoomsResponse{
+			Join: v2JoinTimeline(allRooms...),
+		},
+	})
+	// do the initial request
+	v3.mustDoV3Request(b, aliceToken, sync3.Request{
+		Lists: []sync3.RequestList{{
+			Ranges: sync3.SliceRanges{
+				[2]int64{0, 20}, // first few rooms
+			},
+			RoomSubscription: sync3.RoomSubscription{
+				TimelineLimit: 3,
+			},
+		}},
+	})
+
+	b.ResetTimer() // don't count setup code
+
+	// these should all take roughly the same amount of time, regardless of the value of `numRooms`
+	for n := 0; n < b.N; n++ {
+		v3.mustDoV3Request(b, aliceToken, sync3.Request{
+			Lists: []sync3.RequestList{{
+				// always use a fixed range else we will scale O(n) with the number of rooms
+				Ranges: sync3.SliceRanges{
+					[2]int64{0, 20}, // first few rooms
+				},
+				// include a filter to ensure we loop over rooms
+				Filters: &sync3.RequestFilters{
+					IsEncrypted: &boolFalse,
+				},
+				// include a few required state events to force us to query the database
+				// include a few timeline events to force us to query the database
+				RoomSubscription: sync3.RoomSubscription{
+					TimelineLimit: 3,
+					RequiredState: [][2]string{
+						{"m.room.create", ""},
+						{"m.room.member", alice},
+					},
+				},
+			}},
+		})
+	}
+}

state/storage.go

@@ -650,3 +650,7 @@ func (s *Storage) joinedRoomsAfterPositionWithEvents(membershipEvents []Event, u
 
 	return joinedRooms, nil
 }
+
+func (s *Storage) Teardown() {
+	s.accumulator.db.Close()
+}

sync3/handler/handler.go

@@ -84,6 +84,11 @@ func NewSync3Handler(v2Client sync2.Client, postgresDBURI string) (*SyncLiveHand
 	return sh, nil
 }
 
+// used in tests to close postgres connections
+func (h *SyncLiveHandler) Teardown() {
+	h.Storage.Teardown()
+}
+
 func (h *SyncLiveHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	if req.Method != "POST" {
 		w.WriteHeader(http.StatusMethodNotAllowed)

testutils/db.go

@@ -8,11 +8,17 @@ import (
 	"os/user"
 )
 
+var Quiet = false
+
 func createLocalDB(dbName string) string {
-	fmt.Println("Note: tests require a postgres install accessible to the current user")
+	if !Quiet {
+		fmt.Println("Note: tests require a postgres install accessible to the current user")
+	}
 	createDB := exec.Command("createdb", dbName)
-	createDB.Stdout = os.Stdout
-	createDB.Stderr = os.Stderr
+	if !Quiet {
+		createDB.Stdout = os.Stdout
+		createDB.Stderr = os.Stderr
+	}
 	createDB.Run()
 	return dbName
 }
@@ -20,7 +26,9 @@ func createLocalDB(dbName string) string {
 func currentUser() string {
 	user, err := user.Current()
 	if err != nil {
-		fmt.Println("cannot get current user: ", err)
+		if !Quiet {
+			fmt.Println("cannot get current user: ", err)
+		}
 		os.Exit(2)
 	}
 	return user.Username
@@ -61,5 +69,6 @@ func PrepareDBConnectionString() (connStr string) {
 	if err != nil {
 		panic(err)
 	}
+	db.Close()
 	return
 }

testutils/event.go

@@ -10,6 +10,14 @@ import (
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
+// Common functions between testing.T and testing.B
+type TestBenchInterface interface {
+	Fatalf(s string, args ...interface{})
+	Errorf(s string, args ...interface{})
+	Helper()
+	Name() string
+}
+
 var (
 	eventIDCounter = 0
 	eventIDMu      sync.Mutex
@@ -25,7 +33,7 @@ type eventMock struct {
 	Unsigned       interface{} `json:"unsigned,omitempty"`
 }
 
-func generateEventID(t *testing.T) string {
+func generateEventID(t TestBenchInterface) string {
 	eventIDMu.Lock()
 	defer eventIDMu.Unlock()
 	eventIDCounter++
@@ -48,7 +56,7 @@ func WithUnsigned(unsigned interface{}) eventMockModifier {
 	}
 }
 
-func NewStateEvent(t *testing.T, evType, stateKey, sender string, content interface{}, modifiers ...eventMockModifier) json.RawMessage {
+func NewStateEvent(t TestBenchInterface, evType, stateKey, sender string, content interface{}, modifiers ...eventMockModifier) json.RawMessage {
 	t.Helper()
 	e := &eventMock{
 		Type:           evType,
@@ -68,7 +76,7 @@ func NewStateEvent(t *testing.T, evType, stateKey, sender string, content interf
 	return j
 }
 
-func NewEvent(t *testing.T, evType, sender string, content interface{}, modifiers ...eventMockModifier) json.RawMessage {
+func NewEvent(t TestBenchInterface, evType, sender string, content interface{}, modifiers ...eventMockModifier) json.RawMessage {
 	t.Helper()
 	e := &eventMock{
 		Type:           evType,

v3_test.go

@@ -63,7 +63,9 @@ func (s *testV2Server) queueResponse(userID string, resp sync2.SyncResponse) {
 	ch := s.queues[userID]
 	s.mu.Unlock()
 	ch <- resp
-	log.Printf("testV2Server: enqueued v2 response for %s", userID)
+	if !testutils.Quiet {
+		log.Printf("testV2Server: enqueued v2 response for %s", userID)
+	}
 }
 
 // blocks until nextResponse is called with an empty channel (that is, the server has caught up with v2 responses)
@@ -93,13 +95,17 @@ func (s *testV2Server) nextResponse(userID string) *sync2.SyncResponse {
 	}
 	select {
 	case data := <-ch:
-		log.Printf(
-			"testV2Server: nextResponse %s returning data: [invite=%d,join=%d,leave=%d]",
-			userID, len(data.Rooms.Invite), len(data.Rooms.Join), len(data.Rooms.Leave),
-		)
+		if !testutils.Quiet {
+			log.Printf(
+				"testV2Server: nextResponse %s returning data: [invite=%d,join=%d,leave=%d]",
+				userID, len(data.Rooms.Invite), len(data.Rooms.Join), len(data.Rooms.Leave),
+			)
+		}
 		return &data
 	case <-time.After(1 * time.Second):
-		log.Printf("testV2Server: nextResponse %s waited >1s for data, returning null", userID)
+		if !testutils.Quiet {
+			log.Printf("testV2Server: nextResponse %s waited >1s for data, returning null", userID)
+		}
 		return nil
 	}
 }
@@ -114,7 +120,7 @@ func (s *testV2Server) close() {
 	s.srv.Close()
 }
 
-func runTestV2Server(t *testing.T) *testV2Server {
+func runTestV2Server(t testutils.TestBenchInterface) *testV2Server {
 	t.Helper()
 	server := &testV2Server{
 		tokenToUser: make(map[string]string),
@@ -154,11 +160,13 @@ func runTestV2Server(t *testing.T) *testV2Server {
 }
 
 type testV3Server struct {
-	srv *httptest.Server
+	srv     *httptest.Server
+	handler *handler.SyncLiveHandler
 }
 
 func (s *testV3Server) close() {
 	s.srv.Close()
+	s.handler.Teardown()
 }
 
 func (s *testV3Server) restart(t *testing.T, v2 *testV2Server, pq string) {
@@ -170,12 +178,12 @@ func (s *testV3Server) restart(t *testing.T, v2 *testV2Server, pq string) {
 	v2.srv.CloseClientConnections() // kick-over v2 conns
 }
 
-func (s *testV3Server) mustDoV3Request(t *testing.T, token string, reqBody sync3.Request) (respBody *sync3.Response) {
+func (s *testV3Server) mustDoV3Request(t testutils.TestBenchInterface, token string, reqBody sync3.Request) (respBody *sync3.Response) {
 	t.Helper()
 	return s.mustDoV3RequestWithPos(t, token, "", reqBody)
 }
 
-func (s *testV3Server) mustDoV3RequestWithPos(t *testing.T, token string, pos string, reqBody sync3.Request) (respBody *sync3.Response) {
+func (s *testV3Server) mustDoV3RequestWithPos(t testutils.TestBenchInterface, token string, pos string, reqBody sync3.Request) (respBody *sync3.Response) {
 	t.Helper()
 	resp, respBytes, code := s.doV3Request(t, context.Background(), token, pos, reqBody)
 	if code != 200 {
@@ -184,7 +192,7 @@ func (s *testV3Server) mustDoV3RequestWithPos(t *testing.T, token string, pos st
 	return resp
 }
 
-func (s *testV3Server) doV3Request(t *testing.T, ctx context.Context, token string, pos string, reqBody interface{}) (respBody *sync3.Response, respBytes []byte, statusCode int) {
+func (s *testV3Server) doV3Request(t testutils.TestBenchInterface, ctx context.Context, token string, pos string, reqBody interface{}) (respBody *sync3.Response, respBytes []byte, statusCode int) {
 	t.Helper()
 	var body io.Reader
 	switch v := reqBody.(type) {
@@ -229,7 +237,7 @@ func (s *testV3Server) doV3Request(t *testing.T, ctx context.Context, token stri
 	return &r, respBytes, resp.StatusCode
 }
 
-func runTestServer(t *testing.T, v2Server *testV2Server, postgresConnectionString string) *testV3Server {
+func runTestServer(t testutils.TestBenchInterface, v2Server *testV2Server, postgresConnectionString string) *testV3Server {
 	t.Helper()
 	if postgresConnectionString == "" {
 		postgresConnectionString = testutils.PrepareDBConnectionString()
@@ -248,11 +256,12 @@ func runTestServer(t *testing.T, v2Server *testV2Server, postgresConnectionStrin
 	r.Handle("/_matrix/client/unstable/org.matrix.msc3575/sync", h)
 	srv := httptest.NewServer(r)
 	return &testV3Server{
-		srv: srv,
+		srv:     srv,
+		handler: h,
 	}
 }
 
-func createRoomState(t *testing.T, creator string, baseTimestamp time.Time) []json.RawMessage {
+func createRoomState(t testutils.TestBenchInterface, creator string, baseTimestamp time.Time) []json.RawMessage {
 	t.Helper()
 	var pl gomatrixserverlib.PowerLevelContent
 	pl.Defaults()