mirror of
https://github.com/rqlite/rqlite.git
synced 2026-01-25 04:16:26 +00:00
Snapshot Sink sets MSRW name
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
## v9.3.8 (unreleased)
|
||||
### Implementation changes and bug fixes
|
||||
- [PR #2419](https://github.com/rqlite/rqlite/pull/2419): Improve MSRW Error structure.
|
||||
- [PR #2420](https://github.com/rqlite/rqlite/pull/2420): Snapshot Sink sets owner when taking MSRW.
|
||||
|
||||
## v9.3.7 (January 2nd 2026)
|
||||
### Implementation changes and bug fixes
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package rsync
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
)
|
||||
|
||||
@@ -15,14 +16,14 @@ func (e *ErrMRSWConflict) Error() string {
|
||||
}
|
||||
|
||||
// NewErrMRSWConflict creates a new ErrMRSWConflict with the given message.
|
||||
func NewErrMRSWConflict(msg string) error {
|
||||
return &ErrMRSWConflict{msg: msg}
|
||||
func NewErrMRSWConflict(m string) error {
|
||||
return &ErrMRSWConflict{msg: m}
|
||||
}
|
||||
|
||||
// MultiRSW is a simple concurrency control mechanism that allows
|
||||
// multiple readers or a single writer to execute a critical section at a time.
|
||||
type MultiRSW struct {
|
||||
writerActive bool
|
||||
owner string
|
||||
numReaders int
|
||||
mu sync.Mutex
|
||||
}
|
||||
@@ -36,8 +37,8 @@ func NewMultiRSW() *MultiRSW {
|
||||
func (r *MultiRSW) BeginRead() error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.writerActive {
|
||||
return NewErrMRSWConflict("MSRW conflict")
|
||||
if r.owner != "" {
|
||||
return NewErrMRSWConflict("MSRW conflict owner: " + r.owner)
|
||||
}
|
||||
r.numReaders++
|
||||
return nil
|
||||
@@ -54,13 +55,19 @@ func (r *MultiRSW) EndRead() {
|
||||
}
|
||||
|
||||
// BeginWrite attempts to enter the critical section as a writer.
|
||||
func (r *MultiRSW) BeginWrite() error {
|
||||
func (r *MultiRSW) BeginWrite(owner string) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.writerActive || r.numReaders > 0 {
|
||||
return NewErrMRSWConflict("MSRW conflict")
|
||||
if owner == "" {
|
||||
panic("owner cannot be empty")
|
||||
}
|
||||
r.writerActive = true
|
||||
if r.owner != "" {
|
||||
return NewErrMRSWConflict("MSRW conflict owner: " + r.owner)
|
||||
}
|
||||
if r.numReaders > 0 {
|
||||
return NewErrMRSWConflict(fmt.Sprintf("MSRW conflict %d readers active", r.numReaders))
|
||||
}
|
||||
r.owner = owner
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -68,25 +75,28 @@ func (r *MultiRSW) BeginWrite() error {
|
||||
func (r *MultiRSW) EndWrite() {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if !r.writerActive {
|
||||
if r.owner == "" {
|
||||
panic("write done received but no write is active")
|
||||
}
|
||||
r.writerActive = false
|
||||
r.owner = ""
|
||||
}
|
||||
|
||||
// UpgradeToWriter attempts to upgrade a read lock to a write lock. The
|
||||
// client must be the only reader in order to upgrade, and must already
|
||||
// be in a read lock.
|
||||
func (r *MultiRSW) UpgradeToWriter() error {
|
||||
func (r *MultiRSW) UpgradeToWriter(owner string) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.writerActive || r.numReaders > 1 {
|
||||
return NewErrMRSWConflict("MSRW conflict")
|
||||
if r.owner != "" {
|
||||
return NewErrMRSWConflict("MSRW conflict owner: " + r.owner)
|
||||
}
|
||||
if r.numReaders > 1 {
|
||||
return NewErrMRSWConflict(fmt.Sprintf("MSRW conflict %d readers active", r.numReaders))
|
||||
}
|
||||
if r.numReaders == 0 {
|
||||
panic("upgrade attempted with no readers")
|
||||
}
|
||||
r.writerActive = true
|
||||
r.owner = owner
|
||||
r.numReaders = 0
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -14,20 +14,20 @@ func Test_MultiRSW(t *testing.T) {
|
||||
r.EndRead()
|
||||
|
||||
// Test successful write lock
|
||||
if err := r.BeginWrite(); err != nil {
|
||||
if err := r.BeginWrite("owner1"); err != nil {
|
||||
t.Fatalf("Failed to acquire write lock: %v", err)
|
||||
}
|
||||
r.EndWrite()
|
||||
|
||||
// Test that a write blocks other writers and readers.
|
||||
err := r.BeginWrite()
|
||||
err := r.BeginWrite("owner2")
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to acquire write lock in goroutine: %v", err)
|
||||
}
|
||||
if err := r.BeginRead(); err == nil {
|
||||
t.Fatalf("Expected error when reading during active write, got none")
|
||||
}
|
||||
if err := r.BeginWrite(); err == nil {
|
||||
if err := r.BeginWrite("owner3"); err == nil {
|
||||
t.Fatalf("Expected error when writing during active write, got none")
|
||||
}
|
||||
r.EndWrite()
|
||||
@@ -37,7 +37,7 @@ func Test_MultiRSW(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to acquire read lock in goroutine: %v", err)
|
||||
}
|
||||
if err := r.BeginWrite(); err == nil {
|
||||
if err := r.BeginWrite("owner4"); err == nil {
|
||||
t.Fatalf("Expected error when writing during active read, got none")
|
||||
}
|
||||
r.EndRead()
|
||||
@@ -61,7 +61,7 @@ func Test_MultiRSW_Upgrade(t *testing.T) {
|
||||
if err := r.BeginRead(); err != nil {
|
||||
t.Fatalf("Failed to acquire read lock: %v", err)
|
||||
}
|
||||
if err := r.UpgradeToWriter(); err != nil {
|
||||
if err := r.UpgradeToWriter("owner11"); err != nil {
|
||||
t.Fatalf("Failed to upgrade to write lock: %v", err)
|
||||
}
|
||||
r.EndWrite()
|
||||
@@ -73,7 +73,7 @@ func Test_MultiRSW_Upgrade(t *testing.T) {
|
||||
if err := r.BeginRead(); err != nil {
|
||||
t.Fatalf("Failed to acquire read lock: %v", err)
|
||||
}
|
||||
if err := r.UpgradeToWriter(); err == nil {
|
||||
if err := r.UpgradeToWriter("owner5"); err == nil {
|
||||
t.Fatalf("Expected error when upgrading with multiple readers, got none")
|
||||
}
|
||||
r.EndRead()
|
||||
@@ -83,19 +83,19 @@ func Test_MultiRSW_Upgrade(t *testing.T) {
|
||||
if err := r.BeginRead(); err != nil {
|
||||
t.Fatalf("Failed to acquire read lock: %v", err)
|
||||
}
|
||||
if err := r.UpgradeToWriter(); err != nil {
|
||||
if err := r.UpgradeToWriter("owner6"); err != nil {
|
||||
t.Fatalf("Failed to upgrade to write lock: %v", err)
|
||||
}
|
||||
if err := r.UpgradeToWriter(); err == nil {
|
||||
if err := r.UpgradeToWriter("owner7"); err == nil {
|
||||
t.Fatalf("Expected error when double-ugrading, got none")
|
||||
}
|
||||
r.EndWrite()
|
||||
|
||||
// Test that upgrades are blocked by other writers
|
||||
if err := r.BeginWrite(); err != nil {
|
||||
if err := r.BeginWrite("owner8"); err != nil {
|
||||
t.Fatalf("Failed to acquire write lock: %v", err)
|
||||
}
|
||||
if err := r.UpgradeToWriter(); err == nil {
|
||||
if err := r.UpgradeToWriter("owner9"); err == nil {
|
||||
t.Fatalf("Expected error when upgrading with an active writer, got none")
|
||||
}
|
||||
r.EndWrite()
|
||||
|
||||
@@ -171,7 +171,7 @@ func NewStore(dir string) (*Store, error) {
|
||||
// be a problem, since snapshots are taken infrequently in one at a time.
|
||||
func (s *Store) Create(version raft.SnapshotVersion, index, term uint64, configuration raft.Configuration,
|
||||
configurationIndex uint64, trans raft.Transport) (retSink raft.SnapshotSink, retErr error) {
|
||||
if err := s.mrsw.BeginWrite(); err != nil {
|
||||
if err := s.mrsw.BeginWrite(fmt.Sprintf("snapshot-create-sink:%s", snapshotName(term, index))); err != nil {
|
||||
stats.Add(snapshotCreateMRSWFail, 1)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user