fs

package
v1.8.8-0...-edaca3c Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 22, 2026 License: MIT Imports: 24 Imported by: 0

Documentation

Overview

Package fs contains filesystem-backed implementations used by SOP. This file provides a thin FileIO abstraction over os with retry semantics.

Package fs provides filesystem-backed implementations of SOP storage primitives: registries, blob store, transaction logs and a replication tracker.

Index

Constants

View Source 
const (

	// 250, should generate 1MB file segment. Formula: 250 X 4096 = 1MB
	// Given a 1000 slot size per node (default), should be able to manage ~10,725,000 B-Tree items (key/value pairs).
	//
	// Formula: 250 * 66 * 1000 * 65% (utilization) = 10,725,000
	// Or if you use 5000 slot size per node, 'will give you ~53,625,000 items.
	//
	// Note: This capacity is per segment file. SOP will allocate more segment files as needed.
	// E.g. 1 Billion items @ 250 hashmod ~= 100 segment files (SlotLength 1000).
	// E.g. 1 Billion items @ 250 hashmod ~= 19 segment files (SlotLength 5000).
	MinimumModValue = 250
	// 750k, should generate 3GB file segment.  Formula: 750k X 4096 = 3GB
	MaximumModValue = 750000
)
View Source 
const (
	StoreInfoFilename            = "storeinfo.txt"
	RegistryHashModValueFilename = "reghashmod.txt"
)
View Source 
const (
	// DateHourLayout is the time layout used for hour-bucketed log folders and files.
	DateHourLayout = "2006-01-02T15"
)
View Source 
const (
	FILE_NOT_FOUND = "no such file or directory"
)

Variables

View Source 
var AgeLimit float64 = 70
View Source 
var (
	LockFileRegionDuration = time.Duration(5 * time.Minute)
)

Configurable lock TTL and slack for file-region operations. Defaults retain previous behavior: 5m TTL and ~2% slack (min 2s).

Functions

func Apply4LevelHierarchy 

func Apply4LevelHierarchy(id sop.UUID) string

Apply4LevelHierarchy maps a UUID to a 4-level directory structure using its first four hex digits. Example: abcd-... -> a/b/c/d, enabling broad distribution across subfolders.

func DefaultToFilePath 

func DefaultToFilePath(basePath string, id sop.UUID) string

DefaultToFilePath formats a path by appending a 4-level folder hierarchy derived from the UUID. This reduces per-directory file counts and improves filesystem performance on large datasets.

func GetGlobalErasureConfig 

func GetGlobalErasureConfig() map[string]sop.ErasureCodingConfig

Returns the global Erasure Coding config.

func NewBlobStore 

func NewBlobStore(basePath string, toFilePath ToFilePathFunc, fileIO FileIO) sop.BlobStore

NewBlobStore instantiates a new blobstore for File System storage. If fileIO or toFilePath are nil, sensible defaults are used.

func NewBlobStoreWithEC 

func NewBlobStoreWithEC(toFilePath ToFilePathFunc, fileIO FileIO, erasureConfig map[string]sop.ErasureCodingConfig) (sop.BlobStore, error)

Instantiate a blob store with replication (via Erasure Coding (EC)) capabilities. If a per-table EC config is not supplied, the global configuration is used. Validates that the number of base paths equals data+parity shard count.

func NewManageStoreFolder 

func NewManageStoreFolder(fileIO FileIO) sop.ManageStore

NewManageStoreFolder returns a ManageStore implementation that creates and removes directories on the local filesystem.

func NewRegistry 

func NewRegistry(readWrite bool, hashModValue int, rt *replicationTracker, l2Cache sop.L2Cache) *registryOnDisk

NewRegistry creates a filesystem-backed Registry that manages handles on disk using a hashmap structure. readWrite toggles direct write access to files; hashModValue controls file partitioning (fan-out) for the registry table. rt provides active/passive routing for replication; l2Cache supplies cross-process caching and locking.

func NewReplicationTracker 

func NewReplicationTracker(ctx context.Context, storesBaseFolders []string, replicate bool, l2Cache sop.L2Cache) (*replicationTracker, error)

NewReplicationTracker instantiates a tracker for active/passive folders and replication status. It reads persisted status, initializes global in-memory state, and optionally synchronizes with L2 cache.

