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.

242 lines
5.5 KiB

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