From 27bb38228b647e34fe20a6016fa04c829138c272 Mon Sep 17 00:00:00 2001
From: chrislu <chris.lu@gmail.com>
Date: Sat, 16 Mar 2024 11:38:27 -0700
Subject: [PATCH] only broad cast when there are waiting threads

---
 weed/server/filer_grpc_server_sub_meta.go |  3 +++
 weed/server/filer_server.go               | 17 +++++++++++------
 2 files changed, 14 insertions(+), 6 deletions(-)

diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go
index 8e8b4e5c4..436c4158f 100644
--- a/weed/server/filer_grpc_server_sub_meta.go
+++ b/weed/server/filer_grpc_server_sub_meta.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"google.golang.org/protobuf/proto"
@@ -150,7 +151,9 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 
 		lastReadTime, isDone, readInMemoryLogErr = fs.filer.LocalMetaLogBuffer.LoopProcessLogData("localMeta:"+clientName, lastReadTime, req.UntilNs, func() bool {
 			fs.listenersLock.Lock()
+			atomic.AddInt64(&fs.listenersWaits, 1)
 			fs.listenersCond.Wait()
+			atomic.AddInt64(&fs.listenersWaits, -1)
 			fs.listenersLock.Unlock()
 			if !fs.hasClient(req.ClientId, req.ClientEpoch) {
 				return false
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index 9428c2edf..356761f30 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -7,6 +7,7 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/stats"
@@ -76,7 +77,13 @@ type FilerOption struct {
 }
 
 type FilerServer struct {
-	inFlightDataSize      int64
+	inFlightDataSize int64
+	listenersWaits   int64
+
+	// notifying clients
+	listenersLock sync.Mutex
+	listenersCond *sync.Cond
+
 	inFlightDataLimitCond *sync.Cond
 
 	filer_pb.UnimplementedSeaweedFilerServer
@@ -90,10 +97,6 @@ type FilerServer struct {
 	metricsAddress     string
 	metricsIntervalSec int
 
-	// notifying clients
-	listenersLock sync.Mutex
-	listenersCond *sync.Cond
-
 	// track known metadata listeners
 	knownListenersLock sync.Mutex
 	knownListeners     map[int32]int32
@@ -135,7 +138,9 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	v.SetDefault("filer.options.max_file_name_length", 255)
 	maxFilenameLength := v.GetUint32("filer.options.max_file_name_length")
 	fs.filer = filer.NewFiler(*option.Masters, fs.grpcDialOption, option.Host, option.FilerGroup, option.Collection, option.DefaultReplication, option.DataCenter, maxFilenameLength, func() {
-		fs.listenersCond.Broadcast()
+		if atomic.LoadInt64(&fs.listenersWaits) > 0 {
+			fs.listenersCond.Broadcast()
+		}
 	})
 	fs.filer.Cipher = option.Cipher
 	// we do not support IP whitelist right now