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.

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