Documentation
¶
Overview ¶
Package backend defines the storage abstraction that the ring buffer is built on top of, plus the implementations shmring ships with.
The ring buffer logic in the parent package never talks to OS shared memory directly: it only depends on the Storage interface. This keeps the platform-specific and IPC-specific concerns isolated to this package, so support for additional platforms or transports (POSIX shm, Windows file mappings, /dev/shm, RDMA, ...) can be added as new Storage implementations without touching the ring buffer algorithm.
ShmStorage here is Linux's own direct implementation: a POSIX shared-memory segment, mapped by hand via a file under /dev/shm rather than through github.com/hidez8891/shm (see shm.go, used on macOS/Windows). A /dev/shm-backed file is exactly what glibc's own shm_open is documented to do on Linux (a tmpfs-backed regular file named by the path), so this is equivalent cross-process shared memory, not an approximation of it -- and unlike hidez8891/shm's Memory type, it gives this package the raw mapped address a real OS wakeup primitive needs. See futex_linux.go: that's what lets ShmStorage implement WaiterStorage here, so blocking Write/Read on Linux block on a real futex instead of polling with a sleep-based backoff.
Index ¶
- Variables
- type AtomicStorage
- type MemStorage
- type ShmStorage
- func (s *ShmStorage) Close() error
- func (s *ShmStorage) ReadAt(p []byte, off int64) (int, error)
- func (s *ShmStorage) Size() int64
- func (s *ShmStorage) Wait(off int64, old uint32, timeout time.Duration)
- func (s *ShmStorage) Wake(off int64)
- func (s *ShmStorage) WriteAt(p []byte, off int64) (int, error)
- type Storage
- type WaiterStorage
Constants ¶
This section is empty.
Variables ¶
var ErrIncompleteSegment = errors.New("backend: shared memory segment is smaller than the requested mapping")
ErrIncompleteSegment is returned, wrapped, when a named shared-memory segment exists but is smaller than the mapping being asked for -- most often because its creator has not finished sizing it yet. It is a transient condition for a consumer that opens a segment concurrently with its producer, and the only way such a consumer can tell "come back in a moment" apart from a segment that will never be right.
Only the Linux backend reports it: it maps /dev/shm files itself and so can see their true length. The macOS/Windows backend goes through github.com/hidez8891/shm, which does not expose one.
Functions ¶
This section is empty.
Types ¶
type AtomicStorage ¶
type AtomicStorage interface {
Storage
LoadUint32(off int64) (uint32, error)
StoreUint32(off int64, v uint32) error
}
AtomicStorage is an optional capability a Storage may implement to provide real atomic 32-bit loads/stores for the ring buffer's head/tail/closed counters, at the offsets the ring buffer header defines for them (all 4-byte aligned).
ShmStorage and MemStorage don't implement it: OS shared memory is coherent across processes at the hardware level, so a plain aligned load/store is enough (see their docs). The js/wasm SharedArrayBuffer backend does implement it, using JavaScript's Atomics, because that's the web platform's actual cross-thread visibility guarantee -- an ordinary read/write to a SharedArrayBuffer from two different threads (e.g. a browser main thread and a Web Worker) is a data race under the JavaScript memory model, the same way it would be between two Go goroutines without synchronization.
type MemStorage ¶
type MemStorage struct {
// contains filtered or unexported fields
}
MemStorage is a Storage backed by a plain Go byte slice. It never leaves the process, so it's useful for unit tests, benchmarks, and for platforms where an OS shared-memory backend isn't available: the same ring buffer code path can run against it, with a producer and consumer goroutine sharing one MemStorage instead of two processes sharing shared memory.
Real OS shared memory (see ShmStorage) is coherent across processes at the hardware level, which is what lets the ring buffer's SPSC algorithm use plain aligned loads/stores for its head/tail counters instead of sync/atomic. That guarantee doesn't hold for two goroutines in the same process talking through an ordinary []byte: the Go memory model requires an explicit happens-before edge, or the compiler is free to reorder or cache accesses. MemStorage supplies that edge with a mutex around every ReadAt/WriteAt, so it is safe to share between goroutines even though it isn't lock-free.
func NewMemStorage ¶
func NewMemStorage(size int64) *MemStorage
NewMemStorage allocates a MemStorage of the given size.
func (*MemStorage) Close ¶
func (s *MemStorage) Close() error
Close implements Storage. It is a no-op.
type ShmStorage ¶
type ShmStorage struct {
// contains filtered or unexported fields
}
ShmStorage is a Storage backed by an OS shared-memory segment. It is what makes a ring buffer usable for cross-process communication: one process calls CreateShm, another opens the same named segment with OpenShm.
func CreateShm ¶
func CreateShm(name string, size int64) (*ShmStorage, error)
CreateShm creates a new named shared-memory segment of the given size and returns a Storage backed by it. The segment is removed from the OS when the returned Storage is closed.
size must fit in an int32, matching the limit the macOS/Windows build of ShmStorage imposes (see shm.go), so callers don't need platform- specific size logic.
func OpenShm ¶
func OpenShm(name string, size int64) (*ShmStorage, error)
OpenShm opens a shared-memory segment previously created with CreateShm. size must match the size the segment was created with.
A segment that exists but is not yet that long -- the state CreateShm leaves behind between creating the file and sizing it -- returns an error wrapping ErrIncompleteSegment rather than a Storage. A consumer racing a producer should treat that as "not ready yet" and retry; see mapShm for why it cannot be mapped and read anyway.
func (*ShmStorage) Close ¶
func (s *ShmStorage) Close() error
Close implements Storage: unmaps the segment and, on the creating side, removes the underlying /dev/shm file.
func (*ShmStorage) ReadAt ¶
func (s *ShmStorage) ReadAt(p []byte, off int64) (int, error)
ReadAt implements Storage.
func (*ShmStorage) Wait ¶ added in v0.4.0
func (s *ShmStorage) Wait(off int64, old uint32, timeout time.Duration)
Wait implements WaiterStorage using a real futex(2) FUTEX_WAIT on the shared word, so a blocking Write/Read parked here costs no CPU and wakes as soon as the other side calls Wake.
func (*ShmStorage) Wake ¶ added in v0.4.0
func (s *ShmStorage) Wake(off int64)
Wake implements WaiterStorage via futex(2) FUTEX_WAKE.
type Storage ¶
type Storage interface {
io.ReaderAt
io.WriterAt
// Size returns the total size in bytes of the storage region. It is
// constant for the lifetime of the Storage.
Size() int64
// Close releases resources associated with the storage. For
// process-local backends this is typically a no-op; for shared-memory
// backends it unmaps the segment and, for the creating side, removes
// the underlying OS object.
Close() error
}
Storage is a fixed-size, randomly addressable region of bytes shared between a producer and a consumer. It is the minimal capability the ring buffer needs from its underlying memory.
Implementations must be safe for concurrent use by one reader goroutine and one writer goroutine at the same time (but not by multiple readers or multiple writers), matching the single-producer/single-consumer contract of the ring buffer itself.
type WaiterStorage ¶ added in v0.4.0
type WaiterStorage interface {
Storage
Wait(off int64, old uint32, timeout time.Duration)
Wake(off int64)
}
WaiterStorage is an optional capability a Storage may implement to let a blocking Write/Read sleep on a real OS wakeup primitive tied to one of the ring buffer's header words (head/tail/closed), instead of polling those words with a sleep-based backoff.
This is the abstraction real cross-process wakeup primitives (a Linux futex on the shared word, for example) sit behind: Wait blocks the calling goroutine until the word at off no longer holds old, or until timeout elapses (timeout <= 0 means wait indefinitely); Wake wakes any goroutine -- in this process or another -- currently parked in Wait on off. Wait always returns eventually for some reason (real change, spurious wakeup, or timeout); it's the caller's job to re-read the word and decide what to do next, exactly as with a futex.
Implementing this is optional: a Storage that doesn't implement it gets the existing sleep-and-poll behavior automatically (see Writer/Reader), which remains correct, just busier while waiting.
Source Files