memdb.go 2.9 KB

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