leveldb3_store.go 9.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378
  1. package leveldb
  2. import (
  3. "bytes"
  4. "context"
  5. "crypto/md5"
  6. "fmt"
  7. "io"
  8. "os"
  9. "strings"
  10. "sync"
  11. "github.com/syndtr/goleveldb/leveldb"
  12. leveldb_errors "github.com/syndtr/goleveldb/leveldb/errors"
  13. "github.com/syndtr/goleveldb/leveldb/filter"
  14. "github.com/syndtr/goleveldb/leveldb/opt"
  15. leveldb_util "github.com/syndtr/goleveldb/leveldb/util"
  16. "github.com/chrislusf/seaweedfs/weed/filer"
  17. "github.com/chrislusf/seaweedfs/weed/glog"
  18. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  19. weed_util "github.com/chrislusf/seaweedfs/weed/util"
  20. )
  21. const (
  22. DEFAULT = "_main"
  23. )
  24. func init() {
  25. filer.Stores = append(filer.Stores, &LevelDB3Store{})
  26. }
  27. type LevelDB3Store struct {
  28. dir string
  29. dbs map[string]*leveldb.DB
  30. dbsLock sync.RWMutex
  31. }
  32. func (store *LevelDB3Store) GetName() string {
  33. return "leveldb3"
  34. }
  35. func (store *LevelDB3Store) Initialize(configuration weed_util.Configuration, prefix string) (err error) {
  36. dir := configuration.GetString(prefix + "dir")
  37. return store.initialize(dir)
  38. }
  39. func (store *LevelDB3Store) initialize(dir string) (err error) {
  40. glog.Infof("filer store leveldb3 dir: %s", dir)
  41. os.MkdirAll(dir, 0755)
  42. if err := weed_util.TestFolderWritable(dir); err != nil {
  43. return fmt.Errorf("Check Level Folder %s Writable: %s", dir, err)
  44. }
  45. store.dir = dir
  46. db, loadDbErr := store.loadDB(DEFAULT)
  47. if loadDbErr != nil {
  48. return loadDbErr
  49. }
  50. store.dbs = make(map[string]*leveldb.DB)
  51. store.dbs[DEFAULT] = db
  52. return
  53. }
  54. func (store *LevelDB3Store) loadDB(name string) (*leveldb.DB, error) {
  55. bloom := filter.NewBloomFilter(8) // false positive rate 0.02
  56. opts := &opt.Options{
  57. BlockCacheCapacity: 32 * 1024 * 1024, // default value is 8MiB
  58. WriteBuffer: 16 * 1024 * 1024, // default value is 4MiB
  59. Filter: bloom,
  60. }
  61. if name != DEFAULT {
  62. opts = &opt.Options{
  63. BlockCacheCapacity: 16 * 1024 * 1024, // default value is 8MiB
  64. WriteBuffer: 8 * 1024 * 1024, // default value is 4MiB
  65. Filter: bloom,
  66. }
  67. }
  68. dbFolder := fmt.Sprintf("%s/%s", store.dir, name)
  69. os.MkdirAll(dbFolder, 0755)
  70. db, dbErr := leveldb.OpenFile(dbFolder, opts)
  71. if leveldb_errors.IsCorrupted(dbErr) {
  72. db, dbErr = leveldb.RecoverFile(dbFolder, opts)
  73. }
  74. if dbErr != nil {
  75. glog.Errorf("filer store open dir %s: %v", dbFolder, dbErr)
  76. return nil, dbErr
  77. }
  78. return db, nil
  79. }
  80. func (store *LevelDB3Store) findDB(fullpath weed_util.FullPath, isForChildren bool) (*leveldb.DB, string, weed_util.FullPath, error) {
  81. store.dbsLock.RLock()
  82. defaultDB := store.dbs[DEFAULT]
  83. if !strings.HasPrefix(string(fullpath), "/buckets/") {
  84. store.dbsLock.RUnlock()
  85. return defaultDB, DEFAULT, fullpath, nil
  86. }
  87. // detect bucket
  88. bucketAndObjectKey := string(fullpath)[len("/buckets/"):]
  89. t := strings.Index(bucketAndObjectKey, "/")
  90. if t < 0 && !isForChildren {
  91. store.dbsLock.RUnlock()
  92. return defaultDB, DEFAULT, fullpath, nil
  93. }
  94. bucket := bucketAndObjectKey
  95. shortPath := weed_util.FullPath("/")
  96. if t > 0 {
  97. bucket = bucketAndObjectKey[:t]
  98. shortPath = weed_util.FullPath(bucketAndObjectKey[t:])
  99. }
  100. if db, found := store.dbs[bucket]; found {
  101. store.dbsLock.RUnlock()
  102. return db, bucket, shortPath, nil
  103. }
  104. store.dbsLock.RUnlock()
  105. // upgrade to write lock
  106. store.dbsLock.Lock()
  107. defer store.dbsLock.Unlock()
  108. // double check after getting the write lock
  109. if db, found := store.dbs[bucket]; found {
  110. return db, bucket, shortPath, nil
  111. }
  112. // create db
  113. db, err := store.loadDB(bucket)
  114. if err != nil {
  115. return nil, bucket, shortPath, err
  116. }
  117. store.dbs[bucket] = db
  118. return db, bucket, shortPath, nil
  119. }
  120. func (store *LevelDB3Store) closeDB(bucket string) {
  121. store.dbsLock.Lock()
  122. defer store.dbsLock.Unlock()
  123. if db, found := store.dbs[bucket]; found {
  124. db.Close()
  125. delete(store.dbs, bucket)
  126. }
  127. }
  128. func (store *LevelDB3Store) BeginTransaction(ctx context.Context) (context.Context, error) {
  129. return ctx, nil
  130. }
  131. func (store *LevelDB3Store) CommitTransaction(ctx context.Context) error {
  132. return nil
  133. }
  134. func (store *LevelDB3Store) RollbackTransaction(ctx context.Context) error {
  135. return nil
  136. }
  137. func (store *LevelDB3Store) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
  138. db, _, shortPath, err := store.findDB(entry.FullPath, false)
  139. if err != nil {
  140. return fmt.Errorf("findDB %s : %v", entry.FullPath, err)
  141. }
  142. dir, name := shortPath.DirAndName()
  143. key := genKey(dir, name)
  144. value, err := entry.EncodeAttributesAndChunks()
  145. if err != nil {
  146. return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
  147. }
  148. if len(entry.Chunks) > 50 {
  149. value = weed_util.MaybeGzipData(value)
  150. }
  151. err = db.Put(key, value, nil)
  152. if err != nil {
  153. return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
  154. }
  155. // println("saved", entry.FullPath, "chunks", len(entry.Chunks))
  156. return nil
  157. }
  158. func (store *LevelDB3Store) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
  159. return store.InsertEntry(ctx, entry)
  160. }
  161. func (store *LevelDB3Store) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) {
  162. db, _, shortPath, err := store.findDB(fullpath, false)
  163. if err != nil {
  164. return nil, fmt.Errorf("findDB %s : %v", fullpath, err)
  165. }
  166. dir, name := shortPath.DirAndName()
  167. key := genKey(dir, name)
  168. data, err := db.Get(key, nil)
  169. if err == leveldb.ErrNotFound {
  170. return nil, filer_pb.ErrNotFound
  171. }
  172. if err != nil {
  173. return nil, fmt.Errorf("get %s : %v", fullpath, err)
  174. }
  175. entry = &filer.Entry{
  176. FullPath: fullpath,
  177. }
  178. err = entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(data))
  179. if err != nil {
  180. return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
  181. }
  182. // println("read", entry.FullPath, "chunks", len(entry.Chunks), "data", len(data), string(data))
  183. return entry, nil
  184. }
  185. func (store *LevelDB3Store) DeleteEntry(ctx context.Context, fullpath weed_util.FullPath) (err error) {
  186. db, _, shortPath, err := store.findDB(fullpath, false)
  187. if err != nil {
  188. return fmt.Errorf("findDB %s : %v", fullpath, err)
  189. }
  190. dir, name := shortPath.DirAndName()
  191. key := genKey(dir, name)
  192. err = db.Delete(key, nil)
  193. if err != nil {
  194. return fmt.Errorf("delete %s : %v", fullpath, err)
  195. }
  196. return nil
  197. }
  198. func (store *LevelDB3Store) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
  199. db, bucket, shortPath, err := store.findDB(fullpath, true)
  200. if err != nil {
  201. return fmt.Errorf("findDB %s : %v", fullpath, err)
  202. }
  203. if bucket != DEFAULT && shortPath == "/" {
  204. store.closeDB(bucket)
  205. if bucket != "" { // just to make sure
  206. os.RemoveAll(store.dir + "/" + bucket)
  207. }
  208. return nil
  209. }
  210. directoryPrefix := genDirectoryKeyPrefix(shortPath, "")
  211. batch := new(leveldb.Batch)
  212. iter := db.NewIterator(&leveldb_util.Range{Start: directoryPrefix}, nil)
  213. for iter.Next() {
  214. key := iter.Key()
  215. if !bytes.HasPrefix(key, directoryPrefix) {
  216. break
  217. }
  218. fileName := getNameFromKey(key)
  219. if fileName == "" {
  220. continue
  221. }
  222. batch.Delete(append(directoryPrefix, []byte(fileName)...))
  223. }
  224. iter.Release()
  225. err = db.Write(batch, nil)
  226. if err != nil {
  227. return fmt.Errorf("delete %s : %v", fullpath, err)
  228. }
  229. return nil
  230. }
  231. func (store *LevelDB3Store) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  232. return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", eachEntryFunc)
  233. }
  234. func (store *LevelDB3Store) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  235. db, _, shortPath, err := store.findDB(dirPath, true)
  236. if err != nil {
  237. return lastFileName, fmt.Errorf("findDB %s : %v", dirPath, err)
  238. }
  239. directoryPrefix := genDirectoryKeyPrefix(shortPath, prefix)
  240. lastFileStart := directoryPrefix
  241. if startFileName != "" {
  242. lastFileStart = genDirectoryKeyPrefix(shortPath, startFileName)
  243. }
  244. iter := db.NewIterator(&leveldb_util.Range{Start: lastFileStart}, nil)
  245. for iter.Next() {
  246. key := iter.Key()
  247. if !bytes.HasPrefix(key, directoryPrefix) {
  248. break
  249. }
  250. fileName := getNameFromKey(key)
  251. if fileName == "" {
  252. continue
  253. }
  254. if fileName == startFileName && !includeStartFile {
  255. continue
  256. }
  257. limit--
  258. if limit < 0 {
  259. break
  260. }
  261. lastFileName = fileName
  262. entry := &filer.Entry{
  263. FullPath: weed_util.NewFullPath(string(dirPath), fileName),
  264. }
  265. // println("list", entry.FullPath, "chunks", len(entry.Chunks))
  266. if decodeErr := entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(iter.Value())); decodeErr != nil {
  267. err = decodeErr
  268. glog.V(0).Infof("list %s : %v", entry.FullPath, err)
  269. break
  270. }
  271. if !eachEntryFunc(entry) {
  272. break
  273. }
  274. }
  275. iter.Release()
  276. return lastFileName, err
  277. }
  278. func genKey(dirPath, fileName string) (key []byte) {
  279. key = hashToBytes(dirPath)
  280. key = append(key, []byte(fileName)...)
  281. return key
  282. }
  283. func genDirectoryKeyPrefix(fullpath weed_util.FullPath, startFileName string) (keyPrefix []byte) {
  284. keyPrefix = hashToBytes(string(fullpath))
  285. if len(startFileName) > 0 {
  286. keyPrefix = append(keyPrefix, []byte(startFileName)...)
  287. }
  288. return keyPrefix
  289. }
  290. func getNameFromKey(key []byte) string {
  291. return string(key[md5.Size:])
  292. }
  293. // hash directory
  294. func hashToBytes(dir string) []byte {
  295. h := md5.New()
  296. io.WriteString(h, dir)
  297. b := h.Sum(nil)
  298. return b
  299. }
  300. func (store *LevelDB3Store) Shutdown() {
  301. for _, db := range store.dbs {
  302. db.Close()
  303. }
  304. }