volume.go 8.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286
  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. dirIdx string
  21. Collection string
  22. DataBackend backend.BackendStorageFile
  23. nm NeedleMapper
  24. needleMapKind NeedleMapType
  25. noWriteOrDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
  26. noWriteCanDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
  27. noWriteLock sync.RWMutex
  28. hasRemoteFile bool // if the volume has a remote file
  29. MemoryMapMaxSizeMb uint32
  30. super_block.SuperBlock
  31. dataFileAccessLock sync.RWMutex
  32. asyncRequestsChan chan *needle.AsyncRequest
  33. lastModifiedTsSeconds uint64 // unix time in seconds
  34. lastAppendAtNs uint64 // unix time in nanoseconds
  35. lastCompactIndexOffset uint64
  36. lastCompactRevision uint16
  37. isCompacting bool
  38. volumeInfo *volume_server_pb.VolumeInfo
  39. location *DiskLocation
  40. lastIoError error
  41. }
  42. func NewVolume(dirname string, dirIdx string, collection string, id needle.VolumeId, needleMapKind NeedleMapType, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, memoryMapMaxSizeMb uint32) (v *Volume, e error) {
  43. // if replicaPlacement is nil, the superblock will be loaded from disk
  44. v = &Volume{dir: dirname, dirIdx: dirIdx, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb,
  45. asyncRequestsChan: make(chan *needle.AsyncRequest, 128)}
  46. v.SuperBlock = super_block.SuperBlock{ReplicaPlacement: replicaPlacement, Ttl: ttl}
  47. v.needleMapKind = needleMapKind
  48. e = v.load(true, true, needleMapKind, preallocate)
  49. v.startWorker()
  50. return
  51. }
  52. func (v *Volume) String() string {
  53. v.noWriteLock.RLock()
  54. defer v.noWriteLock.RUnlock()
  55. return fmt.Sprintf("Id:%v dir:%s dirIdx:%s Collection:%s dataFile:%v nm:%v noWrite:%v canDelete:%v", v.Id, v.dir, v.dirIdx, v.Collection, v.DataBackend, v.nm, v.noWriteOrDelete || v.noWriteCanDelete, v.noWriteCanDelete)
  56. }
  57. func VolumeFileName(dir string, collection string, id int) (fileName string) {
  58. idString := strconv.Itoa(id)
  59. if collection == "" {
  60. fileName = path.Join(dir, idString)
  61. } else {
  62. fileName = path.Join(dir, collection+"_"+idString)
  63. }
  64. return
  65. }
  66. func (v *Volume) DataFileName() (fileName string) {
  67. return VolumeFileName(v.dir, v.Collection, int(v.Id))
  68. }
  69. func (v *Volume) IndexFileName() (fileName string) {
  70. return VolumeFileName(v.dirIdx, v.Collection, int(v.Id))
  71. }
  72. func (v *Volume) FileName(ext string) (fileName string) {
  73. switch ext {
  74. case ".idx", ".cpx", ".ldb":
  75. return VolumeFileName(v.dirIdx, v.Collection, int(v.Id)) + ext
  76. }
  77. // .dat, .cpd, .vif
  78. return VolumeFileName(v.dir, v.Collection, int(v.Id)) + ext
  79. }
  80. func (v *Volume) Version() needle.Version {
  81. if v.volumeInfo.Version != 0 {
  82. v.SuperBlock.Version = needle.Version(v.volumeInfo.Version)
  83. }
  84. return v.SuperBlock.Version
  85. }
  86. func (v *Volume) FileStat() (datSize uint64, idxSize uint64, modTime time.Time) {
  87. v.dataFileAccessLock.RLock()
  88. defer v.dataFileAccessLock.RUnlock()
  89. if v.DataBackend == nil {
  90. return
  91. }
  92. datFileSize, modTime, e := v.DataBackend.GetStat()
  93. if e == nil {
  94. return uint64(datFileSize), v.nm.IndexFileSize(), modTime
  95. }
  96. glog.V(0).Infof("Failed to read file size %s %v", v.DataBackend.Name(), e)
  97. return // -1 causes integer overflow and the volume to become unwritable.
  98. }
  99. func (v *Volume) ContentSize() uint64 {
  100. v.dataFileAccessLock.RLock()
  101. defer v.dataFileAccessLock.RUnlock()
  102. if v.nm == nil {
  103. return 0
  104. }
  105. return v.nm.ContentSize()
  106. }
  107. func (v *Volume) DeletedSize() uint64 {
  108. v.dataFileAccessLock.RLock()
  109. defer v.dataFileAccessLock.RUnlock()
  110. if v.nm == nil {
  111. return 0
  112. }
  113. return v.nm.DeletedSize()
  114. }
  115. func (v *Volume) FileCount() uint64 {
  116. v.dataFileAccessLock.RLock()
  117. defer v.dataFileAccessLock.RUnlock()
  118. if v.nm == nil {
  119. return 0
  120. }
  121. return uint64(v.nm.FileCount())
  122. }
  123. func (v *Volume) DeletedCount() uint64 {
  124. v.dataFileAccessLock.RLock()
  125. defer v.dataFileAccessLock.RUnlock()
  126. if v.nm == nil {
  127. return 0
  128. }
  129. return uint64(v.nm.DeletedCount())
  130. }
  131. func (v *Volume) MaxFileKey() types.NeedleId {
  132. v.dataFileAccessLock.RLock()
  133. defer v.dataFileAccessLock.RUnlock()
  134. if v.nm == nil {
  135. return 0
  136. }
  137. return v.nm.MaxFileKey()
  138. }
  139. func (v *Volume) IndexFileSize() uint64 {
  140. v.dataFileAccessLock.RLock()
  141. defer v.dataFileAccessLock.RUnlock()
  142. if v.nm == nil {
  143. return 0
  144. }
  145. return v.nm.IndexFileSize()
  146. }
  147. // Close cleanly shuts down this volume
  148. func (v *Volume) Close() {
  149. v.dataFileAccessLock.Lock()
  150. defer v.dataFileAccessLock.Unlock()
  151. if v.nm != nil {
  152. v.nm.Close()
  153. v.nm = nil
  154. }
  155. if v.DataBackend != nil {
  156. _ = v.DataBackend.Close()
  157. v.DataBackend = nil
  158. stats.VolumeServerVolumeCounter.WithLabelValues(v.Collection, "volume").Dec()
  159. }
  160. }
  161. func (v *Volume) NeedToReplicate() bool {
  162. return v.ReplicaPlacement.GetCopyCount() > 1
  163. }
  164. // volume is expired if modified time + volume ttl < now
  165. // except when volume is empty
  166. // or when the volume does not have a ttl
  167. // or when volumeSizeLimit is 0 when server just starts
  168. func (v *Volume) expired(contentSize uint64, volumeSizeLimit uint64) bool {
  169. if volumeSizeLimit == 0 {
  170. // skip if we don't know size limit
  171. return false
  172. }
  173. if contentSize <= super_block.SuperBlockSize {
  174. return false
  175. }
  176. if v.Ttl == nil || v.Ttl.Minutes() == 0 {
  177. return false
  178. }
  179. glog.V(2).Infof("volume %d now:%v lastModified:%v", v.Id, time.Now().Unix(), v.lastModifiedTsSeconds)
  180. livedMinutes := (time.Now().Unix() - int64(v.lastModifiedTsSeconds)) / 60
  181. glog.V(2).Infof("volume %d ttl:%v lived:%v", v.Id, v.Ttl, livedMinutes)
  182. if int64(v.Ttl.Minutes()) < livedMinutes {
  183. return true
  184. }
  185. return false
  186. }
  187. // wait either maxDelayMinutes or 10% of ttl minutes
  188. func (v *Volume) expiredLongEnough(maxDelayMinutes uint32) bool {
  189. if v.Ttl == nil || v.Ttl.Minutes() == 0 {
  190. return false
  191. }
  192. removalDelay := v.Ttl.Minutes() / 10
  193. if removalDelay > maxDelayMinutes {
  194. removalDelay = maxDelayMinutes
  195. }
  196. if uint64(v.Ttl.Minutes()+removalDelay)*60+v.lastModifiedTsSeconds < uint64(time.Now().Unix()) {
  197. return true
  198. }
  199. return false
  200. }
  201. func (v *Volume) CollectStatus() (maxFileKey types.NeedleId, datFileSize int64, modTime time.Time, fileCount, deletedCount, deletedSize uint64) {
  202. v.dataFileAccessLock.RLock()
  203. defer v.dataFileAccessLock.RUnlock()
  204. glog.V(3).Infof("CollectStatus volume %d", v.Id)
  205. maxFileKey = v.nm.MaxFileKey()
  206. datFileSize, modTime, _ = v.DataBackend.GetStat()
  207. fileCount = uint64(v.nm.FileCount())
  208. deletedCount = uint64(v.nm.DeletedCount())
  209. deletedSize = v.nm.DeletedSize()
  210. fileCount = uint64(v.nm.FileCount())
  211. return
  212. }
  213. func (v *Volume) ToVolumeInformationMessage() (types.NeedleId, *master_pb.VolumeInformationMessage) {
  214. maxFileKey, volumeSize, modTime, fileCount, deletedCount, deletedSize := v.CollectStatus()
  215. volumeInfo := &master_pb.VolumeInformationMessage{
  216. Id: uint32(v.Id),
  217. Size: uint64(volumeSize),
  218. Collection: v.Collection,
  219. FileCount: fileCount,
  220. DeleteCount: deletedCount,
  221. DeletedByteCount: deletedSize,
  222. ReadOnly: v.IsReadOnly(),
  223. ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
  224. Version: uint32(v.Version()),
  225. Ttl: v.Ttl.ToUint32(),
  226. CompactRevision: uint32(v.SuperBlock.CompactionRevision),
  227. ModifiedAtSecond: modTime.Unix(),
  228. }
  229. volumeInfo.RemoteStorageName, volumeInfo.RemoteStorageKey = v.RemoteStorageNameKey()
  230. return maxFileKey, volumeInfo
  231. }
  232. func (v *Volume) RemoteStorageNameKey() (storageName, storageKey string) {
  233. if v.volumeInfo == nil {
  234. return
  235. }
  236. if len(v.volumeInfo.GetFiles()) == 0 {
  237. return
  238. }
  239. return v.volumeInfo.GetFiles()[0].BackendName(), v.volumeInfo.GetFiles()[0].GetKey()
  240. }
  241. func (v *Volume) IsReadOnly() bool {
  242. v.noWriteLock.RLock()
  243. defer v.noWriteLock.RUnlock()
  244. return v.noWriteOrDelete || v.noWriteCanDelete || v.location.isDiskSpaceLow
  245. }