You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

144 lines
3.4 KiB

6 years ago
6 years ago
6 years ago
  1. package operation
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "net/http"
  7. "strings"
  8. "sync"
  9. "time"
  10. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  11. )
  12. type DeleteResult struct {
  13. Fid string `json:"fid"`
  14. Size int `json:"size"`
  15. Status int `json:"status"`
  16. Error string `json:"error,omitempty"`
  17. }
  18. func ParseFileId(fid string) (vid string, key_cookie string, err error) {
  19. commaIndex := strings.Index(fid, ",")
  20. if commaIndex <= 0 {
  21. return "", "", errors.New("Wrong fid format.")
  22. }
  23. return fid[:commaIndex], fid[commaIndex+1:], nil
  24. }
  25. // DeleteFiles batch deletes a list of fileIds
  26. func DeleteFiles(master string, fileIds []string) ([]*volume_server_pb.DeleteResult, error) {
  27. lookupFunc := func(vids []string) (map[string]LookupResult, error) {
  28. return LookupVolumeIds(master, vids)
  29. }
  30. return DeleteFilesWithLookupVolumeId(fileIds, lookupFunc)
  31. }
  32. func DeleteFilesWithLookupVolumeId(fileIds []string, lookupFunc func(vid []string) (map[string]LookupResult, error)) ([]*volume_server_pb.DeleteResult, error) {
  33. var ret []*volume_server_pb.DeleteResult
  34. vid_to_fileIds := make(map[string][]string)
  35. var vids []string
  36. for _, fileId := range fileIds {
  37. vid, _, err := ParseFileId(fileId)
  38. if err != nil {
  39. ret = append(ret, &volume_server_pb.DeleteResult{
  40. FileId: vid,
  41. Status: http.StatusBadRequest,
  42. Error: err.Error()},
  43. )
  44. continue
  45. }
  46. if _, ok := vid_to_fileIds[vid]; !ok {
  47. vid_to_fileIds[vid] = make([]string, 0)
  48. vids = append(vids, vid)
  49. }
  50. vid_to_fileIds[vid] = append(vid_to_fileIds[vid], fileId)
  51. }
  52. lookupResults, err := lookupFunc(vids)
  53. if err != nil {
  54. return ret, err
  55. }
  56. server_to_fileIds := make(map[string][]string)
  57. for vid, result := range lookupResults {
  58. if result.Error != "" {
  59. ret = append(ret, &volume_server_pb.DeleteResult{
  60. FileId: vid,
  61. Status: http.StatusBadRequest,
  62. Error: err.Error()},
  63. )
  64. continue
  65. }
  66. for _, location := range result.Locations {
  67. if _, ok := server_to_fileIds[location.Url]; !ok {
  68. server_to_fileIds[location.Url] = make([]string, 0)
  69. }
  70. server_to_fileIds[location.Url] = append(
  71. server_to_fileIds[location.Url], vid_to_fileIds[vid]...)
  72. }
  73. }
  74. var wg sync.WaitGroup
  75. for server, fidList := range server_to_fileIds {
  76. wg.Add(1)
  77. go func(server string, fidList []string) {
  78. defer wg.Done()
  79. if deleteResults, deleteErr := DeleteFilesAtOneVolumeServer(server, fidList); deleteErr != nil {
  80. err = deleteErr
  81. } else {
  82. ret = append(ret, deleteResults...)
  83. }
  84. }(server, fidList)
  85. }
  86. wg.Wait()
  87. return ret, err
  88. }
  89. // DeleteFilesAtOneVolumeServer deletes a list of files that is on one volume server via gRpc
  90. func DeleteFilesAtOneVolumeServer(volumeServer string, fileIds []string) (ret []*volume_server_pb.DeleteResult, err error) {
  91. err = WithVolumeServerClient(volumeServer, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  92. ctx, cancel := context.WithTimeout(context.Background(), time.Duration(5*time.Second))
  93. defer cancel()
  94. req := &volume_server_pb.BatchDeleteRequest{
  95. FileIds: fileIds,
  96. }
  97. resp, err := volumeServerClient.BatchDelete(ctx, req)
  98. // fmt.Printf("deleted %v %v: %v\n", fileIds, err, resp)
  99. if err != nil {
  100. return err
  101. }
  102. ret = append(ret, resp.Results...)
  103. return nil
  104. })
  105. if err != nil {
  106. return
  107. }
  108. for _, result := range ret {
  109. if result.Error != "" && result.Error != "not found" {
  110. return nil, fmt.Errorf("delete fileId %s: %v", result.FileId, result.Error)
  111. }
  112. }
  113. return
  114. }