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.

401 lines
12 KiB

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