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.

142 lines
3.5 KiB

4 years ago
7 years ago
4 years ago
4 years ago
  1. package filesys
  2. import (
  3. "bytes"
  4. "io"
  5. "runtime"
  6. "sync"
  7. "time"
  8. "github.com/chrislusf/seaweedfs/weed/glog"
  9. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  10. "github.com/chrislusf/seaweedfs/weed/util"
  11. )
  12. var (
  13. concurrentWriterLimit = runtime.NumCPU()
  14. concurrentWriters = util.NewLimitedConcurrentExecutor(4 * concurrentWriterLimit)
  15. )
  16. type ContinuousDirtyPages struct {
  17. intervals *ContinuousIntervals
  18. f *File
  19. writeWaitGroup sync.WaitGroup
  20. chunkSaveErrChan chan error
  21. chunkSaveErrChanClosed bool
  22. lastErr error
  23. lock sync.Mutex
  24. collection string
  25. replication string
  26. }
  27. func newDirtyPages(file *File) *ContinuousDirtyPages {
  28. dirtyPages := &ContinuousDirtyPages{
  29. intervals: &ContinuousIntervals{},
  30. f: file,
  31. chunkSaveErrChan: make(chan error, concurrentWriterLimit),
  32. }
  33. go func() {
  34. for t := range dirtyPages.chunkSaveErrChan {
  35. if t != nil {
  36. dirtyPages.lastErr = t
  37. }
  38. }
  39. }()
  40. return dirtyPages
  41. }
  42. func (pages *ContinuousDirtyPages) AddPage(offset int64, data []byte) {
  43. glog.V(4).Infof("%s AddPage [%d,%d) of %d bytes", pages.f.fullpath(), offset, offset+int64(len(data)), pages.f.entry.Attributes.FileSize)
  44. if len(data) > int(pages.f.wfs.option.ChunkSizeLimit) {
  45. // this is more than what buffer can hold.
  46. pages.flushAndSave(offset, data)
  47. }
  48. pages.intervals.AddInterval(data, offset)
  49. if pages.intervals.TotalSize() >= pages.f.wfs.option.ChunkSizeLimit {
  50. pages.saveExistingLargestPageToStorage()
  51. }
  52. return
  53. }
  54. func (pages *ContinuousDirtyPages) flushAndSave(offset int64, data []byte) {
  55. // flush existing
  56. pages.saveExistingPagesToStorage()
  57. // flush the new page
  58. pages.saveToStorage(bytes.NewReader(data), offset, int64(len(data)))
  59. return
  60. }
  61. func (pages *ContinuousDirtyPages) saveExistingPagesToStorage() {
  62. for pages.saveExistingLargestPageToStorage() {
  63. }
  64. }
  65. func (pages *ContinuousDirtyPages) saveExistingLargestPageToStorage() (hasSavedData bool) {
  66. maxList := pages.intervals.RemoveLargestIntervalLinkedList()
  67. if maxList == nil {
  68. return false
  69. }
  70. fileSize := int64(pages.f.entry.Attributes.FileSize)
  71. chunkSize := min(maxList.Size(), fileSize-maxList.Offset())
  72. if chunkSize == 0 {
  73. return false
  74. }
  75. pages.saveToStorage(maxList.ToReader(), maxList.Offset(), chunkSize)
  76. return true
  77. }
  78. func (pages *ContinuousDirtyPages) saveToStorage(reader io.Reader, offset int64, size int64) {
  79. if pages.chunkSaveErrChanClosed {
  80. pages.chunkSaveErrChan = make(chan error, concurrentWriterLimit)
  81. pages.chunkSaveErrChanClosed = false
  82. }
  83. mtime := time.Now().UnixNano()
  84. pages.writeWaitGroup.Add(1)
  85. go func() {
  86. defer pages.writeWaitGroup.Done()
  87. reader = io.LimitReader(reader, size)
  88. chunk, collection, replication, err := pages.f.wfs.saveDataAsChunk(pages.f.fullpath())(reader, pages.f.Name, offset)
  89. if err != nil {
  90. glog.V(0).Infof("%s saveToStorage [%d,%d): %v", pages.f.fullpath(), offset, offset+size, err)
  91. pages.chunkSaveErrChan <- err
  92. return
  93. }
  94. chunk.Mtime = mtime
  95. pages.collection, pages.replication = collection, replication
  96. pages.f.addChunks([]*filer_pb.FileChunk{chunk})
  97. glog.V(3).Infof("%s saveToStorage [%d,%d)", pages.f.fullpath(), offset, offset+size)
  98. }()
  99. }
  100. func max(x, y int64) int64 {
  101. if x > y {
  102. return x
  103. }
  104. return y
  105. }
  106. func min(x, y int64) int64 {
  107. if x < y {
  108. return x
  109. }
  110. return y
  111. }
  112. func (pages *ContinuousDirtyPages) ReadDirtyDataAt(data []byte, startOffset int64) (maxStop int64) {
  113. return pages.intervals.ReadDataAt(data, startOffset)
  114. }