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.

264 lines
5.8 KiB

7 years ago
6 years ago
7 years ago
7 years ago
7 years ago
6 years ago
6 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
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 filer2
  2. import (
  3. "context"
  4. "fmt"
  5. "os"
  6. "path/filepath"
  7. "strings"
  8. "time"
  9. "github.com/chrislusf/seaweedfs/weed/glog"
  10. "github.com/chrislusf/seaweedfs/weed/operation"
  11. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  12. "github.com/chrislusf/seaweedfs/weed/wdclient"
  13. "github.com/karlseguin/ccache"
  14. "math"
  15. )
  16. type Filer struct {
  17. store FilerStore
  18. directoryCache *ccache.Cache
  19. MasterClient *wdclient.MasterClient
  20. }
  21. func NewFiler(masters []string) *Filer {
  22. return &Filer{
  23. directoryCache: ccache.New(ccache.Configure().MaxSize(1000).ItemsToPrune(100)),
  24. MasterClient: wdclient.NewMasterClient(context.Background(), "filer", masters),
  25. }
  26. }
  27. func (f *Filer) SetStore(store FilerStore) {
  28. f.store = store
  29. }
  30. func (f *Filer) DisableDirectoryCache() {
  31. f.directoryCache = nil
  32. }
  33. func (fs *Filer) GetMaster() string {
  34. return fs.MasterClient.GetMaster()
  35. }
  36. func (fs *Filer) KeepConnectedToMaster() {
  37. fs.MasterClient.KeepConnectedToMaster()
  38. }
  39. func (f *Filer) CreateEntry(entry *Entry) error {
  40. dirParts := strings.Split(string(entry.FullPath), "/")
  41. // fmt.Printf("directory parts: %+v\n", dirParts)
  42. var lastDirectoryEntry *Entry
  43. for i := 1; i < len(dirParts); i++ {
  44. dirPath := "/" + filepath.Join(dirParts[:i]...)
  45. // fmt.Printf("%d directory: %+v\n", i, dirPath)
  46. // first check local cache
  47. dirEntry := f.cacheGetDirectory(dirPath)
  48. // not found, check the store directly
  49. if dirEntry == nil {
  50. glog.V(4).Infof("find uncached directory: %s", dirPath)
  51. dirEntry, _ = f.FindEntry(FullPath(dirPath))
  52. } else {
  53. glog.V(4).Infof("found cached directory: %s", dirPath)
  54. }
  55. // no such existing directory
  56. if dirEntry == nil {
  57. // create the directory
  58. now := time.Now()
  59. dirEntry = &Entry{
  60. FullPath: FullPath(dirPath),
  61. Attr: Attr{
  62. Mtime: now,
  63. Crtime: now,
  64. Mode: os.ModeDir | 0770,
  65. Uid: entry.Uid,
  66. Gid: entry.Gid,
  67. },
  68. }
  69. glog.V(2).Infof("create directory: %s %v", dirPath, dirEntry.Mode)
  70. mkdirErr := f.store.InsertEntry(dirEntry)
  71. if mkdirErr != nil {
  72. return fmt.Errorf("mkdir %s: %v", dirPath, mkdirErr)
  73. }
  74. f.NotifyUpdateEvent(nil, dirEntry)
  75. } else if !dirEntry.IsDirectory() {
  76. return fmt.Errorf("%s is a file", dirPath)
  77. }
  78. // cache the directory entry
  79. f.cacheSetDirectory(dirPath, dirEntry, i)
  80. // remember the direct parent directory entry
  81. if i == len(dirParts)-1 {
  82. lastDirectoryEntry = dirEntry
  83. }
  84. }
  85. if lastDirectoryEntry == nil {
  86. return fmt.Errorf("parent folder not found: %v", entry.FullPath)
  87. }
  88. /*
  89. if !hasWritePermission(lastDirectoryEntry, entry) {
  90. glog.V(0).Infof("directory %s: %v, entry: uid=%d gid=%d",
  91. lastDirectoryEntry.FullPath, lastDirectoryEntry.Attr, entry.Uid, entry.Gid)
  92. return fmt.Errorf("no write permission in folder %v", lastDirectoryEntry.FullPath)
  93. }
  94. */
  95. oldEntry, _ := f.FindEntry(entry.FullPath)
  96. if err := f.store.InsertEntry(entry); err != nil {
  97. return fmt.Errorf("insert entry %s: %v", entry.FullPath, err)
  98. }
  99. f.NotifyUpdateEvent(oldEntry, entry)
  100. f.deleteChunksIfNotNew(oldEntry, entry)
  101. return nil
  102. }
  103. func (f *Filer) UpdateEntry(entry *Entry) (err error) {
  104. return f.store.UpdateEntry(entry)
  105. }
  106. func (f *Filer) FindEntry(p FullPath) (entry *Entry, err error) {
  107. return f.store.FindEntry(p)
  108. }
  109. func (f *Filer) DeleteEntryMetaAndData(p FullPath, isRecursive bool, shouldDeleteChunks bool) (err error) {
  110. entry, err := f.FindEntry(p)
  111. if err != nil {
  112. return err
  113. }
  114. if entry.IsDirectory() {
  115. limit := int(1)
  116. if isRecursive {
  117. limit = math.MaxInt32
  118. }
  119. entries, err := f.ListDirectoryEntries(p, "", false, limit)
  120. if err != nil {
  121. return fmt.Errorf("list folder %s: %v", p, err)
  122. }
  123. if isRecursive {
  124. for _, sub := range entries {
  125. f.DeleteEntryMetaAndData(sub.FullPath, isRecursive, shouldDeleteChunks)
  126. }
  127. } else {
  128. if len(entries) > 0 {
  129. return fmt.Errorf("folder %s is not empty", p)
  130. }
  131. }
  132. f.cacheDelDirectory(string(p))
  133. }
  134. if shouldDeleteChunks {
  135. f.deleteChunks(entry.Chunks)
  136. }
  137. if p == "/" {
  138. return nil
  139. }
  140. glog.V(3).Infof("deleting entry %v", p)
  141. f.NotifyUpdateEvent(entry, nil)
  142. return f.store.DeleteEntry(p)
  143. }
  144. func (f *Filer) ListDirectoryEntries(p FullPath, startFileName string, inclusive bool, limit int) ([]*Entry, error) {
  145. if strings.HasSuffix(string(p), "/") && len(p) > 1 {
  146. p = p[0 : len(p)-1]
  147. }
  148. return f.store.ListDirectoryEntries(p, startFileName, inclusive, limit)
  149. }
  150. func (f *Filer) cacheDelDirectory(dirpath string) {
  151. if f.directoryCache == nil {
  152. return
  153. }
  154. f.directoryCache.Delete(dirpath)
  155. return
  156. }
  157. func (f *Filer) cacheGetDirectory(dirpath string) *Entry {
  158. if f.directoryCache == nil {
  159. return nil
  160. }
  161. item := f.directoryCache.Get(dirpath)
  162. if item == nil {
  163. return nil
  164. }
  165. return item.Value().(*Entry)
  166. }
  167. func (f *Filer) cacheSetDirectory(dirpath string, dirEntry *Entry, level int) {
  168. if f.directoryCache == nil {
  169. return
  170. }
  171. minutes := 60
  172. if level < 10 {
  173. minutes -= level * 6
  174. }
  175. f.directoryCache.Set(dirpath, dirEntry, time.Duration(minutes)*time.Minute)
  176. }
  177. func (f *Filer) deleteChunks(chunks []*filer_pb.FileChunk) {
  178. for _, chunk := range chunks {
  179. f.DeleteFileByFileId(chunk.FileId)
  180. }
  181. }
  182. func (f *Filer) DeleteFileByFileId(fileId string) {
  183. fileUrlOnVolume, err := f.MasterClient.LookupFileId(fileId)
  184. if err != nil {
  185. glog.V(0).Infof("can not find file %s: %v", fileId, err)
  186. }
  187. if err := operation.DeleteFromVolumeServer(fileUrlOnVolume, ""); err != nil {
  188. glog.V(0).Infof("deleting file %s: %v", fileId, err)
  189. }
  190. }
  191. func (f *Filer) deleteChunksIfNotNew(oldEntry, newEntry *Entry) {
  192. if oldEntry == nil {
  193. return
  194. }
  195. if newEntry == nil {
  196. f.deleteChunks(oldEntry.Chunks)
  197. }
  198. var toDelete []*filer_pb.FileChunk
  199. for _, oldChunk := range oldEntry.Chunks {
  200. found := false
  201. for _, newChunk := range newEntry.Chunks {
  202. if oldChunk.FileId == newChunk.FileId {
  203. found = true
  204. break
  205. }
  206. }
  207. if !found {
  208. toDelete = append(toDelete, oldChunk)
  209. }
  210. }
  211. f.deleteChunks(toDelete)
  212. }