benchmark group bugfix

api/chat.proto

@@ -19,6 +19,8 @@ service Chat {
 
   rpc RoomInfo(ReqRoomInfo) returns (ResRoomInfo);
 
+  rpc RoomDissolve(ReqRoomDissolve) returns (google.protobuf.Empty);
+
   rpc RoomList(ReqRoomList) returns (ResRoomList);
 }
 
@@ -35,7 +37,7 @@ message ReqSend {
 }
 
 message ReqRoomSend {
-  string gid = 1;
+  string Rid = 1;
   string message = 2;
 }
 
@@ -83,6 +85,10 @@ message ResRoomInfo {
   Room room = 1;
 }
 
+message ReqRoomDissolve {
+  string rid = 1;
+}
+
 message ReqRoomList {
 }
 

benchmark/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"flag"
 	"fmt"
 	"github.com/bytedance/sonic"
@@ -16,6 +17,7 @@ import (
 	"math/rand"
 	"net"
 	"net/http"
+	"reflect"
 	"runtime"
 	"sonet/api/gen/auth"
 	"sonet/api/gen/chat"
@@ -23,6 +25,7 @@ import (
 	"sonet/pkg/grpc/discovery"
 	"sonet/pkg/protocol"
 	"sonet/pkg/protocol/deliver"
+	"sonet/pkg/utils/collect"
 	"sonet/pkg/utils/logger"
 	"sonet/pkg/utils/security"
 	"sonet/pkg/utils/shutdown"
@@ -41,12 +44,10 @@ var (
 	mockNetUsers    []*NetUser
 	sendCounter     int64 = 0
 	receiverCounter int64 = 0
-	//wsUrls                = []string{"ws://192.168.110.36:7001", "ws://192.168.110.36:7003"}
 	httpClient      *http.Client
 	etcdClient      *clientv3.Client
 	seqId           int32
-	callbacks     map[int32]func(res any)
-	callbackMutex *sync.Mutex
+	callbacks       *collect.ConcurrentMap[int32, func(res any)]
 	useMsSum        int64
 	useMsAvg        int64
 )
@@ -54,8 +55,11 @@ var (
 func init() {
 	config.InitLogger(false)
 
-	callbacks = make(map[int32]func(res any), 128)
-	callbackMutex = &sync.Mutex{}
+	//callbacks = make(map[int32]func(res any), 128)
+	//callbackMutex = &sync.Mutex{}
+	callbacks = collect.NewConcurrentMap[int32, func(res any)](16, func(k int32) string {
+		return strconv.Itoa(int(k))
+	})
 	httpClient = &http.Client{Timeout: 10 * time.Second}
 	var err error
 	etcdClient, err = clientv3.New(clientv3.Config{
@@ -72,10 +76,10 @@ func init() {
 func main() {
 	runtime.GOMAXPROCS(runtime.NumCPU())
 
-	flag.StringVar(&gatewayHttp, "gateway", "http://124.222.131.236:30830", "http gateway address") // 192.168.110.41:7000
-	flag.StringVar(&benchmarkMode, "mode", "deliver", "benchmark mode: deliver/group")
-	flag.IntVar(&mockUsers, "users", 1000, "mock users")
-	flag.IntVar(&eachUserSend, "send", 10000, "each user send msg count")
+	flag.StringVar(&gatewayHttp, "gateway", "http://192.168.110.41:7000", "http gateway address") // 124.222.131.236:30830
+	flag.StringVar(&benchmarkMode, "mode", "group", "benchmark mode: deliver/group")
+	flag.IntVar(&mockUsers, "users", 100, "mock users")
+	flag.IntVar(&eachUserSend, "send", 10, "each user send msg count")
 	flag.Parse()
 
 	if benchmarkMode == "group" {
@@ -114,10 +118,11 @@ func benchmark() {
 	// go prof.StartPprof(":8888")
 
 	ctx, cancel := context.WithCancel(context.Background())
-	mockUsersConnect(ctx)
 
 	go records(ctx)
 
+	mockUsersConnect(ctx)
+
 	for i := 0; i < mockUsers; i++ {
 		netUser := mockNetUsers[i]
 		go sendBatchChatMessage(ctx, netUser.Uid, netUser.Conn, eachUserSend)
@@ -146,32 +151,60 @@ func benchmarkGroup() {
 	groupId := "9527"
 
 	mockUsersConnect(ctx)
+
+	// create room
+	channel, err := send(mockNetUsers[0].Conn, chat.Chat_ServiceDesc.ServiceName, "RoomCreate", &chat.ReqRoomCreate{Rname: "benchmark"})
+	if err != nil {
+		panic(err)
+		return
+	}
+	// wait result
+	res := <-channel
+	if err, failed := res.(error); failed {
+		panic(err)
+	} else {
+		groupId = res.(*chat.ResRoomCreate).Room.Rid
+	}
+
 	for _, user := range mockNetUsers {
-		// join group
-		err := groupDeliver.GroupJoin(context.Background(), user.Uid, []string{groupId})
+		channel, err := send(user.Conn, chat.Chat_ServiceDesc.ServiceName, "RoomJoin", &chat.ReqRoomJoin{Rid: groupId})
 		if err != nil {
 			panic(err)
 		}
+		// wait result
+		if err, failed := (<-channel).(error); failed {
+			panic(err)
+		}
 	}
 
 	shutdown.AddHook(func() {
-		if err := groupDeliver.GroupDissolve(context.Background(), groupId); err != nil {
-			logger.Error("dissolve group error: ", err)
+		// delete room
+		channel, err := send(mockNetUsers[0].Conn, chat.Chat_ServiceDesc.ServiceName, "RoomDissolve", &chat.ReqRoomDissolve{Rid: groupId})
+		if err != nil {
+			panic(err)
 			return
 		}
+		// wait result
+		if err, failed := (<-channel).(error); failed {
+			panic(err)
+		}
 		logger.Info("test group dissolved")
 	})
 
 	go records(ctx)
 
 	// send group message
-	message := &chat.ChatMessage{Sender: "10001", Content: "hi"}
+	args := &chat.ReqRoomSend{Rid: groupId, Message: "hi"}
 	for i := 0; i < eachUserSend; i++ {
-		err := groupDeliver.DeliverGroup(context.Background(), groupId, message)
+		channel, err := send(mockNetUsers[0].Conn, chat.Chat_ServiceDesc.ServiceName, "RoomSend", args)
 		if err != nil {
-			logger.Error("deliver group error:", err)
+			logger.Error("room send error: ", err)
+			continue
+		}
+		// wait result
+		if err, failed := (<-channel).(error); failed {
+			logger.Error("room send failed: ", err)
 		}
-		atomic.AddInt64(&sendCounter, 1)
 	}
 
 	shutdown.Await()
@@ -183,12 +216,17 @@ func mockUsersConnect(ctx context.Context) {
 	mockNetUsers = make([]*NetUser, mockUsers)
 
 	concurrent := 100
-	wg := sync.WaitGroup{}
+	wg := &sync.WaitGroup{}
 	wg.Add(concurrent)
 	each := (mockUsers / concurrent) + 1
 
 	for i := 0; i < concurrent; i++ {
+		//begin, end := i*each, (i+1)*each
+		//fmt.Printf("%d: %d~%d \n", i, begin, end)
+
 		go func(segment int) {
+			defer wg.Done()
+
 			begin, end := segment*each, (segment+1)*each
 			for i := begin; i < end && i < mockUsers; i++ {
 				uid := strconv.Itoa(110000 + i)
@@ -204,12 +242,8 @@ func mockUsersConnect(ctx context.Context) {
 				if err != nil {
 					panic(err)
 				}
-				mockNetUsers[i] = &NetUser{
-					Uid:  uid,
-					Conn: conn,
-				}
+				mockNetUsers[i] = &NetUser{Uid: uid, Conn: conn}
 			}
-			wg.Done()
 		}(i)
 	}
 	wg.Wait()
@@ -221,20 +255,16 @@ func sendBatchChatMessage(ctx context.Context, uid string, conn *websocket.Conn,
 		receiverUid := mockNetUsers[r.Intn(mockUsers)].Uid
 		args := &chat.ReqSend{
 			Receiver: receiverUid,
-			Content:  "hello",
+			Content:  "hi",
 		}
 		channel, err := send(conn, chat.Chat_ServiceDesc.ServiceName, "Send", args)
 		if err != nil {
 			fmt.Println("send failed:", err)
 			return
 		}
-
-		res := <-channel // wait result
-		err, failed := res.(error)
-		if failed {
+		// wait result
+		if err, failed := (<-channel).(error); failed {
 			fmt.Println("send failed:", err)
-		} else {
-			// fmt.Println("send ok:", res)
 		}
 	}
 }
@@ -299,11 +329,16 @@ func handleConn(ctx context.Context, uid string, token string, conn *websocket.C
 	if err != nil {
 		return
 	}
-	res := <-channel
-	// fmt.Printf("%v\n", res)
+	ch := time.After(time.Second * 5)
+
+	select {
+	case <-ch:
+		err = errors.New("req verify timeout")
+	case res := <-channel:
 		if e, failed := res.(error); failed {
 			err = e
 		}
+	}
 	return
 }
 
@@ -349,21 +384,6 @@ func listen(ctx context.Context, conn *websocket.Conn) {
 			atomic.AddInt64(&receiverCounter, 1)
 		}
 
-		if header.Type == 4 {
-			callbackMutex.Lock()
-			callback, ok := callbacks[header.SeqId]
-			if ok {
-				delete(callbacks, header.SeqId)
-			}
-			callbackMutex.Unlock()
-			if !ok {
-				e = fmt.Errorf("callback %d not found", header.SeqId)
-				return
-			}
-			callback(err)
-			continue
-		}
-
 		// notice message
 		if header.Type == protocol.TypeNotice {
 			switch header.Target {
@@ -379,46 +399,55 @@ func listen(ctx context.Context, conn *websocket.Conn) {
 			continue
 		}
 
+		callback, ok := callbacks.LoadAndDelete(header.SeqId)
+		if !ok {
+			logger.Errorf("callback %d not found", header.SeqId)
+			continue
+		}
+
+		if header.Type == protocol.TypeError {
+			err = errors.New(string(payload.Body))
+			callback(err)
+			continue
+		}
+
 		// rpc response
+
 		var msg proto.Message
-		switch header.Svc {
-		case auth.Auth_ServiceDesc.ServiceName:
-			switch header.Target {
-			case "Subject":
-				// fmt.Println("res verify...")
-				msg = &auth.Subject{}
-			}
-		case chat.Chat_ServiceDesc.ServiceName:
-			switch header.Target {
-			case "ResSend":
-				// fmt.Println("res send...")
-				msg = &chat.ResSend{}
+		if svc, ok := protoStructs[header.Svc]; ok {
+			if s, ok := svc[header.Target]; ok {
+				val := reflect.New(reflect.TypeOf(s).Elem())
+				msg = val.Interface().(proto.Message)
 			}
 		}
-		if msg == nil {
-			logger.Warning("unknown rpc response: ", len(payload.Body), header)
+
+		if msg == nil { // protobuf.Empty
+			// logger.Warning("unknown rpc response: ", len(payload.Body), header)
+			callback(nil)
 			continue
 		}
 		err = proto.Unmarshal(payload.Body, msg)
-		if e != nil {
+		if err != nil {
 			logger.Error("unmarshal rcp res body error: ", payload, err)
-			return
-		}
-
-		callbackMutex.Lock()
-		callback, ok := callbacks[header.SeqId]
-		if ok {
-			delete(callbacks, header.SeqId)
-		}
-		callbackMutex.Unlock()
-		if !ok {
-			e = fmt.Errorf("callback %d not found", header.SeqId)
+			callback(fmt.Errorf("unmarshal rcp res body error: %v", err))
 			return
 		}
 		callback(msg)
 	}
 }
 
+var protoStructs = map[string]map[string]proto.Message{
+	auth.Auth_ServiceDesc.ServiceName: {
+		"Subject": &auth.Subject{},
+	},
+	chat.Chat_ServiceDesc.ServiceName: {
+		"ResSend":       &chat.ResSend{},
+		"ResRoomCreate": &chat.ResRoomCreate{},
+		"ResRoomInfo":   &chat.ResRoomInfo{},
+		"ResRoomList":   &chat.ResRoomList{},
+	},
+}
+
 func send(conn *websocket.Conn, svc, method string, reqArgs proto.Message) (channel chan any, err error) {
 	seq := atomic.AddInt32(&seqId, 1)
 	header := &protocol.Header{
@@ -445,16 +474,14 @@ func send(conn *websocket.Conn, svc, method string, reqArgs proto.Message) (chan
 	channel = make(chan any)
 
 	// ready callback
-	callbackMutex.Lock()
-	callbacks[seq] = func(res any) {
+	callbacks.Store(seq, func(res any) {
 		ms := time.Now().UnixMilli() - begin
 		msSum := atomic.AddInt64(&useMsSum, ms)
 		counter := atomic.AddInt64(&receiverCounter, 1)
 		atomic.StoreInt64(&useMsAvg, msSum/counter)
 
 		channel <- res
-	}
-	callbackMutex.Unlock()
+	})
 
 	// send message
 	err = conn.WriteMessage(websocket.BinaryMessage, bytes)

cmd/chat/main.go

@@ -43,6 +43,9 @@ func main() {
 		panic(err)
 	}
 	groupDeli := deliver.NewGroupDeliver(chat.Chat_ServiceDesc.ServiceName, picker, logic.NewRoomLoader(), consumer)
+	if err := groupDeli.Init(ctx); err != nil {
+		panic(err)
+	}
 	chatServer := logic.NewChatServer(deli, groupDeli)
 
 	go func() {

internal/chat/logic/chat_server.go

@@ -98,22 +98,23 @@ func (s *ChatServer) Send(ctx context.Context, req *chat.ReqSend) (*chat.ResSend
 func (s *ChatServer) RoomSend(ctx context.Context, req *chat.ReqRoomSend) (res *chat.ResSend, err error) {
 	subject, err := session.GetSubject(ctx)
 	if err != nil {
-		return nil, err
+		return
 	}
 	gMessage := &chat.ChatMessage{
 		Type:    1,
 		Sender:  subject.Uid,
-		Gid:     req.Gid,
+		Gid:     req.Rid,
 		Content: req.Message,
 	}
-	err = s.groupDeliver.DeliverGroup(ctx, req.Gid, gMessage)
+	err = s.groupDeliver.DeliverGroup(ctx, req.Rid, gMessage)
+	res = &chat.ResSend{}
 	return
 }
 
 func (s *ChatServer) RoomCreate(ctx context.Context, req *chat.ReqRoomCreate) (res *chat.ResRoomCreate, err error) {
 	subject, err := session.GetSubject(ctx)
 	if err != nil {
-		return nil, err
+		return
 	}
 	logger.Infof("create room: %s", req.Rname)
 
@@ -138,10 +139,11 @@ func (s *ChatServer) RoomCreate(ctx context.Context, req *chat.ReqRoomCreate) (r
 	return
 }
 
-func (s *ChatServer) RoomJoin(ctx context.Context, req *chat.ReqRoomJoin) (_ *emptypb.Empty, err error) {
+func (s *ChatServer) RoomJoin(ctx context.Context, req *chat.ReqRoomJoin) (emp *emptypb.Empty, err error) {
+	emp = &emptypb.Empty{}
 	subject, err := session.GetSubject(ctx)
 	if err != nil {
-		return nil, err
+		return
 	}
 
 	room, ok := s.rooms.Load(req.Rid)
@@ -155,10 +157,11 @@ func (s *ChatServer) RoomJoin(ctx context.Context, req *chat.ReqRoomJoin) (_ *em
 	return
 }
 
-func (s *ChatServer) RoomLeave(ctx context.Context, req *chat.ReqRoomLeave) (_ *emptypb.Empty, err error) {
+func (s *ChatServer) RoomLeave(ctx context.Context, req *chat.ReqRoomLeave) (emp *emptypb.Empty, err error) {
+	emp = &emptypb.Empty{}
 	subject, err := session.GetSubject(ctx)
 	if err != nil {
-		return nil, err
+		return
 	}
 
 	room, ok := s.rooms.Load(req.Rid)
@@ -172,7 +175,8 @@ func (s *ChatServer) RoomLeave(ctx context.Context, req *chat.ReqRoomLeave) (_ *
 	return
 }
 
-func (s *ChatServer) RoomKickOut(ctx context.Context, req *chat.ReqRoomKickOut) (_ *emptypb.Empty, err error) {
+func (s *ChatServer) RoomKickOut(ctx context.Context, req *chat.ReqRoomKickOut) (emp *emptypb.Empty, err error) {
+	emp = &emptypb.Empty{}
 	return
 }
 
@@ -187,6 +191,18 @@ func (s *ChatServer) RoomInfo(ctx context.Context, req *chat.ReqRoomInfo) (res *
 	return
 }
 
+// RoomDissolve 解散群组
+func (s *ChatServer) RoomDissolve(ctx context.Context, req *chat.ReqRoomDissolve) (emp *emptypb.Empty, err error) {
+	emp = &emptypb.Empty{}
+
+	err = s.groupDeliver.GroupDissolve(ctx, req.Rid)
+	if err != nil {
+		return
+	}
+	s.rooms.Delete(req.Rid)
+	return
+}
+
 func (s *ChatServer) RoomList(ctx context.Context, req *chat.ReqRoomList) (res *chat.ResRoomList, err error) {
 	rooms := make([]*chat.Room, 16)
 	s.rooms.Range(func(_ string, room *chat.Room) bool {

internal/gateway_ws/gws_server/conn_handler.go

@@ -145,7 +145,7 @@ func (c *GwsHandler) OnClose(socket *gws.Conn, err error) {
 	conn := val.(*gwsConn)
 	if conn.uid != "" {
 		c.sessionStore.Delete(conn.uid)
-		logger.Info("subject offline: ", conn.uid)
+		//logger.Info("subject offline: ", conn.uid)
 	}
 }
 
@@ -251,7 +251,7 @@ func (c *GwsHandler) extractAuthVerify(conn *gwsConn, resp []byte) (sessionUid s
 	// 存储session到内存
 	channel = session.NewChannel(sessionUid, conn)
 	c.sessionStore.Store(sessionUid, channel)
-	logger.Infof("subject online: %s\n", sessionUid)
+	// logger.Infof("subject online: %s\n", sessionUid)
 	return
 }
 

internal/postal/logic/postal_server.go

@@ -95,7 +95,7 @@ func (s *PostalServer) processSessionStoreEvent(ctx context.Context) {
 			return
 
 		case uid := <-s.sessionStore.OnStore():
-			logger.Info("uid %s online", uid)
+			//logger.Infof("uid %s online", uid)
 			// publish mq online event todo delay 5s
 			online := &event.Online{Uid: uid}
 			data, err := online.MarshalBinary()
@@ -108,7 +108,7 @@ func (s *PostalServer) processSessionStoreEvent(ctx context.Context) {
 			}
 
 		case channel := <-s.sessionStore.OnDelete():
-			logger.Info("uid %s offline", channel.Uid)
+			//logger.Infof("uid %s offline", channel.Uid)
 			for _, gid := range channel.Groups() {
 				if g, ok := s.groupStore.Load(gid); ok {
 					g.Leave(channel.Uid)
@@ -179,8 +179,9 @@ func (s *PostalServer) DeliverBatch(ctx context.Context, req *postal.ReqDeliverB
 		}
 	}
 
+	res = &postal.ResDeliver{Ok: true}
 	if len(redeliverReceivers) == 0 {
-		return &postal.ResDeliver{Ok: true}, nil
+		return
 	}
 
 	// 向一致性 hash 下一个节点传递
@@ -191,7 +192,6 @@ func (s *PostalServer) DeliverBatch(ctx context.Context, req *postal.ReqDeliverB
 			Offset:    offset,
 		})
 		if err == nil {
-			res = &postal.ResDeliver{Ok: true}
 			return
 		}
 	}

pkg/grpc/interceptor/recover_interceptor.go

@@ -7,9 +7,10 @@ import (
 	"google.golang.org/grpc/grpclog"
 	"google.golang.org/protobuf/types/known/emptypb"
 	"runtime/debug"
+	"sonet/pkg/utils/logger"
 )
 
-func RecoverInterceptor(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
+func RecoverInterceptor(ctx context.Context, req any, server *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
 	defer func() {
 		if r := recover(); r != nil {
 			switch r.(type) {
@@ -29,6 +30,8 @@ func RecoverInterceptor(ctx context.Context, req any, _ *grpc.UnaryServerInfo, h
 	if err == nil {
 		if empty, ok := resp.(*emptypb.Empty); ok && empty == nil {
 			resp = &emptypb.Empty{} // grpc: error while marshaling: proto: Marshal called with nil
+		} else if resp == nil {
+			logger.Warningf("grpc request: has no response: method=%s, %v", server.FullMethod, req)
 		}
 	}
 	return

pkg/protocol/protocol.go

@@ -61,7 +61,9 @@ func DecodeSo(bytes []byte) (payload *Payload, err error) {
 	header.SeqId = int32(binary.BigEndian.Uint32(bytes[4:8]))
 	// error message
 	if header.Type == TypeError {
-		payload.Body = bytes[8:]
+		body := bytes[8:]
+		payload.Body = make([]byte, len(body))
+		copy(payload.Body, body)
 		return
 	}
 
@@ -84,7 +86,9 @@ func DecodeSo(bytes []byte) (payload *Payload, err error) {
 		return
 	}
 
-	payload.Body = bytes[cursor:]
+	body := bytes[cursor:]
+	payload.Body = make([]byte, len(body))
+	copy(payload.Body, body)
 	return
 }
 

pkg/protocol/session/store.go

@@ -42,7 +42,7 @@ func (c *mapStore) Delete(uid string) {
 	select {
 	case c.deleteCh <- ch:
 	default:
-		logger.Warning("session store delete channel fulled, %s", uid)
+		logger.Warningf("session store delete channel fulled, %s", uid)
 	}
 }
 

pkg/utils/collect/concurrent_map.go

@@ -92,10 +92,12 @@ func (m *ConcurrentMap[K, V]) Range(f func(key K, value V) bool) {
 	}
 }
 
-func (m *ConcurrentMap[K, V]) LoadAndDelete(k K) (prev V, loaded bool) {
+func (m *ConcurrentMap[K, V]) LoadAndDelete(k K) (value V, loaded bool) {
 	m.update(k, func(segment map[K]V) {
-		prev, loaded = segment[k]
+		value, loaded = segment[k]
+		if loaded {
 			delete(segment, k)
+		}
 	})
 	return
 }

pkg/utils/logger/logger.go

@@ -2,6 +2,9 @@ package logger
 
 import (
 	"github.com/sirupsen/logrus"
+	"runtime"
+	"strconv"
+	"strings"
 )
 
 var Logger *SoLogger
@@ -14,51 +17,94 @@ type SoLogger struct {
 	logger *logrus.Logger
 }
 
+func (l *SoLogger) enhance(args []any) (ret []any) {
+	//logrus.SetReportCaller(true)
+	ret = args
+	for i := 2; i < 5; i++ {
+		_, file, line, ok := runtime.Caller(i)
+		if !ok {
+			return
+		}
+		if strings.HasSuffix(file, "/pkg/utils/logger/exported.go") ||
+			strings.HasSuffix(file, "/pkg/utils/logger/logger.go") {
+			continue
+		}
+
+		//caller = fmt.Sprintf("%s:%d ", file, line)
+		caller := file + ":" + strconv.Itoa(line) + " "
+		if arg0, ok := ret[0].(string); ok {
+			ret[0] = caller + arg0
+			return
+		}
+
+		ret = make([]any, len(args)+1)
+		ret[0] = caller
+		copy(ret[1:], args)
+		return
+	}
+	return
+}
+
+func (l *SoLogger) enhancef(format string, args []any) (retf string, ret []any) {
+	args2 := make([]any, len(args)+1)
+	args2[0] = format
+	copy(args2[1:], args)
+
+	ret = l.enhance(args2)
+	retf = ret[0].(string)
+	ret = ret[1:]
+	return
+}
+
 func (l *SoLogger) Info(args ...any) {
-	l.logger.Info(args...)
+	l.logger.Info(l.enhance(args)...)
 }
 
 func (l *SoLogger) Infoln(args ...any) {
-	l.logger.Infoln(args...)
+	l.logger.Infoln(l.enhance(args)...)
 }
 
 func (l *SoLogger) Infof(format string, args ...any) {
+	format, args = l.enhancef(format, args)
 	l.logger.Infof(format, args...)
 }
 
 func (l *SoLogger) Warning(args ...any) {
-	l.logger.Warning(args...)
+	l.logger.Warning(l.enhance(args)...)
 }
 
 func (l *SoLogger) Warningln(args ...any) {
-	l.logger.Warningln(args...)
+	l.logger.Warningln(l.enhance(args)...)
 }
 
 func (l *SoLogger) Warningf(format string, args ...any) {
+	format, args = l.enhancef(format, args)
 	l.logger.Warningf(format, args...)
 }
 
 func (l *SoLogger) Error(args ...any) {
-	l.logger.Error(args...)
+	l.logger.Error(l.enhance(args)...)
 }
 
 func (l *SoLogger) Errorln(args ...any) {
-	l.logger.Errorln(args...)
+	l.logger.Errorln(l.enhance(args)...)
 }
 
 func (l *SoLogger) Errorf(format string, args ...any) {
+	format, args = l.enhancef(format, args)
 	l.logger.Errorf(format, args...)
 }
 
 func (l *SoLogger) Fatal(args ...any) {
-	l.logger.Fatal(args...)
+	l.logger.Fatal(l.enhance(args)...)
 }
 
 func (l *SoLogger) Fatalln(args ...any) {
-	l.logger.Fatalln(args...)
+	l.logger.Fatalln(l.enhance(args)...)
 }
 
 func (l *SoLogger) Fatalf(format string, args ...any) {
+	format, args = l.enhancef(format, args)
 	l.logger.Fatalf(format, args...)
 }