segment

package
v1.13.0 Latest Latest
Warning

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

 Go to latest Published: Sep 8, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// Header layout (fixed 18 bytes + key)
	//  0..7  : uint64 value length
	//  8..11 : uint32 key length
	// 12..15 : uint32 CRC32 checksum of value bytes
	// 16..17 : uint16 header format version
	// 18..N  : key bytes
	ValueHeaderSize = 18

	// Footer layout (20 bytes)
	//  0..5  : ASCII magic "SEGEOF"
	//  6..7  : ASCII version "01", "02", ...
	//  8..11 : uint32 entries in segment
	// 12..19 : uint64 total value bytes
	SegmentFooterMagicPrefix = "SEGEOF" // First 6 bytes – static prefix
	SegmentFooterSize        = 20

	// Segment versioning constants.
	CurrentSegmentVersion = 1 // Increment when format changes

	// Current header version – increment when ValueHeader layout changes.
	CurrentValueHeaderVersion = 1
)

Constants defining the on-disk segment format.

Variables

This section is empty.

Functions

func BuildSegmentFooterWithVersion 

func BuildSegmentFooterWithVersion(version int, entries uint32, dataBytes int64) []byte

BuildSegmentFooterWithVersion builds a segment footer that encodes the given version, entry count and total dataBytes. The returned slice is owned by the caller.

func BuildValueHeader 

func BuildValueHeader(key string, valueLen int64, checksum uint32, version uint16) []byte

BuildValueHeader returns a header buffer for the given key and value length. The caller owns the returned slice.

func CalculateValueHeaderSize 

func CalculateValueHeaderSize(key string) int64

CalculateValueHeaderSize calculates the size of the header for a given key.

func ParseSegmentFooter 

func ParseSegmentFooter(footer []byte) (version int, entries uint32, dataBytes int64, ok bool)

ParseSegmentFooter parses a 20-byte segment footer and returns the encoded version, entry count and total dataBytes. ok will be false when the footer does not start with the expected magic prefix.

func ReadValueHeader 

func ReadValueHeader(f *os.File) (valueLen int64, headerSize int64, keyLen int64, version uint16, checksum uint32, err error)

ReadValueHeader parses the header at the beginning of a segment/raw file and returns the value length, total header size and key length. It expects the file cursor to be at the beginning of the file or supports random access via Pread.

func ReadValueHeaderAt 

func ReadValueHeaderAt(f *os.File, offset int64) (valueLen int64, headerSize int64, keyLen int64, version uint16, checksum uint32, err error)

ReadValueHeaderAt parses the header at a specific offset in a segment file and returns the value length, total header size and key length. This function uses pread to read from the specified offset without modifying the file's current position, making it safe for concurrent reads.

func UpdateValueHeader 

func UpdateValueHeader(header []byte, valueLen int64, checksum uint32) []byte

UpdateValueHeader updates both value length and checksum in the supplied header slice.

Types

type EntryInfo 

type EntryInfo struct {
	Key         string
	Offset      int64  // Offset in the segment file where this entry starts
	HeaderSize  int64  // Size of the header (including key)
	ValueLength int64  // Length of the value
	Checksum    uint32 // CRC32 checksum of the value
	Version     uint16 // Header format version
}

EntryInfo contains information about a single entry in a segment

type Iterator 

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

Iterator provides sequential access to entries in a segment

func (*Iterator) CurrentPosition 

func (it *Iterator) CurrentPosition() int64

CurrentPosition returns the current position in the segment file

func (*Iterator) LastError 

func (it *Iterator) LastError() error

LastError returns the last error encountered during iteration

func (*Iterator) Next 

func (it *Iterator) Next() (*EntryInfo, error)

Next advances the iterator to the next entry and returns its information Returns io.EOF when there are no more entries

func (*Iterator) Reset 

func (it *Iterator) Reset()

Reset resets the iterator to the beginning of the segment

