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.

322 lines
8.3 KiB

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