func SetGlobalErasureConfig 

func SetGlobalErasureConfig(erasureConfig map[string]sop.ErasureCodingConfig)

Invoke SetGlobalErasureConfig to set the application global Erasure Coding Config lookup.

func TriggerFailover 

func TriggerFailover(ctx context.Context, storesBaseFolders []string, replicate bool, l2Cache sop.L2Cache) error

TriggerFailover is a helper to manually trigger failover for testing purposes.

Types

type BlobStoreWithEC 

type BlobStoreWithEC struct {
	// contains filtered or unexported fields
}

BlobStore has no caching built in because blobs are huge, caller code can apply caching on top of it. BlobStoreWithEC adds Erasure Coding (EC) for replication/tolerance across multiple drives.

func (*BlobStoreWithEC) Add 

func (b *BlobStoreWithEC) Add(ctx context.Context, storesblobs []sop.BlobsPayload[sop.KeyValuePair[sop.UUID, []byte]]) error

Add splits blob content into shards, writes each shard with per-shard metadata prefix, and tolerates up to ParityShardsCount write failures per blob. Errors beyond tolerance trigger rollback.

func (*BlobStoreWithEC) GetOne 

func (b *BlobStoreWithEC) GetOne(ctx context.Context, blobFilePath string, blobID sop.UUID) ([]byte, error)

GetOne reads shards across drives, extracts per-shard metadata, and decodes via EC. If some shards are missing but enough remain (>= data shards), decoding still succeeds. Optionally repairs corrupted/bitrotted shards by recomputing and rewriting them.

func (*BlobStoreWithEC) Remove 

func (b *BlobStoreWithEC) Remove(ctx context.Context, storesBlobsIDs []sop.BlobsPayload[sop.UUID]) error

Remove deletes shard files across all configured drives. Errors are tolerated and logged when replication is expected to handle eventual consistency.

func (*BlobStoreWithEC) RemoveStore 

func (b *BlobStoreWithEC) RemoveStore(ctx context.Context, blobStoreName string) error

RemoveStore recursively deletes the base folder for a store and all of its contents.

func (*BlobStoreWithEC) Update 

func (b *BlobStoreWithEC) Update(ctx context.Context, storesblobs []sop.BlobsPayload[sop.KeyValuePair[sop.UUID, []byte]]) error

type ByModTime 

type ByModTime []FileInfoWithModTime

ByModTime sorts FileInfoWithModTime by modification time.

func (ByModTime) Len 

func (fis ByModTime) Len() int

func (ByModTime) Less 

func (fis ByModTime) Less(i, j int) bool

func (ByModTime) Swap 

func (fis ByModTime) Swap(i, j int)

type DirectIO 

type DirectIO interface {
	// Open opens a file with the given name and flags using direct I/O when possible.
	Open(ctx context.Context, filename string, flag int, permission os.FileMode) (*os.File, error)
	// WriteAt writes a block at the given offset.
	WriteAt(ctx context.Context, file *os.File, block []byte, offset int64) (int, error)
	// ReadAt reads a block at the given offset.
	ReadAt(ctx context.Context, file *os.File, block []byte, offset int64) (int, error)
	// Close closes the provided file handle.
	Close(file *os.File) error
}

DirectIO exposes unbuffered file operations using O_DIRECT semantics where supported. It is intended for large, block-aligned I/O on segment files. Implementations should be used with directio.AlignedBlock buffers and block-aligned offsets. 

var DirectIOSim DirectIO

DirectIOSim is a simulated DirectIO implementation for testing.

func NewDirectIO 

func NewDirectIO() DirectIO

NewDirectIO returns a DirectIO implementation backed by github.com/ncw/directio.

type FileIO 

type FileIO interface {
	WriteFile(ctx context.Context, name string, data []byte, perm os.FileMode) error
	ReadFile(ctx context.Context, name string) ([]byte, error)
	Remove(ctx context.Context, name string) error
	Stat(ctx context.Context, path string) (os.FileInfo, error)
	Exists(ctx context.Context, path string) bool

	// Directory API.
	RemoveAll(ctx context.Context, path string) error
	MkdirAll(ctx context.Context, path string, perm os.FileMode) error
	ReadDir(ctx context.Context, sourceDir string) ([]os.DirEntry, error)
}

