seaweedfs/weed/storage/volume.go

250 lines
6.8 KiB
Go
Raw Normal View History

package storage
import (
2012-11-20 09:45:36 +00:00
"fmt"
"path"
"strconv"
"sync"
"time"
2019-04-19 04:43:36 +00:00
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
"github.com/chrislusf/seaweedfs/weed/stats"
"github.com/chrislusf/seaweedfs/weed/storage/backend"
"github.com/chrislusf/seaweedfs/weed/storage/needle"
2019-12-23 20:48:20 +00:00
"github.com/chrislusf/seaweedfs/weed/storage/super_block"
"github.com/chrislusf/seaweedfs/weed/storage/types"
2019-04-19 04:43:36 +00:00
"github.com/chrislusf/seaweedfs/weed/glog"
)
type Volume struct {
Id needle.VolumeId
dir string
Collection string
DataBackend backend.BackendStorageFile
nm NeedleMapper
needleMapKind NeedleMapType
noWriteOrDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
noWriteCanDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
hasRemoteFile bool // if the volume has a remote file
2019-10-22 05:57:01 +00:00
MemoryMapMaxSizeMb uint32
2019-12-23 20:48:20 +00:00
super_block.SuperBlock
2012-11-20 09:45:36 +00:00
dataFileAccessLock sync.RWMutex
asyncRequestsChan chan *needle.AsyncRequest
2020-07-03 23:34:31 +00:00
lastModifiedTsSeconds uint64 // unix time in seconds
lastAppendAtNs uint64 // unix time in nanoseconds
2016-10-07 08:22:24 +00:00
lastCompactIndexOffset uint64
lastCompactRevision uint16
isCompacting bool
2019-12-28 19:21:49 +00:00
volumeInfo *volume_server_pb.VolumeInfo
2020-07-03 23:34:31 +00:00
location *DiskLocation
}
2019-12-23 20:48:20 +00:00
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) {
2018-06-23 23:48:19 +00:00
// if replicaPlacement is nil, the superblock will be loaded from disk
v = &Volume{dir: dirname, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb,
asyncRequestsChan: make(chan *needle.AsyncRequest, 128)}
2019-12-23 20:48:20 +00:00
v.SuperBlock = super_block.SuperBlock{ReplicaPlacement: replicaPlacement, Ttl: ttl}
v.needleMapKind = needleMapKind
e = v.load(true, true, needleMapKind, preallocate)
v.startWorker()
return
}
func (v *Volume) String() string {
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)
}
2019-06-03 09:26:31 +00:00
func VolumeFileName(dir string, collection string, id int) (fileName string) {
idString := strconv.Itoa(id)
if collection == "" {
fileName = path.Join(dir, idString)
} else {
fileName = path.Join(dir, collection+"_"+idString)
}
return
2014-01-22 04:51:46 +00:00
}
func (v *Volume) FileName() (fileName string) {
2019-06-03 09:26:31 +00:00
return VolumeFileName(v.dir, v.Collection, int(v.Id))
}
2019-04-19 04:43:36 +00:00
func (v *Volume) Version() needle.Version {
if v.volumeInfo.Version != 0 {
v.SuperBlock.Version = needle.Version(v.volumeInfo.Version)
}
2019-12-23 20:48:20 +00:00
return v.SuperBlock.Version
}
2019-04-19 07:39:34 +00:00
func (v *Volume) FileStat() (datSize uint64, idxSize uint64, modTime time.Time) {
v.dataFileAccessLock.RLock()
defer v.dataFileAccessLock.RUnlock()
if v.DataBackend == nil {
2019-04-19 07:39:34 +00:00
return
}
datFileSize, modTime, e := v.DataBackend.GetStat()
if e == nil {
return uint64(datFileSize), v.nm.IndexFileSize(), modTime
}
2019-12-09 03:44:16 +00:00
glog.V(0).Infof("Failed to read file size %s %v", v.DataBackend.Name(), e)
2019-04-19 07:39:34 +00:00
return // -1 causes integer overflow and the volume to become unwritable.
}
func (v *Volume) ContentSize() uint64 {
v.dataFileAccessLock.RLock()
defer v.dataFileAccessLock.RUnlock()
if v.nm == nil {
return 0
}
return v.nm.ContentSize()
}
func (v *Volume) DeletedSize() uint64 {
v.dataFileAccessLock.RLock()
defer v.dataFileAccessLock.RUnlock()
if v.nm == nil {
return 0
}
return v.nm.DeletedSize()
}
2019-04-18 07:19:18 +00:00
func (v *Volume) FileCount() uint64 {
v.dataFileAccessLock.RLock()
defer v.dataFileAccessLock.RUnlock()
if v.nm == nil {
return 0
}
2019-04-11 06:39:53 +00:00
return uint64(v.nm.FileCount())
}
func (v *Volume) DeletedCount() uint64 {
v.dataFileAccessLock.RLock()
defer v.dataFileAccessLock.RUnlock()
if v.nm == nil {
return 0
}
return uint64(v.nm.DeletedCount())
}
func (v *Volume) MaxFileKey() types.NeedleId {
v.dataFileAccessLock.RLock()
defer v.dataFileAccessLock.RUnlock()
if v.nm == nil {
return 0
}
return v.nm.MaxFileKey()
}
func (v *Volume) IndexFileSize() uint64 {
v.dataFileAccessLock.RLock()
defer v.dataFileAccessLock.RUnlock()
if v.nm == nil {
return 0
}
return v.nm.IndexFileSize()
}
// Close cleanly shuts down this volume
func (v *Volume) Close() {
2015-05-23 17:16:01 +00:00
v.dataFileAccessLock.Lock()
defer v.dataFileAccessLock.Unlock()
if v.nm != nil {
v.nm.Close()
v.nm = nil
}
if v.DataBackend != nil {
_ = v.DataBackend.Close()
v.DataBackend = nil
2019-06-18 04:02:50 +00:00
stats.VolumeServerVolumeCounter.WithLabelValues(v.Collection, "volume").Dec()
}
}
2012-11-20 08:54:37 +00:00
func (v *Volume) NeedToReplicate() bool {
return v.ReplicaPlacement.GetCopyCount() > 1
}
// volume is expired if modified time + volume ttl < now
// except when volume is empty
// or when the volume does not have a ttl
// or when volumeSizeLimit is 0 when server just starts
func (v *Volume) expired(volumeSizeLimit uint64) bool {
if volumeSizeLimit == 0 {
2020-07-03 23:34:31 +00:00
// skip if we don't know size limit
return false
}
if v.ContentSize() == 0 {
return false
}
if v.Ttl == nil || v.Ttl.Minutes() == 0 {
return false
}
glog.V(2).Infof("now:%v lastModified:%v", time.Now().Unix(), v.lastModifiedTsSeconds)
2019-04-19 07:39:34 +00:00
livedMinutes := (time.Now().Unix() - int64(v.lastModifiedTsSeconds)) / 60
glog.V(2).Infof("ttl:%v lived:%v", v.Ttl, livedMinutes)
if int64(v.Ttl.Minutes()) < livedMinutes {
return true
}
return false
}
// wait either maxDelayMinutes or 10% of ttl minutes
2019-01-17 01:17:19 +00:00
func (v *Volume) expiredLongEnough(maxDelayMinutes uint32) bool {
if v.Ttl == nil || v.Ttl.Minutes() == 0 {
return false
}
removalDelay := v.Ttl.Minutes() / 10
if removalDelay > maxDelayMinutes {
removalDelay = maxDelayMinutes
}
2019-04-19 07:39:34 +00:00
if uint64(v.Ttl.Minutes()+removalDelay)*60+v.lastModifiedTsSeconds < uint64(time.Now().Unix()) {
return true
}
return false
}
2019-03-18 03:27:08 +00:00
func (v *Volume) ToVolumeInformationMessage() *master_pb.VolumeInformationMessage {
size, _, modTime := v.FileStat()
volumInfo := &master_pb.VolumeInformationMessage{
2019-03-18 03:27:08 +00:00
Id: uint32(v.Id),
2019-04-19 07:39:34 +00:00
Size: size,
2019-03-18 03:27:08 +00:00
Collection: v.Collection,
2019-12-25 02:13:45 +00:00
FileCount: v.FileCount(),
DeleteCount: v.DeletedCount(),
DeletedByteCount: v.DeletedSize(),
ReadOnly: v.IsReadOnly(),
2019-03-18 03:27:08 +00:00
ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
Version: uint32(v.Version()),
Ttl: v.Ttl.ToUint32(),
CompactRevision: uint32(v.SuperBlock.CompactionRevision),
ModifiedAtSecond: modTime.Unix(),
2019-03-18 03:27:08 +00:00
}
volumInfo.RemoteStorageName, volumInfo.RemoteStorageKey = v.RemoteStorageNameKey()
return volumInfo
}
func (v *Volume) RemoteStorageNameKey() (storageName, storageKey string) {
2019-12-28 19:21:49 +00:00
if v.volumeInfo == nil {
2019-12-26 00:17:58 +00:00
return
}
2019-12-28 19:21:49 +00:00
if len(v.volumeInfo.GetFiles()) == 0 {
return
}
2019-12-28 19:21:49 +00:00
return v.volumeInfo.GetFiles()[0].BackendName(), v.volumeInfo.GetFiles()[0].GetKey()
2019-03-18 03:27:08 +00:00
}
func (v *Volume) IsReadOnly() bool {
2020-07-03 23:34:31 +00:00
return v.noWriteOrDelete || v.noWriteCanDelete || v.location.isDiskSpaceLow
}