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.

272 lines
6.1 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
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. if _, err := f.FindEntry(FullPath(dirPath)); err == ErrNotFound {
  73. return fmt.Errorf("mkdir %s: %v", dirPath, mkdirErr)
  74. }
  75. } else {
  76. f.NotifyUpdateEvent(nil, dirEntry, false)
  77. }
  78. } else if !dirEntry.IsDirectory() {
  79. return fmt.Errorf("%s is a file", dirPath)
  80. }
  81. // cache the directory entry
  82. f.cacheSetDirectory(dirPath, dirEntry, i)
  83. // remember the direct parent directory entry
  84. if i == len(dirParts)-1 {
  85. lastDirectoryEntry = dirEntry
  86. }
  87. }
  88. if lastDirectoryEntry == nil {
  89. return fmt.Errorf("parent folder not found: %v", entry.FullPath)
  90. }
  91. /*
  92. if !hasWritePermission(lastDirectoryEntry, entry) {
  93. glog.V(0).Infof("directory %s: %v, entry: uid=%d gid=%d",
  94. lastDirectoryEntry.FullPath, lastDirectoryEntry.Attr, entry.Uid, entry.Gid)
  95. return fmt.Errorf("no write permission in folder %v", lastDirectoryEntry.FullPath)
  96. }
  97. */
  98. oldEntry, _ := f.FindEntry(entry.FullPath)
  99. if oldEntry == nil {
  100. if err := f.store.InsertEntry(entry); err != nil {
  101. return fmt.Errorf("insert entry %s: %v", entry.FullPath, err)
  102. }
  103. } else {
  104. if err := f.store.UpdateEntry(entry); err != nil {
  105. return fmt.Errorf("update entry %s: %v", entry.FullPath, err)
  106. }
  107. }
  108. f.NotifyUpdateEvent(oldEntry, entry, true)
  109. f.deleteChunksIfNotNew(oldEntry, entry)
  110. return nil
  111. }
  112. func (f *Filer) UpdateEntry(entry *Entry) (err error) {
  113. return f.store.UpdateEntry(entry)
  114. }
  115. func (f *Filer) FindEntry(p FullPath) (entry *Entry, err error) {
  116. return f.store.FindEntry(p)
  117. }
  118. func (f *Filer) DeleteEntryMetaAndData(p FullPath, isRecursive bool, shouldDeleteChunks bool) (err error) {
  119. entry, err := f.FindEntry(p)
  120. if err != nil {
  121. return err
  122. }
  123. if entry.IsDirectory() {
  124. limit := int(1)
  125. if isRecursive {
  126. limit = math.MaxInt32
  127. }
  128. entries, err := f.ListDirectoryEntries(p, "", false, limit)
  129. if err != nil {
  130. return fmt.Errorf("list folder %s: %v", p, err)
  131. }
  132. if isRecursive {
  133. for _, sub := range entries {
  134. f.DeleteEntryMetaAndData(sub.FullPath, isRecursive, shouldDeleteChunks)
  135. }
  136. } else {
  137. if len(entries) > 0 {
  138. return fmt.Errorf("folder %s is not empty", p)
  139. }
  140. }
  141. f.cacheDelDirectory(string(p))
  142. }
  143. if shouldDeleteChunks {
  144. f.deleteChunks(entry.Chunks)
  145. }
  146. if p == "/" {
  147. return nil
  148. }
  149. glog.V(3).Infof("deleting entry %v", p)
  150. f.NotifyUpdateEvent(entry, nil, shouldDeleteChunks)
  151. return f.store.DeleteEntry(p)
  152. }
  153. func (f *Filer) ListDirectoryEntries(p FullPath, startFileName string, inclusive bool, limit int) ([]*Entry, error) {
  154. if strings.HasSuffix(string(p), "/") && len(p) > 1 {
  155. p = p[0 : len(p)-1]
  156. }
  157. return f.store.ListDirectoryEntries(p, startFileName, inclusive, limit)
  158. }
  159. func (f *Filer) cacheDelDirectory(dirpath string) {
  160. if f.directoryCache == nil {
  161. return
  162. }
  163. f.directoryCache.Delete(dirpath)
  164. return
  165. }
  166. func (f *Filer) cacheGetDirectory(dirpath string) *Entry {
  167. if f.directoryCache == nil {
  168. return nil
  169. }
  170. item := f.directoryCache.Get(dirpath)
  171. if item == nil {
  172. return nil
  173. }
  174. return item.Value().(*Entry)
  175. }
  176. func (f *Filer) cacheSetDirectory(dirpath string, dirEntry *Entry, level int) {
  177. if f.directoryCache == nil {
  178. return
  179. }
  180. minutes := 60
  181. if level < 10 {
  182. minutes -= level * 6
  183. }
  184. f.directoryCache.Set(dirpath, dirEntry, time.Duration(minutes)*time.Minute)
  185. }
  186. func (f *Filer) deleteChunks(chunks []*filer_pb.FileChunk) {
  187. for _, chunk := range chunks {
  188. f.DeleteFileByFileId(chunk.FileId)
  189. }
  190. }
  191. func (f *Filer) DeleteFileByFileId(fileId string) {
  192. fileUrlOnVolume, err := f.MasterClient.LookupFileId(fileId)
  193. if err != nil {
  194. glog.V(0).Infof("can not find file %s: %v", fileId, err)
  195. }
  196. if err := operation.DeleteFromVolumeServer(fileUrlOnVolume, ""); err != nil {
  197. glog.V(0).Infof("deleting file %s: %v", fileId, err)
  198. }
  199. }
  200. func (f *Filer) deleteChunksIfNotNew(oldEntry, newEntry *Entry) {
  201. if oldEntry == nil {
  202. return
  203. }
  204. if newEntry == nil {
  205. f.deleteChunks(oldEntry.Chunks)
  206. }
  207. var toDelete []*filer_pb.FileChunk
  208. for _, oldChunk := range oldEntry.Chunks {
  209. found := false
  210. for _, newChunk := range newEntry.Chunks {
  211. if oldChunk.FileId == newChunk.FileId {
  212. found = true
  213. break
  214. }
  215. }
  216. if !found {
  217. toDelete = append(toDelete, oldChunk)
  218. }
  219. }
  220. f.deleteChunks(toDelete)
  221. }