FileIO defines filesystem operations used by this package. The default implementation delegates to the standard library's os package with retry semantics for transient errors.

func NewFileIO 

func NewFileIO() FileIO

NewFileIO returns a FileIO that performs I/O via the os package with basic retry handling for transient errors (e.g., NFS hiccups). Directories are created on-demand for writes.

type FileInfoWithModTime 

type FileInfoWithModTime struct {
	os.DirEntry
	ModTime time.Time
}

FileInfoWithModTime associates a DirEntry with its modification timestamp for sorting purposes.

type Registry 

type Registry interface {
	sop.Registry
	io.Closer
}

Registry extends sop.Registry with io.Closer to allow releasing resources when the transaction completes.

type ReplicationTrackedDetails 

type ReplicationTrackedDetails struct {
	FailedToReplicate bool
	// If true, folder as specified in storesBaseFolders[0] will be the active folder,
	// otherwise the 2nd folder, as specified in storesBaseFolders[1].
	ActiveFolderToggler bool

	// If true, Transactions should log commit changes so they can be used for "fast forward" functionality
	// of ReinstateFailedDrives functionality.
	LogCommitChanges bool
}

ReplicationTrackedDetails captures the replication state shared across processes, including the active/passive folder selection and whether replication has failed. 

var GlobalReplicationDetails *ReplicationTrackedDetails

type StoreRepository 

type StoreRepository struct {
	// contains filtered or unexported fields
}

StoreRepository is a filesystem-backed implementation of sop.StoreRepository. It manages store metadata and coordinates replication via a replicationTracker.

func NewStoreRepository 

func NewStoreRepository(ctx context.Context, rt *replicationTracker, manageStore sop.ManageStore, cache sop.L2Cache, registryHashModVal int) (*StoreRepository, error)

NewStoreRepository creates a StoreRepository that persists store info to disk. When replication is enabled, it validates base-folder configuration and writes the global registry-hash-mod value once, replicating it to the passive drive.

func (*StoreRepository) Add 

func (sr *StoreRepository) Add(ctx context.Context, stores ...sop.StoreInfo) error

Add appends new stores to the repository, updating the store list, creating folders, writing per-store metadata, and replicating the changes when configured. A cache entry is written for each added store. The store list is guarded by a cache-based lock.

func (*StoreRepository) CopyToPassiveFolders 

func (sr *StoreRepository) CopyToPassiveFolders(ctx context.Context) error

CopyToPassiveFolders copies store metadata (store list and per-store info) and registry segment files from the active folder to passive targets. It temporarily flips the active folder toggler to write into the passive side via the fileIO replication wrapper.

func (*StoreRepository) Get 

func (sr *StoreRepository) Get(ctx context.Context, names ...string) ([]sop.StoreInfo, error)

Get returns store info for the named stores, consulting the cache first.

func (*StoreRepository) GetAll 

func (sr *StoreRepository) GetAll(ctx context.Context) ([]string, error)

GetAll returns the list of all store names. If the store list file is absent it returns nil to indicate an empty repository. The list itself is not cached by design.

func (*StoreRepository) GetRegistryHashModValue 

func (sr *StoreRepository) GetRegistryHashModValue(ctx context.Context) (int, error)

GetRegistryHashModValue returns the configured registry hash modulus value, reading from disk if needed. Uses the replication-aware file IO wrapper to read the value written during initialization.

func (*StoreRepository) GetStoreFileStat 

func (sr *StoreRepository) GetStoreFileStat(ctx context.Context, storeName string) (os.FileInfo, error)

GetStoreFileStat returns the FileInfo of the store's metadata file.

func (*StoreRepository) GetStoresBaseFolder 

func (sr *StoreRepository) GetStoresBaseFolder() string

GetStoresBaseFolder returns the currently active base folder path used for store files.

func (*StoreRepository) GetWithTTL 

func (sr *StoreRepository) GetWithTTL(ctx context.Context, isCacheTTL bool, cacheDuration time.Duration, names ...string) ([]sop.StoreInfo, error)

GetWithTTL returns store info, optionally using TTL-aware cache lookups. Any misses are loaded from disk and then cached with the store's configured cache duration.

func (*StoreRepository) Remove 

func (sr *StoreRepository) Remove(ctx context.Context, storeNames ...string) error

