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.

146 lines
3.6 KiB

7 years ago
4 years ago
5 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. maxList.Destroy()
  77. return true
  78. }
  79. func (pages *ContinuousDirtyPages) saveToStorage(reader io.Reader, offset int64, size int64) {
  80. if pages.chunkSaveErrChanClosed {
  81. pages.chunkSaveErrChan = make(chan error, concurrentWriterLimit)
  82. pages.chunkSaveErrChanClosed = false
  83. }
  84. mtime := time.Now().UnixNano()
  85. pages.writeWaitGroup.Add(1)
  86. concurrentWriters.Execute(func() {
  87. defer pages.writeWaitGroup.Done()
  88. dir, _ := pages.f.fullpath().DirAndName()
  89. reader = io.LimitReader(reader, size)
  90. chunk, collection, replication, err := pages.f.wfs.saveDataAsChunk(dir)(reader, pages.f.Name, offset)
  91. if err != nil {
  92. glog.V(0).Infof("%s saveToStorage [%d,%d): %v", pages.f.fullpath(), offset, offset+size, err)
  93. pages.chunkSaveErrChan <- err
  94. return
  95. }
  96. chunk.Mtime = mtime
  97. pages.collection, pages.replication = collection, replication
  98. pages.f.addChunks([]*filer_pb.FileChunk{chunk})
  99. glog.V(3).Infof("%s saveToStorage [%d,%d)", pages.f.fullpath(), offset, offset+size)
  100. })
  101. }
  102. func max(x, y int64) int64 {
  103. if x > y {
  104. return x
  105. }
  106. return y
  107. }
  108. func min(x, y int64) int64 {
  109. if x < y {
  110. return x
  111. }
  112. return y
  113. }
  114. func (pages *ContinuousDirtyPages) ReadDirtyDataAt(data []byte, startOffset int64) (maxStop int64) {
  115. return pages.intervals.ReadDataAt(data, startOffset)
  116. }