volume_vacuum.go 16 KB

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