volume_vacuum.go 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451
  1. package storage
  2. import (
  3. "fmt"
  4. "os"
  5. "runtime"
  6. "time"
  7. "github.com/chrislusf/seaweedfs/weed/glog"
  8. "github.com/chrislusf/seaweedfs/weed/stats"
  9. "github.com/chrislusf/seaweedfs/weed/storage/backend"
  10. idx2 "github.com/chrislusf/seaweedfs/weed/storage/idx"
  11. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  12. "github.com/chrislusf/seaweedfs/weed/storage/needle_map"
  13. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  14. . "github.com/chrislusf/seaweedfs/weed/storage/types"
  15. "github.com/chrislusf/seaweedfs/weed/util"
  16. )
  17. func (v *Volume) garbageLevel() float64 {
  18. if v.ContentSize() == 0 {
  19. return 0
  20. }
  21. deletedSize := v.DeletedSize()
  22. fileSize := v.ContentSize()
  23. if v.DeletedCount() > 0 && v.DeletedSize() == 0 {
  24. // this happens for .sdx converted back to normal .idx
  25. // where deleted entry size is missing
  26. datFileSize, _, _ := v.FileStat()
  27. deletedSize = datFileSize - fileSize - super_block.SuperBlockSize
  28. fileSize = datFileSize
  29. }
  30. return float64(deletedSize) / float64(fileSize)
  31. }
  32. // compact a volume based on deletions in .dat files
  33. func (v *Volume) Compact(preallocate int64, compactionBytePerSecond int64) error {
  34. if v.MemoryMapMaxSizeMb != 0 { //it makes no sense to compact in memory
  35. return nil
  36. }
  37. glog.V(3).Infof("Compacting volume %d ...", v.Id)
  38. //no need to lock for copy on write
  39. //v.accessLock.Lock()
  40. //defer v.accessLock.Unlock()
  41. //glog.V(3).Infof("Got Compaction lock...")
  42. v.isCompacting = true
  43. defer func() {
  44. v.isCompacting = false
  45. }()
  46. v.lastCompactIndexOffset = v.IndexFileSize()
  47. v.lastCompactRevision = v.SuperBlock.CompactionRevision
  48. glog.V(3).Infof("creating copies for volume %d ,last offset %d...", v.Id, v.lastCompactIndexOffset)
  49. if err := v.DataBackend.Sync(); err != nil {
  50. glog.V(0).Infof("compact fail to sync volume %d", v.Id)
  51. }
  52. if err := v.nm.Sync(); err != nil {
  53. glog.V(0).Infof("compact fail to sync volume idx %d", v.Id)
  54. }
  55. return v.copyDataAndGenerateIndexFile(v.FileName(".cpd"), v.FileName(".cpx"), preallocate, compactionBytePerSecond)
  56. }
  57. // compact a volume based on deletions in .idx files
  58. func (v *Volume) Compact2(preallocate int64, compactionBytePerSecond int64) error {
  59. if v.MemoryMapMaxSizeMb != 0 { //it makes no sense to compact in memory
  60. return nil
  61. }
  62. glog.V(3).Infof("Compact2 volume %d ...", v.Id)
  63. v.isCompacting = true
  64. defer func() {
  65. v.isCompacting = false
  66. }()
  67. v.lastCompactIndexOffset = v.IndexFileSize()
  68. v.lastCompactRevision = v.SuperBlock.CompactionRevision
  69. glog.V(3).Infof("creating copies for volume %d ...", v.Id)
  70. if err := v.DataBackend.Sync(); err != nil {
  71. glog.V(0).Infof("compact2 fail to sync volume dat %d: %v", v.Id, err)
  72. }
  73. if err := v.nm.Sync(); err != nil {
  74. glog.V(0).Infof("compact2 fail to sync volume idx %d: %v", v.Id, err)
  75. }
  76. return copyDataBasedOnIndexFile(v.FileName(".dat"), v.FileName(".idx"), v.FileName(".cpd"), v.FileName(".cpx"), v.SuperBlock, v.Version(), preallocate, compactionBytePerSecond)
  77. }
  78. func (v *Volume) CommitCompact() error {
  79. if v.MemoryMapMaxSizeMb != 0 { //it makes no sense to compact in memory
  80. return nil
  81. }
  82. glog.V(0).Infof("Committing volume %d vacuuming...", v.Id)
  83. v.isCompacting = true
  84. defer func() {
  85. v.isCompacting = false
  86. }()
  87. v.dataFileAccessLock.Lock()
  88. defer v.dataFileAccessLock.Unlock()
  89. glog.V(3).Infof("Got volume %d committing lock...", v.Id)
  90. v.nm.Close()
  91. if v.DataBackend != nil {
  92. if err := v.DataBackend.Close(); err != nil {
  93. glog.V(0).Infof("fail to close volume %d", v.Id)
  94. }
  95. }
  96. v.DataBackend = nil
  97. stats.VolumeServerVolumeCounter.WithLabelValues(v.Collection, "volume").Dec()
  98. var e error
  99. if e = v.makeupDiff(v.FileName(".cpd"), v.FileName(".cpx"), v.FileName(".dat"), v.FileName(".idx")); e != nil {
  100. glog.V(0).Infof("makeupDiff in CommitCompact volume %d failed %v", v.Id, e)
  101. e = os.Remove(v.FileName(".cpd"))
  102. if e != nil {
  103. return e
  104. }
  105. e = os.Remove(v.FileName(".cpx"))
  106. if e != nil {
  107. return e
  108. }
  109. } else {
  110. if runtime.GOOS == "windows" {
  111. e = os.RemoveAll(v.FileName(".dat"))
  112. if e != nil {
  113. return e
  114. }
  115. e = os.RemoveAll(v.FileName(".idx"))
  116. if e != nil {
  117. return e
  118. }
  119. }
  120. var e error
  121. if e = os.Rename(v.FileName(".cpd"), v.FileName(".dat")); e != nil {
  122. return fmt.Errorf("rename %s: %v", v.FileName(".cpd"), e)
  123. }
  124. if e = os.Rename(v.FileName(".cpx"), v.FileName(".idx")); e != nil {
  125. return fmt.Errorf("rename %s: %v", v.FileName(".cpx"), e)
  126. }
  127. }
  128. //glog.V(3).Infof("Pretending to be vacuuming...")
  129. //time.Sleep(20 * time.Second)
  130. os.RemoveAll(v.FileName(".ldb"))
  131. glog.V(3).Infof("Loading volume %d commit file...", v.Id)
  132. if e = v.load(true, false, v.needleMapKind, 0); e != nil {
  133. return e
  134. }
  135. return nil
  136. }
  137. func (v *Volume) cleanupCompact() error {
  138. glog.V(0).Infof("Cleaning up volume %d vacuuming...", v.Id)
  139. e1 := os.Remove(v.FileName(".cpd"))
  140. e2 := os.Remove(v.FileName(".cpx"))
  141. if e1 != nil {
  142. return e1
  143. }
  144. if e2 != nil {
  145. return e2
  146. }
  147. return nil
  148. }
  149. func fetchCompactRevisionFromDatFile(datBackend backend.BackendStorageFile) (compactRevision uint16, err error) {
  150. superBlock, err := super_block.ReadSuperBlock(datBackend)
  151. if err != nil {
  152. return 0, err
  153. }
  154. return superBlock.CompactionRevision, nil
  155. }
  156. // if old .dat and .idx files are updated, this func tries to apply the same changes to new files accordingly
  157. func (v *Volume) makeupDiff(newDatFileName, newIdxFileName, oldDatFileName, oldIdxFileName string) (err error) {
  158. var indexSize int64
  159. oldIdxFile, err := os.Open(oldIdxFileName)
  160. if err != nil {
  161. return fmt.Errorf("makeupDiff open %s failed: %v", oldIdxFileName, err)
  162. }
  163. defer oldIdxFile.Close()
  164. oldDatFile, err := os.Open(oldDatFileName)
  165. if err != nil {
  166. return fmt.Errorf("makeupDiff open %s failed: %v", oldDatFileName, err)
  167. }
  168. oldDatBackend := backend.NewDiskFile(oldDatFile)
  169. defer oldDatBackend.Close()
  170. // skip if the old .idx file has not changed
  171. if indexSize, err = verifyIndexFileIntegrity(oldIdxFile); err != nil {
  172. return fmt.Errorf("verifyIndexFileIntegrity %s failed: %v", oldIdxFileName, err)
  173. }
  174. if indexSize == 0 || uint64(indexSize) <= v.lastCompactIndexOffset {
  175. return nil
  176. }
  177. // fail if the old .dat file has changed to a new revision
  178. oldDatCompactRevision, err := fetchCompactRevisionFromDatFile(oldDatBackend)
  179. if err != nil {
  180. return fmt.Errorf("fetchCompactRevisionFromDatFile src %s failed: %v", oldDatFile.Name(), err)
  181. }
  182. if oldDatCompactRevision != v.lastCompactRevision {
  183. return fmt.Errorf("current old dat file's compact revision %d is not the expected one %d", oldDatCompactRevision, v.lastCompactRevision)
  184. }
  185. type keyField struct {
  186. offset Offset
  187. size Size
  188. }
  189. incrementedHasUpdatedIndexEntry := make(map[NeedleId]keyField)
  190. for idxOffset := indexSize - NeedleMapEntrySize; uint64(idxOffset) >= v.lastCompactIndexOffset; idxOffset -= NeedleMapEntrySize {
  191. var IdxEntry []byte
  192. if IdxEntry, err = readIndexEntryAtOffset(oldIdxFile, idxOffset); err != nil {
  193. return fmt.Errorf("readIndexEntry %s at offset %d failed: %v", oldIdxFileName, idxOffset, err)
  194. }
  195. key, offset, size := idx2.IdxFileEntry(IdxEntry)
  196. glog.V(4).Infof("key %d offset %d size %d", key, offset, size)
  197. if _, found := incrementedHasUpdatedIndexEntry[key]; !found {
  198. incrementedHasUpdatedIndexEntry[key] = keyField{
  199. offset: offset,
  200. size: size,
  201. }
  202. }
  203. }
  204. // no updates during commit step
  205. if len(incrementedHasUpdatedIndexEntry) == 0 {
  206. return nil
  207. }
  208. // deal with updates during commit step
  209. var (
  210. dst, idx *os.File
  211. )
  212. if dst, err = os.OpenFile(newDatFileName, os.O_RDWR, 0644); err != nil {
  213. return fmt.Errorf("open dat file %s failed: %v", newDatFileName, err)
  214. }
  215. dstDatBackend := backend.NewDiskFile(dst)
  216. defer dstDatBackend.Close()
  217. if idx, err = os.OpenFile(newIdxFileName, os.O_RDWR, 0644); err != nil {
  218. return fmt.Errorf("open idx file %s failed: %v", newIdxFileName, err)
  219. }
  220. defer idx.Close()
  221. var newDatCompactRevision uint16
  222. newDatCompactRevision, err = fetchCompactRevisionFromDatFile(dstDatBackend)
  223. if err != nil {
  224. return fmt.Errorf("fetchCompactRevisionFromDatFile dst %s failed: %v", dst.Name(), err)
  225. }
  226. if oldDatCompactRevision+1 != newDatCompactRevision {
  227. return fmt.Errorf("oldDatFile %s 's compact revision is %d while newDatFile %s 's compact revision is %d", oldDatFileName, oldDatCompactRevision, newDatFileName, newDatCompactRevision)
  228. }
  229. for key, increIdxEntry := range incrementedHasUpdatedIndexEntry {
  230. idxEntryBytes := needle_map.ToBytes(key, increIdxEntry.offset, increIdxEntry.size)
  231. var offset int64
  232. if offset, err = dst.Seek(0, 2); err != nil {
  233. glog.V(0).Infof("failed to seek the end of file: %v", err)
  234. return
  235. }
  236. //ensure file writing starting from aligned positions
  237. if offset%NeedlePaddingSize != 0 {
  238. offset = offset + (NeedlePaddingSize - offset%NeedlePaddingSize)
  239. if offset, err = dst.Seek(offset, 0); err != nil {
  240. glog.V(0).Infof("failed to align in datafile %s: %v", dst.Name(), err)
  241. return
  242. }
  243. }
  244. //updated needle
  245. if !increIdxEntry.offset.IsZero() && increIdxEntry.size != 0 && increIdxEntry.size.IsValid() {
  246. //even the needle cache in memory is hit, the need_bytes is correct
  247. glog.V(4).Infof("file %d offset %d size %d", key, increIdxEntry.offset.ToActualOffset(), increIdxEntry.size)
  248. var needleBytes []byte
  249. needleBytes, err = needle.ReadNeedleBlob(oldDatBackend, increIdxEntry.offset.ToActualOffset(), increIdxEntry.size, v.Version())
  250. if err != nil {
  251. return fmt.Errorf("ReadNeedleBlob %s key %d offset %d size %d failed: %v", oldDatFile.Name(), key, increIdxEntry.offset.ToActualOffset(), increIdxEntry.size, err)
  252. }
  253. dstDatBackend.Write(needleBytes)
  254. util.Uint32toBytes(idxEntryBytes[8:12], uint32(offset/NeedlePaddingSize))
  255. } else { //deleted needle
  256. //fakeDelNeedle 's default Data field is nil
  257. fakeDelNeedle := new(needle.Needle)
  258. fakeDelNeedle.Id = key
  259. fakeDelNeedle.Cookie = 0x12345678
  260. fakeDelNeedle.AppendAtNs = uint64(time.Now().UnixNano())
  261. _, _, _, err = fakeDelNeedle.Append(dstDatBackend, v.Version())
  262. if err != nil {
  263. return fmt.Errorf("append deleted %d failed: %v", key, err)
  264. }
  265. util.Uint32toBytes(idxEntryBytes[8:12], uint32(0))
  266. }
  267. if _, err := idx.Seek(0, 2); err != nil {
  268. return fmt.Errorf("cannot seek end of indexfile %s: %v",
  269. newIdxFileName, err)
  270. }
  271. _, err = idx.Write(idxEntryBytes)
  272. }
  273. return nil
  274. }
  275. type VolumeFileScanner4Vacuum struct {
  276. version needle.Version
  277. v *Volume
  278. dstBackend backend.BackendStorageFile
  279. nm *needle_map.MemDb
  280. newOffset int64
  281. now uint64
  282. writeThrottler *util.WriteThrottler
  283. }
  284. func (scanner *VolumeFileScanner4Vacuum) VisitSuperBlock(superBlock super_block.SuperBlock) error {
  285. scanner.version = superBlock.Version
  286. superBlock.CompactionRevision++
  287. _, err := scanner.dstBackend.WriteAt(superBlock.Bytes(), 0)
  288. scanner.newOffset = int64(superBlock.BlockSize())
  289. return err
  290. }
  291. func (scanner *VolumeFileScanner4Vacuum) ReadNeedleBody() bool {
  292. return true
  293. }
  294. func (scanner *VolumeFileScanner4Vacuum) VisitNeedle(n *needle.Needle, offset int64, needleHeader, needleBody []byte) error {
  295. if n.HasTtl() && scanner.now >= n.LastModified+uint64(scanner.v.Ttl.Minutes()*60) {
  296. return nil
  297. }
  298. nv, ok := scanner.v.nm.Get(n.Id)
  299. glog.V(4).Infoln("needle expected offset ", offset, "ok", ok, "nv", nv)
  300. if ok && nv.Offset.ToActualOffset() == offset && nv.Size > 0 && nv.Size.IsValid() {
  301. if err := scanner.nm.Set(n.Id, ToOffset(scanner.newOffset), n.Size); err != nil {
  302. return fmt.Errorf("cannot put needle: %s", err)
  303. }
  304. if _, _, _, err := n.Append(scanner.dstBackend, scanner.v.Version()); err != nil {
  305. return fmt.Errorf("cannot append needle: %s", err)
  306. }
  307. delta := n.DiskSize(scanner.version)
  308. scanner.newOffset += delta
  309. scanner.writeThrottler.MaybeSlowdown(delta)
  310. glog.V(4).Infoln("saving key", n.Id, "volume offset", offset, "=>", scanner.newOffset, "data_size", n.Size)
  311. }
  312. return nil
  313. }
  314. func (v *Volume) copyDataAndGenerateIndexFile(dstName, idxName string, preallocate int64, compactionBytePerSecond int64) (err error) {
  315. var (
  316. dst backend.BackendStorageFile
  317. )
  318. if dst, err = backend.CreateVolumeFile(dstName, preallocate, 0); err != nil {
  319. return
  320. }
  321. defer dst.Close()
  322. nm := needle_map.NewMemDb()
  323. defer nm.Close()
  324. scanner := &VolumeFileScanner4Vacuum{
  325. v: v,
  326. now: uint64(time.Now().Unix()),
  327. nm: nm,
  328. dstBackend: dst,
  329. writeThrottler: util.NewWriteThrottler(compactionBytePerSecond),
  330. }
  331. err = ScanVolumeFile(v.dir, v.Collection, v.Id, v.needleMapKind, scanner)
  332. if err != nil {
  333. return nil
  334. }
  335. err = nm.SaveToIdx(idxName)
  336. return
  337. }
  338. func copyDataBasedOnIndexFile(srcDatName, srcIdxName, dstDatName, datIdxName string, sb super_block.SuperBlock, version needle.Version, preallocate int64, compactionBytePerSecond int64) (err error) {
  339. var (
  340. srcDatBackend, dstDatBackend backend.BackendStorageFile
  341. dataFile *os.File
  342. )
  343. if dstDatBackend, err = backend.CreateVolumeFile(dstDatName, preallocate, 0); err != nil {
  344. return
  345. }
  346. defer dstDatBackend.Close()
  347. oldNm := needle_map.NewMemDb()
  348. defer oldNm.Close()
  349. newNm := needle_map.NewMemDb()
  350. defer newNm.Close()
  351. if err = oldNm.LoadFromIdx(srcIdxName); err != nil {
  352. return
  353. }
  354. if dataFile, err = os.Open(srcDatName); err != nil {
  355. return err
  356. }
  357. srcDatBackend = backend.NewDiskFile(dataFile)
  358. defer srcDatBackend.Close()
  359. now := uint64(time.Now().Unix())
  360. sb.CompactionRevision++
  361. dstDatBackend.WriteAt(sb.Bytes(), 0)
  362. newOffset := int64(sb.BlockSize())
  363. writeThrottler := util.NewWriteThrottler(compactionBytePerSecond)
  364. oldNm.AscendingVisit(func(value needle_map.NeedleValue) error {
  365. offset, size := value.Offset, value.Size
  366. if offset.IsZero() || size.IsDeleted() {
  367. return nil
  368. }
  369. n := new(needle.Needle)
  370. err := n.ReadData(srcDatBackend, offset.ToActualOffset(), size, version)
  371. if err != nil {
  372. return nil
  373. }
  374. if n.HasTtl() && now >= n.LastModified+uint64(sb.Ttl.Minutes()*60) {
  375. return nil
  376. }
  377. if err = newNm.Set(n.Id, ToOffset(newOffset), n.Size); err != nil {
  378. return fmt.Errorf("cannot put needle: %s", err)
  379. }
  380. if _, _, _, err = n.Append(dstDatBackend, sb.Version); err != nil {
  381. return fmt.Errorf("cannot append needle: %s", err)
  382. }
  383. delta := n.DiskSize(version)
  384. newOffset += delta
  385. writeThrottler.MaybeSlowdown(delta)
  386. glog.V(4).Infoln("saving key", n.Id, "volume offset", offset, "=>", newOffset, "data_size", n.Size)
  387. return nil
  388. })
  389. newNm.SaveToIdx(datIdxName)
  390. return
  391. }