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.

396 lines
12 KiB

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