chunk_cache_on_disk.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145
  1. package chunk_cache
  2. import (
  3. "fmt"
  4. "os"
  5. "time"
  6. "github.com/syndtr/goleveldb/leveldb/opt"
  7. "github.com/chrislusf/seaweedfs/weed/util/log"
  8. "github.com/chrislusf/seaweedfs/weed/storage"
  9. "github.com/chrislusf/seaweedfs/weed/storage/backend"
  10. "github.com/chrislusf/seaweedfs/weed/storage/types"
  11. "github.com/chrislusf/seaweedfs/weed/util"
  12. )
  13. // This implements an on disk cache
  14. // The entries are an FIFO with a size limit
  15. type ChunkCacheVolume struct {
  16. DataBackend backend.BackendStorageFile
  17. nm storage.NeedleMapper
  18. fileName string
  19. smallBuffer []byte
  20. sizeLimit int64
  21. lastModTime time.Time
  22. fileSize int64
  23. }
  24. func LoadOrCreateChunkCacheVolume(fileName string, preallocate int64) (*ChunkCacheVolume, error) {
  25. v := &ChunkCacheVolume{
  26. smallBuffer: make([]byte, types.NeedlePaddingSize),
  27. fileName: fileName,
  28. sizeLimit: preallocate,
  29. }
  30. var err error
  31. if exists, canRead, canWrite, modTime, fileSize := util.CheckFile(v.fileName + ".dat"); exists {
  32. if !canRead {
  33. return nil, fmt.Errorf("cannot read cache file %s.dat", v.fileName)
  34. }
  35. if !canWrite {
  36. return nil, fmt.Errorf("cannot write cache file %s.dat", v.fileName)
  37. }
  38. if dataFile, err := os.OpenFile(v.fileName+".dat", os.O_RDWR|os.O_CREATE, 0644); err != nil {
  39. return nil, fmt.Errorf("cannot create cache file %s.dat: %v", v.fileName, err)
  40. } else {
  41. v.DataBackend = backend.NewDiskFile(dataFile)
  42. v.lastModTime = modTime
  43. v.fileSize = fileSize
  44. }
  45. } else {
  46. if v.DataBackend, err = backend.CreateVolumeFile(v.fileName+".dat", preallocate, 0); err != nil {
  47. return nil, fmt.Errorf("cannot create cache file %s.dat: %v", v.fileName, err)
  48. }
  49. v.lastModTime = time.Now()
  50. }
  51. var indexFile *os.File
  52. if indexFile, err = os.OpenFile(v.fileName+".idx", os.O_RDWR|os.O_CREATE, 0644); err != nil {
  53. return nil, fmt.Errorf("cannot write cache index %s.idx: %v", v.fileName, err)
  54. }
  55. log.Debug("loading leveldb", v.fileName+".ldb")
  56. opts := &opt.Options{
  57. BlockCacheCapacity: 2 * 1024 * 1024, // default value is 8MiB
  58. WriteBuffer: 1 * 1024 * 1024, // default value is 4MiB
  59. CompactionTableSizeMultiplier: 10, // default value is 1
  60. }
  61. if v.nm, err = storage.NewLevelDbNeedleMap(v.fileName+".ldb", indexFile, opts); err != nil {
  62. return nil, fmt.Errorf("loading leveldb %s error: %v", v.fileName+".ldb", err)
  63. }
  64. return v, nil
  65. }
  66. func (v *ChunkCacheVolume) Shutdown() {
  67. if v.DataBackend != nil {
  68. v.DataBackend.Close()
  69. v.DataBackend = nil
  70. }
  71. if v.nm != nil {
  72. v.nm.Close()
  73. v.nm = nil
  74. }
  75. }
  76. func (v *ChunkCacheVolume) destroy() {
  77. v.Shutdown()
  78. os.Remove(v.fileName + ".dat")
  79. os.Remove(v.fileName + ".idx")
  80. os.RemoveAll(v.fileName + ".ldb")
  81. }
  82. func (v *ChunkCacheVolume) Reset() (*ChunkCacheVolume, error) {
  83. v.destroy()
  84. return LoadOrCreateChunkCacheVolume(v.fileName, v.sizeLimit)
  85. }
  86. func (v *ChunkCacheVolume) GetNeedle(key types.NeedleId) ([]byte, error) {
  87. nv, ok := v.nm.Get(key)
  88. if !ok {
  89. return nil, storage.ErrorNotFound
  90. }
  91. data := make([]byte, nv.Size)
  92. if readSize, readErr := v.DataBackend.ReadAt(data, nv.Offset.ToAcutalOffset()); readErr != nil {
  93. return nil, fmt.Errorf("read %s.dat [%d,%d): %v",
  94. v.fileName, nv.Offset.ToAcutalOffset(), nv.Offset.ToAcutalOffset()+int64(nv.Size), readErr)
  95. } else {
  96. if readSize != int(nv.Size) {
  97. return nil, fmt.Errorf("read %d, expected %d", readSize, nv.Size)
  98. }
  99. }
  100. return data, nil
  101. }
  102. func (v *ChunkCacheVolume) WriteNeedle(key types.NeedleId, data []byte) error {
  103. offset := v.fileSize
  104. written, err := v.DataBackend.WriteAt(data, offset)
  105. if err != nil {
  106. return err
  107. } else if written != len(data) {
  108. return fmt.Errorf("partial written %d, expected %d", written, len(data))
  109. }
  110. v.fileSize += int64(written)
  111. extraSize := written % types.NeedlePaddingSize
  112. if extraSize != 0 {
  113. v.DataBackend.WriteAt(v.smallBuffer[:types.NeedlePaddingSize-extraSize], offset+int64(written))
  114. v.fileSize += int64(types.NeedlePaddingSize - extraSize)
  115. }
  116. if err := v.nm.Put(key, types.ToOffset(offset), types.Size(len(data))); err != nil {
  117. return err
  118. }
  119. return nil
  120. }