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.

290 lines
7.8 KiB

7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
6 years ago
6 years ago
6 years ago
7 years ago
6 years ago
  1. package filesys
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/filer2"
  6. "github.com/chrislusf/seaweedfs/weed/glog"
  7. "github.com/chrislusf/seaweedfs/weed/operation"
  8. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  9. "github.com/chrislusf/seaweedfs/weed/util"
  10. "github.com/gabriel-vasile/mimetype"
  11. "github.com/seaweedfs/fuse"
  12. "github.com/seaweedfs/fuse/fs"
  13. "google.golang.org/grpc"
  14. "mime"
  15. "path"
  16. "strings"
  17. "sync"
  18. "time"
  19. )
  20. type FileHandle struct {
  21. // cache file has been written to
  22. dirtyPages *ContinuousDirtyPages
  23. contentType string
  24. dirtyMetadata bool
  25. handle uint64
  26. f *File
  27. RequestId fuse.RequestID // unique ID for request
  28. NodeId fuse.NodeID // file or directory the request is about
  29. Uid uint32 // user ID of process making request
  30. Gid uint32 // group ID of process making request
  31. }
  32. func newFileHandle(file *File, uid, gid uint32) *FileHandle {
  33. return &FileHandle{
  34. f: file,
  35. dirtyPages: newDirtyPages(file),
  36. Uid: uid,
  37. Gid: gid,
  38. }
  39. }
  40. var _ = fs.Handle(&FileHandle{})
  41. // var _ = fs.HandleReadAller(&FileHandle{})
  42. var _ = fs.HandleReader(&FileHandle{})
  43. var _ = fs.HandleFlusher(&FileHandle{})
  44. var _ = fs.HandleWriter(&FileHandle{})
  45. var _ = fs.HandleReleaser(&FileHandle{})
  46. func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) error {
  47. glog.V(4).Infof("%s read fh %d: [%d,%d)", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size))
  48. // this value should come from the filer instead of the old f
  49. if len(fh.f.entry.Chunks) == 0 {
  50. glog.V(1).Infof("empty fh %v/%v", fh.f.dir.Path, fh.f.Name)
  51. return nil
  52. }
  53. buff := make([]byte, req.Size)
  54. if fh.f.entryViewCache == nil {
  55. fh.f.entryViewCache = filer2.NonOverlappingVisibleIntervals(fh.f.entry.Chunks)
  56. }
  57. chunkViews := filer2.ViewFromVisibleIntervals(fh.f.entryViewCache, req.Offset, req.Size)
  58. var vids []string
  59. for _, chunkView := range chunkViews {
  60. vids = append(vids, volumeId(chunkView.FileId))
  61. }
  62. vid2Locations := make(map[string]*filer_pb.Locations)
  63. err := fh.f.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  64. glog.V(4).Infof("read fh lookup volume id locations: %v", vids)
  65. resp, err := client.LookupVolume(ctx, &filer_pb.LookupVolumeRequest{
  66. VolumeIds: vids,
  67. })
  68. if err != nil {
  69. return err
  70. }
  71. vid2Locations = resp.LocationsMap
  72. return nil
  73. })
  74. if err != nil {
  75. glog.V(4).Infof("%v/%v read fh lookup volume ids: %v", fh.f.dir.Path, fh.f.Name, err)
  76. return fmt.Errorf("failed to lookup volume ids %v: %v", vids, err)
  77. }
  78. var totalRead int64
  79. var wg sync.WaitGroup
  80. for _, chunkView := range chunkViews {
  81. wg.Add(1)
  82. go func(chunkView *filer2.ChunkView) {
  83. defer wg.Done()
  84. glog.V(4).Infof("read fh reading chunk: %+v", chunkView)
  85. locations := vid2Locations[volumeId(chunkView.FileId)]
  86. if locations == nil || len(locations.Locations) == 0 {
  87. glog.V(0).Infof("failed to locate %s", chunkView.FileId)
  88. err = fmt.Errorf("failed to locate %s", chunkView.FileId)
  89. return
  90. }
  91. var n int64
  92. n, err = util.ReadUrl(
  93. fmt.Sprintf("http://%s/%s", locations.Locations[0].Url, chunkView.FileId),
  94. chunkView.Offset,
  95. int(chunkView.Size),
  96. buff[chunkView.LogicOffset-req.Offset:chunkView.LogicOffset-req.Offset+int64(chunkView.Size)],
  97. !chunkView.IsFullChunk)
  98. if err != nil {
  99. 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)
  100. err = fmt.Errorf("failed to read http://%s/%s: %v",
  101. locations.Locations[0].Url, chunkView.FileId, err)
  102. return
  103. }
  104. glog.V(4).Infof("read fh read %d bytes: %+v", n, chunkView)
  105. totalRead += n
  106. }(chunkView)
  107. }
  108. wg.Wait()
  109. resp.Data = buff[:totalRead]
  110. return err
  111. }
  112. // Write to the file handle
  113. func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
  114. // write the request to volume servers
  115. 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)))
  116. chunks, err := fh.dirtyPages.AddPage(ctx, req.Offset, req.Data)
  117. if err != nil {
  118. glog.Errorf("%+v/%v write fh %d: [%d,%d): %v", fh.f.dir.Path, fh.f.Name, fh.handle, req.Offset, req.Offset+int64(len(req.Data)), err)
  119. 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)
  120. }
  121. resp.Size = len(req.Data)
  122. if req.Offset == 0 {
  123. // detect mime type
  124. var possibleExt string
  125. fh.contentType, possibleExt = mimetype.Detect(req.Data)
  126. if ext := path.Ext(fh.f.Name); ext != possibleExt {
  127. fh.contentType = mime.TypeByExtension(ext)
  128. }
  129. fh.dirtyMetadata = true
  130. }
  131. fh.f.addChunks(chunks)
  132. if len(chunks) > 0 {
  133. fh.dirtyMetadata = true
  134. }
  135. return nil
  136. }
  137. func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) error {
  138. glog.V(4).Infof("%v release fh %d", fh.f.fullpath(), fh.handle)
  139. fh.dirtyPages.releaseResource()
  140. fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
  141. fh.f.isOpen = false
  142. return nil
  143. }
  144. func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
  145. // fflush works at fh level
  146. // send the data to the OS
  147. glog.V(4).Infof("%s fh %d flush %v", fh.f.fullpath(), fh.handle, req)
  148. chunk, err := fh.dirtyPages.FlushToStorage(ctx)
  149. if err != nil {
  150. glog.Errorf("flush %s/%s: %v", fh.f.dir.Path, fh.f.Name, err)
  151. return fmt.Errorf("flush %s/%s: %v", fh.f.dir.Path, fh.f.Name, err)
  152. }
  153. fh.f.addChunk(chunk)
  154. if !fh.dirtyMetadata {
  155. return nil
  156. }
  157. return fh.f.wfs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  158. if fh.f.entry.Attributes != nil {
  159. fh.f.entry.Attributes.Mime = fh.contentType
  160. fh.f.entry.Attributes.Uid = req.Uid
  161. fh.f.entry.Attributes.Gid = req.Gid
  162. fh.f.entry.Attributes.Mtime = time.Now().Unix()
  163. fh.f.entry.Attributes.Crtime = time.Now().Unix()
  164. fh.f.entry.Attributes.FileMode = uint32(0770)
  165. }
  166. request := &filer_pb.CreateEntryRequest{
  167. Directory: fh.f.dir.Path,
  168. Entry: fh.f.entry,
  169. }
  170. //glog.V(1).Infof("%s/%s set chunks: %v", fh.f.dir.Path, fh.f.Name, len(fh.f.entry.Chunks))
  171. //for i, chunk := range fh.f.entry.Chunks {
  172. // glog.V(4).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))
  173. //}
  174. chunks, garbages := filer2.CompactFileChunks(fh.f.entry.Chunks)
  175. fh.f.entry.Chunks = chunks
  176. // fh.f.entryViewCache = nil
  177. fh.f.wfs.deleteFileChunks(ctx, garbages)
  178. if _, err := client.CreateEntry(ctx, request); err != nil {
  179. return fmt.Errorf("update fh: %v", err)
  180. }
  181. return nil
  182. })
  183. }
  184. func deleteFileIds(ctx context.Context, grpcDialOption grpc.DialOption, client filer_pb.SeaweedFilerClient, fileIds []string) error {
  185. var vids []string
  186. for _, fileId := range fileIds {
  187. vids = append(vids, volumeId(fileId))
  188. }
  189. lookupFunc := func(vids []string) (map[string]operation.LookupResult, error) {
  190. m := make(map[string]operation.LookupResult)
  191. glog.V(4).Infof("remove file lookup volume id locations: %v", vids)
  192. resp, err := client.LookupVolume(ctx, &filer_pb.LookupVolumeRequest{
  193. VolumeIds: vids,
  194. })
  195. if err != nil {
  196. return m, err
  197. }
  198. for _, vid := range vids {
  199. lr := operation.LookupResult{
  200. VolumeId: vid,
  201. Locations: nil,
  202. }
  203. locations := resp.LocationsMap[vid]
  204. for _, loc := range locations.Locations {
  205. lr.Locations = append(lr.Locations, operation.Location{
  206. Url: loc.Url,
  207. PublicUrl: loc.PublicUrl,
  208. })
  209. }
  210. m[vid] = lr
  211. }
  212. return m, err
  213. }
  214. _, err := operation.DeleteFilesWithLookupVolumeId(grpcDialOption, fileIds, lookupFunc)
  215. return err
  216. }
  217. func volumeId(fileId string) string {
  218. lastCommaIndex := strings.LastIndex(fileId, ",")
  219. if lastCommaIndex > 0 {
  220. return fileId[:lastCommaIndex]
  221. }
  222. return fileId
  223. }