gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

refactoring

Browse source
This commit is contained in:
Chris Lu 2020-07-12 17:31:24 -07:00
parent fc9af962ec
commit 1dd2c76532
3 changed files with 22 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
weed/filer2/filer.go
View file

@ -38,9 +38,11 @@ type Filer struct {
LocalMetaLogBuffer *log_buffer.LogBuffer LocalMetaLogBuffer *log_buffer.LogBuffer
metaLogCollection string metaLogCollection string
metaLogReplication string metaLogReplication string
MetaAggregator *MetaAggregator
} }
func NewFiler(masters []string, grpcDialOption grpc.DialOption, filerHost string, filerGrpcPort uint32, collection string, replication string, notifyFn func()) *Filer { func NewFiler(masters []string, grpcDialOption grpc.DialOption,
filerHost string, filerGrpcPort uint32, collection string, replication string, notifyFn func()) *Filer {
f := &Filer{ f := &Filer{
directoryCache: ccache.New(ccache.Configure().MaxSize(1000).ItemsToPrune(100)), directoryCache: ccache.New(ccache.Configure().MaxSize(1000).ItemsToPrune(100)),
MasterClient: wdclient.NewMasterClient(grpcDialOption, "filer", filerHost, filerGrpcPort, masters), MasterClient: wdclient.NewMasterClient(grpcDialOption, "filer", filerHost, filerGrpcPort, masters),
@ -56,6 +58,20 @@ func NewFiler(masters []string, grpcDialOption grpc.DialOption, filerHost string
return f return f
} }
func (f *Filer) AggregateFromPeers(self string, filers []string) {
// set peers
if strings.HasPrefix(f.GetStore().GetName(), "leveldb") && len(filers) > 0 {
glog.Fatalf("filers using separate leveldb stores should not configure %d peers %+v", len(filers), filers)
}
if len(filers) == 0 {
filers = append(filers, self)
}
f.MetaAggregator = NewMetaAggregator(filers, f.GrpcDialOption)
f.MetaAggregator.StartLoopSubscribe(time.Now().UnixNano())
}
func (f *Filer) SetStore(store FilerStore) { func (f *Filer) SetStore(store FilerStore) {
f.store = NewFilerStoreWrapper(store) f.store = NewFilerStoreWrapper(store)
} }

8
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) lastReadTime = time.Unix(0, processedTsNs)
} }
err = fs.metaAggregator.MetaLogBuffer.LoopProcessLogData(lastReadTime, func() bool { err = fs.filer.MetaAggregator.MetaLogBuffer.LoopProcessLogData(lastReadTime, func() bool {
fs.metaAggregator.ListenersLock.Lock() fs.filer.MetaAggregator.ListenersLock.Lock()
fs.metaAggregator.ListenersCond.Wait() fs.filer.MetaAggregator.ListenersCond.Wait()
fs.metaAggregator.ListenersLock.Unlock() fs.filer.MetaAggregator.ListenersLock.Unlock()
return true return true
}, eachLogEntryFn) }, eachLogEntryFn)

12
weed/server/filer_server.go
View file

@ -5,7 +5,6 @@ import (
"fmt" "fmt"
"net/http" "net/http"
"os" "os"
"strings"
"sync" "sync"
"time" "time"
@ -60,7 +59,6 @@ type FilerServer struct {
option *FilerOption option *FilerOption
secret security.SigningKey secret security.SigningKey
filer *filer2.Filer filer *filer2.Filer
metaAggregator *filer2.MetaAggregator
grpcDialOption grpc.DialOption grpcDialOption grpc.DialOption
// notifying clients // notifying clients
@ -121,15 +119,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
readonlyMux.HandleFunc("/", fs.readonlyFilerHandler) readonlyMux.HandleFunc("/", fs.readonlyFilerHandler)
} }
// set peers fs.filer.AggregateFromPeers(fmt.Sprintf("%s:%d", option.Host, option.Port), option.Filers)
if strings.HasPrefix(fs.filer.GetStore().GetName(), "leveldb") && len(option.Filers) > 0 {
glog.Fatalf("filers using separate leveldb stores should not configure %d peers %+v", len(option.Filers), option.Filers)
}
if len(option.Filers) == 0 {
option.Filers = append(option.Filers, fmt.Sprintf("%s:%d", option.Host, option.Port))
}
fs.metaAggregator = filer2.NewMetaAggregator(option.Filers, fs.grpcDialOption)
fs.metaAggregator.StartLoopSubscribe(time.Now().UnixNano())
fs.filer.LoadBuckets() fs.filer.LoadBuckets()