volume.go 8.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307
  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 NeedleMapKind
  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 NeedleMapKind, 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. func (v *Volume) DiskType() types.DiskType {
  148. return v.location.DiskType
  149. }
  150. // Close cleanly shuts down this volume
  151. func (v *Volume) Close() {
  152. v.dataFileAccessLock.Lock()
  153. defer v.dataFileAccessLock.Unlock()
  154. if v.nm != nil {
  155. if err := v.nm.Sync(); err != nil {
  156. glog.Warningf("Volume Close fail to sync volume idx %d", v.Id)
  157. }
  158. v.nm.Close()
  159. v.nm = nil
  160. }
  161. if v.DataBackend != nil {
  162. if err := v.DataBackend.Sync(); err != nil {
  163. glog.Warningf("Volume Close fail to sync volume %d", v.Id)
  164. }
  165. _ = v.DataBackend.Close()
  166. v.DataBackend = nil
  167. stats.VolumeServerVolumeCounter.WithLabelValues(v.Collection, "volume").Dec()
  168. }
  169. }
  170. func (v *Volume) NeedToReplicate() bool {
  171. return v.ReplicaPlacement.GetCopyCount() > 1
  172. }
  173. // volume is expired if modified time + volume ttl < now
  174. // except when volume is empty
  175. // or when the volume does not have a ttl
  176. // or when volumeSizeLimit is 0 when server just starts
  177. func (v *Volume) expired(contentSize uint64, volumeSizeLimit uint64) bool {
  178. if volumeSizeLimit == 0 {
  179. // skip if we don't know size limit
  180. return false
  181. }
  182. if contentSize <= super_block.SuperBlockSize {
  183. return false
  184. }
  185. if v.Ttl == nil || v.Ttl.Minutes() == 0 {
  186. return false
  187. }
  188. glog.V(2).Infof("volume %d now:%v lastModified:%v", v.Id, time.Now().Unix(), v.lastModifiedTsSeconds)
  189. livedMinutes := (time.Now().Unix() - int64(v.lastModifiedTsSeconds)) / 60
  190. glog.V(2).Infof("volume %d ttl:%v lived:%v", v.Id, v.Ttl, livedMinutes)
  191. if int64(v.Ttl.Minutes()) < livedMinutes {
  192. return true
  193. }
  194. return false
  195. }
  196. // wait either maxDelayMinutes or 10% of ttl minutes
  197. func (v *Volume) expiredLongEnough(maxDelayMinutes uint32) bool {
  198. if v.Ttl == nil || v.Ttl.Minutes() == 0 {
  199. return false
  200. }
  201. removalDelay := v.Ttl.Minutes() / 10
  202. if removalDelay > maxDelayMinutes {
  203. removalDelay = maxDelayMinutes
  204. }
  205. if uint64(v.Ttl.Minutes()+removalDelay)*60+v.lastModifiedTsSeconds < uint64(time.Now().Unix()) {
  206. return true
  207. }
  208. return false
  209. }
  210. func (v *Volume) collectStatus() (maxFileKey types.NeedleId, datFileSize int64, modTime time.Time, fileCount, deletedCount, deletedSize uint64, ok bool) {
  211. v.dataFileAccessLock.RLock()
  212. defer v.dataFileAccessLock.RUnlock()
  213. glog.V(3).Infof("collectStatus volume %d", v.Id)
  214. if v.nm == nil {
  215. return
  216. }
  217. ok = true
  218. maxFileKey = v.nm.MaxFileKey()
  219. datFileSize, modTime, _ = v.DataBackend.GetStat()
  220. fileCount = uint64(v.nm.FileCount())
  221. deletedCount = uint64(v.nm.DeletedCount())
  222. deletedSize = v.nm.DeletedSize()
  223. fileCount = uint64(v.nm.FileCount())
  224. return
  225. }
  226. func (v *Volume) ToVolumeInformationMessage() (types.NeedleId, *master_pb.VolumeInformationMessage) {
  227. maxFileKey, volumeSize, modTime, fileCount, deletedCount, deletedSize, ok := v.collectStatus()
  228. if !ok {
  229. return 0, nil
  230. }
  231. volumeInfo := &master_pb.VolumeInformationMessage{
  232. Id: uint32(v.Id),
  233. Size: uint64(volumeSize),
  234. Collection: v.Collection,
  235. FileCount: fileCount,
  236. DeleteCount: deletedCount,
  237. DeletedByteCount: deletedSize,
  238. ReadOnly: v.IsReadOnly(),
  239. ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
  240. Version: uint32(v.Version()),
  241. Ttl: v.Ttl.ToUint32(),
  242. CompactRevision: uint32(v.SuperBlock.CompactionRevision),
  243. ModifiedAtSecond: modTime.Unix(),
  244. DiskType: string(v.location.DiskType),
  245. }
  246. volumeInfo.RemoteStorageName, volumeInfo.RemoteStorageKey = v.RemoteStorageNameKey()
  247. return maxFileKey, volumeInfo
  248. }
  249. func (v *Volume) RemoteStorageNameKey() (storageName, storageKey string) {
  250. if v.volumeInfo == nil {
  251. return
  252. }
  253. if len(v.volumeInfo.GetFiles()) == 0 {
  254. return
  255. }
  256. return v.volumeInfo.GetFiles()[0].BackendName(), v.volumeInfo.GetFiles()[0].GetKey()
  257. }
  258. func (v *Volume) IsReadOnly() bool {
  259. v.noWriteLock.RLock()
  260. defer v.noWriteLock.RUnlock()
  261. return v.noWriteOrDelete || v.noWriteCanDelete || v.location.isDiskSpaceLow
  262. }