loadtest

Author: makiuchi-d

server/cmd/wsnet2-bot/cmd/load.go

@@ -13,6 +13,8 @@ import (
 
 	"github.com/spf13/cobra"
 
+	"wsnet2/binary"
+	"wsnet2/client"
 	"wsnet2/pb"
 )
 
@@ -89,7 +91,7 @@ func runLoad(ctx context.Context, roomCount, players, watchers int, withWatchabl
 			wg.Done()
 		}(roomCount)
 	}
-	for i := 0; i < roomCount; i++ {
+	for i := range roomCount {
 		time.Sleep(5 * time.Millisecond)
 		go func(i int) {
 			wid := fmt.Sprintf("%v-%v", pid, i)
@@ -157,13 +159,24 @@ func runLoadRoom(ctx context.Context, p, w int, group uint32, lifetime time.Dura
 		wg.Wait()
 	}()
 
-	masterId := fmt.Sprintf("master-%v", cidsuffix)
+	playerIds := make([]string, 0, p)
+	for n := range p {
+		playerIds = append(playerIds, fmt.Sprintf("player-%v-%d", cidsuffix, n))
+	}
+	propKey := "cidsuffix"
+	propVal := binary.MarshalStr8(cidsuffix)
+	query := client.NewQuery().Equal(propKey, propVal)
+	props := binary.MarshalDict(binary.Dict{propKey: propVal})
+	ready := make(chan struct{})
+
+	masterId := playerIds[0]
 	logger.Debugf("%s create %s", logprefix, masterId)
 	room, master, err := createRoom(context.Background(), masterId, &pb.RoomOption{
 		Visible:     true,
 		Joinable:    true,
 		Watchable:   true,
 		SearchGroup: group,
+		PublicProps: props,
 	})
 	if err != nil {
 		return err
@@ -172,45 +185,236 @@ func runLoadRoom(ctx context.Context, p, w int, group uint32, lifetime time.Dura
 
 	wg.Add(1)
 	go func() {
-		rttSum, rttCnt, rttMax, avg := runMaster(ctx, master, lifetime, group, logprefix)
+		rttSum, rttCnt, rttMax, avg := runLoadMaster(ctx, master, lifetime, ready, playerIds, logprefix)
 		logger.Infof("%s end RTT sum=%v cnt=%v avg=%v max=%v", logprefix, rttSum, rttCnt, avg, rttMax)
 		wg.Done()
 	}()
 
 	time.Sleep(time.Second)
 
-	for i := 0; i < p; i++ {
-		playerId := fmt.Sprintf("player-%v-%v", cidsuffix, i)
-
-		logger.Debugf("%s watch %s", logprefix, playerId)
-		_, player, err := joinRoom(context.Background(), playerId, room.Id, nil)
+	for _, playerId := range playerIds[1:] {
+		time.Sleep(5 * time.Millisecond)
+		logger.Debugf("%s join %s", logprefix, playerId)
+		_, player, err := joinRandom(context.Background(), playerId, group, query)
 		if err != nil {
 			return err
 		}
 
 		wg.Add(1)
 		go func() {
-			runPlayer(ctx, player, masterId, logprefix)
+			runLoadPlayer(ctx, player, lifetime, ready, playerIds, logprefix)
 			wg.Done()
 		}()
 	}
 
-	for i := 0; i < w; i++ {
+	for i := range w {
 		watcherId := fmt.Sprintf("watcher-%v-%v", cidsuffix, i)
 
-		logger.Debugf("%s join %s", logprefix, watcherId)
+		logger.Debugf("%s watch %s", logprefix, watcherId)
 		_, watcher, err := watchRoom(context.Background(), watcherId, room.Id, nil)
 		if err != nil {
 			return err
 		}
 
 		wg.Add(1)
 		go func() {
-			runWatcher(ctx, watcher, logprefix)
+			runLoadWatcher(ctx, watcher, logprefix)
 			wg.Done()
 		}()
 	}
 
 	wg.Wait()
 	return nil
 }
+
+// runLoadMaster runs master player for loadtest
+//
+// 1. 全員入ってくるまで待つ
+//   - ready chanで通知
+//
+// 2. メッセージ送信
+//   - size: 300±150b、5%の確率で+1kb
+//   - freq: 15msg / 1sec
+//   - type: broadcastとtargets(全員)を交互に
+//
+// 3. lifetime経過でLeave
+func runLoadMaster(ctx context.Context, conn *client.Connection, lifetime time.Duration, ready chan struct{}, pids []string, logprefix string) (rttSum, rttCnt, rttMax int64, rttAvg float64) {
+	logger.Debugf("%s %s start", logprefix, conn.UserId())
+	sendctx, cancel := context.WithCancel(ctx)
+	var wg sync.WaitGroup
+	wg.Add(3)
+
+	go func() {
+		defer wg.Done()
+
+		var c <-chan time.Time
+		if lifetime > 0 {
+			c = time.After(lifetime)
+		}
+		var msg string
+		select {
+		case <-ctx.Done():
+			msg = "context done"
+		case <-c:
+			msg = "lifetime elapsed"
+		}
+		cancel()
+		logger.Debugf("%v master leave", logprefix)
+		conn.Leave(msg)
+	}()
+
+	go func() {
+		defer wg.Done()
+
+		for range len(pids) {
+			_, ok := waitEvent(conn, lifetime, binary.EvTypeJoined)
+			if !ok {
+				return
+			}
+		}
+		logger.Debugf("%v all players joined", logprefix)
+		close(ready)
+
+		for {
+			ev, ok := waitEvent(conn, lifetime, binary.EvTypePong)
+			if !ok {
+				return
+			}
+			p, _ := binary.UnmarshalEvPongPayload(ev.Payload())
+			rtt := time.Now().UnixMilli() - int64(p.Timestamp)
+			if rtt > RttThreshold {
+				logger.Warnf("%s master rtt=%d", logprefix, rtt)
+			}
+			rttSum += rtt
+			rttCnt++
+			if rttMax < rtt {
+				rttMax = rtt
+			}
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+
+		select {
+		case <-sendctx.Done():
+			return
+		case <-ready:
+		}
+
+		logger.Debugf("%v change room unjoinable", logprefix)
+		conn.Send(binary.MsgTypeRoomProp, binary.MarshalRoomPropPayload(
+			true, false, true, LoadSearchGroup, uint32(len(pids)), 0, nil, nil))
+
+		tick := time.NewTicker(time.Second / 10)
+		defer tick.Stop()
+
+		broadcast := true
+		for {
+			select {
+			case <-sendctx.Done():
+				return
+			case <-tick.C:
+			}
+
+			// 300±150、300周辺が高頻度になるように
+			size := 150 + rand.IntN(101) + rand.IntN(101) + rand.IntN(101)
+			if rand.IntN(20) == 0 { // 5%
+				size += 1000
+			}
+
+			if broadcast {
+				conn.Broadcast(msgBody[:size])
+			} else {
+				conn.ToTargets(msgBody[:size], pids...)
+			}
+			broadcast = !broadcast
+		}
+	}()
+
+	msg, err := conn.Wait(ctx)
+	if err != nil {
+		logger.Errorf("%s %v error: %v", logprefix, conn.UserId(), err)
+	}
+	logger.Debugf("%s %v end: %v", logprefix, conn.UserId(), msg)
+
+	wg.Wait()
+	return rttSum, rttCnt, rttMax, float64(rttSum) / float64(rttCnt)
+}
+
+// runLoadPlayer runs player for loadtest
+//
+// 1. 全員入室を待つ
+//   - ready chanを待つ
+//
+// 2. メッセージ送信
+//   - size: 250±150b、5%の確率で+1kb
+//   - freq: 10msg / 1sec
+//   - type: broadcastとtargets(全員)を交互に
+//
+// 3. 誰かが抜けたら終了
+//   - 正常系ならMasterが最初に抜ける
+func runLoadPlayer(ctx context.Context, conn *client.Connection, lifetime time.Duration, ready chan struct{}, pids []string, logprefix string) {
+	logger.Debugf("%s %s start", logprefix, conn.UserId())
+
+	sendctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	go func() {
+		_, ok := waitEvent(conn, lifetime, binary.EvTypeLeft)
+		cancel()
+		if !ok {
+			return
+		}
+		conn.Leave("leave")
+		discardEvents(conn)
+	}()
+
+	go func() {
+		select {
+		case <-sendctx.Done():
+			return
+		case <-ready:
+		}
+
+		tick := time.NewTicker(time.Second / 10)
+		defer tick.Stop()
+
+		broadcast := true
+		for {
+			select {
+			case <-sendctx.Done():
+				return
+			case <-tick.C:
+			}
+
+			// 200±150
+			size := 50 + rand.IntN(101) + rand.IntN(101) + rand.IntN(101)
+			if rand.IntN(20) == 0 { // 5%
+				size += 1000
+			}
+
+			logger.Debugf("%v msgsize = %v", logprefix, size)
+
+			if broadcast {
+				conn.Broadcast(msgBody[:size])
+			} else {
+				conn.ToTargets(msgBody[:size], pids...)
+			}
+			broadcast = !broadcast
+		}
+	}()
+
+	msg, err := conn.Wait(ctx)
+	if err != nil {
+		logger.Errorf("%s %v error: %v", logprefix, conn.UserId(), err)
+	}
+	logger.Debugf("%s %v end: %v", logprefix, conn.UserId(), msg)
+}
+
+// runLoadWatcher runs watcher for loadtest
+//
+// pattern: same as soak test
+func runLoadWatcher(ctx context.Context, conn *client.Connection, logprefix string) {
+	runSoakWatcher(ctx, conn, logprefix)
+}

server/cmd/wsnet2-bot/cmd/root.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"errors"
+	"math/rand/v2"
 	"net/http"
 	"net/url"
 	"os"
@@ -94,6 +95,10 @@ func init() {
 	rootCmd.PersistentFlags().BoolVarP(&skipTLSVerify, "skip-tls-verify", "s", false, "Skip TLS verify")
 	rootCmd.PersistentFlags().DurationVarP(&timeout, "timeout", "t", 5*time.Second, "Lobby request timeout")
 	rootCmd.PersistentFlags().BoolVarP(&verbose, "verbose", "v", false, "Verbose log output")
+
+	for i := range msgBody {
+		msgBody[i] = byte(rand.IntN(256))
+	}
 }
 
 func setupLogger() error {