You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

131 lines
3.6 KiB

7 years ago
  1. package storage
  2. import (
  3. "fmt"
  4. "os"
  5. "sync/atomic"
  6. . "github.com/chrislusf/seaweedfs/weed/storage/types"
  7. "github.com/willf/bloom"
  8. )
  9. type mapMetric struct {
  10. DeletionCounter uint32 `json:"DeletionCounter"`
  11. FileCounter uint32 `json:"FileCounter"`
  12. DeletionByteCounter uint64 `json:"DeletionByteCounter"`
  13. FileByteCounter uint64 `json:"FileByteCounter"`
  14. MaximumFileKey uint64 `json:"MaxFileKey"`
  15. }
  16. func (mm *mapMetric) logDelete(deletedByteCount uint32) {
  17. mm.LogDeletionCounter(deletedByteCount)
  18. }
  19. func (mm *mapMetric) logPut(key NeedleId, oldSize uint32, newSize uint32) {
  20. mm.MaybeSetMaxFileKey(key)
  21. mm.LogFileCounter(newSize)
  22. if oldSize > 0 && oldSize != TombstoneFileSize {
  23. mm.LogDeletionCounter(oldSize)
  24. }
  25. }
  26. func (mm *mapMetric) LogFileCounter(newSize uint32) {
  27. atomic.AddUint32(&mm.FileCounter, 1)
  28. atomic.AddUint64(&mm.FileByteCounter, uint64(newSize))
  29. }
  30. func (mm *mapMetric) LogDeletionCounter(oldSize uint32) {
  31. if oldSize > 0 {
  32. atomic.AddUint32(&mm.DeletionCounter, 1)
  33. atomic.AddUint64(&mm.DeletionByteCounter, uint64(oldSize))
  34. }
  35. }
  36. func (mm *mapMetric) ContentSize() uint64 {
  37. return atomic.LoadUint64(&mm.FileByteCounter)
  38. }
  39. func (mm *mapMetric) DeletedSize() uint64 {
  40. return atomic.LoadUint64(&mm.DeletionByteCounter)
  41. }
  42. func (mm *mapMetric) FileCount() int {
  43. return int(atomic.LoadUint32(&mm.FileCounter))
  44. }
  45. func (mm *mapMetric) DeletedCount() int {
  46. return int(atomic.LoadUint32(&mm.DeletionCounter))
  47. }
  48. func (mm *mapMetric) MaxFileKey() NeedleId {
  49. t := uint64(mm.MaximumFileKey)
  50. return NeedleId(t)
  51. }
  52. func (mm *mapMetric) MaybeSetMaxFileKey(key NeedleId) {
  53. if key > mm.MaxFileKey() {
  54. atomic.StoreUint64(&mm.MaximumFileKey, uint64(key))
  55. }
  56. }
  57. func newNeedleMapMetricFromIndexFile(r *os.File) (mm *mapMetric, err error) {
  58. mm = &mapMetric{}
  59. var bf *bloom.BloomFilter
  60. buf := make([]byte, NeedleIdSize)
  61. err = reverseWalkIndexFile(r, func(entryCount int64) {
  62. bf = bloom.NewWithEstimates(uint(entryCount), 0.001)
  63. }, func(key NeedleId, offset Offset, size uint32) error {
  64. mm.MaybeSetMaxFileKey(key)
  65. NeedleIdToBytes(buf, key)
  66. if size != TombstoneFileSize {
  67. mm.FileByteCounter += uint64(size)
  68. }
  69. if !bf.Test(buf) {
  70. mm.FileCounter++
  71. bf.Add(buf)
  72. } else {
  73. // deleted file
  74. mm.DeletionCounter++
  75. if size != TombstoneFileSize {
  76. // previously already deleted file
  77. mm.DeletionByteCounter += uint64(size)
  78. }
  79. }
  80. return nil
  81. })
  82. return
  83. }
  84. func reverseWalkIndexFile(r *os.File, initFn func(entryCount int64), fn func(key NeedleId, offset Offset, size uint32) error) error {
  85. fi, err := r.Stat()
  86. if err != nil {
  87. return fmt.Errorf("file %s stat error: %v", r.Name(), err)
  88. }
  89. fileSize := fi.Size()
  90. if fileSize%NeedleMapEntrySize != 0 {
  91. return fmt.Errorf("unexpected file %s size: %d", r.Name(), fileSize)
  92. }
  93. entryCount := fileSize / NeedleMapEntrySize
  94. initFn(entryCount)
  95. batchSize := int64(1024 * 4)
  96. bytes := make([]byte, NeedleMapEntrySize*batchSize)
  97. nextBatchSize := entryCount % batchSize
  98. if nextBatchSize == 0 {
  99. nextBatchSize = batchSize
  100. }
  101. remainingCount := entryCount - nextBatchSize
  102. for remainingCount >= 0 {
  103. _, e := r.ReadAt(bytes[:NeedleMapEntrySize*nextBatchSize], NeedleMapEntrySize*remainingCount)
  104. // glog.V(0).Infoln("file", r.Name(), "readerOffset", NeedleMapEntrySize*remainingCount, "count", count, "e", e)
  105. if e != nil {
  106. return e
  107. }
  108. for i := int(nextBatchSize) - 1; i >= 0; i-- {
  109. key, offset, size := IdxFileEntry(bytes[i*NeedleMapEntrySize : i*NeedleMapEntrySize+NeedleMapEntrySize])
  110. if e = fn(key, offset, size); e != nil {
  111. return e
  112. }
  113. }
  114. nextBatchSize = batchSize
  115. remainingCount -= nextBatchSize
  116. }
  117. return nil
  118. }