mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-07 12:37:41 +01:00
* volume: a delete cancels the writes of its needle still in flight A write queued behind the volume locks, for example behind another write's fsync on a busy disk, is not in the needle map yet. A delete of that needle in the meantime finds nothing, and the write lands afterwards, so the needle comes back on this copy only while the other replicas have it deleted. Each volume now tracks its writes from entry until they return. A delete cancels the ones for its needle, and a cancelled write that landed is tombstoned before it returns, unless the needle was deleted or rewritten since. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: a replicate delete that finds nothing cancels the pending write Rolling back a failed chunk upload sends DELETE ?type=replicate to every holder, including the slow one whose write timed out on the client. That holder has not indexed the write yet, so the delete answered 404 and the write landed live afterwards. Cancel the needle's pending writes, then look again in case one finished in between. A plain delete that finds nothing still answers 404 without cancelling: it is not sent on to the replicas. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * seaweed-volume: a delete cancels the writes of its needle still in flight Mirrors the Go volume. Volume::track_pending_write registers a write until its guard drops, delete_needle cancels the needle's pending writes, and settle_pending_write tombstones a cancelled write that landed unless the needle was deleted or rewritten since. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * seaweed-volume: track uploads and cancel them on a replicate delete An upload is tracked from the handler, and a durable one from submission to the write queue, where the guard travels with the request: a handler dropped while the write waits in the queue does not untrack it, and the worker settles it after the batch. A replicate delete that finds nothing cancels the needle's pending writes and looks again, like the Go handler. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: take a cancelled write back in the lock span that publishes it The write tombstoned itself only after it returned, with the volume locks released. In that gap a compaction could move the record so the offset check skipped it, a close could drop it, and an identical write sent after the delete could match the live record and report success before the tombstone removed it. The check now runs right after the publish, under the same dataFileSyncLock and dataFileAccessLock hold, in the plain, fsync and batch write paths. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: a delete cancels only the writes carrying its cookie A delete with the right needle id but the wrong cookie could cancel an in-flight write and get it tombstoned. Each pending write now records its cookie, and a cancel only marks the matching ones. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * seaweed-volume: take a cancelled write back in the lock span that publishes it Mirror the Go change: settle_pending_write ran after write_needles_grouped returned, so a compaction, a close or an identical write later in the same batch could slip in before the tombstone. drop_cancelled_write now runs right after the publish, under the same sync and access lock hold, in the synced run, the unsynced run and write_needle_unsynced_or_refused. write_pending_needles carries each write's PendingWrite down to that point. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * seaweed-volume: a delete cancels only the writes carrying its cookie Mirror the Go change: a delete with the right needle id and the wrong cookie no longer cancels an in-flight write. Each tracked write keeps its cookie next to its cancel flag. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * seaweed-volume: refuse a queued write whose volume was remounted The write queue tracks a write on the volume found at submit, but the worker looks the volume up again by id. After a remount, a delete reaches the new volume and cannot cancel the write registered on the old one, so the worker now refuses such a write instead of landing it on the new volume. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * seaweed-volume: decode the delete's type parameter like Go's FormValue The delete handler matched the raw query text for type=replicate, so a percent-encoded name or value was not treated as a replica delete and did not cancel the write it undoes. Parse the query the way the upload handler does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: test a replica delete landing in a write's fsync window Drive DeleteHandler while a durable write is parked in its .dat fsync, appended but not indexed: the replica delete answers 404 and the write is taken back when it publishes, while a plain delete leaves it live. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: flush a take-back tombstone before the index row that points at it A durable write taken back during its fsync appends the tombstone after the .dat sync, and the .idx sync that follows made its row durable while the record itself could still be lost, which reloads the volume read only. The fsync and batch paths now sync the .dat again first when a take-back appended a tombstone; a failure stops the volume taking writes like a failed index sync. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * seaweed-volume: flush a take-back tombstone before the index row that points at it A durable write taken back in write_synced_run appends its tombstone after the .dat sync, so the .idx flush that follows could make the row durable while the tombstone record itself was still in the page cache. The run now syncs the .dat again first when a take-back appended a tombstone; a failure stops the volume taking writes like a failed index flush. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: a skip-cookie batch delete cancels the writes of the file id it names BatchDelete with SkipCookieCheck left the needle's cookie at zero, so the delete's cancel matched none of the in-flight writes and a retry queued behind it could land afterwards. The non-EC delete now carries the file id's cookie, as the Rust volume server already does; the EC path keeps cookie 0, which DeleteEcShardNeedle reads as "skip the check". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: release a held test write before the store closes A failed wait ended the test before release ran, so the writer stayed parked in its fsync and the store cleanup blocked on its lock until the package timed out. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
692 lines
23 KiB
Go
692 lines
23 KiB
Go
package storage
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"path"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/stats"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/backend"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/types"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
)
|
|
|
|
type Volume struct {
|
|
Id needle.VolumeId
|
|
dir string
|
|
dirIdx string
|
|
Collection string
|
|
DataBackend backend.BackendStorageFile
|
|
nm NeedleMapper
|
|
tmpNm TempNeedleMapper
|
|
needleMapKind NeedleMapKind
|
|
noWriteOrDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
|
|
noWriteCanDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
|
|
ioUnavailable bool
|
|
ioUnavailableError string
|
|
noWriteLock sync.RWMutex
|
|
hasRemoteFile atomic.Bool // if the volume is tiered: data lives in a remote backend
|
|
MemoryMapMaxSizeMb uint32
|
|
|
|
super_block.SuperBlock
|
|
|
|
dataFileAccessLock sync.RWMutex
|
|
// Serializes a data-file mutation with the sync that makes it durable:
|
|
// write, delete, lifecycle and compaction paths hold it for their whole
|
|
// critical section, while the fsync itself runs with dataFileAccessLock
|
|
// released so readers of this volume do not wait on it.
|
|
dataFileSyncLock sync.Mutex
|
|
superBlockAccessLock sync.Mutex
|
|
|
|
// The batch worker exists only once the volume takes a durable write. Most
|
|
// never do -- read-only, remote-tiered, or written without fsync -- and a
|
|
// parked worker costs its goroutine stack plus a 128-slot channel, which a
|
|
// server holding millions of volumes cannot pay for all of them.
|
|
asyncWorkerLock sync.Mutex
|
|
asyncRequestsChan chan *needle.AsyncRequest
|
|
asyncWorkerClosed bool
|
|
|
|
pendingWrites pendingWrites
|
|
|
|
lastModifiedTsSeconds uint64 // unix time in seconds
|
|
lastAppendAtNs uint64 // unix time in nanoseconds
|
|
lastWriteAppendAtNs uint64 // AppendAtNs of the newest write; tombstones don't move it
|
|
lastWriteNeedleKey types.NeedleId // the write behind the watermark
|
|
lastWriteDeleted bool // that write was deleted, so recovery has to rescan
|
|
keepLastModifiedTsOnLoad bool
|
|
|
|
lastCompactIndexOffset uint64
|
|
lastCompactRevision uint16
|
|
ldbTimeout int64
|
|
|
|
isCompactionInProgress atomic.Bool
|
|
lastDiskCheckNs atomic.Int64 // unix time in nanoseconds for phantom volume detection
|
|
|
|
volumeInfoRWLock sync.RWMutex
|
|
volumeInfo *volume_server_pb.VolumeInfo
|
|
location *DiskLocation
|
|
diskId uint32 // ID of this volume's disk in Store.Locations array
|
|
|
|
IoErrorTracker
|
|
}
|
|
|
|
func NewVolume(dirname string, dirIdx string, collection string, id needle.VolumeId, needleMapKind NeedleMapKind, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, ver needle.Version, memoryMapMaxSizeMb uint32, ldbTimeout int64) (v *Volume, e error) {
|
|
// if replicaPlacement is nil, the superblock will be loaded from disk
|
|
v = &Volume{dir: dirname, dirIdx: dirIdx, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb}
|
|
v.SuperBlock = super_block.SuperBlock{ReplicaPlacement: replicaPlacement, Ttl: ttl}
|
|
v.needleMapKind = needleMapKind
|
|
v.ldbTimeout = ldbTimeout
|
|
e = v.load(true, true, needleMapKind, preallocate, ver)
|
|
return
|
|
}
|
|
|
|
func (v *Volume) String() string {
|
|
v.noWriteLock.RLock()
|
|
defer v.noWriteLock.RUnlock()
|
|
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)
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
func (v *Volume) DataFileName() (fileName string) {
|
|
return VolumeFileName(v.dir, v.Collection, int(v.Id))
|
|
}
|
|
|
|
func (v *Volume) IndexFileName() (fileName string) {
|
|
return VolumeFileName(v.dirIdx, v.Collection, int(v.Id))
|
|
}
|
|
|
|
func (v *Volume) FileName(ext string) (fileName string) {
|
|
switch ext {
|
|
case ".idx", ".cpx", ".ldb", ".cpldb", ".rdb":
|
|
return VolumeFileName(v.dirIdx, v.Collection, int(v.Id)) + ext
|
|
}
|
|
// .dat, .cpd, .vif
|
|
return VolumeFileName(v.dir, v.Collection, int(v.Id)) + ext
|
|
}
|
|
|
|
// RelocateIndexTo moves the volume's index to newIdxDir and reopens the volume
|
|
// against it in place, without unmounting. It takes the data-file write lock —
|
|
// so a concurrent read blocks briefly instead of failing — closes the needle
|
|
// map and data backend, moves the .idx (and the derived .sdx best-effort), then
|
|
// retargets dirIdx and reloads, mirroring CommitCompact's close-swap-load. A
|
|
// decode co-locates the rebuilt index with the data so the on-demand mount can
|
|
// find the volume; this returns it to the -dir.idx tier once the EC shards are
|
|
// gone. A no-op when the index already lives in newIdxDir. A derived .ldb is
|
|
// not moved: the reload rebuilds it in newIdxDir from the .idx.
|
|
func (v *Volume) RelocateIndexTo(newIdxDir string) error {
|
|
v.dataFileSyncLock.Lock()
|
|
defer v.dataFileSyncLock.Unlock()
|
|
v.dataFileAccessLock.Lock()
|
|
defer v.dataFileAccessLock.Unlock()
|
|
|
|
if v.dirIdx == newIdxDir {
|
|
return nil
|
|
}
|
|
oldBase := VolumeFileName(v.dirIdx, v.Collection, int(v.Id))
|
|
if _, err := os.Stat(oldBase + ".idx"); err != nil {
|
|
return nil // nothing co-located to move
|
|
}
|
|
newBase := VolumeFileName(newIdxDir, v.Collection, int(v.Id))
|
|
|
|
if v.nm != nil {
|
|
_ = v.nm.Sync()
|
|
v.nm.Close()
|
|
v.nm = nil
|
|
}
|
|
if v.DataBackend != nil {
|
|
_ = v.DataBackend.Sync()
|
|
_ = v.DataBackend.Close()
|
|
v.DataBackend = nil
|
|
}
|
|
|
|
if err := RenameOrCopyFile(oldBase+".idx", newBase+".idx"); err != nil {
|
|
// Reopen against the old dir so the volume is not left down; surface a
|
|
// failed reopen since it leaves the volume unusable until the next load.
|
|
if reopenErr := v.load(true, false, v.needleMapKind, 0, v.Version()); reopenErr != nil {
|
|
glog.Errorf("relocate volume %d: reopen after failed .idx move: %v", v.Id, reopenErr)
|
|
}
|
|
return fmt.Errorf("relocate index for volume %d: move .idx: %w", v.Id, err)
|
|
}
|
|
// The .sdx is a derived sorted index; move it when present, but a failure is
|
|
// not fatal — drop the stale copy so the reload rebuilds it in the new dir.
|
|
if _, err := os.Stat(oldBase + ".sdx"); err == nil {
|
|
if err := RenameOrCopyFile(oldBase+".sdx", newBase+".sdx"); err != nil {
|
|
glog.Warningf("relocate volume %d: move .sdx: %v (will rebuild)", v.Id, err)
|
|
_ = os.Remove(oldBase + ".sdx")
|
|
}
|
|
}
|
|
v.dirIdx = newIdxDir
|
|
return v.load(true, false, v.needleMapKind, 0, v.Version())
|
|
}
|
|
|
|
func (v *Volume) Version() needle.Version {
|
|
v.superBlockAccessLock.Lock()
|
|
defer v.superBlockAccessLock.Unlock()
|
|
if v.volumeInfo.Version != 0 {
|
|
v.SuperBlock.Version = needle.Version(v.volumeInfo.Version)
|
|
}
|
|
return v.SuperBlock.Version
|
|
}
|
|
|
|
// UpdateReplicaPlacement records rp in the live volume — the in-memory .vif
|
|
// copy and the superblock — and rewrites the .dat header when the data file
|
|
// is local. Keeps a mounted volume consistent with the .vif the caller just
|
|
// configured on disk, so heartbeats and later compactions report and rewrite
|
|
// the same placement.
|
|
func (v *Volume) UpdateReplicaPlacement(rp *super_block.ReplicaPlacement) error {
|
|
v.dataFileSyncLock.Lock()
|
|
defer v.dataFileSyncLock.Unlock()
|
|
v.dataFileAccessLock.Lock()
|
|
defer v.dataFileAccessLock.Unlock()
|
|
v.superBlockAccessLock.Lock()
|
|
defer v.superBlockAccessLock.Unlock()
|
|
v.volumeInfoRWLock.Lock()
|
|
defer v.volumeInfoRWLock.Unlock()
|
|
|
|
v.SuperBlock.ReplicaPlacement = rp
|
|
if v.volumeInfo != nil {
|
|
v.volumeInfo.Replication = rp.String()
|
|
}
|
|
diskFile, ok := v.DataBackend.(*backend.DiskFile)
|
|
if !ok {
|
|
return nil // remote-tiered: the .vif carries the placement
|
|
}
|
|
if _, err := diskFile.WriteAt(v.SuperBlock.Bytes(), 0); err != nil {
|
|
return fmt.Errorf("volume %d failed to write superblock: %v", v.Id, err)
|
|
}
|
|
return diskFile.Sync()
|
|
}
|
|
|
|
func (v *Volume) FileStat() (datSize uint64, idxSize uint64, modTime time.Time) {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
|
|
if v.DataBackend == nil {
|
|
return
|
|
}
|
|
|
|
datFileSize, modTime, e := v.DataBackend.GetStat()
|
|
if e == nil {
|
|
return uint64(datFileSize), v.nm.IndexFileSize(), modTime
|
|
}
|
|
glog.V(0).Infof("Failed to read file size %s %v", v.DataBackend.Name(), e)
|
|
return // -1 causes integer overflow and the volume to become unwritable.
|
|
}
|
|
|
|
// datAllocatedSize returns the disk space held by the .dat file, or 0 when it is unknown.
|
|
func (v *Volume) datAllocatedSize() uint64 {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
|
|
df, ok := v.DataBackend.(*backend.DiskFile)
|
|
if !ok {
|
|
return 0
|
|
}
|
|
fi, err := df.File.Stat()
|
|
if err != nil {
|
|
return 0
|
|
}
|
|
return AllocatedSize(fi)
|
|
}
|
|
|
|
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) doIsEmpty() (bool, error) {
|
|
// check v.DataBackend.GetStat()
|
|
if v.DataBackend == nil {
|
|
return false, fmt.Errorf("v.DataBackend is nil")
|
|
} else {
|
|
datFileSize, _, e := v.DataBackend.GetStat()
|
|
if e != nil {
|
|
glog.V(0).Infof("Failed to read file size %s %v", v.DataBackend.Name(), e)
|
|
return false, fmt.Errorf("v.DataBackend.GetStat(): %v", e)
|
|
}
|
|
if datFileSize > super_block.SuperBlockSize {
|
|
return false, nil
|
|
}
|
|
}
|
|
// check v.nm.ContentSize()
|
|
if v.nm != nil {
|
|
if v.nm.ContentSize() > 0 {
|
|
return false, nil
|
|
}
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
// doIsGarbage reports whether every byte ever written is already deleted —
|
|
// the same all-garbage state vacuum measures, checked against the byte
|
|
// counters because the count counters drift on reload.
|
|
func (v *Volume) doIsGarbage() bool {
|
|
if v.nm == nil {
|
|
return false
|
|
}
|
|
contentSize := v.nm.ContentSize()
|
|
return contentSize > 0 && v.nm.DeletedSize() >= contentSize
|
|
}
|
|
|
|
func (v *Volume) DeletedSize() uint64 {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
if v.nm == nil {
|
|
return 0
|
|
}
|
|
return v.nm.DeletedSize()
|
|
}
|
|
|
|
func (v *Volume) FileCount() uint64 {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
if v.nm == nil {
|
|
return 0
|
|
}
|
|
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()
|
|
}
|
|
|
|
func (v *Volume) DiskType() types.DiskType {
|
|
return v.location.DiskType
|
|
}
|
|
|
|
func (v *Volume) SyncToDisk() {
|
|
v.dataFileSyncLock.Lock()
|
|
defer v.dataFileSyncLock.Unlock()
|
|
v.dataFileAccessLock.Lock()
|
|
defer v.dataFileAccessLock.Unlock()
|
|
// Flush with the access lock dropped so readers are not stalled by the
|
|
// fsync; writers and file swaps stay excluded by dataFileSyncLock.
|
|
nm, backend := v.nm, v.DataBackend
|
|
v.dataFileAccessLock.Unlock()
|
|
if nm != nil {
|
|
if err := nm.Sync(); err != nil {
|
|
glog.Warningf("Volume Close fail to sync volume idx %d", v.Id)
|
|
}
|
|
}
|
|
if backend != nil {
|
|
if err := backend.Sync(); err != nil {
|
|
glog.Warningf("Volume Close fail to sync volume %d", v.Id)
|
|
}
|
|
}
|
|
v.dataFileAccessLock.Lock()
|
|
}
|
|
|
|
// Close cleanly shuts down this volume
|
|
func (v *Volume) Close() {
|
|
// Wait for any in-progress compaction to finish and claim the flag so no
|
|
// new compaction can start. This must happen BEFORE acquiring
|
|
// dataFileSyncLock and dataFileAccessLock to avoid deadlocking with
|
|
// CommitCompact which holds the flag while waiting for the locks.
|
|
for !v.isCompactionInProgress.CompareAndSwap(false, true) {
|
|
time.Sleep(521 * time.Millisecond)
|
|
glog.Warningf("Volume Close wait for compaction %d", v.Id)
|
|
}
|
|
defer v.isCompactionInProgress.Store(false)
|
|
|
|
v.dataFileSyncLock.Lock()
|
|
defer v.dataFileSyncLock.Unlock()
|
|
v.dataFileAccessLock.Lock()
|
|
defer v.dataFileAccessLock.Unlock()
|
|
|
|
v.doClose()
|
|
}
|
|
|
|
// SwapDataBackend atomically replaces the data backend and updates the
|
|
// remote-tier flag under dataFileAccessLock, closing the old backend. Both tier
|
|
// directions go through here so hasRemoteFile always matches the live backend:
|
|
// tier-down passes hasRemoteFile=false (now serving a local .dat), tier-up
|
|
// passes true. Keeping the swap and the flag under one lock means the heartbeat
|
|
// never observes a half-swapped backend or a flag that disagrees with it — a
|
|
// stale-false flag would make doDeleteRequest skip the new .dat's tombstones and
|
|
// disable the phantom-.dat guard.
|
|
func (v *Volume) SwapDataBackend(newBackend backend.BackendStorageFile, hasRemoteFile bool) {
|
|
v.dataFileSyncLock.Lock()
|
|
defer v.dataFileSyncLock.Unlock()
|
|
v.dataFileAccessLock.Lock()
|
|
defer v.dataFileAccessLock.Unlock()
|
|
v.swapDataBackendLocked(newBackend, hasRemoteFile)
|
|
}
|
|
|
|
// swapDataBackendLocked is the body of SwapDataBackend for callers that already
|
|
// hold dataFileAccessLock (e.g. load() reached while CommitCompact holds the
|
|
// lock). Reusing it from those under-lock paths avoids re-entering the
|
|
// non-reentrant lock, which would deadlock.
|
|
func (v *Volume) swapDataBackendLocked(newBackend backend.BackendStorageFile, hasRemoteFile bool) {
|
|
if v.DataBackend != nil {
|
|
v.DataBackend.Close()
|
|
}
|
|
v.DataBackend = newBackend
|
|
v.hasRemoteFile.Store(hasRemoteFile)
|
|
}
|
|
|
|
func (v *Volume) doClose() {
|
|
if v.nm != nil {
|
|
if err := v.nm.Sync(); err != nil {
|
|
glog.Warningf("Volume Close fail to sync volume idx %d", v.Id)
|
|
}
|
|
v.nm.Close()
|
|
v.nm = nil
|
|
}
|
|
if v.DataBackend != nil {
|
|
if err := v.DataBackend.Close(); err != nil {
|
|
glog.Warningf("Volume Close fail to sync volume %d", v.Id)
|
|
}
|
|
v.DataBackend = nil
|
|
stats.VolumeServerVolumeGauge.WithLabelValues(v.Collection, "volume").Dec()
|
|
}
|
|
}
|
|
|
|
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(contentSize uint64, volumeSizeLimit uint64) bool {
|
|
if volumeSizeLimit == 0 {
|
|
// skip if we don't know size limit
|
|
return false
|
|
}
|
|
if contentSize <= super_block.SuperBlockSize {
|
|
return false
|
|
}
|
|
if v.Ttl == nil || v.Ttl.Minutes() == 0 {
|
|
return false
|
|
}
|
|
glog.V(2).Infof("volume %d now:%v lastModified:%v", v.Id, time.Now().Unix(), v.lastModifiedTsSeconds)
|
|
livedMinutes := (time.Now().Unix() - int64(v.lastModifiedTsSeconds)) / 60
|
|
glog.V(2).Infof("volume %d ttl:%v lived:%v", v.Id, v.Ttl, livedMinutes)
|
|
if int64(v.Ttl.Minutes()) < livedMinutes {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
// ExpireAtSec is when this volume's data becomes garbage, counted from its last
|
|
// write. Counting from the current time instead let every .vif rewrite -- a
|
|
// read-only mark, a tier upload, an EC encode -- hand an already expiring volume
|
|
// another full TTL. Zero when the volume has no TTL.
|
|
func (v *Volume) ExpireAtSec() uint64 {
|
|
if v.Ttl == nil {
|
|
return 0
|
|
}
|
|
ttlSeconds := v.Ttl.ToSeconds()
|
|
if ttlSeconds == 0 {
|
|
return 0
|
|
}
|
|
lastWriteSec := v.lastModifiedTsSeconds
|
|
if lastWriteSec == 0 {
|
|
lastWriteSec = uint64(time.Now().Unix())
|
|
}
|
|
return lastWriteSec + ttlSeconds
|
|
}
|
|
|
|
// wait either maxDelayMinutes or 10% of ttl minutes
|
|
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
|
|
}
|
|
|
|
if uint64(v.Ttl.Minutes()+removalDelay)*60+v.lastModifiedTsSeconds < uint64(time.Now().Unix()) {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (v *Volume) collectStatus() (maxFileKey types.NeedleId, datFileSize int64, modTime time.Time, fileCount, deletedCount, deletedSize uint64, ok bool) {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
|
|
if v.nm == nil || v.DataBackend == nil {
|
|
return
|
|
}
|
|
|
|
ok = true
|
|
|
|
maxFileKey = v.nm.MaxFileKey()
|
|
datFileSize, modTime, _ = v.DataBackend.GetStat()
|
|
fileCount = uint64(v.nm.FileCount())
|
|
deletedCount = uint64(v.nm.DeletedCount())
|
|
deletedSize = v.nm.DeletedSize()
|
|
|
|
return
|
|
}
|
|
|
|
// ToVolumeInformationMessage fills into with what the master is told about this
|
|
// volume, allocating a message when into is nil. A heartbeat that keeps only
|
|
// the volumes it reports fills the same message for all the rest.
|
|
func (v *Volume) ToVolumeInformationMessage(into *master_pb.VolumeInformationMessage) (types.NeedleId, *master_pb.VolumeInformationMessage) {
|
|
|
|
maxFileKey, volumeSize, modTime, fileCount, deletedCount, deletedSize, ok := v.collectStatus()
|
|
|
|
if !ok {
|
|
return 0, nil
|
|
}
|
|
|
|
// Detect phantom volumes: the .dat was unlinked from disk but is still held
|
|
// open as a deleted FD, so the volume keeps serving and heartbeating while no
|
|
// disk-path operation can ever succeed. Skip remote-tiered volumes, whose .dat
|
|
// legitimately lives in cloud storage. Only a present .dat is cached for 30s; a
|
|
// missing one is re-checked every heartbeat so the volume stays suppressed until
|
|
// the file returns.
|
|
if fileCount > 0 && !v.HasRemoteFile() {
|
|
const diskCheckIntervalNs = 30 * int64(time.Second)
|
|
now := time.Now().UnixNano()
|
|
if now-v.lastDiskCheckNs.Load() > diskCheckIntervalNs {
|
|
if _, err := os.Stat(v.FileName(".dat")); os.IsNotExist(err) {
|
|
glog.Warningf("Volume %d: data file %s missing (held open as deleted FD) - not reporting to master", v.Id, v.FileName(".dat"))
|
|
return 0, nil
|
|
}
|
|
v.lastDiskCheckNs.Store(now)
|
|
}
|
|
}
|
|
|
|
volumeInfo := into
|
|
if volumeInfo == nil {
|
|
volumeInfo = &master_pb.VolumeInformationMessage{}
|
|
}
|
|
volumeInfo.Id = uint32(v.Id)
|
|
volumeInfo.Size = uint64(volumeSize)
|
|
volumeInfo.Collection = v.Collection
|
|
volumeInfo.FileCount = fileCount
|
|
volumeInfo.DeleteCount = deletedCount
|
|
volumeInfo.DeletedByteCount = deletedSize
|
|
volumeInfo.ReadOnly, _, volumeInfo.ReadOnlyCanDelete, _ = v.ReadOnlyReasons()
|
|
volumeInfo.ReplicaPlacement = uint32(v.ReplicaPlacement.Byte())
|
|
volumeInfo.Version = uint32(v.Version())
|
|
volumeInfo.Ttl = v.Ttl.ToUint32()
|
|
volumeInfo.CompactRevision = uint32(v.SuperBlock.CompactionRevision)
|
|
volumeInfo.ModifiedAtSecond = modTime.Unix()
|
|
volumeInfo.DiskType = string(v.location.DiskType)
|
|
volumeInfo.DiskId = v.diskId
|
|
volumeInfo.RemoteStorageName, volumeInfo.RemoteStorageKey = v.RemoteStorageNameKey()
|
|
|
|
return maxFileKey, volumeInfo
|
|
}
|
|
|
|
func (v *Volume) RemoteStorageNameKey() (storageName, storageKey string) {
|
|
if v.volumeInfo == nil {
|
|
return
|
|
}
|
|
if len(v.volumeInfo.GetFiles()) == 0 {
|
|
return
|
|
}
|
|
return v.volumeInfo.GetFiles()[0].BackendName(), v.volumeInfo.GetFiles()[0].GetKey()
|
|
}
|
|
|
|
func (v *Volume) IsReadOnly() bool {
|
|
readOnly, _, _, _ := v.ReadOnlyReasons()
|
|
return readOnly
|
|
}
|
|
|
|
// ReadOnlyReasons reports whether the volume refuses writes and why, reading the
|
|
// flags once so the reasons cannot disagree with the verdict.
|
|
func (v *Volume) ReadOnlyReasons() (readOnly, noWriteOrDelete, noWriteCanDelete, diskSpaceLow bool) {
|
|
v.noWriteLock.RLock()
|
|
noWriteOrDelete, noWriteCanDelete = v.noWriteOrDelete, v.noWriteCanDelete
|
|
if v.ioUnavailable {
|
|
noWriteOrDelete = true
|
|
}
|
|
v.noWriteLock.RUnlock()
|
|
// The location is attached when the volume joins a disk location, which is
|
|
// after NewVolume hands it back.
|
|
diskSpaceLow = v.location != nil && v.location.isDiskSpaceLow.Load()
|
|
return noWriteOrDelete || noWriteCanDelete || diskSpaceLow, noWriteOrDelete, noWriteCanDelete, diskSpaceLow
|
|
}
|
|
|
|
var errVolumeUnavailable = errors.New("volume unavailable")
|
|
|
|
func (v *Volume) UnavailableError() error {
|
|
v.noWriteLock.RLock()
|
|
unavailable := v.ioUnavailable
|
|
reason := v.ioUnavailableError
|
|
v.noWriteLock.RUnlock()
|
|
if !unavailable {
|
|
return nil
|
|
}
|
|
if reason == "" {
|
|
return fmt.Errorf("volume %d is unavailable: %w", v.Id, errVolumeUnavailable)
|
|
}
|
|
return fmt.Errorf("volume %d is unavailable: %s: %w", v.Id, reason, errVolumeUnavailable)
|
|
}
|
|
|
|
func (v *Volume) markIoUnavailable(err error) {
|
|
v.noWriteLock.Lock()
|
|
v.noWriteOrDelete = true
|
|
v.ioUnavailable = true
|
|
v.ioUnavailableError = err.Error()
|
|
v.noWriteLock.Unlock()
|
|
v.markIoQuarantined()
|
|
|
|
if persistErr := v.persistUnavailable(err.Error()); persistErr != nil {
|
|
glog.Warningf("volume %d: failed to persist unavailable marker: %v", v.Id, persistErr)
|
|
}
|
|
if v.volumeInfo != nil {
|
|
if persistErr := v.PersistReadOnly(true, false); persistErr != nil {
|
|
glog.Warningf("volume %d: failed to persist unavailable state: %v", v.Id, persistErr)
|
|
}
|
|
}
|
|
glog.Errorf("volume %d entered unavailable state after failed recovery: %v", v.Id, err)
|
|
}
|
|
|
|
// persistUnavailable records the failed-recovery state so a reload keeps the
|
|
// volume unavailable instead of serving an unverified .dat/index pair.
|
|
func (v *Volume) persistUnavailable(reason string) error {
|
|
marker := v.FileName(".unavailable")
|
|
f, err := os.OpenFile(marker, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := f.WriteString(reason); err != nil {
|
|
f.Close()
|
|
return err
|
|
}
|
|
if err := f.Sync(); err != nil {
|
|
f.Close()
|
|
return err
|
|
}
|
|
if err := f.Close(); err != nil {
|
|
return err
|
|
}
|
|
return util.FsyncDir(v.dir)
|
|
}
|
|
|
|
// restoreUnavailable re-arms the in-memory state the .unavailable marker
|
|
// recorded. The marker is deleted manually once the volume pair is verified.
|
|
func (v *Volume) restoreUnavailable() {
|
|
reason, err := os.ReadFile(v.FileName(".unavailable"))
|
|
if err != nil {
|
|
return
|
|
}
|
|
v.noWriteLock.Lock()
|
|
v.noWriteOrDelete = true
|
|
v.ioUnavailable = true
|
|
v.ioUnavailableError = strings.TrimSpace(string(reason))
|
|
v.noWriteLock.Unlock()
|
|
v.markIoQuarantined()
|
|
glog.Warningf("volume %d is unavailable: %s", v.Id, v.ioUnavailableError)
|
|
}
|
|
|
|
func (v *Volume) PersistReadOnly(readOnly bool, canDelete bool) error {
|
|
v.volumeInfoRWLock.Lock()
|
|
defer v.volumeInfoRWLock.Unlock()
|
|
prevReadOnly := v.volumeInfo.ReadOnly
|
|
prevReadOnlyCanDelete := v.volumeInfo.ReadOnlyCanDelete
|
|
v.volumeInfo.ReadOnly = readOnly
|
|
v.volumeInfo.ReadOnlyCanDelete = readOnly && canDelete
|
|
if err := v.SaveVolumeInfo(); err != nil {
|
|
// A pre-commit failure (write/sync/close/rename) leaves the old
|
|
// .vif intact, so roll back in-memory state to match it. A
|
|
// NotCrashDurableError means the rename already committed the
|
|
// new mode to disk; rolling back would split in-memory state
|
|
// from the durable file, so keep the new state and propagate.
|
|
var ndErr *volume_info.NotCrashDurableError
|
|
if !errors.As(err, &ndErr) {
|
|
v.volumeInfo.ReadOnly = prevReadOnly
|
|
v.volumeInfo.ReadOnlyCanDelete = prevReadOnlyCanDelete
|
|
}
|
|
return fmt.Errorf("persist volume read-only state: %w", err)
|
|
}
|
|
return nil
|
|
}
|