Remove deletes the specified stores and their metadata, updates the store list, and replicates the removal when configured. Missing stores are tolerated with a warning.

func (*StoreRepository) Replicate 

func (sr *StoreRepository) Replicate(ctx context.Context, stores []sop.StoreInfo) error

Replicate writes the updated per-store metadata to the passive target. Any write error disables the current operation, signaling the caller to handle replication failures upstream.

func (*StoreRepository) Update 

func (sr *StoreRepository) Update(ctx context.Context, stores []sop.StoreInfo) ([]sop.StoreInfo, error)

type ToFilePathFunc 

type ToFilePathFunc func(basePath string, id sop.UUID) string

ToFilePathFunc formats a base path and UUID into a filesystem path optimized for I/O locality. 

var ToFilePath ToFilePathFunc = DefaultToFilePath

ToFilePath holds the global path formatting function used by the blob stores. Applications may override this to control file placement and partitioning.

type TransactionLog 

type TransactionLog struct {
	// contains filtered or unexported fields
}

TransactionLog writes per-transaction logs to local storage and supports scanning and cleanup of aged records. It also coordinates with a priority log for replication scenarios.

func NewTransactionLog 

func NewTransactionLog(cache sop.L2Cache, rt *replicationTracker) *TransactionLog

NewTransactionLog constructs a TransactionLog bound to the provided cache and replication tracker.

func (*TransactionLog) Add 

func (tl *TransactionLog) Add(ctx context.Context, tid sop.UUID, commitFunction int, payload []byte) error

Add appends a commit record with payload to this transaction's log file, creating the file if needed.

func (TransactionLog) Get 

func (l TransactionLog) Get(ctx context.Context, tid sop.UUID) ([]sop.RegistryPayload[sop.Handle], error)

Get loads the priority log payload for a transaction, if present.

func (TransactionLog) GetBatch 

func (l TransactionLog) GetBatch(ctx context.Context, batchSize int) ([]sop.KeyValuePair[sop.UUID, []sop.RegistryPayload[sop.Handle]], error)

GetBatch returns up to batchSize oldest priority log entries ready for processing. Entries are considered ready when their last-modified time is older than priorityLogMinAgeInMin from the current hour (capped to the hour).

func (*TransactionLog) GetOne 

func (tl *TransactionLog) GetOne(ctx context.Context) (sop.UUID, string, []sop.KeyValuePair[int, []byte], error)

GetOne claims one expired transaction hour bucket and returns a tid and its records for cleanup processing.

func (*TransactionLog) GetOneOfHour 

func (tl *TransactionLog) GetOneOfHour(ctx context.Context, hour string) (sop.UUID, []sop.KeyValuePair[int, []byte], error)

GetOneOfHour returns a tid and records for the specified hour if the bucket is within the TTL window.

func (TransactionLog) IsEnabled 

func (l TransactionLog) IsEnabled() bool

IsEnabled reports whether priority logging is enabled.

func (TransactionLog) LogCommitChanges 

func (l TransactionLog) LogCommitChanges(ctx context.Context, stores []sop.StoreInfo, newRootNodesHandles, addedNodesHandles, updatedNodesHandles, removedNodesHandles []sop.RegistryPayload[sop.Handle]) error

LogCommitChanges persists commit-change metadata used when reinstating failed drives.

func (*TransactionLog) NewUUID 

func (tl *TransactionLog) NewUUID() sop.UUID

NewUUID returns a new unique identifier for correlating log files.

func (*TransactionLog) PriorityLog 

func (tl *TransactionLog) PriorityLog() sop.TransactionPriorityLog

PriorityLog returns the FS-backed priority log for commit-change logging.

func (TransactionLog) ProcessNewer 

func (l TransactionLog) ProcessNewer(ctx context.Context, processor func(tid sop.UUID, payload []sop.RegistryPayload[sop.Handle]) error) error

ProcessNewer iterates over all priority logs newer than 5 mins and invokes the processor callback for each.

func (*TransactionLog) Remove 

func (tl *TransactionLog) Remove(ctx context.Context, tid sop.UUID) error

Remove deletes the log file for the specified transaction ID.

Source Files

View all Source files

Directories

Path Synopsis
The decoder reverses the process done by "encoder.go"
 The decoder reverses the process done by "encoder.go"

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.