memdb.go 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118
  1. package needle_map
  2. import (
  3. "fmt"
  4. "os"
  5. "github.com/syndtr/goleveldb/leveldb"
  6. "github.com/syndtr/goleveldb/leveldb/opt"
  7. "github.com/syndtr/goleveldb/leveldb/storage"
  8. "github.com/chrislusf/seaweedfs/weed/glog"
  9. "github.com/chrislusf/seaweedfs/weed/storage/idx"
  10. . "github.com/chrislusf/seaweedfs/weed/storage/types"
  11. )
  12. //This map uses in memory level db
  13. type MemDb struct {
  14. db *leveldb.DB
  15. }
  16. func NewMemDb() *MemDb {
  17. opts := &opt.Options{}
  18. var err error
  19. t := &MemDb{}
  20. if t.db, err = leveldb.Open(storage.NewMemStorage(), opts); err != nil {
  21. glog.V(0).Infof("MemDb fails to open: %v", err)
  22. return nil
  23. }
  24. return t
  25. }
  26. func (cm *MemDb) Set(key NeedleId, offset Offset, size Size) error {
  27. bytes := ToBytes(key, offset, size)
  28. if err := cm.db.Put(bytes[0:NeedleIdSize], bytes[NeedleIdSize:NeedleIdSize+OffsetSize+SizeSize], nil); err != nil {
  29. return fmt.Errorf("failed to write temp leveldb: %v", err)
  30. }
  31. return nil
  32. }
  33. func (cm *MemDb) Delete(key NeedleId) error {
  34. bytes := make([]byte, NeedleIdSize)
  35. NeedleIdToBytes(bytes, key)
  36. return cm.db.Delete(bytes, nil)
  37. }
  38. func (cm *MemDb) Get(key NeedleId) (*NeedleValue, bool) {
  39. bytes := make([]byte, NeedleIdSize)
  40. NeedleIdToBytes(bytes[0:NeedleIdSize], key)
  41. data, err := cm.db.Get(bytes, nil)
  42. if err != nil || len(data) != OffsetSize+SizeSize {
  43. return nil, false
  44. }
  45. offset := BytesToOffset(data[0:OffsetSize])
  46. size := BytesToSize(data[OffsetSize : OffsetSize+SizeSize])
  47. return &NeedleValue{Key: key, Offset: offset, Size: size}, true
  48. }
  49. // Visit visits all entries or stop if any error when visiting
  50. func (cm *MemDb) AscendingVisit(visit func(NeedleValue) error) (ret error) {
  51. iter := cm.db.NewIterator(nil, nil)
  52. for iter.Next() {
  53. key := BytesToNeedleId(iter.Key())
  54. data := iter.Value()
  55. offset := BytesToOffset(data[0:OffsetSize])
  56. size := BytesToSize(data[OffsetSize : OffsetSize+SizeSize])
  57. needle := NeedleValue{Key: key, Offset: offset, Size: size}
  58. ret = visit(needle)
  59. if ret != nil {
  60. return
  61. }
  62. }
  63. iter.Release()
  64. ret = iter.Error()
  65. return
  66. }
  67. func (cm *MemDb) SaveToIdx(idxName string) (ret error) {
  68. idxFile, err := os.OpenFile(idxName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
  69. if err != nil {
  70. return
  71. }
  72. defer idxFile.Close()
  73. return cm.AscendingVisit(func(value NeedleValue) error {
  74. if value.Offset.IsZero() || value.Size.IsDeleted() {
  75. return nil
  76. }
  77. _, err := idxFile.Write(value.ToBytes())
  78. return err
  79. })
  80. }
  81. func (cm *MemDb) LoadFromIdx(idxName string) (ret error) {
  82. idxFile, err := os.OpenFile(idxName, os.O_RDONLY, 0644)
  83. if err != nil {
  84. return
  85. }
  86. defer idxFile.Close()
  87. return idx.WalkIndexFile(idxFile, func(key NeedleId, offset Offset, size Size) error {
  88. if offset.IsZero() || size.IsDeleted() {
  89. return cm.Delete(key)
  90. }
  91. return cm.Set(key, offset, size)
  92. })
  93. }
  94. func (cm *MemDb) Close() {
  95. cm.db.Close()
  96. }