diff options
| author | Chris Lu <chris.lu@gmail.com> | 2020-02-25 21:50:12 -0800 |
|---|---|---|
| committer | Chris Lu <chris.lu@gmail.com> | 2020-02-25 21:50:12 -0800 |
| commit | 892e726eb9c2427634c46f8ae9b7bcf0b6d1b082 (patch) | |
| tree | 3bf821356579902219633c6f6d42739deb1edd2d /weed/server | |
| parent | bd3254b53f78b8f42e31ea50cbf2e0d7e87b2bbc (diff) | |
| download | seaweedfs-892e726eb9c2427634c46f8ae9b7bcf0b6d1b082.tar.xz seaweedfs-892e726eb9c2427634c46f8ae9b7bcf0b6d1b082.zip | |
avoid reusing context object
fix https://github.com/chrislusf/seaweedfs/issues/1182
Diffstat (limited to 'weed/server')
| -rw-r--r-- | weed/server/filer_grpc_server.go | 4 | ||||
| -rw-r--r-- | weed/server/filer_server.go | 4 | ||||
| -rw-r--r-- | weed/server/master_grpc_server_collection.go | 9 | ||||
| -rw-r--r-- | weed/server/master_server.go | 3 | ||||
| -rw-r--r-- | weed/server/master_server_handlers_admin.go | 4 | ||||
| -rw-r--r-- | weed/server/volume_grpc_copy.go | 4 | ||||
| -rw-r--r-- | weed/server/volume_grpc_erasure_coding.go | 2 | ||||
| -rw-r--r-- | weed/server/volume_grpc_file.go | 3 | ||||
| -rw-r--r-- | weed/server/volume_server_handlers_read.go | 6 | ||||
| -rw-r--r-- | weed/server/webdav_server.go | 34 |
10 files changed, 34 insertions, 39 deletions
diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 9bbabec26..c6ef04e85 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -300,8 +300,8 @@ func (fs *FilerServer) AssignVolume(ctx context.Context, req *filer_pb.AssignVol func (fs *FilerServer) DeleteCollection(ctx context.Context, req *filer_pb.DeleteCollectionRequest) (resp *filer_pb.DeleteCollectionResponse, err error) { - err = fs.filer.MasterClient.WithClient(ctx, func(client master_pb.SeaweedClient) error { - _, err := client.CollectionDelete(ctx, &master_pb.CollectionDeleteRequest{ + err = fs.filer.MasterClient.WithClient(func(client master_pb.SeaweedClient) error { + _, err := client.CollectionDelete(context.Background(), &master_pb.CollectionDeleteRequest{ Name: req.GetCollection(), }) return err diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index b0df851c9..5fc038e17 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -127,8 +127,8 @@ func maybeStartMetrics(fs *FilerServer, option *FilerOption) { } func readFilerConfiguration(grpcDialOption grpc.DialOption, masterGrpcAddress string) (metricsAddress string, metricsIntervalSec int, err error) { - err = operation.WithMasterServerClient(masterGrpcAddress, grpcDialOption, func(ctx context.Context, masterClient master_pb.SeaweedClient) error { - resp, err := masterClient.GetMasterConfiguration(ctx, &master_pb.GetMasterConfigurationRequest{}) + err = operation.WithMasterServerClient(masterGrpcAddress, grpcDialOption, func(masterClient master_pb.SeaweedClient) error { + resp, err := masterClient.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{}) if err != nil { return fmt.Errorf("get master %s configuration: %v", masterGrpcAddress, err) } diff --git a/weed/server/master_grpc_server_collection.go b/weed/server/master_grpc_server_collection.go index f02b0f242..b92d6bcbe 100644 --- a/weed/server/master_grpc_server_collection.go +++ b/weed/server/master_grpc_server_collection.go @@ -4,6 +4,7 @@ import ( "context" "github.com/chrislusf/raft" + "github.com/chrislusf/seaweedfs/weed/operation" "github.com/chrislusf/seaweedfs/weed/pb/master_pb" "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb" @@ -57,8 +58,8 @@ func (ms *MasterServer) doDeleteNormalCollection(collectionName string) error { } for _, server := range collection.ListVolumeServers() { - err := operation.WithVolumeServerClient(server.Url(), ms.grpcDialOption, func(ctx context.Context, client volume_server_pb.VolumeServerClient) error { - _, deleteErr := client.DeleteCollection(ctx, &volume_server_pb.DeleteCollectionRequest{ + err := operation.WithVolumeServerClient(server.Url(), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + _, deleteErr := client.DeleteCollection(context.Background(), &volume_server_pb.DeleteCollectionRequest{ Collection: collectionName, }) return deleteErr @@ -77,8 +78,8 @@ func (ms *MasterServer) doDeleteEcCollection(collectionName string) error { listOfEcServers := ms.Topo.ListEcServersByCollection(collectionName) for _, server := range listOfEcServers { - err := operation.WithVolumeServerClient(server, ms.grpcDialOption, func(ctx context.Context, client volume_server_pb.VolumeServerClient) error { - _, deleteErr := client.DeleteCollection(ctx, &volume_server_pb.DeleteCollectionRequest{ + err := operation.WithVolumeServerClient(server, ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + _, deleteErr := client.DeleteCollection(context.Background(), &volume_server_pb.DeleteCollectionRequest{ Collection: collectionName, }) return deleteErr diff --git a/weed/server/master_server.go b/weed/server/master_server.go index f5f547a32..095008339 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -1,7 +1,6 @@ package weed_server import ( - "context" "fmt" "net/http" "net/http/httputil" @@ -89,7 +88,7 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers []string) *Maste preallocateSize: preallocateSize, clientChans: make(map[string]chan *master_pb.VolumeLocation), grpcDialOption: grpcDialOption, - MasterClient: wdclient.NewMasterClient(context.Background(), grpcDialOption, "master", peers), + MasterClient: wdclient.NewMasterClient(grpcDialOption, "master", peers), } ms.bounedLeaderChan = make(chan int, 16) diff --git a/weed/server/master_server_handlers_admin.go b/weed/server/master_server_handlers_admin.go index 44a04cb86..2965a4863 100644 --- a/weed/server/master_server_handlers_admin.go +++ b/weed/server/master_server_handlers_admin.go @@ -25,8 +25,8 @@ func (ms *MasterServer) collectionDeleteHandler(w http.ResponseWriter, r *http.R return } for _, server := range collection.ListVolumeServers() { - err := operation.WithVolumeServerClient(server.Url(), ms.grpcDialOption, func(ctx context.Context, client volume_server_pb.VolumeServerClient) error { - _, deleteErr := client.DeleteCollection(ctx, &volume_server_pb.DeleteCollectionRequest{ + err := operation.WithVolumeServerClient(server.Url(), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + _, deleteErr := client.DeleteCollection(context.Background(), &volume_server_pb.DeleteCollectionRequest{ Collection: collection.Name, }) return deleteErr diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 6d74f8171..fcd13d364 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -41,9 +41,9 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo // confirm size and timestamp var volFileInfoResp *volume_server_pb.ReadVolumeFileStatusResponse var volumeFileName, idxFileName, datFileName string - err := operation.WithVolumeServerClient(req.SourceDataNode, vs.grpcDialOption, func(ctx context.Context, client volume_server_pb.VolumeServerClient) error { + err := operation.WithVolumeServerClient(req.SourceDataNode, vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { var err error - volFileInfoResp, err = client.ReadVolumeFileStatus(ctx, + volFileInfoResp, err = client.ReadVolumeFileStatus(context.Background(), &volume_server_pb.ReadVolumeFileStatusRequest{ VolumeId: req.VolumeId, }) diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 256e7c447..67efc0f6d 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -106,7 +106,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv baseFileName := storage.VolumeFileName(location.Directory, req.Collection, int(req.VolumeId)) - err := operation.WithVolumeServerClient(req.SourceDataNode, vs.grpcDialOption, func(ctx context.Context, client volume_server_pb.VolumeServerClient) error { + err := operation.WithVolumeServerClient(req.SourceDataNode, vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { // copy ec data slices for _, shardId := range req.ShardIds { diff --git a/weed/server/volume_grpc_file.go b/weed/server/volume_grpc_file.go index c20aeb60f..4d71ddeb1 100644 --- a/weed/server/volume_grpc_file.go +++ b/weed/server/volume_grpc_file.go @@ -1,7 +1,6 @@ package weed_server import ( - "context" "encoding/json" "net/http" "strings" @@ -45,7 +44,7 @@ func (vs *VolumeServer) FileGet(req *volume_server_pb.FileGetRequest, stream vol if hasVolume { count, err = vs.store.ReadVolumeNeedle(volumeId, n) } else if hasEcVolume { - count, err = vs.store.ReadEcShardNeedle(context.Background(), volumeId, n) + count, err = vs.store.ReadEcShardNeedle(volumeId, n) } if err != nil || count < 0 { diff --git a/weed/server/volume_server_handlers_read.go b/weed/server/volume_server_handlers_read.go index d89d13a0d..40a951a74 100644 --- a/weed/server/volume_server_handlers_read.go +++ b/weed/server/volume_server_handlers_read.go @@ -2,7 +2,7 @@ package weed_server import ( "bytes" - "context" + "encoding/json" "errors" "fmt" "io" @@ -15,8 +15,6 @@ import ( "strings" "time" - "encoding/json" - "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/images" "github.com/chrislusf/seaweedfs/weed/operation" @@ -86,7 +84,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) if hasVolume { count, err = vs.store.ReadVolumeNeedle(volumeId, n) } else if hasEcVolume { - count, err = vs.store.ReadEcShardNeedle(context.Background(), volumeId, n) + count, err = vs.store.ReadEcShardNeedle(volumeId, n) } // glog.V(4).Infoln("read bytes", count, "error", err) if err != nil || count < 0 { diff --git a/weed/server/webdav_server.go b/weed/server/webdav_server.go index 7a7ab7a4b..959e50128 100644 --- a/weed/server/webdav_server.go +++ b/weed/server/webdav_server.go @@ -96,11 +96,11 @@ func NewWebDavFileSystem(option *WebDavOption) (webdav.FileSystem, error) { }, nil } -func (fs *WebDavFileSystem) WithFilerClient(ctx context.Context, fn func(context.Context, filer_pb.SeaweedFilerClient) error) error { +func (fs *WebDavFileSystem) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error { - return util.WithCachedGrpcClient(ctx, func(ctx2 context.Context, grpcConnection *grpc.ClientConn) error { + return util.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) - return fn(ctx2, client) + return fn(client) }, fs.option.FilerGrpcAddress, fs.option.GrpcDialOption) } @@ -135,7 +135,7 @@ func (fs *WebDavFileSystem) Mkdir(ctx context.Context, fullDirPath string, perm return os.ErrExist } - return fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + return fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error { dir, name := filer2.FullPath(fullDirPath).DirAndName() request := &filer_pb.CreateEntryRequest{ Directory: dir, @@ -153,7 +153,7 @@ func (fs *WebDavFileSystem) Mkdir(ctx context.Context, fullDirPath string, perm } glog.V(1).Infof("mkdir: %v", request) - if err := filer_pb.CreateEntry(ctx, client, request); err != nil { + if err := filer_pb.CreateEntry(client, request); err != nil { return fmt.Errorf("mkdir %s/%s: %v", dir, name, err) } @@ -184,8 +184,8 @@ func (fs *WebDavFileSystem) OpenFile(ctx context.Context, fullFilePath string, f } dir, name := filer2.FullPath(fullFilePath).DirAndName() - err = fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { - if err := filer_pb.CreateEntry(ctx, client, &filer_pb.CreateEntryRequest{ + err = fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error { + if err := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{ Directory: dir, Entry: &filer_pb.Entry{ Name: name, @@ -249,7 +249,7 @@ func (fs *WebDavFileSystem) removeAll(ctx context.Context, fullFilePath string) //_, err = fs.db.Exec(`delete from filesystem where fullFilePath = ?`, fullFilePath) } dir, name := filer2.FullPath(fullFilePath).DirAndName() - err = fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + err = fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.DeleteEntryRequest{ Directory: dir, @@ -308,7 +308,7 @@ func (fs *WebDavFileSystem) Rename(ctx context.Context, oldName, newName string) oldDir, oldBaseName := filer2.FullPath(oldName).DirAndName() newDir, newBaseName := filer2.FullPath(newName).DirAndName() - return fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + return fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.AtomicRenameEntryRequest{ OldDirectory: oldDir, @@ -336,7 +336,7 @@ func (fs *WebDavFileSystem) stat(ctx context.Context, fullFilePath string) (os.F fullpath := filer2.FullPath(fullFilePath) var fi FileInfo - entry, err := filer2.GetEntry(ctx, fs, fullpath) + entry, err := filer2.GetEntry(fs, fullpath) if entry == nil { return nil, os.ErrNotExist } @@ -372,7 +372,7 @@ func (f *WebDavFile) Write(buf []byte) (int, error) { var err error ctx := context.Background() if f.entry == nil { - f.entry, err = filer2.GetEntry(ctx, f.fs, filer2.FullPath(f.name)) + f.entry, err = filer2.GetEntry(f.fs, filer2.FullPath(f.name)) } if f.entry == nil { @@ -386,7 +386,7 @@ func (f *WebDavFile) Write(buf []byte) (int, error) { var auth security.EncodedJwt var collection, replication string - if err = f.fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + if err = f.fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.AssignVolumeRequest{ Count: 1, @@ -434,7 +434,7 @@ func (f *WebDavFile) Write(buf []byte) (int, error) { f.entry.Chunks = append(f.entry.Chunks, chunk) - err = f.fs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + err = f.fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error { f.entry.Attributes.Mtime = time.Now().Unix() f.entry.Attributes.Collection = collection f.entry.Attributes.Replication = replication @@ -474,10 +474,9 @@ func (f *WebDavFile) Close() error { func (f *WebDavFile) Read(p []byte) (readSize int, err error) { glog.V(2).Infof("WebDavFileSystem.Read %v", f.name) - ctx := context.Background() if f.entry == nil { - f.entry, err = filer2.GetEntry(ctx, f.fs, filer2.FullPath(f.name)) + f.entry, err = filer2.GetEntry(f.fs, filer2.FullPath(f.name)) } if f.entry == nil { return 0, err @@ -493,7 +492,7 @@ func (f *WebDavFile) Read(p []byte) (readSize int, err error) { } chunkViews := filer2.ViewFromVisibleIntervals(f.entryViewCache, f.off, len(p)) - totalRead, err := filer2.ReadIntoBuffer(ctx, f.fs, filer2.FullPath(f.name), p, chunkViews, f.off) + totalRead, err := filer2.ReadIntoBuffer(f.fs, filer2.FullPath(f.name), p, chunkViews, f.off) if err != nil { return 0, err } @@ -512,11 +511,10 @@ func (f *WebDavFile) Read(p []byte) (readSize int, err error) { func (f *WebDavFile) Readdir(count int) (ret []os.FileInfo, err error) { glog.V(2).Infof("WebDavFileSystem.Readdir %v count %d", f.name, count) - ctx := context.Background() dir, _ := filer2.FullPath(f.name).DirAndName() - err = filer2.ReadDirAllEntries(ctx, f.fs, filer2.FullPath(dir), "", func(entry *filer_pb.Entry, isLast bool) { + err = filer2.ReadDirAllEntries(f.fs, filer2.FullPath(dir), "", func(entry *filer_pb.Entry, isLast bool) { fi := FileInfo{ size: int64(filer2.TotalSize(entry.GetChunks())), name: entry.Name, |
