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.

233 lines
6.4 KiB

7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
  1. package filesys
  2. import (
  3. "bazil.org/fuse"
  4. "bazil.org/fuse/fs"
  5. "context"
  6. "fmt"
  7. "github.com/chrislusf/seaweedfs/weed/filer2"
  8. "github.com/chrislusf/seaweedfs/weed/glog"
  9. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  10. "github.com/chrislusf/seaweedfs/weed/util"
  11. "strings"
  12. "sync"
  13. "net/http"
  14. )
  15. type FileHandle struct {
  16. // cache file has been written to
  17. dirtyPages *ContinuousDirtyPages
  18. dirtyMetadata bool
  19. handle uint64
  20. f *File
  21. RequestId fuse.RequestID // unique ID for request
  22. NodeId fuse.NodeID // file or directory the request is about
  23. Uid uint32 // user ID of process making request
  24. Gid uint32 // group ID of process making request
  25. }
  26. func newFileHandle(file *File, uid, gid uint32) *FileHandle {
  27. return &FileHandle{
  28. f: file,
  29. dirtyPages: newDirtyPages(file),
  30. Uid: uid,
  31. Gid: gid,
  32. }
  33. }
  34. var _ = fs.Handle(&FileHandle{})
  35. // var _ = fs.HandleReadAller(&FileHandle{})
  36. var _ = fs.HandleReader(&FileHandle{})
  37. var _ = fs.HandleFlusher(&FileHandle{})
  38. var _ = fs.HandleWriter(&FileHandle{})
  39. var _ = fs.HandleReleaser(&FileHandle{})
  40. func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) error {
  41. glog.V(4).Infof("%s read fh %d: [%d,%d)", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size))
  42. // this value should come from the filer instead of the old f
  43. if len(fh.f.Chunks) == 0 {
  44. glog.V(0).Infof("empty fh %v/%v", fh.f.dir.Path, fh.f.Name)
  45. return fmt.Errorf("empty file %v/%v", fh.f.dir.Path, fh.f.Name)
  46. }
  47. buff := make([]byte, req.Size)
  48. chunkViews := filer2.ViewFromChunks(fh.f.Chunks, req.Offset, req.Size)
  49. var vids []string
  50. for _, chunkView := range chunkViews {
  51. vids = append(vids, volumeId(chunkView.FileId))
  52. }
  53. vid2Locations := make(map[string]*filer_pb.Locations)
  54. err := fh.f.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  55. glog.V(4).Infof("read fh lookup volume id locations: %v", vids)
  56. resp, err := client.LookupVolume(ctx, &filer_pb.LookupVolumeRequest{
  57. VolumeIds: vids,
  58. })
  59. if err != nil {
  60. return err
  61. }
  62. vid2Locations = resp.LocationsMap
  63. return nil
  64. })
  65. if err != nil {
  66. glog.V(4).Infof("%v/%v read fh lookup volume ids: %v", fh.f.dir.Path, fh.f.Name, err)
  67. return fmt.Errorf("failed to lookup volume ids %v: %v", vids, err)
  68. }
  69. var totalRead int64
  70. var wg sync.WaitGroup
  71. for _, chunkView := range chunkViews {
  72. wg.Add(1)
  73. go func(chunkView *filer2.ChunkView) {
  74. defer wg.Done()
  75. glog.V(4).Infof("read fh reading chunk: %+v", chunkView)
  76. locations := vid2Locations[volumeId(chunkView.FileId)]
  77. if locations == nil || len(locations.Locations) == 0 {
  78. glog.V(0).Infof("failed to locate %s", chunkView.FileId)
  79. err = fmt.Errorf("failed to locate %s", chunkView.FileId)
  80. return
  81. }
  82. var n int64
  83. n, err = util.ReadUrl(
  84. fmt.Sprintf("http://%s/%s", locations.Locations[0].Url, chunkView.FileId),
  85. chunkView.Offset,
  86. int(chunkView.Size),
  87. buff[chunkView.LogicOffset-req.Offset:chunkView.LogicOffset-req.Offset+int64(chunkView.Size)])
  88. if err != nil {
  89. glog.V(0).Infof("%v/%v read http://%s/%v %v bytes: %v", fh.f.dir.Path, fh.f.Name, locations.Locations[0].Url, chunkView.FileId, n, err)
  90. err = fmt.Errorf("failed to read http://%s/%s: %v",
  91. locations.Locations[0].Url, chunkView.FileId, err)
  92. return
  93. }
  94. glog.V(4).Infof("read fh read %d bytes: %+v", n, chunkView)
  95. totalRead += n
  96. }(chunkView)
  97. }
  98. wg.Wait()
  99. resp.Data = buff[:totalRead]
  100. return err
  101. }
  102. // Write to the file handle
  103. func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
  104. // write the request to volume servers
  105. glog.V(4).Infof("%+v/%v write fh %d: [%d,%d)", fh.f.dir.Path, fh.f.Name, fh.handle, req.Offset, req.Offset+int64(len(req.Data)))
  106. chunks, err := fh.dirtyPages.AddPage(ctx, req.Offset, req.Data)
  107. if err != nil {
  108. return fmt.Errorf("write %s/%s at [%d,%d): %v", fh.f.dir.Path, fh.f.Name, req.Offset, req.Offset+int64(len(req.Data)), err)
  109. }
  110. resp.Size = len(req.Data)
  111. if req.Offset == 0 {
  112. fh.f.attributes.Mime = http.DetectContentType(req.Data)
  113. fh.dirtyMetadata = true
  114. }
  115. for _, chunk := range chunks {
  116. fh.f.Chunks = append(fh.f.Chunks, chunk)
  117. glog.V(1).Infof("uploaded %s/%s to %s [%d,%d)", fh.f.dir.Path, fh.f.Name, chunk.FileId, chunk.Offset, chunk.Offset+int64(chunk.Size))
  118. fh.dirtyMetadata = true
  119. }
  120. return nil
  121. }
  122. func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) error {
  123. glog.V(4).Infof("%v release fh %d", fh.f.fullpath(), fh.handle)
  124. fh.f.wfs.ReleaseHandle(fuse.HandleID(fh.handle))
  125. fh.f.isOpen = false
  126. return nil
  127. }
  128. // Flush - experimenting with uploading at flush, this slows operations down till it has been
  129. // completely flushed
  130. func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
  131. // fflush works at fh level
  132. // send the data to the OS
  133. glog.V(4).Infof("%s fh %d flush %v", fh.f.fullpath(), fh.handle, req)
  134. chunk, err := fh.dirtyPages.FlushToStorage(ctx)
  135. if err != nil {
  136. glog.V(0).Infof("flush %s/%s to %s [%d,%d): %v", fh.f.dir.Path, fh.f.Name, chunk.FileId, chunk.Offset, chunk.Offset+int64(chunk.Size), err)
  137. return fmt.Errorf("flush %s/%s to %s [%d,%d): %v", fh.f.dir.Path, fh.f.Name, chunk.FileId, chunk.Offset, chunk.Offset+int64(chunk.Size), err)
  138. }
  139. if chunk != nil {
  140. fh.f.Chunks = append(fh.f.Chunks, chunk)
  141. fh.dirtyMetadata = true
  142. }
  143. if !fh.dirtyMetadata {
  144. return nil
  145. }
  146. if len(fh.f.Chunks) == 0 {
  147. glog.V(2).Infof("fh %s/%s flush skipping empty: %v", fh.f.dir.Path, fh.f.Name, req)
  148. return nil
  149. }
  150. err = fh.f.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  151. request := &filer_pb.UpdateEntryRequest{
  152. Directory: fh.f.dir.Path,
  153. Entry: &filer_pb.Entry{
  154. Name: fh.f.Name,
  155. Attributes: fh.f.attributes,
  156. Chunks: fh.f.Chunks,
  157. },
  158. }
  159. glog.V(1).Infof("%s/%s set chunks: %v", fh.f.dir.Path, fh.f.Name, len(fh.f.Chunks))
  160. for i, chunk := range fh.f.Chunks {
  161. glog.V(1).Infof("%s/%s chunks %d: %v [%d,%d)", fh.f.dir.Path, fh.f.Name, i, chunk.FileId, chunk.Offset, chunk.Offset+int64(chunk.Size))
  162. }
  163. if _, err := client.UpdateEntry(ctx, request); err != nil {
  164. return fmt.Errorf("update fh: %v", err)
  165. }
  166. return nil
  167. })
  168. if err == nil {
  169. fh.dirtyMetadata = false
  170. }
  171. return err
  172. }
  173. func volumeId(fileId string) string {
  174. lastCommaIndex := strings.LastIndex(fileId, ",")
  175. if lastCommaIndex > 0 {
  176. return fileId[:lastCommaIndex]
  177. }
  178. return fileId
  179. }