dirty_pages_chunked.go 2.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697
  1. package mount
  2. import (
  3. "fmt"
  4. "github.com/seaweedfs/seaweedfs/weed/glog"
  5. "github.com/seaweedfs/seaweedfs/weed/mount/page_writer"
  6. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  7. "io"
  8. "sync"
  9. "time"
  10. )
  11. type ChunkedDirtyPages struct {
  12. fh *FileHandle
  13. writeWaitGroup sync.WaitGroup
  14. lastErr error
  15. collection string
  16. replication string
  17. uploadPipeline *page_writer.UploadPipeline
  18. hasWrites bool
  19. }
  20. var (
  21. _ = page_writer.DirtyPages(&ChunkedDirtyPages{})
  22. )
  23. func newMemoryChunkPages(fh *FileHandle, chunkSize int64) *ChunkedDirtyPages {
  24. dirtyPages := &ChunkedDirtyPages{
  25. fh: fh,
  26. }
  27. swapFileDir := fh.wfs.option.getTempFilePageDir()
  28. dirtyPages.uploadPipeline = page_writer.NewUploadPipeline(fh.wfs.concurrentWriters, chunkSize,
  29. dirtyPages.saveChunkedFileIntervalToStorage, fh.wfs.option.ConcurrentWriters, swapFileDir)
  30. return dirtyPages
  31. }
  32. func (pages *ChunkedDirtyPages) AddPage(offset int64, data []byte, isSequential bool) {
  33. pages.hasWrites = true
  34. glog.V(4).Infof("%v memory AddPage [%d, %d)", pages.fh.fh, offset, offset+int64(len(data)))
  35. pages.uploadPipeline.SaveDataAt(data, offset, isSequential)
  36. return
  37. }
  38. func (pages *ChunkedDirtyPages) FlushData() error {
  39. if !pages.hasWrites {
  40. return nil
  41. }
  42. pages.uploadPipeline.FlushAll()
  43. if pages.lastErr != nil {
  44. return fmt.Errorf("flush data: %v", pages.lastErr)
  45. }
  46. return nil
  47. }
  48. func (pages *ChunkedDirtyPages) ReadDirtyDataAt(data []byte, startOffset int64) (maxStop int64) {
  49. if !pages.hasWrites {
  50. return
  51. }
  52. return pages.uploadPipeline.MaybeReadDataAt(data, startOffset)
  53. }
  54. func (pages *ChunkedDirtyPages) saveChunkedFileIntervalToStorage(reader io.Reader, offset int64, size int64, cleanupFn func()) {
  55. mtime := time.Now().UnixNano()
  56. defer cleanupFn()
  57. fileFullPath := pages.fh.FullPath()
  58. fileName := fileFullPath.Name()
  59. chunk, err := pages.fh.wfs.saveDataAsChunk(fileFullPath)(reader, fileName, offset)
  60. if err != nil {
  61. glog.V(0).Infof("%v saveToStorage [%d,%d): %v", fileFullPath, offset, offset+size, err)
  62. pages.lastErr = err
  63. return
  64. }
  65. chunk.Mtime = mtime
  66. pages.fh.AddChunks([]*filer_pb.FileChunk{chunk})
  67. pages.fh.entryViewCache = nil
  68. glog.V(3).Infof("%v saveToStorage %s [%d,%d)", fileFullPath, chunk.FileId, offset, offset+size)
  69. }
  70. func (pages ChunkedDirtyPages) Destroy() {
  71. pages.uploadPipeline.Shutdown()
  72. }
  73. func (pages *ChunkedDirtyPages) LockForRead(startOffset, stopOffset int64) {
  74. pages.uploadPipeline.LockForRead(startOffset, stopOffset)
  75. }
  76. func (pages *ChunkedDirtyPages) UnlockForRead(startOffset, stopOffset int64) {
  77. pages.uploadPipeline.UnlockForRead(startOffset, stopOffset)
  78. }