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.

131 lines
3.0 KiB

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. var ret []*volume_server_pb.DeleteResult
  27. vid_to_fileIds := make(map[string][]string)
  28. var vids []string
  29. for _, fileId := range fileIds {
  30. vid, _, err := ParseFileId(fileId)
  31. if err != nil {
  32. ret = append(ret, &volume_server_pb.DeleteResult{
  33. FileId: vid,
  34. Status: http.StatusBadRequest,
  35. Error: err.Error()},
  36. )
  37. continue
  38. }
  39. if _, ok := vid_to_fileIds[vid]; !ok {
  40. vid_to_fileIds[vid] = make([]string, 0)
  41. vids = append(vids, vid)
  42. }
  43. vid_to_fileIds[vid] = append(vid_to_fileIds[vid], fileId)
  44. }
  45. lookupResults, err := LookupVolumeIds(master, vids)
  46. if err != nil {
  47. return ret, err
  48. }
  49. server_to_fileIds := make(map[string][]string)
  50. for vid, result := range lookupResults {
  51. if result.Error != "" {
  52. ret = append(ret, &volume_server_pb.DeleteResult{
  53. FileId: vid,
  54. Status: http.StatusBadRequest,
  55. Error: err.Error()},
  56. )
  57. continue
  58. }
  59. for _, location := range result.Locations {
  60. if _, ok := server_to_fileIds[location.Url]; !ok {
  61. server_to_fileIds[location.Url] = make([]string, 0)
  62. }
  63. server_to_fileIds[location.Url] = append(
  64. server_to_fileIds[location.Url], vid_to_fileIds[vid]...)
  65. }
  66. }
  67. var wg sync.WaitGroup
  68. for server, fidList := range server_to_fileIds {
  69. wg.Add(1)
  70. go func(server string, fidList []string) {
  71. defer wg.Done()
  72. if deleteResults, deleteErr := DeleteFilesAtOneVolumeServer(server, fidList); deleteErr != nil {
  73. err = deleteErr
  74. } else {
  75. ret = append(ret, deleteResults...)
  76. }
  77. }(server, fidList)
  78. }
  79. wg.Wait()
  80. return ret, err
  81. }
  82. // DeleteFilesAtOneVolumeServer deletes a list of files that is on one volume server via gRpc
  83. func DeleteFilesAtOneVolumeServer(volumeServer string, fileIds []string) (ret []*volume_server_pb.DeleteResult, err error) {
  84. err = WithVolumeServerClient(volumeServer, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
  85. req := &volume_server_pb.BatchDeleteRequest{
  86. FileIds: fileIds,
  87. }
  88. resp, err := volumeServerClient.BatchDelete(context.Background(), req)
  89. fmt.Printf("deleted %v %v: %v\n", fileIds, err, resp)
  90. if err != nil {
  91. return err
  92. }
  93. ret = append(ret, resp.Results...)
  94. return nil
  95. })
  96. if err != nil {
  97. return
  98. }
  99. for _, result := range ret {
  100. if result.Error != "" {
  101. return nil, fmt.Errorf("delete fileId %s: %v", result.FileId, result.Error)
  102. }
  103. }
  104. return
  105. }