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.

291 lines
7.7 KiB

7 years ago
7 years ago
7 years ago
7 years ago
4 years ago
4 years ago
5 years ago
4 years ago
4 years ago
6 years ago
4 years ago
4 years ago
4 years ago
4 years ago
6 years ago
6 years ago
4 years ago
6 years ago
  1. package filesys
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "math"
  7. "net/http"
  8. "os"
  9. "sync"
  10. "time"
  11. "github.com/seaweedfs/fuse"
  12. "github.com/seaweedfs/fuse/fs"
  13. "github.com/chrislusf/seaweedfs/weed/filer"
  14. "github.com/chrislusf/seaweedfs/weed/glog"
  15. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  16. )
  17. type FileHandle struct {
  18. // cache file has been written to
  19. dirtyPages *ContinuousDirtyPages
  20. contentType string
  21. handle uint64
  22. sync.RWMutex
  23. f *File
  24. RequestId fuse.RequestID // unique ID for request
  25. NodeId fuse.NodeID // file or directory the request is about
  26. Uid uint32 // user ID of process making request
  27. Gid uint32 // group ID of process making request
  28. }
  29. func newFileHandle(file *File, uid, gid uint32) *FileHandle {
  30. fh := &FileHandle{
  31. f: file,
  32. dirtyPages: newDirtyPages(file),
  33. Uid: uid,
  34. Gid: gid,
  35. }
  36. if fh.f.entry != nil {
  37. fh.f.entry.Attributes.FileSize = filer.FileSize(fh.f.entry)
  38. }
  39. return fh
  40. }
  41. var _ = fs.Handle(&FileHandle{})
  42. // var _ = fs.HandleReadAller(&FileHandle{})
  43. var _ = fs.HandleReader(&FileHandle{})
  44. var _ = fs.HandleFlusher(&FileHandle{})
  45. var _ = fs.HandleWriter(&FileHandle{})
  46. var _ = fs.HandleReleaser(&FileHandle{})
  47. func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) error {
  48. glog.V(4).Infof("%s read fh %d: [%d,%d) size %d resp.Data cap=%d", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size), req.Size, cap(resp.Data))
  49. fh.RLock()
  50. defer fh.RUnlock()
  51. if req.Size <= 0 {
  52. return nil
  53. }
  54. buff := resp.Data[:cap(resp.Data)]
  55. if req.Size > cap(resp.Data) {
  56. // should not happen
  57. buff = make([]byte, req.Size)
  58. }
  59. totalRead, err := fh.readFromChunks(buff, req.Offset)
  60. if err == nil {
  61. maxStop := fh.readFromDirtyPages(buff, req.Offset)
  62. totalRead = max(maxStop-req.Offset, totalRead)
  63. }
  64. if err == io.EOF {
  65. err = nil
  66. }
  67. if err != nil {
  68. glog.Warningf("file handle read %s %d: %v", fh.f.fullpath(), totalRead, err)
  69. return fuse.EIO
  70. }
  71. if totalRead > int64(len(buff)) {
  72. glog.Warningf("%s FileHandle Read %d: [%d,%d) size %d totalRead %d", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size), req.Size, totalRead)
  73. totalRead = min(int64(len(buff)), totalRead)
  74. }
  75. // resp.Data = buff[:totalRead]
  76. resp.Data = buff
  77. return err
  78. }
  79. func (fh *FileHandle) readFromDirtyPages(buff []byte, startOffset int64) (maxStop int64) {
  80. maxStop = fh.dirtyPages.ReadDirtyDataAt(buff, startOffset)
  81. return
  82. }
  83. func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
  84. fileSize := int64(filer.FileSize(fh.f.entry))
  85. if fileSize == 0 {
  86. glog.V(1).Infof("empty fh %v", fh.f.fullpath())
  87. return 0, io.EOF
  88. }
  89. var chunkResolveErr error
  90. if fh.f.entryViewCache == nil {
  91. fh.f.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(filer.LookupFn(fh.f.wfs), fh.f.entry.Chunks)
  92. if chunkResolveErr != nil {
  93. return 0, fmt.Errorf("fail to resolve chunk manifest: %v", chunkResolveErr)
  94. }
  95. fh.f.reader = nil
  96. }
  97. if fh.f.reader == nil {
  98. chunkViews := filer.ViewFromVisibleIntervals(fh.f.entryViewCache, 0, math.MaxInt64)
  99. fh.f.reader = filer.NewChunkReaderAtFromClient(fh.f.wfs, chunkViews, fh.f.wfs.chunkCache, fileSize)
  100. }
  101. totalRead, err := fh.f.reader.ReadAt(buff, offset)
  102. if err != nil && err != io.EOF {
  103. glog.Errorf("file handle read %s: %v", fh.f.fullpath(), err)
  104. }
  105. glog.V(4).Infof("file handle read %s [%d,%d] %d : %v", fh.f.fullpath(), offset, offset+int64(totalRead), totalRead, err)
  106. return int64(totalRead), err
  107. }
  108. // Write to the file handle
  109. func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
  110. fh.Lock()
  111. defer fh.Unlock()
  112. // write the request to volume servers
  113. data := req.Data
  114. if len(data) <= 512 {
  115. // fuse message cacheable size
  116. data = make([]byte, len(req.Data))
  117. copy(data, req.Data)
  118. }
  119. fh.f.entry.Attributes.FileSize = uint64(max(req.Offset+int64(len(data)), int64(fh.f.entry.Attributes.FileSize)))
  120. glog.V(4).Infof("%v write [%d,%d) %d", fh.f.fullpath(), req.Offset, req.Offset+int64(len(req.Data)), len(req.Data))
  121. fh.dirtyPages.AddPage(req.Offset, data)
  122. resp.Size = len(data)
  123. if req.Offset == 0 {
  124. // detect mime type
  125. fh.contentType = http.DetectContentType(data)
  126. fh.f.dirtyMetadata = true
  127. }
  128. fh.f.dirtyMetadata = true
  129. return nil
  130. }
  131. func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) error {
  132. glog.V(4).Infof("Release %v fh %d", fh.f.fullpath(), fh.handle)
  133. fh.Lock()
  134. defer fh.Unlock()
  135. fh.f.isOpen--
  136. if fh.f.isOpen < 0 {
  137. glog.V(0).Infof("Release reset %s open count %d => %d", fh.f.Name, fh.f.isOpen, 0)
  138. fh.f.isOpen = 0
  139. return nil
  140. }
  141. if fh.f.isOpen == 0 {
  142. if err := fh.doFlush(ctx, req.Header); err != nil {
  143. glog.Errorf("Release doFlush %s: %v", fh.f.Name, err)
  144. }
  145. // stop the goroutine
  146. if !fh.dirtyPages.chunkSaveErrChanClosed {
  147. fh.dirtyPages.chunkSaveErrChanClosed = true
  148. close(fh.dirtyPages.chunkSaveErrChan)
  149. }
  150. fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
  151. }
  152. return nil
  153. }
  154. func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
  155. fh.Lock()
  156. defer fh.Unlock()
  157. return fh.doFlush(ctx, req.Header)
  158. }
  159. func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {
  160. // flush works at fh level
  161. // send the data to the OS
  162. glog.V(4).Infof("doFlush %s fh %d", fh.f.fullpath(), fh.handle)
  163. fh.dirtyPages.saveExistingPagesToStorage()
  164. fh.dirtyPages.writeWaitGroup.Wait()
  165. if fh.dirtyPages.lastErr != nil {
  166. return fh.dirtyPages.lastErr
  167. }
  168. if !fh.f.dirtyMetadata {
  169. return nil
  170. }
  171. err := fh.f.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  172. if fh.f.entry.Attributes != nil {
  173. fh.f.entry.Attributes.Mime = fh.contentType
  174. if fh.f.entry.Attributes.Uid == 0 {
  175. fh.f.entry.Attributes.Uid = header.Uid
  176. }
  177. if fh.f.entry.Attributes.Gid == 0 {
  178. fh.f.entry.Attributes.Gid = header.Gid
  179. }
  180. if fh.f.entry.Attributes.Crtime == 0 {
  181. fh.f.entry.Attributes.Crtime = time.Now().Unix()
  182. }
  183. fh.f.entry.Attributes.Mtime = time.Now().Unix()
  184. fh.f.entry.Attributes.FileMode = uint32(os.FileMode(fh.f.entry.Attributes.FileMode) &^ fh.f.wfs.option.Umask)
  185. fh.f.entry.Attributes.Collection = fh.dirtyPages.collection
  186. fh.f.entry.Attributes.Replication = fh.dirtyPages.replication
  187. }
  188. request := &filer_pb.CreateEntryRequest{
  189. Directory: fh.f.dir.FullPath(),
  190. Entry: fh.f.entry,
  191. Signatures: []int32{fh.f.wfs.signature},
  192. }
  193. glog.V(4).Infof("%s set chunks: %v", fh.f.fullpath(), len(fh.f.entry.Chunks))
  194. for i, chunk := range fh.f.entry.Chunks {
  195. glog.V(4).Infof("%s chunks %d: %v [%d,%d)", fh.f.fullpath(), i, chunk.GetFileIdString(), chunk.Offset, chunk.Offset+int64(chunk.Size))
  196. }
  197. manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(fh.f.entry.Chunks)
  198. chunks, _ := filer.CompactFileChunks(filer.LookupFn(fh.f.wfs), nonManifestChunks)
  199. chunks, manifestErr := filer.MaybeManifestize(fh.f.wfs.saveDataAsChunk(fh.f.fullpath()), chunks)
  200. if manifestErr != nil {
  201. // not good, but should be ok
  202. glog.V(0).Infof("MaybeManifestize: %v", manifestErr)
  203. }
  204. fh.f.entry.Chunks = append(chunks, manifestChunks...)
  205. fh.f.wfs.mapPbIdFromLocalToFiler(request.Entry)
  206. defer fh.f.wfs.mapPbIdFromFilerToLocal(request.Entry)
  207. if err := filer_pb.CreateEntry(client, request); err != nil {
  208. glog.Errorf("fh flush create %s: %v", fh.f.fullpath(), err)
  209. return fmt.Errorf("fh flush create %s: %v", fh.f.fullpath(), err)
  210. }
  211. fh.f.wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry))
  212. return nil
  213. })
  214. if err == nil {
  215. fh.f.dirtyMetadata = false
  216. }
  217. if err != nil {
  218. glog.Errorf("%v fh %d flush: %v", fh.f.fullpath(), fh.handle, err)
  219. return fuse.EIO
  220. }
  221. return nil
  222. }