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.

256 lines
7.9 KiB

3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
  1. package page_writer
  2. import (
  3. "fmt"
  4. "github.com/chrislusf/seaweedfs/weed/glog"
  5. "github.com/chrislusf/seaweedfs/weed/util"
  6. "github.com/chrislusf/seaweedfs/weed/util/mem"
  7. "sync"
  8. "sync/atomic"
  9. "time"
  10. )
  11. type LogicChunkIndex int
  12. type UploadPipeline struct {
  13. filepath util.FullPath
  14. ChunkSize int64
  15. writers *util.LimitedConcurrentExecutor
  16. writableChunks map[LogicChunkIndex]*MemChunk
  17. writableChunksLock sync.Mutex
  18. sealedChunks map[LogicChunkIndex]*SealedChunk
  19. sealedChunksLock sync.Mutex
  20. activeWriterCond *sync.Cond
  21. activeWriterCount int32
  22. activeReadChunks map[LogicChunkIndex]int
  23. activeReadChunksLock sync.Mutex
  24. saveToStorageFn SaveToStorageFunc
  25. }
  26. type SealedChunk struct {
  27. chunk *MemChunk
  28. referenceCounter int // track uploading or reading processes
  29. }
  30. func (sc *SealedChunk) FreeReference(messageOnFree string) {
  31. sc.referenceCounter--
  32. if sc.referenceCounter == 0 {
  33. glog.V(4).Infof("Free sealed chunk: %s", messageOnFree)
  34. mem.Free(sc.chunk.buf)
  35. }
  36. }
  37. func NewUploadPipeline(filepath util.FullPath, writers *util.LimitedConcurrentExecutor, chunkSize int64, saveToStorageFn SaveToStorageFunc) *UploadPipeline {
  38. return &UploadPipeline{
  39. ChunkSize: chunkSize,
  40. writableChunks: make(map[LogicChunkIndex]*MemChunk),
  41. sealedChunks: make(map[LogicChunkIndex]*SealedChunk),
  42. writers: writers,
  43. activeWriterCond: sync.NewCond(&sync.Mutex{}),
  44. saveToStorageFn: saveToStorageFn,
  45. filepath: filepath,
  46. activeReadChunks: make(map[LogicChunkIndex]int),
  47. }
  48. }
  49. func (cw *UploadPipeline) SaveDataAt(p []byte, off int64) (n int) {
  50. cw.writableChunksLock.Lock()
  51. defer cw.writableChunksLock.Unlock()
  52. logicChunkIndex := LogicChunkIndex(off / cw.ChunkSize)
  53. offsetRemainder := off % cw.ChunkSize
  54. memChunk, found := cw.writableChunks[logicChunkIndex]
  55. if !found {
  56. memChunk = &MemChunk{
  57. buf: mem.Allocate(int(cw.ChunkSize)),
  58. usage: newChunkWrittenIntervalList(),
  59. }
  60. cw.writableChunks[logicChunkIndex] = memChunk
  61. }
  62. n = copy(memChunk.buf[offsetRemainder:], p)
  63. memChunk.usage.MarkWritten(offsetRemainder, offsetRemainder+int64(n))
  64. cw.maybeMoveToSealed(memChunk, logicChunkIndex)
  65. return
  66. }
  67. func (cw *UploadPipeline) MaybeReadDataAt(p []byte, off int64) (maxStop int64) {
  68. logicChunkIndex := LogicChunkIndex(off / cw.ChunkSize)
  69. // read from sealed chunks first
  70. cw.sealedChunksLock.Lock()
  71. sealedChunk, found := cw.sealedChunks[logicChunkIndex]
  72. if found {
  73. sealedChunk.referenceCounter++
  74. }
  75. cw.sealedChunksLock.Unlock()
  76. if found {
  77. maxStop = readMemChunk(sealedChunk.chunk, p, off, logicChunkIndex, cw.ChunkSize)
  78. glog.V(4).Infof("%s read sealed memchunk [%d,%d)", cw.filepath, off, maxStop)
  79. sealedChunk.FreeReference(fmt.Sprintf("%s finish reading chunk %d", cw.filepath, logicChunkIndex))
  80. }
  81. // read from writable chunks last
  82. cw.writableChunksLock.Lock()
  83. defer cw.writableChunksLock.Unlock()
  84. writableChunk, found := cw.writableChunks[logicChunkIndex]
  85. if !found {
  86. return
  87. }
  88. writableMaxStop := readMemChunk(writableChunk, p, off, logicChunkIndex, cw.ChunkSize)
  89. glog.V(4).Infof("%s read writable memchunk [%d,%d)", cw.filepath, off, writableMaxStop)
  90. maxStop = max(maxStop, writableMaxStop)
  91. return
  92. }
  93. func (cw *UploadPipeline) FlushAll() {
  94. cw.writableChunksLock.Lock()
  95. defer cw.writableChunksLock.Unlock()
  96. for logicChunkIndex, memChunk := range cw.writableChunks {
  97. cw.moveToSealed(memChunk, logicChunkIndex)
  98. }
  99. cw.waitForCurrentWritersToComplete()
  100. }
  101. func (cw *UploadPipeline) LockForRead(startOffset, stopOffset int64) {
  102. startLogicChunkIndex := LogicChunkIndex(startOffset / cw.ChunkSize)
  103. stopLogicChunkIndex := LogicChunkIndex(stopOffset / cw.ChunkSize)
  104. if stopOffset%cw.ChunkSize > 0 {
  105. stopLogicChunkIndex += 1
  106. }
  107. cw.activeReadChunksLock.Lock()
  108. defer cw.activeReadChunksLock.Unlock()
  109. for i := startLogicChunkIndex; i < stopLogicChunkIndex; i++ {
  110. if count, found := cw.activeReadChunks[i]; found {
  111. cw.activeReadChunks[i] = count + 1
  112. } else {
  113. cw.activeReadChunks[i] = 1
  114. }
  115. }
  116. }
  117. func (cw *UploadPipeline) UnlockForRead(startOffset, stopOffset int64) {
  118. startLogicChunkIndex := LogicChunkIndex(startOffset / cw.ChunkSize)
  119. stopLogicChunkIndex := LogicChunkIndex(stopOffset / cw.ChunkSize)
  120. if stopOffset%cw.ChunkSize > 0 {
  121. stopLogicChunkIndex += 1
  122. }
  123. cw.activeReadChunksLock.Lock()
  124. defer cw.activeReadChunksLock.Unlock()
  125. for i := startLogicChunkIndex; i < stopLogicChunkIndex; i++ {
  126. if count, found := cw.activeReadChunks[i]; found {
  127. if count == 1 {
  128. delete(cw.activeReadChunks, i)
  129. } else {
  130. cw.activeReadChunks[i] = count - 1
  131. }
  132. }
  133. }
  134. }
  135. func (cw *UploadPipeline) IsLocked(logicChunkIndex LogicChunkIndex) bool {
  136. cw.activeReadChunksLock.Lock()
  137. defer cw.activeReadChunksLock.Unlock()
  138. if count, found := cw.activeReadChunks[logicChunkIndex]; found {
  139. return count > 0
  140. }
  141. return false
  142. }
  143. func (cw *UploadPipeline) waitForCurrentWritersToComplete() {
  144. cw.activeWriterCond.L.Lock()
  145. t := int32(100)
  146. for {
  147. t = atomic.LoadInt32(&cw.activeWriterCount)
  148. if t <= 0 {
  149. break
  150. }
  151. glog.V(4).Infof("activeWriterCond is %d", t)
  152. cw.activeWriterCond.Wait()
  153. }
  154. cw.activeWriterCond.L.Unlock()
  155. }
  156. func (cw *UploadPipeline) maybeMoveToSealed(memChunk *MemChunk, logicChunkIndex LogicChunkIndex) {
  157. if memChunk.usage.IsComplete(cw.ChunkSize) {
  158. cw.moveToSealed(memChunk, logicChunkIndex)
  159. }
  160. }
  161. func (cw *UploadPipeline) moveToSealed(memChunk *MemChunk, logicChunkIndex LogicChunkIndex) {
  162. atomic.AddInt32(&cw.activeWriterCount, 1)
  163. glog.V(4).Infof("%s activeWriterCount %d ++> %d", cw.filepath, cw.activeWriterCount-1, cw.activeWriterCount)
  164. cw.sealedChunksLock.Lock()
  165. if oldMemChunk, found := cw.sealedChunks[logicChunkIndex]; found {
  166. oldMemChunk.FreeReference(fmt.Sprintf("%s replace chunk %d", cw.filepath, logicChunkIndex))
  167. }
  168. sealedChunk := &SealedChunk{
  169. chunk: memChunk,
  170. referenceCounter: 1, // default 1 is for uploading process
  171. }
  172. cw.sealedChunks[logicChunkIndex] = sealedChunk
  173. delete(cw.writableChunks, logicChunkIndex)
  174. cw.sealedChunksLock.Unlock()
  175. cw.writers.Execute(func() {
  176. // first add to the file chunks
  177. cw.saveOneChunk(sealedChunk.chunk, logicChunkIndex)
  178. // notify waiting process
  179. atomic.AddInt32(&cw.activeWriterCount, -1)
  180. glog.V(4).Infof("%s activeWriterCount %d --> %d", cw.filepath, cw.activeWriterCount+1, cw.activeWriterCount)
  181. // Lock and Unlock are not required,
  182. // but it may signal multiple times during one wakeup,
  183. // and the waiting goroutine may miss some of them!
  184. cw.activeWriterCond.L.Lock()
  185. cw.activeWriterCond.Broadcast()
  186. cw.activeWriterCond.L.Unlock()
  187. // wait for readers
  188. for cw.IsLocked(logicChunkIndex) {
  189. time.Sleep(59 * time.Millisecond)
  190. }
  191. // then remove from sealed chunks
  192. cw.sealedChunksLock.Lock()
  193. defer cw.sealedChunksLock.Unlock()
  194. delete(cw.sealedChunks, logicChunkIndex)
  195. sealedChunk.FreeReference(fmt.Sprintf("%s finished uploading chunk %d", cw.filepath, logicChunkIndex))
  196. })
  197. }
  198. func (cw *UploadPipeline) saveOneChunk(memChunk *MemChunk, logicChunkIndex LogicChunkIndex) {
  199. if cw.saveToStorageFn == nil {
  200. return
  201. }
  202. for t := memChunk.usage.head.next; t != memChunk.usage.tail; t = t.next {
  203. reader := util.NewBytesReader(memChunk.buf[t.StartOffset:t.stopOffset])
  204. cw.saveToStorageFn(reader, int64(logicChunkIndex)*cw.ChunkSize+t.StartOffset, t.Size(), func() {
  205. })
  206. }
  207. }
  208. func readMemChunk(memChunk *MemChunk, p []byte, off int64, logicChunkIndex LogicChunkIndex, chunkSize int64) (maxStop int64) {
  209. memChunkBaseOffset := int64(logicChunkIndex) * chunkSize
  210. for t := memChunk.usage.head.next; t != memChunk.usage.tail; t = t.next {
  211. logicStart := max(off, int64(logicChunkIndex)*chunkSize+t.StartOffset)
  212. logicStop := min(off+int64(len(p)), memChunkBaseOffset+t.stopOffset)
  213. if logicStart < logicStop {
  214. copy(p[logicStart-off:logicStop-off], memChunk.buf[logicStart-memChunkBaseOffset:logicStop-memChunkBaseOffset])
  215. maxStop = max(maxStop, logicStop)
  216. }
  217. }
  218. return
  219. }
  220. func (p2 *UploadPipeline) Shutdown() {
  221. }