on_disk_cache_layer.go 2.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091
  1. package chunk_cache
  2. import (
  3. "fmt"
  4. "path"
  5. "sort"
  6. "github.com/chrislusf/seaweedfs/weed/util/log"
  7. "github.com/chrislusf/seaweedfs/weed/storage"
  8. "github.com/chrislusf/seaweedfs/weed/storage/types"
  9. )
  10. type OnDiskCacheLayer struct {
  11. diskCaches []*ChunkCacheVolume
  12. }
  13. func NewOnDiskCacheLayer(dir, namePrefix string, diskSize int64, segmentCount int) *OnDiskCacheLayer {
  14. volumeCount, volumeSize := int(diskSize/(30000*1024*1024)), int64(30000*1024*1024)
  15. if volumeCount < segmentCount {
  16. volumeCount, volumeSize = segmentCount, diskSize/int64(segmentCount)
  17. }
  18. c := &OnDiskCacheLayer{}
  19. for i := 0; i < volumeCount; i++ {
  20. fileName := path.Join(dir, fmt.Sprintf("%s_%d", namePrefix, i))
  21. diskCache, err := LoadOrCreateChunkCacheVolume(fileName, volumeSize)
  22. if err != nil {
  23. log.Errorf("failed to add cache %s : %v", fileName, err)
  24. } else {
  25. c.diskCaches = append(c.diskCaches, diskCache)
  26. }
  27. }
  28. // keep newest cache to the front
  29. sort.Slice(c.diskCaches, func(i, j int) bool {
  30. return c.diskCaches[i].lastModTime.After(c.diskCaches[j].lastModTime)
  31. })
  32. return c
  33. }
  34. func (c *OnDiskCacheLayer) setChunk(needleId types.NeedleId, data []byte) {
  35. if c.diskCaches[0].fileSize+int64(len(data)) > c.diskCaches[0].sizeLimit {
  36. t, resetErr := c.diskCaches[len(c.diskCaches)-1].Reset()
  37. if resetErr != nil {
  38. log.Errorf("failed to reset cache file %s", c.diskCaches[len(c.diskCaches)-1].fileName)
  39. return
  40. }
  41. for i := len(c.diskCaches) - 1; i > 0; i-- {
  42. c.diskCaches[i] = c.diskCaches[i-1]
  43. }
  44. c.diskCaches[0] = t
  45. }
  46. if err := c.diskCaches[0].WriteNeedle(needleId, data); err != nil {
  47. log.Infof("cache write %v size %d: %v", needleId, len(data), err)
  48. }
  49. }
  50. func (c *OnDiskCacheLayer) getChunk(needleId types.NeedleId) (data []byte) {
  51. var err error
  52. for _, diskCache := range c.diskCaches {
  53. data, err = diskCache.GetNeedle(needleId)
  54. if err == storage.ErrorNotFound {
  55. continue
  56. }
  57. if err != nil {
  58. log.Errorf("failed to read cache file %s id %d", diskCache.fileName, needleId)
  59. continue
  60. }
  61. if len(data) != 0 {
  62. return
  63. }
  64. }
  65. return nil
  66. }
  67. func (c *OnDiskCacheLayer) shutdown() {
  68. for _, diskCache := range c.diskCaches {
  69. diskCache.Shutdown()
  70. }
  71. }