type Manager 

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

Manager manages the segments on disk.

func NewManager 

func NewManager(basePath string, segmentSize int64) (*Manager, error)

NewManager creates a new segment manager

func (*Manager) AcquireOpenSegmentWithReservation 

func (sm *Manager) AcquireOpenSegmentWithReservation(callerID string, needed int64) (*Segment, error)

AcquireOpenSegmentWithReservation returns an open segment reserved for the caller The callerID should be unique per goroutine/thread (e.g., "compactor", "recompactor-1") The segment will be reserved exclusively for this caller until released

func (*Manager) Close 

func (sm *Manager) Close()

Close closes all segment files

func (*Manager) FinalizeSegment 

func (sm *Manager) FinalizeSegment(seg *Segment) error

FinalizeSegment writes a footer to the segment file and closes it so that no further writes are possible.

func (*Manager) GetFragmentationRatio 

func (sm *Manager) GetFragmentationRatio(segmentPath string, deletedBytes int64) float64

GetFragmentationRatio calculates the fragmentation ratio for a segment Returns the ratio of dead space to total segment size (0.0 to 1.0)

func (*Manager) GetOpenSegments 

func (sm *Manager) GetOpenSegments() []*Segment

GetOpenSegments returns all currently open segments (for testing/debugging)

func (*Manager) GetSegmentByPath 

func (sm *Manager) GetSegmentByPath(path string) *Segment

GetSegmentByPath returns a segment by its path

func (*Manager) GetSegmentCount 

func (sm *Manager) GetSegmentCount() int

GetSegmentCount returns the number of segments currently managed.

func (*Manager) GetSegments 

func (sm *Manager) GetSegments() []*Segment

GetSegments returns a copy of the current segments slice for testing/inspection. The returned segments are safe to read but should not be modified.

func (*Manager) IsSegmentFragmented 

func (sm *Manager) IsSegmentFragmented(segmentPath string, deletedBytes int64, threshold float64) bool

IsSegmentFragmented checks if a segment exceeds the fragmentation threshold

func (*Manager) ReadEntry 

func (sm *Manager) ReadEntry(userKey string, segPath string, offset, length int64) (io.ReadCloser, error)

ReadEntry returns an io.ReadCloser over a slice of a segment file.

func (*Manager) RefreshMetrics added in v1.8.0 

func (sm *Manager) RefreshMetrics()

RefreshMetrics publishes the current segment count and total segment size to the gauges. Called periodically (e.g. from the cleaner tick) so the segment gauges track live contents rather than only the value at startup.

func (*Manager) RegisterSegment 

func (sm *Manager) RegisterSegment(path string, entries uint32, bytes int64)

RegisterSegment allows helper code to add new segments without poking into internal maps externally.

func (*Manager) ReleaseAllSegments 

func (sm *Manager) ReleaseAllSegments(callerID string) error

ReleaseAllSegments releases all segments reserved by the given caller. Snapshot the open set under sm.mu and release it BEFORE calling seg.Release (which takes seg.mu): holding sm.mu while acquiring a segment lock would block for the duration of a rate-limited WriteEntry copy and — via Go's writer-preferring RWMutex — convoy Manager.ReadEntry behind it, the same hazard fixed in AcquireOpenSegmentWithReservation.

func (*Manager) RemoveSegment 

func (sm *Manager) RemoveSegment(segmentPath string) *Segment

RemoveSegment removes a segment from the manager's tracking This should be called when a segment is being deleted permanently Returns the removed segment if found, nil otherwise

type Segment 

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

Segment is a file on disk that contains key/value pairs.

func NewSegment 

func NewSegment(path string, entries uint32, dataBytes int64, size int64, maxSupportedSize int64) *Segment

NewSegment creates a new segment with the given path and size.

func NewSegmentWithReservation 

func NewSegmentWithReservation(path string, entries uint32, dataBytes int64, size int64, maxSupportedSize int64, callerID string) *Segment

