feat: update BroadcastMessage to avoid alloc (#40)

* add performance test

* print test system info

* update BroadcastMessage

Author: szpnygo

.gitignore

@@ -1 +1,4 @@
-coverage.out
+coverage.out
+mem.pprof
+cpu.pprof
+trace.out

agent/agent.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/metagogs/gogs/acceptor"
 	"github.com/metagogs/gogs/message"
+	"github.com/metagogs/gogs/packet"
 	"github.com/metagogs/gogs/proto"
 	"github.com/metagogs/gogs/session"
 	"github.com/metagogs/gogs/utils/bytebuffer"
@@ -183,6 +184,13 @@ func (a *Agent) Send(in interface{}, name ...string) error {
 	return nil
 }
 
+func (a *Agent) SendPacket(data *packet.Packet) {
+	select {
+	case a.chSend <- data.ToData():
+	case <-a.chDie:
+	}
+}
+
 func (a *Agent) SendData(data []byte) {
 	select {
 	case a.chSendByte <- data:

app.go

@@ -42,7 +42,6 @@ type App struct {
 	LatencyServer *latency.LatencyServer // 延时服务管理
 	GroupServer   *group.GroupServer     // 组管理
 
-	helper *appHelper //一些便捷操作
 	Config *config.Config
 }
 
@@ -71,11 +70,6 @@ func NewApp(config *config.Config) *App {
 		MessageServer: appBuidler.messageServer,
 		webServer:     appBuidler.webServer,
 	}
-	message.DefaultMessageServer = app.MessageServer
-
-	//初始化便捷服务
-	helper := newAppHelper(app)
-	app.helper = helper
 
 	system.RegisterSystemComponent(app.MessageServer, NewNetworkComponent(app))
 
@@ -121,10 +115,6 @@ func (app *App) RegisterWebHandler(port int, f func(gin *gin.Engine)) {
 	app.webServer.RegisterWebHandler(port, f)
 }
 
-func (app *App) Helper() *appHelper {
-	return app.helper
-}
-
 func (app *App) Start() {
 	deploymentFlag := flag.Bool("deployment", false, "deployment mode")
 	deploymentScv := flag.Bool("svc", false, "use the k8s svc")

builder.go

@@ -40,7 +40,7 @@ func NewBuilder(config *config.Config) *builder {
 func newSessionPool(config *config.Config) session.SessionPool {
 	// session池管理
 	appSessionPool := session.NewSessionPool(config)
-	session.DefaultSessionPool = appSessionPool
+	DefaultSessionPool = appSessionPool
 	return appSessionPool
 }
 
@@ -63,7 +63,9 @@ func newAdminServer(config *config.Config, sessionPool session.SessionPool, code
 }
 
 func newMessageServer(config *config.Config, codecHelper *codec.CodecHelper, dispatchServer *dispatch.DispatchServer) *message.MessageServer {
-	return message.NewMessageServer(codecHelper, dispatchServer)
+	messageServer := message.NewMessageServer(codecHelper, dispatchServer)
+	DefaultMessageServer = messageServer
+	return messageServer
 }
 
 func newWebServer(config *config.Config) *webserver.WebServer {

e2e/e2e_test.go

@@ -5,19 +5,19 @@ import (
 	"time"
 
 	"github.com/gorilla/websocket"
+	"github.com/metagogs/gogs"
 	"github.com/metagogs/gogs/config"
 	"github.com/metagogs/gogs/e2e/testdata"
 	"github.com/metagogs/gogs/e2e/testdata/fakeinternal/logic/baseworld"
 	"github.com/metagogs/gogs/e2e/testdata/game"
 	"github.com/metagogs/gogs/global"
-	"github.com/metagogs/gogs/session"
 	"github.com/stretchr/testify/assert"
 )
 
 var (
 	defaultConfig = config.NewConfig("default.yaml")
-	testClient    *client
-	testClient2   *client
+	testClient    *TestClient
+	testClient2   *TestClient
 	uid           string
 	uid2          string
 )
@@ -38,10 +38,10 @@ func TestSendConnectWS(t *testing.T) {
 	// wait for started
 	<-started
 	t.Log("websocket testing")
-	clients := []*client{}
+	clients := []*TestClient{}
 	// try to make 10 websocket client
 	for i := 0; i < 10; i++ {
-		client, err := newWSClinet("ws://127.0.0.1:8888/base")
+		client, err := NewWSClinet("ws://127.0.0.1:8888/base")
 		assert.Nil(t, err)
 		clients = append(clients, client)
 		defer client.Close()
@@ -73,22 +73,22 @@ func TestSendConnectWS(t *testing.T) {
 	var err error
 	t.Log("websocket message testing")
 	t.Log("user1 login")
-	testClient, err = newWSClinet("ws://127.0.0.1:8888/base")
+	testClient, err = NewWSClinet("ws://127.0.0.1:8888/base")
 	assert.Nil(t, err)
 	go testClient.Start(t)
 
 	<-time.After(1 * time.Second)
 
 	// the session pool should only have one session
-	sessions := session.ListSessions()
+	sessions := gogs.ListSessions()
 	assert.Equal(t, 1, len(sessions))
 
 	//test send data
 	sessions[0].SendData([]byte("hello world"))
 	select {
 	case <-time.After(1 * time.Second):
 		t.Fatal("server get the data timeout")
-	case msg := <-testClient.datas:
+	case msg := <-testClient.Datas:
 		assert.Equal(t, "hello world", string(msg))
 	}
 
@@ -99,14 +99,14 @@ func TestSendConnectWS(t *testing.T) {
 	joinWorld(t, testClient)
 
 	t.Log("user2 login")
-	testClient2, err = newWSClinet("ws://127.0.0.1:8888/base")
+	testClient2, err = NewWSClinet("ws://127.0.0.1:8888/base")
 	assert.Nil(t, err)
 	go testClient2.Start(t)
 
 	<-time.After(1 * time.Second)
 
 	// with the two users logined, we should have to sessions
-	sessions = session.ListSessions()
+	sessions = gogs.ListSessions()
 	assert.Equal(t, 2, len(sessions))
 
 	uid2 = userLogin(t, "e2e2")
@@ -121,7 +121,7 @@ func TestSendConnectWS(t *testing.T) {
 	select {
 	case <-time.After(1 * time.Second):
 		t.Fatal("client get the JoinWorldNotify message timeout")
-	case msg := <-testClient.datas:
+	case msg := <-testClient.Datas:
 		t.Log("client get the JoinWorldNotify message")
 		p, err := testdata.TestApp.MessageServer.DecodeMessage(msg)
 		assert.Nil(t, err)
@@ -147,7 +147,7 @@ func TestSendConnectWS(t *testing.T) {
 	select {
 	case <-time.After(1 * time.Second):
 		t.Fatal("client get t he UpdateUserInWorld message timeout")
-	case msg := <-testClient2.datas:
+	case msg := <-testClient2.Datas:
 		t.Log("client get the UpdateUserInWorld message")
 		p, err := testdata.TestApp.MessageServer.DecodeMessage(msg)
 		assert.Nil(t, err)
@@ -165,7 +165,7 @@ func TestSendConnectWS(t *testing.T) {
 
 // bindUser when we create a client, we should send the BindUser message to th server
 // to bind the user to the connection. The servet should send the BindUserSuccess message to the client
-func bindUser(t *testing.T, sendClient *client, id string) {
+func bindUser(t *testing.T, sendClient *TestClient, id string) {
 	t.Helper()
 	t.Log("client send BindUser message")
 	err := sendClient.WriteMessage(websocket.BinaryMessage, encodeMessage(t, &game.BindUser{
@@ -183,7 +183,7 @@ func bindUser(t *testing.T, sendClient *client, id string) {
 	select {
 	case <-time.After(1 * time.Second):
 		t.Fatal("client get the BindSuccess message timeout")
-	case msg := <-sendClient.datas:
+	case msg := <-sendClient.Datas:
 		t.Log("client get the BindSuccess message success")
 		p, err := testdata.TestApp.MessageServer.DecodeMessage(msg)
 		assert.Nil(t, err)
@@ -194,7 +194,7 @@ func bindUser(t *testing.T, sendClient *client, id string) {
 
 // joinWorld user can send the JoinWorld message to the server to join the world
 // the server should send the JoinWorldSuccess message to the client
-func joinWorld(t *testing.T, sendClient *client) {
+func joinWorld(t *testing.T, sendClient *TestClient) {
 	t.Helper()
 	t.Log("client send JoinWorld message")
 	err := sendClient.WriteMessage(websocket.BinaryMessage, encodeMessage(t, &game.JoinWorld{
@@ -211,7 +211,7 @@ func joinWorld(t *testing.T, sendClient *client) {
 	select {
 	case <-time.After(1 * time.Second):
 		t.Fatal("client get the JoinWorldSuccess message timeout")
-	case msg := <-sendClient.datas:
+	case msg := <-sendClient.Datas:
 		t.Log("client get the JoinWorldSuccess message success")
 		p, err := testdata.TestApp.MessageServer.DecodeMessage(msg)
 		assert.Nil(t, err)

e2e/helper.go

@@ -16,50 +16,59 @@ import (
 	"github.com/tidwall/gjson"
 )
 
-func startServer(t *testing.T, config *config.Config) (context.CancelFunc, chan int) { //nolint
+func startServer(t *testing.T, config *config.Config) (context.CancelFunc, chan struct{}) { //nolint
 	defer func() {
 		if err := recover(); err != nil {
 			t.Fatal("start error", err)
 		}
 	}()
-	startTest := make(chan int)
+	startTest := make(chan struct{})
 	config.StaredCallback = func() {
-		start := time.After(1 * time.Second)
-		<-start
-		startTest <- 1
+		<-time.After(1 * time.Second)
+		startTest <- struct{}{}
 	}
 	t.Helper()
 	ctx, cancel := context.WithCancel(context.Background())
 	go testdata.StartServer(ctx, config)
 	return cancel, startTest
 }
 
-type client struct {
+type TestClient struct {
 	*websocket.Conn
-	datas chan []byte
+	Datas chan []byte
 }
 
-func (c *client) Start(t *testing.T) {
+func (c *TestClient) Start(t *testing.T) {
 	t.Helper()
 	for {
 		_, data, err := c.ReadMessage()
 		if err != nil {
 			break
 		}
-		c.datas <- data
+		c.Datas <- data
+
+	}
+}
+func (c *TestClient) Start2() {
+	for {
+		_, data, err := c.ReadMessage()
+		if err != nil {
+			break
+		}
+		c.Datas <- data
 
 	}
 }
 
-func newWSClinet(address string) (*client, error) { //nolint
+func NewWSClinet(address string) (*TestClient, error) { //nolint
 	conn, _, err := websocket.DefaultDialer.Dial(address, nil) //nolint
 	if err != nil {
 		return nil, err
 	}
 
-	return &client{
+	return &TestClient{
 		Conn:  conn,
-		datas: make(chan []byte),
+		Datas: make(chan []byte),
 	}, nil
 
 }