ec_volume_delete.go 2.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798
  1. package erasure_coding
  2. import (
  3. "fmt"
  4. "io"
  5. "os"
  6. "github.com/seaweedfs/seaweedfs/weed/storage/types"
  7. "github.com/seaweedfs/seaweedfs/weed/util"
  8. )
  9. var (
  10. MarkNeedleDeleted = func(file *os.File, offset int64) error {
  11. b := make([]byte, types.SizeSize)
  12. types.SizeToBytes(b, types.TombstoneFileSize)
  13. n, err := file.WriteAt(b, offset+types.NeedleIdSize+types.OffsetSize)
  14. if err != nil {
  15. return fmt.Errorf("sorted needle write error: %v", err)
  16. }
  17. if n != types.SizeSize {
  18. return fmt.Errorf("sorted needle written %d bytes, expecting %d", n, types.SizeSize)
  19. }
  20. return nil
  21. }
  22. )
  23. func (ev *EcVolume) DeleteNeedleFromEcx(needleId types.NeedleId) (err error) {
  24. _, _, err = SearchNeedleFromSortedIndex(ev.ecxFile, ev.ecxFileSize, needleId, MarkNeedleDeleted)
  25. if err != nil {
  26. if err == NotFoundError {
  27. return nil
  28. }
  29. return err
  30. }
  31. b := make([]byte, types.NeedleIdSize)
  32. types.NeedleIdToBytes(b, needleId)
  33. ev.ecjFileAccessLock.Lock()
  34. ev.ecjFile.Seek(0, io.SeekEnd)
  35. ev.ecjFile.Write(b)
  36. ev.ecjFileAccessLock.Unlock()
  37. return
  38. }
  39. func RebuildEcxFile(baseFileName string) error {
  40. if !util.FileExists(baseFileName + ".ecj") {
  41. return nil
  42. }
  43. ecxFile, err := os.OpenFile(baseFileName+".ecx", os.O_RDWR, 0644)
  44. if err != nil {
  45. return fmt.Errorf("rebuild: failed to open ecx file: %v", err)
  46. }
  47. defer ecxFile.Close()
  48. fstat, err := ecxFile.Stat()
  49. if err != nil {
  50. return err
  51. }
  52. ecxFileSize := fstat.Size()
  53. ecjFile, err := os.OpenFile(baseFileName+".ecj", os.O_RDWR, 0644)
  54. if err != nil {
  55. return fmt.Errorf("rebuild: failed to open ecj file: %v", err)
  56. }
  57. buf := make([]byte, types.NeedleIdSize)
  58. for {
  59. n, _ := ecjFile.Read(buf)
  60. if n != types.NeedleIdSize {
  61. break
  62. }
  63. needleId := types.BytesToNeedleId(buf)
  64. _, _, err = SearchNeedleFromSortedIndex(ecxFile, ecxFileSize, needleId, MarkNeedleDeleted)
  65. if err != nil && err != NotFoundError {
  66. ecxFile.Close()
  67. return err
  68. }
  69. }
  70. ecxFile.Close()
  71. os.Remove(baseFileName + ".ecj")
  72. return nil
  73. }