lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-10-21 16:37:25 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

aggregate multiple filer metadata chagne events

Browse Source
This commit is contained in:
Chris Lu
2020-07-05 23:05:02 -07:00
parent 49929e0869
commit 0b8cc9b626
3 changed files with 102 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
weed/server/filer_grpc_server_sub_meta.go
View File

@@ -37,10 +37,10 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
lastReadTime = time.Unix(0, processedTsNs)
}
err = fs.filer.LocalMetaLogBuffer.LoopProcessLogData(lastReadTime, func() bool {
fs.listenersLock.Lock()
fs.listenersCond.Wait()
fs.listenersLock.Unlock()
err = fs.metaAggregator.MetaLogBuffer.LoopProcessLogData(lastReadTime, func() bool {
fs.metaAggregator.ListenersLock.Lock()
fs.metaAggregator.ListenersCond.Wait()
fs.metaAggregator.ListenersLock.Unlock()
return true
}, eachLogEntryFn)
@@ -134,7 +134,3 @@ func (fs *FilerServer) addClient(clientType string, clientAddress string) (clien
func (fs *FilerServer) deleteClient(clientName string) {
glog.V(0).Infof("- listener %v", clientName)
}
func (fs *FilerServer) notifyMetaListeners() {
fs.listenersCond.Broadcast()
}

8
weed/server/filer_server.go
View File

@@ -52,12 +52,14 @@ type FilerOption struct {
Port uint32
recursiveDelete bool
Cipher bool
Filers []string
}
type FilerServer struct {
option *FilerOption
secret security.SigningKey
filer *filer2.Filer
metaAggregator *filer2.MetaAggregator
grpcDialOption grpc.DialOption
// notifying clients
@@ -81,12 +83,16 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
glog.Fatal("master list is required!")
}
fs.filer = filer2.NewFiler(option.Masters, fs.grpcDialOption, option.Host, option.Port, option.Collection, option.DefaultReplication, fs.notifyMetaListeners)
fs.filer = filer2.NewFiler(option.Masters, fs.grpcDialOption, option.Host, option.Port, option.Collection, option.DefaultReplication, func() {
fs.listenersCond.Broadcast()
})
fs.metaAggregator = filer2.NewMetaAggregator(append(option.Filers, fmt.Sprintf("%s:%d", option.Host, option.Port)), fs.grpcDialOption)
fs.filer.Cipher = option.Cipher
maybeStartMetrics(fs, option)
go fs.filer.KeepConnectedToMaster()
fs.metaAggregator.StartLoopSubscribe(time.Now().UnixNano())
v := util.GetViper()
if !util.LoadConfiguration("filer", false) {