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.

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