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.

315 lines
9.2 KiB

10 years ago
10 years ago
8 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/util"
  8. )
  9. func (v *Volume) garbageLevel() float64 {
  10. return float64(v.nm.DeletedSize()) / float64(v.ContentSize())
  11. }
  12. func (v *Volume) Compact() error {
  13. glog.V(3).Infof("Compacting ...")
  14. //no need to lock for copy on write
  15. //v.accessLock.Lock()
  16. //defer v.accessLock.Unlock()
  17. //glog.V(3).Infof("Got Compaction lock...")
  18. filePath := v.FileName()
  19. v.lastCompactIndexOffset = v.nm.IndexFileSize()
  20. v.lastCompactRevision = v.SuperBlock.CompactRevision
  21. glog.V(3).Infof("creating copies for volume %d ,last offset %d...", v.Id, v.lastCompactIndexOffset)
  22. return v.copyDataAndGenerateIndexFile(filePath+".cpd", filePath+".cpx", v.dataFileSize)
  23. }
  24. func (v *Volume) Compact2() error {
  25. glog.V(3).Infof("Compact2 ...")
  26. filePath := v.FileName()
  27. glog.V(3).Infof("creating copies for volume %d ...", v.Id)
  28. return v.copyDataBasedOnIndexFile(filePath+".cpd", filePath+".cpx")
  29. }
  30. func (v *Volume) commitCompact() error {
  31. glog.V(0).Infof("Committing vacuuming...")
  32. v.dataFileAccessLock.Lock()
  33. defer v.dataFileAccessLock.Unlock()
  34. glog.V(3).Infof("Got Committing lock...")
  35. v.nm.Close()
  36. _ = v.dataFile.Close()
  37. var e error
  38. if e = v.makeupDiff(v.FileName()+".cpd", v.FileName()+".cpx", v.FileName()+".dat", v.FileName()+".idx"); e != nil {
  39. glog.V(0).Infof("makeupDiff in commitCompact failed %v", e)
  40. e = os.Remove(v.FileName() + ".cpd")
  41. if e != nil {
  42. return e
  43. }
  44. e = os.Remove(v.FileName() + ".cpx")
  45. if e != nil {
  46. return e
  47. }
  48. } else {
  49. var e error
  50. if e = os.Rename(v.FileName()+".cpd", v.FileName()+".dat"); e != nil {
  51. return e
  52. }
  53. if e = os.Rename(v.FileName()+".cpx", v.FileName()+".idx"); e != nil {
  54. return e
  55. }
  56. }
  57. //glog.V(3).Infof("Pretending to be vacuuming...")
  58. //time.Sleep(20 * time.Second)
  59. glog.V(3).Infof("Loading Commit file...")
  60. if e = v.load(true, false, v.needleMapKind, 0); e != nil {
  61. return e
  62. }
  63. return nil
  64. }
  65. func fetchCompactRevisionFromDatFile(file *os.File) (compactRevision uint16, err error) {
  66. if _, err = file.Seek(0, 0); err != nil {
  67. return 0, fmt.Errorf("cannot seek to the beginning of %s: %v", file.Name(), err)
  68. }
  69. header := make([]byte, SuperBlockSize)
  70. if _, e := file.Read(header); e != nil {
  71. return 0, fmt.Errorf("cannot read file %s 's super block: %v", file.Name(), e)
  72. }
  73. superBlock, err := ParseSuperBlock(header)
  74. if err != nil {
  75. return 0, err
  76. }
  77. return superBlock.CompactRevision, nil
  78. }
  79. func (v *Volume) makeupDiff(newDatFileName, newIdxFileName, oldDatFileName, oldIdxFileName string) (err error) {
  80. var indexSize int64
  81. oldIdxFile, err := os.Open(oldIdxFileName)
  82. defer oldIdxFile.Close()
  83. oldDatFile, err := os.Open(oldDatFileName)
  84. defer oldDatFile.Close()
  85. if indexSize, err = verifyIndexFileIntegrity(oldIdxFile); err != nil {
  86. return fmt.Errorf("verifyIndexFileIntegrity %s failed: %v", oldIdxFileName, err)
  87. }
  88. if indexSize == 0 || uint64(indexSize) <= v.lastCompactIndexOffset {
  89. return nil
  90. }
  91. oldDatCompactRevision, err := fetchCompactRevisionFromDatFile(oldDatFile)
  92. if err != nil {
  93. return
  94. }
  95. if oldDatCompactRevision != v.lastCompactRevision {
  96. return fmt.Errorf("current old dat file's compact revision %d is not the expected one %d", oldDatCompactRevision, v.lastCompactRevision)
  97. }
  98. type keyField struct {
  99. offset uint32
  100. size uint32
  101. }
  102. incrementedHasUpdatedIndexEntry := make(map[uint64]keyField)
  103. for idx_offset := indexSize - NeedleIndexSize; uint64(idx_offset) >= v.lastCompactIndexOffset; idx_offset -= NeedleIndexSize {
  104. var IdxEntry []byte
  105. if IdxEntry, err = readIndexEntryAtOffset(oldIdxFile, idx_offset); err != nil {
  106. return fmt.Errorf("readIndexEntry %s at offset %d failed: %v", oldIdxFileName, idx_offset, err)
  107. }
  108. key, offset, size := idxFileEntry(IdxEntry)
  109. if _, found := incrementedHasUpdatedIndexEntry[key]; !found {
  110. incrementedHasUpdatedIndexEntry[key] = keyField{
  111. offset: offset,
  112. size: size,
  113. }
  114. }
  115. }
  116. if len(incrementedHasUpdatedIndexEntry) > 0 {
  117. var (
  118. dst, idx *os.File
  119. )
  120. if dst, err = os.OpenFile(newDatFileName, os.O_RDWR, 0644); err != nil {
  121. return
  122. }
  123. defer dst.Close()
  124. if idx, err = os.OpenFile(newIdxFileName, os.O_RDWR, 0644); err != nil {
  125. return
  126. }
  127. defer idx.Close()
  128. var newDatCompactRevision uint16
  129. newDatCompactRevision, err = fetchCompactRevisionFromDatFile(dst)
  130. if err != nil {
  131. return
  132. }
  133. if oldDatCompactRevision+1 != newDatCompactRevision {
  134. return fmt.Errorf("oldDatFile %s 's compact revision is %d while newDatFile %s 's compact revision is %d", oldDatFileName, oldDatCompactRevision, newDatFileName, newDatCompactRevision)
  135. }
  136. idx_entry_bytes := make([]byte, 16)
  137. for key, incre_idx_entry := range incrementedHasUpdatedIndexEntry {
  138. util.Uint64toBytes(idx_entry_bytes[0:8], key)
  139. util.Uint32toBytes(idx_entry_bytes[8:12], incre_idx_entry.offset)
  140. util.Uint32toBytes(idx_entry_bytes[12:16], incre_idx_entry.size)
  141. var offset int64
  142. if offset, err = dst.Seek(0, 2); err != nil {
  143. glog.V(0).Infof("failed to seek the end of file: %v", err)
  144. return
  145. }
  146. //ensure file writing starting from aligned positions
  147. if offset%NeedlePaddingSize != 0 {
  148. offset = offset + (NeedlePaddingSize - offset%NeedlePaddingSize)
  149. if offset, err = v.dataFile.Seek(offset, 0); err != nil {
  150. glog.V(0).Infof("failed to align in datafile %s: %v", v.dataFile.Name(), err)
  151. return
  152. }
  153. }
  154. //updated needle
  155. if incre_idx_entry.offset != 0 && incre_idx_entry.size != 0 {
  156. //even the needle cache in memory is hit, the need_bytes is correct
  157. var needle_bytes []byte
  158. needle_bytes, _, err = ReadNeedleBlob(oldDatFile, int64(incre_idx_entry.offset)*NeedlePaddingSize, incre_idx_entry.size)
  159. if err != nil {
  160. return
  161. }
  162. dst.Write(needle_bytes)
  163. util.Uint32toBytes(idx_entry_bytes[8:12], uint32(offset/NeedlePaddingSize))
  164. } else { //deleted needle
  165. //fakeDelNeedle 's default Data field is nil
  166. fakeDelNeedle := new(Needle)
  167. fakeDelNeedle.Id = key
  168. fakeDelNeedle.Cookie = 0x12345678
  169. _, _, err = fakeDelNeedle.Append(dst, v.Version())
  170. if err != nil {
  171. return
  172. }
  173. util.Uint32toBytes(idx_entry_bytes[8:12], uint32(0))
  174. }
  175. if _, err := idx.Seek(0, 2); err != nil {
  176. return fmt.Errorf("cannot seek end of indexfile %s: %v",
  177. newIdxFileName, err)
  178. }
  179. _, err = idx.Write(idx_entry_bytes)
  180. }
  181. }
  182. return nil
  183. }
  184. func (v *Volume) copyDataAndGenerateIndexFile(dstName, idxName string, preallocate int64) (err error) {
  185. var (
  186. dst, idx *os.File
  187. )
  188. if dst, err = createVolumeFile(dstName, preallocate); err != nil {
  189. return
  190. }
  191. defer dst.Close()
  192. if idx, err = os.OpenFile(idxName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644); err != nil {
  193. return
  194. }
  195. defer idx.Close()
  196. nm := NewNeedleMap(idx)
  197. new_offset := int64(SuperBlockSize)
  198. now := uint64(time.Now().Unix())
  199. err = ScanVolumeFile(v.dir, v.Collection, v.Id, v.needleMapKind,
  200. func(superBlock SuperBlock) error {
  201. superBlock.CompactRevision++
  202. _, err = dst.Write(superBlock.Bytes())
  203. return err
  204. }, true, func(n *Needle, offset int64) error {
  205. if n.HasTtl() && now >= n.LastModified+uint64(v.Ttl.Minutes()*60) {
  206. return nil
  207. }
  208. nv, ok := v.nm.Get(n.Id)
  209. glog.V(4).Infoln("needle expected offset ", offset, "ok", ok, "nv", nv)
  210. if ok && int64(nv.Offset)*NeedlePaddingSize == offset && nv.Size > 0 {
  211. if err = nm.Put(n.Id, uint32(new_offset/NeedlePaddingSize), n.Size); err != nil {
  212. return fmt.Errorf("cannot put needle: %s", err)
  213. }
  214. if _, _, err := n.Append(dst, v.Version()); err != nil {
  215. return fmt.Errorf("cannot append needle: %s", err)
  216. }
  217. new_offset += n.DiskSize()
  218. glog.V(3).Infoln("saving key", n.Id, "volume offset", offset, "=>", new_offset, "data_size", n.Size)
  219. }
  220. return nil
  221. })
  222. return
  223. }
  224. func (v *Volume) copyDataBasedOnIndexFile(dstName, idxName string) (err error) {
  225. var (
  226. dst, idx, oldIndexFile *os.File
  227. )
  228. if dst, err = os.OpenFile(dstName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644); err != nil {
  229. return
  230. }
  231. defer dst.Close()
  232. if idx, err = os.OpenFile(idxName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644); err != nil {
  233. return
  234. }
  235. defer idx.Close()
  236. if oldIndexFile, err = os.OpenFile(v.FileName()+".idx", os.O_RDONLY, 0644); err != nil {
  237. return
  238. }
  239. defer oldIndexFile.Close()
  240. nm := NewNeedleMap(idx)
  241. now := uint64(time.Now().Unix())
  242. v.SuperBlock.CompactRevision++
  243. dst.Write(v.SuperBlock.Bytes())
  244. new_offset := int64(SuperBlockSize)
  245. WalkIndexFile(oldIndexFile, func(key uint64, offset, size uint32) error {
  246. if offset == 0 || size == TombstoneFileSize {
  247. return nil
  248. }
  249. nv, ok := v.nm.Get(key)
  250. if !ok {
  251. return nil
  252. }
  253. n := new(Needle)
  254. n.ReadData(v.dataFile, int64(offset)*NeedlePaddingSize, size, v.Version())
  255. defer n.ReleaseMemory()
  256. if n.HasTtl() && now >= n.LastModified+uint64(v.Ttl.Minutes()*60) {
  257. return nil
  258. }
  259. glog.V(4).Infoln("needle expected offset ", offset, "ok", ok, "nv", nv)
  260. if nv.Offset == offset && nv.Size > 0 {
  261. if err = nm.Put(n.Id, uint32(new_offset/NeedlePaddingSize), n.Size); err != nil {
  262. return fmt.Errorf("cannot put needle: %s", err)
  263. }
  264. if _, _, err = n.Append(dst, v.Version()); err != nil {
  265. return fmt.Errorf("cannot append needle: %s", err)
  266. }
  267. new_offset += n.DiskSize()
  268. glog.V(3).Infoln("saving key", n.Id, "volume offset", offset, "=>", new_offset, "data_size", n.Size)
  269. }
  270. return nil
  271. })
  272. return
  273. }