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.

140 lines
4.3 KiB

10 years ago
  1. package main
  2. import (
  3. "flag"
  4. "fmt"
  5. "io"
  6. "os"
  7. "path"
  8. "strconv"
  9. "github.com/chrislusf/seaweedfs/weed/glog"
  10. "github.com/chrislusf/seaweedfs/weed/storage"
  11. "github.com/chrislusf/seaweedfs/weed/util"
  12. "github.com/chrislusf/seaweedfs/weed/storage/types"
  13. )
  14. var (
  15. fixVolumePath = flag.String("dir", "/tmp", "data directory to store files")
  16. fixVolumeCollection = flag.String("collection", "", "the volume collection name")
  17. fixVolumeId = flag.Int("volumeId", -1, "a volume id. The volume should already exist in the dir. The volume index file should not exist.")
  18. )
  19. /*
  20. This is to resolve an one-time issue that caused inconsistency with .dat and .idx files.
  21. In this case, the .dat file contains all data, but some of deletion caused incorrect offset.
  22. The .idx has all correct offsets.
  23. 1. fix the .dat file, a new .dat_fixed file will be generated.
  24. go run fix_dat.go -volumeId=9 -dir=/Users/chrislu/Downloads
  25. 2. move the original .dat and .idx files to some backup folder, and rename .dat_fixed to .dat file
  26. mv 9.dat_fixed 9.dat
  27. 3. fix the .idx file with the "weed fix"
  28. weed fix -volumeId=9 -dir=/Users/chrislu/Downloads
  29. */
  30. func main() {
  31. flag.Parse()
  32. fileName := strconv.Itoa(*fixVolumeId)
  33. if *fixVolumeCollection != "" {
  34. fileName = *fixVolumeCollection + "_" + fileName
  35. }
  36. indexFile, err := os.OpenFile(path.Join(*fixVolumePath, fileName+".idx"), os.O_RDONLY, 0644)
  37. if err != nil {
  38. glog.Fatalf("Read Volume Index %v", err)
  39. }
  40. defer indexFile.Close()
  41. datFile, err := os.OpenFile(path.Join(*fixVolumePath, fileName+".dat"), os.O_RDONLY, 0644)
  42. if err != nil {
  43. glog.Fatalf("Read Volume Data %v", err)
  44. }
  45. defer datFile.Close()
  46. newDatFile, err := os.Create(path.Join(*fixVolumePath, fileName+".dat_fixed"))
  47. if err != nil {
  48. glog.Fatalf("Write New Volume Data %v", err)
  49. }
  50. defer newDatFile.Close()
  51. superBlock, err := storage.ReadSuperBlock(datFile)
  52. if err != nil {
  53. glog.Fatalf("Read Volume Data superblock %v", err)
  54. }
  55. newDatFile.Write(superBlock.Bytes())
  56. iterateEntries(datFile, indexFile, func(n *storage.Needle, offset int64) {
  57. fmt.Printf("file id=%d name=%s size=%d dataSize=%d\n", n.Id, string(n.Name), n.Size, n.DataSize)
  58. s, _, e := n.Append(newDatFile, superBlock.Version())
  59. fmt.Printf("size %d error %v\n", s, e)
  60. })
  61. }
  62. func iterateEntries(datFile, idxFile *os.File, visitNeedle func(n *storage.Needle, offset int64)) {
  63. // start to read index file
  64. var readerOffset int64
  65. bytes := make([]byte, 16)
  66. count, _ := idxFile.ReadAt(bytes, readerOffset)
  67. readerOffset += int64(count)
  68. // start to read dat file
  69. superblock, err := storage.ReadSuperBlock(datFile)
  70. if err != nil {
  71. fmt.Printf("cannot read dat file super block: %v", err)
  72. return
  73. }
  74. offset := int64(superblock.BlockSize())
  75. version := superblock.Version()
  76. n, rest, err := storage.ReadNeedleHeader(datFile, version, offset)
  77. if err != nil {
  78. fmt.Printf("cannot read needle header: %v", err)
  79. return
  80. }
  81. fmt.Printf("Needle %+v, rest %d\n", n, rest)
  82. for n != nil && count > 0 {
  83. // parse index file entry
  84. key := util.BytesToUint64(bytes[0:8])
  85. offsetFromIndex := util.BytesToUint32(bytes[8:12])
  86. sizeFromIndex := util.BytesToUint32(bytes[12:16])
  87. count, _ = idxFile.ReadAt(bytes, readerOffset)
  88. readerOffset += int64(count)
  89. if offsetFromIndex != 0 && offset != int64(offsetFromIndex)*8 {
  90. //t := offset
  91. offset = int64(offsetFromIndex) * 8
  92. //fmt.Printf("Offset change %d => %d\n", t, offset)
  93. }
  94. fmt.Printf("key: %d offsetFromIndex %d n.Size %d sizeFromIndex:%d\n", key, offsetFromIndex, n.Size, sizeFromIndex)
  95. rest = storage.NeedleBodyLength(sizeFromIndex, version)
  96. func() {
  97. defer func() {
  98. if r := recover(); r != nil {
  99. fmt.Println("Recovered in f", r)
  100. }
  101. }()
  102. if err = n.ReadNeedleBody(datFile, version, offset+int64(types.NeedleEntrySize), rest); err != nil {
  103. fmt.Printf("cannot read needle body: offset %d body %d %v\n", offset, rest, err)
  104. }
  105. }()
  106. if n.Size <= n.DataSize {
  107. continue
  108. }
  109. visitNeedle(n, offset)
  110. offset += types.NeedleEntrySize + rest
  111. //fmt.Printf("==> new entry offset %d\n", offset)
  112. if n, rest, err = storage.ReadNeedleHeader(datFile, version, offset); err != nil {
  113. if err == io.EOF {
  114. return
  115. }
  116. fmt.Printf("cannot read needle header: %v\n", err)
  117. return
  118. }
  119. //fmt.Printf("new entry needle size:%d rest:%d\n", n.Size, rest)
  120. }
  121. }