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.

282 lines
7.6 KiB

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