NewSegmentWithReservation creates a new segment with the given path and size, atomically reserved for the caller. This ensures the segment is created with a reservation in place, preventing race conditions.

func (*Segment) File 

func (s *Segment) File() *os.File

File returns the underlying file (must be called while holding lock)

func (*Segment) Finalize 

func (s *Segment) Finalize() error

Finalize finalizes the segment by writing the footer and closing the file.

func (*Segment) GetDataBytes added in v1.11.0 

func (s *Segment) GetDataBytes() int64

GetDataBytes returns the total bytes of value payloads stored in the segment, excluding entry headers. It is restored from the footer when a finalized segment is loaded, so it is the payload-unit total to compare live payload bytes against.

func (*Segment) GetNumEntries 

func (s *Segment) GetNumEntries() uint32

GetNumEntries returns the number of entries in the segment.

func (*Segment) GetReservedBy 

func (s *Segment) GetReservedBy() string

GetReservedBy returns who has reserved this segment (empty string if not reserved)

func (*Segment) GetSize 

func (s *Segment) GetSize() int64

GetSize returns the current size of data written to the segment.

func (*Segment) GetSizeUnsafe 

func (s *Segment) GetSizeUnsafe() int64

GetSizeUnsafe returns the size without locking (must be called while holding lock)

func (*Segment) HasOpenFile 

func (s *Segment) HasOpenFile() bool

HasOpenFile returns true if the segment has an open file.

func (*Segment) IncrementDataBytes 

func (s *Segment) IncrementDataBytes(delta int64)

IncrementDataBytes increments the data bytes count (must be called while holding lock)

func (*Segment) IncrementEntries 

func (s *Segment) IncrementEntries()

IncrementEntries increments the entry count (must be called while holding lock)

func (*Segment) IncrementSize 

func (s *Segment) IncrementSize(delta int64)

IncrementSize increments the segment size (must be called while holding lock)

func (*Segment) IsReserved 

func (s *Segment) IsReserved() bool

IsReserved checks if the segment is reserved by anyone

func (*Segment) IsReservedBy 

func (s *Segment) IsReservedBy(callerID string) bool

IsReservedBy checks if the segment is reserved by the given caller

func (*Segment) Lock 

func (s *Segment) Lock()

Lock locks the segment for exclusive access

func (*Segment) NewIterator 

func (s *Segment) NewIterator(file *os.File) (*Iterator, error)

NewIterator creates a new iterator for the segment The caller is responsible for closing the iterator when done

func (*Segment) Path 

func (s *Segment) Path() string

Path returns the path of the segment.

func (*Segment) ReadEntry 

func (s *Segment) ReadEntry(key string, offset, length int64, fdCache *fd.FdCache) (io.ReadCloser, error)

ReadEntry reads a value from the segment. The caller is responsible for closing the reader when done.

func (*Segment) Release 

func (s *Segment) Release(callerID string) error

Release releases the reservation on the segment

func (*Segment) Remaining 

func (s *Segment) Remaining() int64

Remaining returns the remaining space in the segment.

func (*Segment) Reserve 

func (s *Segment) Reserve(callerID string) bool

Reserve attempts to reserve the segment for exclusive use by the caller Returns true if successful, false if already reserved or if segment is closed

func (*Segment) SetOpenFile 

func (s *Segment) SetOpenFile(file *os.File)

SetOpenFile sets the open file for the segment.

func (*Segment) Sync 

func (s *Segment) Sync() error

Sync syncs the segment to disk. This is a no-op if the segment is not open.

func (*Segment) Unlock 

func (s *Segment) Unlock()

Unlock unlocks the segment

func (*Segment) WriteEntry 

func (s *Segment) WriteEntry(key string, r io.Reader, vm *pb.ValueMessage) (int64, error)

WriteEntry writes an entry to a segment from an io.Reader

Source Files

View all Source files

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.