lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-05-06 22:57:49 +08:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

only broad cast when there are waiting threads

Browse Source
This commit is contained in:
chrislu 2024-03-16 11:38:27 -07:00
parent 054d0b3487
commit 27bb38228b
2 changed files with 14 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
weed/server/filer_grpc_server_sub_meta.go
View File

@ -4,6 +4,7 @@ import (
"fmt" "fmt"
"github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/stats"
"strings" "strings"
"sync/atomic"
"time" "time"
"google.golang.org/protobuf/proto" "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 { lastReadTime, isDone, readInMemoryLogErr = fs.filer.LocalMetaLogBuffer.LoopProcessLogData("localMeta:"+clientName, lastReadTime, req.UntilNs, func() bool {
fs.listenersLock.Lock() fs.listenersLock.Lock()
atomic.AddInt64(&fs.listenersWaits, 1)
fs.listenersCond.Wait() fs.listenersCond.Wait()
atomic.AddInt64(&fs.listenersWaits, -1)
fs.listenersLock.Unlock() fs.listenersLock.Unlock()
if !fs.hasClient(req.ClientId, req.ClientEpoch) { if !fs.hasClient(req.ClientId, req.ClientEpoch) {
return false return false

17
weed/server/filer_server.go
View File

@ -7,6 +7,7 @@ import (
"os" "os"
"strings" "strings"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/stats"
@ -76,7 +77,13 @@ type FilerOption struct {
} }
type FilerServer struct { type FilerServer struct {
inFlightDataSize int64 inFlightDataSize int64
listenersWaits int64
// notifying clients
listenersLock sync.Mutex
listenersCond *sync.Cond
inFlightDataLimitCond *sync.Cond inFlightDataLimitCond *sync.Cond
filer_pb.UnimplementedSeaweedFilerServer filer_pb.UnimplementedSeaweedFilerServer
@ -90,10 +97,6 @@ type FilerServer struct {
metricsAddress string metricsAddress string
metricsIntervalSec int metricsIntervalSec int
// notifying clients
listenersLock sync.Mutex
listenersCond *sync.Cond
// track known metadata listeners // track known metadata listeners
knownListenersLock sync.Mutex knownListenersLock sync.Mutex
knownListeners map[int32]int32 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) v.SetDefault("filer.options.max_file_name_length", 255)
maxFilenameLength := v.GetUint32("filer.options.max_file_name_length") 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.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 fs.filer.Cipher = option.Cipher
// we do not support IP whitelist right now // we do not support IP whitelist right now