page_chunk_swapfile.go 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198
  1. package page_writer
  2. import (
  3. "github.com/seaweedfs/seaweedfs/weed/glog"
  4. "github.com/seaweedfs/seaweedfs/weed/util"
  5. "github.com/seaweedfs/seaweedfs/weed/util/mem"
  6. "os"
  7. "sync"
  8. )
  9. var (
  10. _ = PageChunk(&SwapFileChunk{})
  11. )
  12. type ActualChunkIndex int
  13. type SwapFile struct {
  14. dir string
  15. file *os.File
  16. chunkSize int64
  17. chunkTrackingLock sync.Mutex
  18. activeChunkCount int
  19. freeActualChunkList []ActualChunkIndex
  20. }
  21. type SwapFileChunk struct {
  22. sync.RWMutex
  23. swapfile *SwapFile
  24. usage *ChunkWrittenIntervalList
  25. logicChunkIndex LogicChunkIndex
  26. actualChunkIndex ActualChunkIndex
  27. activityScore *ActivityScore
  28. //memChunk *MemChunk
  29. }
  30. func NewSwapFile(dir string, chunkSize int64) *SwapFile {
  31. return &SwapFile{
  32. dir: dir,
  33. file: nil,
  34. chunkSize: chunkSize,
  35. }
  36. }
  37. func (sf *SwapFile) FreeResource() {
  38. if sf.file != nil {
  39. sf.file.Close()
  40. os.Remove(sf.file.Name())
  41. }
  42. }
  43. func (sf *SwapFile) NewSwapFileChunk(logicChunkIndex LogicChunkIndex) (tc *SwapFileChunk) {
  44. if sf.file == nil {
  45. var err error
  46. sf.file, err = os.CreateTemp(sf.dir, "")
  47. if err != nil {
  48. glog.Errorf("create swap file: %v", err)
  49. return nil
  50. }
  51. }
  52. sf.chunkTrackingLock.Lock()
  53. defer sf.chunkTrackingLock.Unlock()
  54. sf.activeChunkCount++
  55. // assign a new physical chunk
  56. var actualChunkIndex ActualChunkIndex
  57. if len(sf.freeActualChunkList) > 0 {
  58. actualChunkIndex = sf.freeActualChunkList[0]
  59. sf.freeActualChunkList = sf.freeActualChunkList[1:]
  60. } else {
  61. actualChunkIndex = ActualChunkIndex(sf.activeChunkCount)
  62. }
  63. swapFileChunk := &SwapFileChunk{
  64. swapfile: sf,
  65. usage: newChunkWrittenIntervalList(),
  66. logicChunkIndex: logicChunkIndex,
  67. actualChunkIndex: actualChunkIndex,
  68. activityScore: NewActivityScore(),
  69. // memChunk: NewMemChunk(logicChunkIndex, sf.chunkSize),
  70. }
  71. // println(logicChunkIndex, "|", "++++", swapFileChunk.actualChunkIndex, swapFileChunk, sf)
  72. return swapFileChunk
  73. }
  74. func (sc *SwapFileChunk) FreeResource() {
  75. sc.Lock()
  76. defer sc.Unlock()
  77. sc.swapfile.chunkTrackingLock.Lock()
  78. defer sc.swapfile.chunkTrackingLock.Unlock()
  79. sc.swapfile.freeActualChunkList = append(sc.swapfile.freeActualChunkList, sc.actualChunkIndex)
  80. sc.swapfile.activeChunkCount--
  81. // println(sc.logicChunkIndex, "|", "----", sc.actualChunkIndex, sc, sc.swapfile)
  82. }
  83. func (sc *SwapFileChunk) WriteDataAt(src []byte, offset int64, tsNs int64) (n int) {
  84. sc.Lock()
  85. defer sc.Unlock()
  86. // println(sc.logicChunkIndex, "|", tsNs, "write at", offset, len(src), sc.actualChunkIndex)
  87. innerOffset := offset % sc.swapfile.chunkSize
  88. var err error
  89. n, err = sc.swapfile.file.WriteAt(src, int64(sc.actualChunkIndex)*sc.swapfile.chunkSize+innerOffset)
  90. sc.usage.MarkWritten(innerOffset, innerOffset+int64(n), tsNs)
  91. if err != nil {
  92. glog.Errorf("failed to write swap file %s: %v", sc.swapfile.file.Name(), err)
  93. }
  94. //sc.memChunk.WriteDataAt(src, offset, tsNs)
  95. sc.activityScore.MarkWrite()
  96. return
  97. }
  98. func (sc *SwapFileChunk) ReadDataAt(p []byte, off int64, tsNs int64) (maxStop int64) {
  99. sc.RLock()
  100. defer sc.RUnlock()
  101. // println(sc.logicChunkIndex, "|", tsNs, "read at", off, len(p), sc.actualChunkIndex)
  102. //memCopy := make([]byte, len(p))
  103. //copy(memCopy, p)
  104. chunkStartOffset := int64(sc.logicChunkIndex) * sc.swapfile.chunkSize
  105. for t := sc.usage.head.next; t != sc.usage.tail; t = t.next {
  106. logicStart := max(off, chunkStartOffset+t.StartOffset)
  107. logicStop := min(off+int64(len(p)), chunkStartOffset+t.stopOffset)
  108. if logicStart < logicStop {
  109. if t.TsNs >= tsNs {
  110. actualStart := logicStart - chunkStartOffset + int64(sc.actualChunkIndex)*sc.swapfile.chunkSize
  111. if _, err := sc.swapfile.file.ReadAt(p[logicStart-off:logicStop-off], actualStart); err != nil {
  112. glog.Errorf("failed to reading swap file %s: %v", sc.swapfile.file.Name(), err)
  113. break
  114. }
  115. maxStop = max(maxStop, logicStop)
  116. } else {
  117. println("read old data2", tsNs-t.TsNs, "ns")
  118. }
  119. }
  120. }
  121. //sc.memChunk.ReadDataAt(memCopy, off, tsNs)
  122. //if bytes.Compare(memCopy, p) != 0 {
  123. // println("read wrong data from swap file", off, sc.logicChunkIndex)
  124. //}
  125. sc.activityScore.MarkRead()
  126. return
  127. }
  128. func (sc *SwapFileChunk) IsComplete() bool {
  129. sc.RLock()
  130. defer sc.RUnlock()
  131. return sc.usage.IsComplete(sc.swapfile.chunkSize)
  132. }
  133. func (sc *SwapFileChunk) ActivityScore() int64 {
  134. return sc.activityScore.ActivityScore()
  135. }
  136. func (sc *SwapFileChunk) WrittenSize() int64 {
  137. sc.RLock()
  138. defer sc.RUnlock()
  139. return sc.usage.WrittenSize()
  140. }
  141. func (sc *SwapFileChunk) SaveContent(saveFn SaveToStorageFunc) {
  142. sc.RLock()
  143. defer sc.RUnlock()
  144. if saveFn == nil {
  145. return
  146. }
  147. // println(sc.logicChunkIndex, "|", "save")
  148. for t := sc.usage.head.next; t != sc.usage.tail; t = t.next {
  149. startOffset := t.StartOffset
  150. stopOffset := t.stopOffset
  151. tsNs := t.TsNs
  152. for t != sc.usage.tail && t.next.StartOffset == stopOffset {
  153. stopOffset = t.next.stopOffset
  154. t = t.next
  155. tsNs = max(tsNs, t.TsNs)
  156. }
  157. data := mem.Allocate(int(stopOffset - startOffset))
  158. n, _ := sc.swapfile.file.ReadAt(data, startOffset+int64(sc.actualChunkIndex)*sc.swapfile.chunkSize)
  159. if n > 0 {
  160. reader := util.NewBytesReader(data[:n])
  161. saveFn(reader, int64(sc.logicChunkIndex)*sc.swapfile.chunkSize+startOffset, int64(n), tsNs, func() {
  162. })
  163. }
  164. mem.Free(data)
  165. }
  166. }