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.

232 lines
6.4 KiB

6 years ago
6 years ago
6 years ago
5 years ago
6 years ago
10 years ago
6 years ago
12 years ago
  1. package storage
  2. import (
  3. "fmt"
  4. "path"
  5. "strconv"
  6. "sync"
  7. "time"
  8. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  9. "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
  10. "github.com/chrislusf/seaweedfs/weed/stats"
  11. "github.com/chrislusf/seaweedfs/weed/storage/backend"
  12. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  13. "github.com/chrislusf/seaweedfs/weed/storage/super_block"
  14. "github.com/chrislusf/seaweedfs/weed/storage/types"
  15. "github.com/chrislusf/seaweedfs/weed/glog"
  16. )
  17. type Volume struct {
  18. Id needle.VolumeId
  19. dir string
  20. Collection string
  21. DataBackend backend.BackendStorageFile
  22. nm NeedleMapper
  23. needleMapKind NeedleMapType
  24. noWriteOrDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
  25. noWriteCanDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
  26. MemoryMapMaxSizeMb uint32
  27. super_block.SuperBlock
  28. dataFileAccessLock sync.RWMutex
  29. lastModifiedTsSeconds uint64 //unix time in seconds
  30. lastAppendAtNs uint64 //unix time in nanoseconds
  31. lastCompactIndexOffset uint64
  32. lastCompactRevision uint16
  33. isCompacting bool
  34. volumeTierInfo *volume_server_pb.VolumeTierInfo
  35. }
  36. func NewVolume(dirname string, collection string, id needle.VolumeId, needleMapKind NeedleMapType, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, memoryMapMaxSizeMb uint32) (v *Volume, e error) {
  37. // if replicaPlacement is nil, the superblock will be loaded from disk
  38. v = &Volume{dir: dirname, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb}
  39. v.SuperBlock = super_block.SuperBlock{ReplicaPlacement: replicaPlacement, Ttl: ttl}
  40. v.needleMapKind = needleMapKind
  41. e = v.load(true, true, needleMapKind, preallocate)
  42. return
  43. }
  44. func (v *Volume) String() string {
  45. return fmt.Sprintf("Id:%v, dir:%s, Collection:%s, dataFile:%v, nm:%v, noWrite:%v canDelete:%v", v.Id, v.dir, v.Collection, v.DataBackend, v.nm, v.noWriteOrDelete || v.noWriteCanDelete, v.noWriteCanDelete)
  46. }
  47. func VolumeFileName(dir string, collection string, id int) (fileName string) {
  48. idString := strconv.Itoa(id)
  49. if collection == "" {
  50. fileName = path.Join(dir, idString)
  51. } else {
  52. fileName = path.Join(dir, collection+"_"+idString)
  53. }
  54. return
  55. }
  56. func (v *Volume) FileName() (fileName string) {
  57. return VolumeFileName(v.dir, v.Collection, int(v.Id))
  58. }
  59. func (v *Volume) Version() needle.Version {
  60. return v.SuperBlock.Version
  61. }
  62. func (v *Volume) FileStat() (datSize uint64, idxSize uint64, modTime time.Time) {
  63. v.dataFileAccessLock.RLock()
  64. defer v.dataFileAccessLock.RUnlock()
  65. if v.DataBackend == nil {
  66. return
  67. }
  68. datFileSize, modTime, e := v.DataBackend.GetStat()
  69. if e == nil {
  70. return uint64(datFileSize), v.nm.IndexFileSize(), modTime
  71. }
  72. glog.V(0).Infof("Failed to read file size %s %v", v.DataBackend.Name(), e)
  73. return // -1 causes integer overflow and the volume to become unwritable.
  74. }
  75. func (v *Volume) ContentSize() uint64 {
  76. v.dataFileAccessLock.RLock()
  77. defer v.dataFileAccessLock.RUnlock()
  78. if v.nm == nil {
  79. return 0
  80. }
  81. return v.nm.ContentSize()
  82. }
  83. func (v *Volume) DeletedSize() uint64 {
  84. v.dataFileAccessLock.RLock()
  85. defer v.dataFileAccessLock.RUnlock()
  86. if v.nm == nil {
  87. return 0
  88. }
  89. return v.nm.DeletedSize()
  90. }
  91. func (v *Volume) FileCount() uint64 {
  92. v.dataFileAccessLock.RLock()
  93. defer v.dataFileAccessLock.RUnlock()
  94. if v.nm == nil {
  95. return 0
  96. }
  97. return uint64(v.nm.FileCount())
  98. }
  99. func (v *Volume) DeletedCount() uint64 {
  100. v.dataFileAccessLock.RLock()
  101. defer v.dataFileAccessLock.RUnlock()
  102. if v.nm == nil {
  103. return 0
  104. }
  105. return uint64(v.nm.DeletedCount())
  106. }
  107. func (v *Volume) MaxFileKey() types.NeedleId {
  108. v.dataFileAccessLock.RLock()
  109. defer v.dataFileAccessLock.RUnlock()
  110. if v.nm == nil {
  111. return 0
  112. }
  113. return v.nm.MaxFileKey()
  114. }
  115. func (v *Volume) IndexFileSize() uint64 {
  116. v.dataFileAccessLock.RLock()
  117. defer v.dataFileAccessLock.RUnlock()
  118. if v.nm == nil {
  119. return 0
  120. }
  121. return v.nm.IndexFileSize()
  122. }
  123. // Close cleanly shuts down this volume
  124. func (v *Volume) Close() {
  125. v.dataFileAccessLock.Lock()
  126. defer v.dataFileAccessLock.Unlock()
  127. if v.nm != nil {
  128. v.nm.Close()
  129. v.nm = nil
  130. }
  131. if v.DataBackend != nil {
  132. _ = v.DataBackend.Close()
  133. v.DataBackend = nil
  134. stats.VolumeServerVolumeCounter.WithLabelValues(v.Collection, "volume").Dec()
  135. }
  136. }
  137. func (v *Volume) NeedToReplicate() bool {
  138. return v.ReplicaPlacement.GetCopyCount() > 1
  139. }
  140. // volume is expired if modified time + volume ttl < now
  141. // except when volume is empty
  142. // or when the volume does not have a ttl
  143. // or when volumeSizeLimit is 0 when server just starts
  144. func (v *Volume) expired(volumeSizeLimit uint64) bool {
  145. if volumeSizeLimit == 0 {
  146. //skip if we don't know size limit
  147. return false
  148. }
  149. if v.ContentSize() == 0 {
  150. return false
  151. }
  152. if v.Ttl == nil || v.Ttl.Minutes() == 0 {
  153. return false
  154. }
  155. glog.V(1).Infof("now:%v lastModified:%v", time.Now().Unix(), v.lastModifiedTsSeconds)
  156. livedMinutes := (time.Now().Unix() - int64(v.lastModifiedTsSeconds)) / 60
  157. glog.V(1).Infof("ttl:%v lived:%v", v.Ttl, livedMinutes)
  158. if int64(v.Ttl.Minutes()) < livedMinutes {
  159. return true
  160. }
  161. return false
  162. }
  163. // wait either maxDelayMinutes or 10% of ttl minutes
  164. func (v *Volume) expiredLongEnough(maxDelayMinutes uint32) bool {
  165. if v.Ttl == nil || v.Ttl.Minutes() == 0 {
  166. return false
  167. }
  168. removalDelay := v.Ttl.Minutes() / 10
  169. if removalDelay > maxDelayMinutes {
  170. removalDelay = maxDelayMinutes
  171. }
  172. if uint64(v.Ttl.Minutes()+removalDelay)*60+v.lastModifiedTsSeconds < uint64(time.Now().Unix()) {
  173. return true
  174. }
  175. return false
  176. }
  177. func (v *Volume) ToVolumeInformationMessage() *master_pb.VolumeInformationMessage {
  178. size, _, modTime := v.FileStat()
  179. volumInfo := &master_pb.VolumeInformationMessage{
  180. Id: uint32(v.Id),
  181. Size: size,
  182. Collection: v.Collection,
  183. FileCount: uint64(v.FileCount()),
  184. DeleteCount: uint64(v.DeletedCount()),
  185. DeletedByteCount: v.DeletedSize(),
  186. ReadOnly: v.noWriteOrDelete,
  187. ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
  188. Version: uint32(v.Version()),
  189. Ttl: v.Ttl.ToUint32(),
  190. CompactRevision: uint32(v.SuperBlock.CompactionRevision),
  191. ModifiedAtSecond: modTime.Unix(),
  192. }
  193. volumInfo.RemoteStorageName, volumInfo.RemoteStorageKey = v.RemoteStorageNameKey()
  194. return volumInfo
  195. }
  196. func (v *Volume) RemoteStorageNameKey() (storageName, storageKey string) {
  197. if len(v.volumeTierInfo.GetFiles()) == 0 {
  198. return
  199. }
  200. return v.volumeTierInfo.GetFiles()[0].BackendName(), v.volumeTierInfo.GetFiles()[0].GetKey()
  201. }