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

Merge pull request #829 from PapaYofen/fix-823

Browse source 
Fix https://github.com/chrislusf/seaweedfs/issues/823
This commit is contained in:
Chris Lu 2019-01-10 09:02:13 -08:00 committed by GitHub
parent 2a75a36b27 39c7455881
commit e978f3aaab
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 58 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
weed/operation/assign_file_id.go
View file

@ -3,6 +3,7 @@ package operation
import (
"context"
"fmt"
"time"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
)
@ -40,6 +41,9 @@ func Assign(server string, primaryRequest *VolumeAssignRequest, alternativeReque
}
lastError = withMasterServerClient(server, func(masterClient master_pb.SeaweedClient) error {
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
defer cancel()
req := &master_pb.AssignRequest{
Count: primaryRequest.Count,
Replication: primaryRequest.Replication,
@ -49,7 +53,7 @@ func Assign(server string, primaryRequest *VolumeAssignRequest, alternativeReque
Rack: primaryRequest.Rack,
DataNode: primaryRequest.DataNode,
}
resp, grpcErr := masterClient.Assign(context.Background(), req)
resp, grpcErr := masterClient.Assign(ctx, req)
if grpcErr != nil {
return grpcErr
}

5
weed/operation/delete_content.go
View file

@ -7,6 +7,7 @@ import (
"net/http"
"strings"
"sync"
"time"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
)
@ -108,12 +109,14 @@ func DeleteFilesWithLookupVolumeId(fileIds []string, lookupFunc func(vid []strin
func DeleteFilesAtOneVolumeServer(volumeServer string, fileIds []string) (ret []*volume_server_pb.DeleteResult, err error) {
err = WithVolumeServerClient(volumeServer, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
defer cancel()
req := &volume_server_pb.BatchDeleteRequest{
FileIds: fileIds,
}
resp, err := volumeServerClient.BatchDelete(context.Background(), req)
resp, err := volumeServerClient.BatchDelete(ctx, req)
// fmt.Printf("deleted %v %v: %v\n", fileIds, err, resp)

5
weed/operation/lookup.go
View file

@ -99,10 +99,13 @@ func LookupVolumeIds(server string, vids []string) (map[string]LookupResult, err
//only query unknown_vids
err := withMasterServerClient(server, func(masterClient master_pb.SeaweedClient) error {
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
defer cancel()
req := &master_pb.LookupVolumeRequest{
VolumeIds: unknown_vids,
}
resp, grpcErr := masterClient.LookupVolume(context.Background(), req)
resp, grpcErr := masterClient.LookupVolume(ctx, req)
if grpcErr != nil {
return grpcErr
}

6
weed/operation/stats.go
View file

@ -2,6 +2,7 @@ package operation
import (
"context"
"time"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
)
@ -9,7 +10,10 @@ import (
func Statistics(server string, req *master_pb.StatisticsRequest) (resp *master_pb.StatisticsResponse, err error) {
err = withMasterServerClient(server, func(masterClient master_pb.SeaweedClient) error {
grpcResponse, grpcErr := masterClient.Statistics(context.Background(), req)
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
defer cancel()
grpcResponse, grpcErr := masterClient.Statistics(ctx, req)
if grpcErr != nil {
return grpcErr
}

6
weed/operation/sync_volume.go
View file

@ -4,6 +4,7 @@ import (
"context"
"fmt"
"io"
"time"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
. "github.com/chrislusf/seaweedfs/weed/storage/types"
@ -13,7 +14,10 @@ import (
func GetVolumeSyncStatus(server string, vid uint32) (resp *volume_server_pb.VolumeSyncStatusResponse, err error) {
WithVolumeServerClient(server, func(client volume_server_pb.VolumeServerClient) error {
resp, err = client.VolumeSyncStatus(context.Background(), &volume_server_pb.VolumeSyncStatusRequest{
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
defer cancel()
resp, err = client.VolumeSyncStatus(ctx, &volume_server_pb.VolumeSyncStatusRequest{
VolumdId: vid,
})
return nil

8
weed/operation/upload_content.go
View file

@ -13,6 +13,7 @@ import (
"net/textproto"
"path/filepath"
"strings"
"time"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/security"
@ -30,9 +31,10 @@ var (
)
func init() {
client = &http.Client{Transport: &http.Transport{
MaxIdleConnsPerHost: 1024,
}}
client = &http.Client{
Transport: &http.Transport{MaxIdleConnsPerHost: 1024},
Timeout: 5 * time.Second,
}
}
var fileNameEscaper = strings.NewReplacer("\\", "\\\\", "\"", "\\\"")

6
weed/server/master_server_handlers_admin.go
View file

@ -7,6 +7,7 @@ import (
"math/rand"
"net/http"
"strconv"
"time"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/operation"
@ -24,7 +25,10 @@ func (ms *MasterServer) collectionDeleteHandler(w http.ResponseWriter, r *http.R
}
for _, server := range collection.ListVolumeServers() {
err := operation.WithVolumeServerClient(server.Url(), func(client volume_server_pb.VolumeServerClient) error {
_, deleteErr := client.DeleteCollection(context.Background(), &volume_server_pb.DeleteCollectionRequest{
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
defer cancel()
_, deleteErr := client.DeleteCollection(ctx, &volume_server_pb.DeleteCollectionRequest{
Collection: collection.Name,
})
return deleteErr

6
weed/topology/allocate_volume.go
View file

@ -2,6 +2,7 @@ package topology
import (
"context"
"time"
"github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
@ -15,7 +16,10 @@ type AllocateVolumeResult struct {
func AllocateVolume(dn *DataNode, vid storage.VolumeId, option *VolumeGrowOption) error {
return operation.WithVolumeServerClient(dn.Url(), func(client volume_server_pb.VolumeServerClient) error {
_, deleteErr := client.AssignVolume(context.Background(), &volume_server_pb.AssignVolumeRequest{
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
defer cancel()
_, deleteErr := client.AssignVolume(ctx, &volume_server_pb.AssignVolumeRequest{
VolumdId: uint32(vid),
Collection: option.Collection,
Replication: option.ReplicaPlacement.String(),

16
weed/topology/topology_vacuum.go
View file

@ -15,7 +15,10 @@ func batchVacuumVolumeCheck(vl *VolumeLayout, vid storage.VolumeId, locationlist
for index, dn := range locationlist.list {
go func(index int, url string, vid storage.VolumeId) {
err := operation.WithVolumeServerClient(url, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
resp, err := volumeServerClient.VacuumVolumeCheck(context.Background(), &volume_server_pb.VacuumVolumeCheckRequest{
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
defer cancel()
resp, err := volumeServerClient.VacuumVolumeCheck(ctx, &volume_server_pb.VacuumVolumeCheckRequest{
VolumdId: uint32(vid),
})
if err != nil {
@ -48,6 +51,7 @@ func batchVacuumVolumeCompact(vl *VolumeLayout, vid storage.VolumeId, locationli
ch := make(chan bool, locationlist.Length())
for index, dn := range locationlist.list {
go func(index int, url string, vid storage.VolumeId) {
// TODO: set timeout according to actual compact duration
glog.V(0).Infoln(index, "Start vacuuming", vid, "on", url)
err := operation.WithVolumeServerClient(url, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
_, err := volumeServerClient.VacuumVolumeCompact(context.Background(), &volume_server_pb.VacuumVolumeCompactRequest{
@ -81,7 +85,10 @@ func batchVacuumVolumeCommit(vl *VolumeLayout, vid storage.VolumeId, locationlis
for _, dn := range locationlist.list {
glog.V(0).Infoln("Start Commiting vacuum", vid, "on", dn.Url())
err := operation.WithVolumeServerClient(dn.Url(), func(volumeServerClient volume_server_pb.VolumeServerClient) error {
_, err := volumeServerClient.VacuumVolumeCommit(context.Background(), &volume_server_pb.VacuumVolumeCommitRequest{
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
defer cancel()
_, err := volumeServerClient.VacuumVolumeCommit(ctx, &volume_server_pb.VacuumVolumeCommitRequest{
VolumdId: uint32(vid),
})
return err
@ -102,7 +109,10 @@ func batchVacuumVolumeCleanup(vl *VolumeLayout, vid storage.VolumeId, locationli
for _, dn := range locationlist.list {
glog.V(0).Infoln("Start cleaning up", vid, "on", dn.Url())
err := operation.WithVolumeServerClient(dn.Url(), func(volumeServerClient volume_server_pb.VolumeServerClient) error {
_, err := volumeServerClient.VacuumVolumeCleanup(context.Background(), &volume_server_pb.VacuumVolumeCleanupRequest{
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
defer cancel()
_, err := volumeServerClient.VacuumVolumeCleanup(ctx, &volume_server_pb.VacuumVolumeCleanupRequest{
VolumdId: uint32(vid),
})
return err

8
weed/util/http_util.go
View file

@ -11,6 +11,7 @@ import (
"net/http"
"net/url"
"strings"
"time"
"github.com/chrislusf/seaweedfs/weed/security"
)
@ -24,7 +25,10 @@ func init() {
Transport = &http.Transport{
MaxIdleConnsPerHost: 1024,
}
client = &http.Client{Transport: Transport}
client = &http.Client{
Transport: Transport,
Timeout: 5 * time.Second,
}
}
func PostBytes(url string, body []byte) ([]byte, error) {
@ -63,6 +67,8 @@ func Post(url string, values url.Values) ([]byte, error) {
return b, nil
}
// github.com/chrislusf/seaweedfs/unmaintained/repeated_vacuum/repeated_vacuum.go
// may need increasing http.Client.Timeout
func Get(url string) ([]byte, error) {
r, err := client.Get(url)
if err != nil {