implement least active load balance

pkg/remoting/getty/getty_remoting.go

@@ -19,6 +19,7 @@ package getty
 
 import (
 	"fmt"
+	"github.com/seata/seata-go/pkg/remoting/rpc"
 	"sync"
 	"time"
 
@@ -61,14 +62,19 @@ func (g *GettyRemoting) SendSync(msg message.RpcMessage, s getty.Session, callba
 	if s == nil {
 		s = sessionManager.selectSession(msg)
 	}
-	return g.sendAsync(s, msg, callback)
+	rpc.BeginCount(s.RemoteAddr())
+	result, err := g.sendAsync(s, msg, callback)
+	rpc.EndCount(s.RemoteAddr())
+	return result, err
 }
 
 func (g *GettyRemoting) SendASync(msg message.RpcMessage, s getty.Session, callback callbackMethod) error {
 	if s == nil {
 		s = sessionManager.selectSession(msg)
 	}
+	rpc.BeginCount(s.RemoteAddr())
 	_, err := g.sendAsync(s, msg, callback)
+	rpc.EndCount(s.RemoteAddr())
 	return err
 }
 

pkg/remoting/loadbalance/least_active_loadbalance.go

@@ -18,6 +18,7 @@
 package loadbalance
 
 import (
+	"github.com/seata/seata-go/pkg/remoting/rpc"
 	"math/rand"
 	"sync"
 	"time"
@@ -27,23 +28,23 @@ import (
 
 func LeastActiveLoadBalance(sessions *sync.Map, xid string) getty.Session {
 	var session getty.Session
-	var leastActive int64 = -1
+	var leastActive int32 = -1
 	leastCount := 0
 	var leastIndexes []getty.Session
 	sessions.Range(func(key, value interface{}) bool {
 		session = key.(getty.Session)
 		if session.IsClosed() {
 			sessions.Delete(session)
 		} else {
-			interval := session.GetActive().UnixNano()
-			if leastActive == -1 || interval < leastActive {
-				leastActive = interval
+			active := rpc.GetStatus(session.RemoteAddr()).GetActive()
+			if leastActive == -1 || active < leastActive {
+				leastActive = active
 				leastCount = 1
 				if len(leastIndexes) > 0 {
 					leastIndexes = leastIndexes[:0]
 				}
 				leastIndexes = append(leastIndexes, session)
-			} else if interval == leastActive {
+			} else if active == leastActive {
 				leastIndexes = append(leastIndexes, session)
 				leastCount++
 			}

pkg/remoting/loadbalance/least_active_loadbalance_test.go

@@ -19,47 +19,44 @@ package loadbalance
 
 import (
 	"fmt"
-	"sync"
-	"testing"
-	"time"
-
 	"github.com/golang/mock/gomock"
-	"github.com/stretchr/testify/assert"
-
 	"github.com/seata/seata-go/pkg/remoting/mock"
+	"github.com/seata/seata-go/pkg/remoting/rpc"
+	"github.com/stretchr/testify/assert"
+	"strconv"
+	"sync"
+	"testing"
 )
 
 func TestLeastActiveLoadBalance(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	sessions := &sync.Map{}
 
-	now := time.Now()
+	for i := 1; i <= 3; i++ {
+		session := mock.NewMockTestSession(ctrl)
+		session.EXPECT().IsClosed().Return(false).AnyTimes()
+		addr := "127.0.0." + strconv.Itoa(i) + ":8000"
+		session.EXPECT().RemoteAddr().AnyTimes().DoAndReturn(func() string {
+			return addr
+		})
+		sessions.Store(session, fmt.Sprintf("session-%d", i))
+		rpc.BeginCount(addr)
+	}
 
 	session := mock.NewMockTestSession(ctrl)
-	session.EXPECT().GetActive().Return(now.Add(-time.Second * 7)).AnyTimes()
-	session.EXPECT().RemoteAddr().AnyTimes().DoAndReturn(func() string {
-		return "127.0.0.1:8000"
-	})
-	session.EXPECT().IsClosed().Return(false).AnyTimes()
-	sessions.Store(session, fmt.Sprintf("session-%d", 1))
-
-	session = mock.NewMockTestSession(ctrl)
-	session.EXPECT().GetActive().Return(now.Add(-time.Second * 5)).AnyTimes()
+	session.EXPECT().IsClosed().Return(true).AnyTimes()
+	addr := "127.0.0.5:8000"
 	session.EXPECT().RemoteAddr().AnyTimes().DoAndReturn(func() string {
-		return "127.0.0.1:8001"
+		return addr
 	})
-	session.EXPECT().IsClosed().Return(false).AnyTimes()
-	sessions.Store(session, fmt.Sprintf("session-%d", 2))
+	sessions.Store(session, "session-5")
+	rpc.BeginCount(addr)
 
-	session = mock.NewMockTestSession(ctrl)
-	session.EXPECT().GetActive().Return(now.Add(-time.Second * 10)).AnyTimes()
-	session.EXPECT().RemoteAddr().AnyTimes().DoAndReturn(func() string {
-		return "127.0.0.1:8002"
-	})
-	session.EXPECT().IsClosed().Return(true).AnyTimes()
-	sessions.Store(session, fmt.Sprintf("session-%d", 3))
+	countTwo := "127.0.0.1:8000"
+	rpc.BeginCount(countTwo)
 
 	result := LeastActiveLoadBalance(sessions, "test_xid")
-	assert.True(t, result.RemoteAddr() == "127.0.0.1:8000")
+	assert.False(t, result.RemoteAddr() == countTwo)
+	assert.False(t, result.RemoteAddr() == addr)
 	assert.False(t, result.IsClosed())
 }

pkg/remoting/rpc/rpc_status.go

@@ -0,0 +1,47 @@
+package rpc
+
+import (
+	"sync"
+	"sync/atomic"
+)
+
+var ServiceStatusMap sync.Map
+
+type Status struct {
+	Active int32
+	Total  int32
+}
+
+// RemoveStatus remove the RpcStatus of this service
+func RemoveStatus(service string) {
+	ServiceStatusMap.Delete(service)
+}
+
+// BeginCount begin count
+func BeginCount(service string) {
+	status := GetStatus(service)
+	atomic.AddInt32(&status.Active, 1)
+}
+
+// EndCount end count
+func EndCount(service string) {
+	status := GetStatus(service)
+	atomic.AddInt32(&status.Active, -1)
+	atomic.AddInt32(&status.Total, 1)
+}
+
+// GetStatus get status
+func GetStatus(service string) *Status {
+	a, _ := ServiceStatusMap.LoadOrStore(service, new(Status))
+	return a.(*Status)
+}
+
+// GetActive get active.
+func (s *Status) GetActive() int32 {
+	return s.Active
+}
+
+// GetTotal get total.
+func (s *Status) GetTotal() int32 {
+	return s.Total
+}

pkg/remoting/rpc/rpc_status_test.go

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rpc
+
+import (
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+var service = "127.0.0.1:8000"
+
+func TestStatus(t *testing.T) {
+	rpcStatus1 := GetStatus(service)
+	assert.NotNil(t, rpcStatus1)
+	rpcStatus2 := GetStatus(service)
+	assert.Equal(t, rpcStatus1, rpcStatus2)
+}
+
+func TestRemoveStatus(t *testing.T) {
+	old := GetStatus(service)
+	RemoveStatus(service)
+	assert.Equal(t, GetStatus(service), old)
+}
+
+func TestBeginCount(t *testing.T) {
+	BeginCount(service)
+	assert.Equal(t, GetStatus(service).GetActive(), int32(1))
+}
+
+func TestEndCount(t *testing.T) {
+	EndCount(service)
+	assert.Equal(t, GetStatus(service).GetActive(), int32(0))
+	assert.Equal(t, GetStatus(service).GetTotal(), int32(1))
+}