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.

321 lines
8.5 KiB

7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
5 years ago
7 years ago
7 years ago
5 years ago
7 years ago
7 years ago
5 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
7 years ago
5 years ago
7 years ago
5 years ago
6 years ago
5 years ago
6 years ago
5 years ago
7 years ago
6 years ago
6 years ago
7 years ago
5 years ago
7 years ago
5 years ago
5 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
5 years ago
  1. package filer2
  2. import (
  3. "context"
  4. "fmt"
  5. "os"
  6. "strings"
  7. "time"
  8. "google.golang.org/grpc"
  9. "github.com/karlseguin/ccache"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  12. "github.com/chrislusf/seaweedfs/weed/util"
  13. "github.com/chrislusf/seaweedfs/weed/util/log_buffer"
  14. "github.com/chrislusf/seaweedfs/weed/wdclient"
  15. )
  16. const PaginationSize = 1024 * 256
  17. var (
  18. OS_UID = uint32(os.Getuid())
  19. OS_GID = uint32(os.Getgid())
  20. )
  21. type Filer struct {
  22. store *FilerStoreWrapper
  23. directoryCache *ccache.Cache
  24. MasterClient *wdclient.MasterClient
  25. fileIdDeletionQueue *util.UnboundedQueue
  26. GrpcDialOption grpc.DialOption
  27. DirBucketsPath string
  28. FsyncBuckets []string
  29. buckets *FilerBuckets
  30. Cipher bool
  31. MetaLogBuffer *log_buffer.LogBuffer
  32. metaLogCollection string
  33. metaLogReplication string
  34. }
  35. func NewFiler(masters []string, grpcDialOption grpc.DialOption, filerHost string, filerGrpcPort uint32, collection string, replication string, notifyFn func()) *Filer {
  36. f := &Filer{
  37. directoryCache: ccache.New(ccache.Configure().MaxSize(1000).ItemsToPrune(100)),
  38. MasterClient: wdclient.NewMasterClient(grpcDialOption, "filer", filerHost, filerGrpcPort, masters),
  39. fileIdDeletionQueue: util.NewUnboundedQueue(),
  40. GrpcDialOption: grpcDialOption,
  41. }
  42. f.MetaLogBuffer = log_buffer.NewLogBuffer(time.Minute, f.logFlushFunc, notifyFn)
  43. f.metaLogCollection = collection
  44. f.metaLogReplication = replication
  45. go f.loopProcessingDeletion()
  46. return f
  47. }
  48. func (f *Filer) SetStore(store FilerStore) {
  49. f.store = NewFilerStoreWrapper(store)
  50. }
  51. func (f *Filer) DisableDirectoryCache() {
  52. f.directoryCache = nil
  53. }
  54. func (fs *Filer) GetMaster() string {
  55. return fs.MasterClient.GetMaster()
  56. }
  57. func (fs *Filer) KeepConnectedToMaster() {
  58. fs.MasterClient.KeepConnectedToMaster()
  59. }
  60. func (f *Filer) BeginTransaction(ctx context.Context) (context.Context, error) {
  61. return f.store.BeginTransaction(ctx)
  62. }
  63. func (f *Filer) CommitTransaction(ctx context.Context) error {
  64. return f.store.CommitTransaction(ctx)
  65. }
  66. func (f *Filer) RollbackTransaction(ctx context.Context) error {
  67. return f.store.RollbackTransaction(ctx)
  68. }
  69. func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, o_excl bool) error {
  70. if string(entry.FullPath) == "/" {
  71. return nil
  72. }
  73. dirParts := strings.Split(string(entry.FullPath), "/")
  74. // fmt.Printf("directory parts: %+v\n", dirParts)
  75. var lastDirectoryEntry *Entry
  76. for i := 1; i < len(dirParts); i++ {
  77. dirPath := "/" + util.Join(dirParts[:i]...)
  78. // fmt.Printf("%d directory: %+v\n", i, dirPath)
  79. // first check local cache
  80. dirEntry := f.cacheGetDirectory(dirPath)
  81. // not found, check the store directly
  82. if dirEntry == nil {
  83. glog.V(4).Infof("find uncached directory: %s", dirPath)
  84. dirEntry, _ = f.FindEntry(ctx, util.FullPath(dirPath))
  85. } else {
  86. // glog.V(4).Infof("found cached directory: %s", dirPath)
  87. }
  88. // no such existing directory
  89. if dirEntry == nil {
  90. // create the directory
  91. now := time.Now()
  92. dirEntry = &Entry{
  93. FullPath: util.FullPath(dirPath),
  94. Attr: Attr{
  95. Mtime: now,
  96. Crtime: now,
  97. Mode: os.ModeDir | 0770,
  98. Uid: entry.Uid,
  99. Gid: entry.Gid,
  100. Collection: entry.Collection,
  101. Replication: entry.Replication,
  102. },
  103. }
  104. glog.V(2).Infof("create directory: %s %v", dirPath, dirEntry.Mode)
  105. mkdirErr := f.store.InsertEntry(ctx, dirEntry)
  106. if mkdirErr != nil {
  107. if _, err := f.FindEntry(ctx, util.FullPath(dirPath)); err == filer_pb.ErrNotFound {
  108. glog.V(3).Infof("mkdir %s: %v", dirPath, mkdirErr)
  109. return fmt.Errorf("mkdir %s: %v", dirPath, mkdirErr)
  110. }
  111. } else {
  112. f.maybeAddBucket(dirEntry)
  113. f.NotifyUpdateEvent(nil, dirEntry, false)
  114. }
  115. } else if !dirEntry.IsDirectory() {
  116. glog.Errorf("CreateEntry %s: %s should be a directory", entry.FullPath, dirPath)
  117. return fmt.Errorf("%s is a file", dirPath)
  118. }
  119. // cache the directory entry
  120. f.cacheSetDirectory(dirPath, dirEntry, i)
  121. // remember the direct parent directory entry
  122. if i == len(dirParts)-1 {
  123. lastDirectoryEntry = dirEntry
  124. }
  125. }
  126. if lastDirectoryEntry == nil {
  127. glog.Errorf("CreateEntry %s: lastDirectoryEntry is nil", entry.FullPath)
  128. return fmt.Errorf("parent folder not found: %v", entry.FullPath)
  129. }
  130. /*
  131. if !hasWritePermission(lastDirectoryEntry, entry) {
  132. glog.V(0).Infof("directory %s: %v, entry: uid=%d gid=%d",
  133. lastDirectoryEntry.FullPath, lastDirectoryEntry.Attr, entry.Uid, entry.Gid)
  134. return fmt.Errorf("no write permission in folder %v", lastDirectoryEntry.FullPath)
  135. }
  136. */
  137. oldEntry, _ := f.FindEntry(ctx, entry.FullPath)
  138. glog.V(4).Infof("CreateEntry %s: old entry: %v exclusive:%v", entry.FullPath, oldEntry, o_excl)
  139. if oldEntry == nil {
  140. if err := f.store.InsertEntry(ctx, entry); err != nil {
  141. glog.Errorf("insert entry %s: %v", entry.FullPath, err)
  142. return fmt.Errorf("insert entry %s: %v", entry.FullPath, err)
  143. }
  144. } else {
  145. if o_excl {
  146. glog.V(3).Infof("EEXIST: entry %s already exists", entry.FullPath)
  147. return fmt.Errorf("EEXIST: entry %s already exists", entry.FullPath)
  148. }
  149. if err := f.UpdateEntry(ctx, oldEntry, entry); err != nil {
  150. glog.Errorf("update entry %s: %v", entry.FullPath, err)
  151. return fmt.Errorf("update entry %s: %v", entry.FullPath, err)
  152. }
  153. }
  154. f.maybeAddBucket(entry)
  155. f.NotifyUpdateEvent(oldEntry, entry, true)
  156. f.deleteChunksIfNotNew(oldEntry, entry)
  157. glog.V(4).Infof("CreateEntry %s: created", entry.FullPath)
  158. return nil
  159. }
  160. func (f *Filer) UpdateEntry(ctx context.Context, oldEntry, entry *Entry) (err error) {
  161. if oldEntry != nil {
  162. if oldEntry.IsDirectory() && !entry.IsDirectory() {
  163. glog.Errorf("existing %s is a directory", entry.FullPath)
  164. return fmt.Errorf("existing %s is a directory", entry.FullPath)
  165. }
  166. if !oldEntry.IsDirectory() && entry.IsDirectory() {
  167. glog.Errorf("existing %s is a file", entry.FullPath)
  168. return fmt.Errorf("existing %s is a file", entry.FullPath)
  169. }
  170. }
  171. return f.store.UpdateEntry(ctx, entry)
  172. }
  173. func (f *Filer) FindEntry(ctx context.Context, p util.FullPath) (entry *Entry, err error) {
  174. now := time.Now()
  175. if string(p) == "/" {
  176. return &Entry{
  177. FullPath: p,
  178. Attr: Attr{
  179. Mtime: now,
  180. Crtime: now,
  181. Mode: os.ModeDir | 0755,
  182. Uid: OS_UID,
  183. Gid: OS_GID,
  184. },
  185. }, nil
  186. }
  187. entry, err = f.store.FindEntry(ctx, p)
  188. if entry != nil && entry.TtlSec > 0 {
  189. if entry.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) {
  190. f.store.DeleteEntry(ctx, p.Child(entry.Name()))
  191. return nil, filer_pb.ErrNotFound
  192. }
  193. }
  194. return
  195. }
  196. func (f *Filer) ListDirectoryEntries(ctx context.Context, p util.FullPath, startFileName string, inclusive bool, limit int) ([]*Entry, error) {
  197. if strings.HasSuffix(string(p), "/") && len(p) > 1 {
  198. p = p[0 : len(p)-1]
  199. }
  200. var makeupEntries []*Entry
  201. entries, expiredCount, lastFileName, err := f.doListDirectoryEntries(ctx, p, startFileName, inclusive, limit)
  202. for expiredCount > 0 && err == nil {
  203. makeupEntries, expiredCount, lastFileName, err = f.doListDirectoryEntries(ctx, p, lastFileName, false, expiredCount)
  204. if err == nil {
  205. entries = append(entries, makeupEntries...)
  206. }
  207. }
  208. return entries, err
  209. }
  210. func (f *Filer) doListDirectoryEntries(ctx context.Context, p util.FullPath, startFileName string, inclusive bool, limit int) (entries []*Entry, expiredCount int, lastFileName string, err error) {
  211. listedEntries, listErr := f.store.ListDirectoryEntries(ctx, p, startFileName, inclusive, limit)
  212. if listErr != nil {
  213. return listedEntries, expiredCount, "", listErr
  214. }
  215. for _, entry := range listedEntries {
  216. lastFileName = entry.Name()
  217. if entry.TtlSec > 0 {
  218. if entry.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) {
  219. f.store.DeleteEntry(ctx, p.Child(entry.Name()))
  220. expiredCount++
  221. continue
  222. }
  223. }
  224. entries = append(entries, entry)
  225. }
  226. return
  227. }
  228. func (f *Filer) cacheDelDirectory(dirpath string) {
  229. if dirpath == "/" {
  230. return
  231. }
  232. if f.directoryCache == nil {
  233. return
  234. }
  235. f.directoryCache.Delete(dirpath)
  236. return
  237. }
  238. func (f *Filer) cacheGetDirectory(dirpath string) *Entry {
  239. if f.directoryCache == nil {
  240. return nil
  241. }
  242. item := f.directoryCache.Get(dirpath)
  243. if item == nil {
  244. return nil
  245. }
  246. return item.Value().(*Entry)
  247. }
  248. func (f *Filer) cacheSetDirectory(dirpath string, dirEntry *Entry, level int) {
  249. if f.directoryCache == nil {
  250. return
  251. }
  252. minutes := 60
  253. if level < 10 {
  254. minutes -= level * 6
  255. }
  256. f.directoryCache.Set(dirpath, dirEntry, time.Duration(minutes)*time.Minute)
  257. }
  258. func (f *Filer) Shutdown() {
  259. f.MetaLogBuffer.Shutdown()
  260. f.store.Shutdown()
  261. }