implement least active load balance

pkg/remoting/loadbalance/least_active_loadbalance.go

@@ -0,0 +1,63 @@
+/*
+ * 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 loadbalance
+
+import (
+	"math/rand"
+	"sync"
+	"time"
+
+	getty "github.com/apache/dubbo-getty"
+)
+
+func LeastActiveLoadBalance(sessions *sync.Map, xid string) getty.Session {
+	var session getty.Session
+	var leastActive int64 = -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
+				leastCount = 1
+				if len(leastIndexes) > 0 {
+					leastIndexes = leastIndexes[:0]
+				}
+				leastIndexes = append(leastIndexes, session)
+			} else if interval == leastActive {
+				leastIndexes = append(leastIndexes, session)
+				leastCount++
+			}
+		}
+		return true
+	})
+
+	if leastCount == 0 {
+		return nil
+	}
+
+	if leastCount == 1 {
+		return leastIndexes[0]
+	} else {
+		return leastIndexes[rand.New(rand.NewSource(time.Now().UnixNano())).Intn(leastCount)]
+	}
+}

pkg/remoting/loadbalance/least_active_loadbalance_test.go

@@ -0,0 +1,65 @@
+/*
+ * 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 loadbalance
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/seata/seata-go/pkg/remoting/mock"
+)
+
+func TestConsistentHashLoadBalance(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	sessions := &sync.Map{}
+
+	now := time.Now()
+
+	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().RemoteAddr().AnyTimes().DoAndReturn(func() string {
+		return "127.0.0.1:8001"
+	})
+	session.EXPECT().IsClosed().Return(false).AnyTimes()
+	sessions.Store(session, fmt.Sprintf("session-%d", 2))
+
+	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))
+
+	result := LeastActiveLoadBalance(sessions, "test_xid")
+	assert.True(t, result.RemoteAddr() == "127.0.0.1:8000")
+	assert.False(t, result.IsClosed())
+}

pkg/remoting/loadbalance/loadbalance.go

@@ -37,6 +37,8 @@ func Select(loadBalanceType string, sessions *sync.Map, xid string) getty.Sessio
 		return RandomLoadBalance(sessions, xid)
 	case xidLoadBalance:
 		return XidLoadBalance(sessions, xid)
+	case leastActiveLoadBalance:
+		return LeastActiveLoadBalance(sessions, xid)	
 	default:
 		return RandomLoadBalance(sessions, xid)
 	}