sync: Made doc comment on Gate.Close make more sense

Locker is an interface type in the stdlib sync package.

sync/doc.go

@@ -0,0 +1,2 @@
+// Package usync extends sync.
+package usync

sync/gate.go

@@ -1,6 +1,6 @@
 package usync
 
-import "sync"
+import "sync/atomic"
 
 // Error defines errors that this package can produce
 type Error string; const (
@@ -12,83 +12,74 @@ func (err Error) Error () string {
 	return string(err)
 }
 
+// if you know of a better way, i'd like to hear of it. otherwise, STFU!!!!
+
 // Gate wraps a channel and allows the receiver to abruptly stop receiving
 // messages without causing the sender to lock up.
 type Gate[T any] struct {
-	channel    chan T
-	lock       sync.RWMutex
-	open       bool
+	channel    atomic.Value
 	bufferSize int
 }
 
 // NewGate creates a new gate with no buffer.
 func NewGate[T any] () Gate[T] {
-	return Gate[T] {
-		channel: make(chan T),
-		open:    true,
-	}
+	return NewBufferedGate[T](0)
 }
 
 // NewBufferedGate creates a new gate with a buffer.
 func NewBufferedGate[T any] (buffer int) Gate[T] {
-	return Gate[T] {
-		channel:    make(chan T, buffer),
-		open:       true,
+	gate := Gate[T] {
 		bufferSize: buffer,
 	}
+	gate.channel.Store(make(chan T, buffer))
+	return gate
 }
 
 // Send sends and item to the channel, returning whether the item was sent.
-func (this *Gate[T]) Send (item T) bool {
-	this.lock.RLock()
-	defer this.lock.RUnlock()
-	if !this.open { return false }
-	this.channel <- item
+func (this *Gate[T]) Send (item T) (ok bool) {
+	channel := this.load()
+	if channel == nil { return false }
+	// if the channel send panics, "return true" never gets run so it just
+	// returns false after recovering
+	defer func() { recover() }()
+	channel <- item
 	return true
 }
 
 // Receive returns a receive-only channel that can be used to receive items.
 func (this *Gate[T]) Receive () <- chan T {
-	this.lock.RLock()
-	defer this.lock.RUnlock()
-	return this.channel
+	return this.load()
 }
 
-// Close closes the gate, drains all remaining messages, and closes the channel.
+// Close closes the gate, unblocking any send or receive operations.
 func (this *Gate[T]) Close () error {
-	this.lock.Lock()
-	if !this.open { return ErrAlreadyClosed }
-	this.open = false
-	this.lock.Unlock()
-	for len(this.channel) > 0 { <- this.channel }
-	close(this.channel)
+	channel := this.channel.Swap((chan T)(nil)).(chan T)
+	if channel == nil { return ErrAlreadyClosed }
+	close(channel)
 	return nil
 }
 
 // Reset re-opens the gate if it is closed, and creates a new channel.
 func (this *Gate[T]) Reset () error {
-	this.lock.Lock()
-	defer this.lock.Unlock()
-	this.open = true
-	if this.channel != nil { close(this.channel) }
-	this.channel = make(chan T, this.bufferSize)
+	this.channel.Store(make(chan T, this.bufferSize))
 	return nil
 }
 
 // Open returns whether the gate is open.
 func (this *Gate[T]) Open () bool {
-	this.lock.RLock()
-	defer this.lock.RUnlock()
-	return this.open
+	return this.load() != nil
 }
 
 // Len returns the amount of items in the channel.
 func (this *Gate[T]) Len () int {
-	return len(this.channel)
+	return len(this.load())
 }
 
 // Cap returns the amount of items the channel can hold.
 func (this *Gate[T]) Cap () int {
-	return cap(this.channel)
+	return cap(this.load())
 }
 
+func (this *Gate[T]) load() chan T {
+	return this.channel.Load().(chan T)
+}

sync/gate_test.go

@@ -0,0 +1,63 @@
+package usync
+
+import "time"
+import "errors"
+import "testing"
+
+func TestGate(test *testing.T) {
+	gate := NewGate[int]()
+	go func() {
+		for number := range 21 {
+			test.Log("send", number)
+			gate.Send(number)
+		}
+		test.Log("send done")
+	}()
+	for correct := range 21 {
+		got := <- gate.Receive()
+		test.Log("RECV", correct, got)
+		if correct != got {
+			test.Fatal("RECV not equal")
+		}
+	}
+	test.Log("RECV done")
+}
+
+func TestGateCloseEarly(test *testing.T) {
+	gate := NewGate[int]()
+	sendDone := make(chan struct { })
+	go func() {
+		for number := range 21 {
+			test.Log("send", number)
+			gate.Send(number)
+		}
+		test.Log("send done")
+		close(sendDone)
+	}()
+	for correct := range 15 {
+		got := <- gate.Receive()
+		test.Log("RECV", correct, got)
+		if correct != got {
+			test.Fatal("RECV not equal")
+		}
+	}
+	test.Log("RECV done, closing gate... DOOR STUCK!! PLS I BEG OF YOU!!")
+	gate.Close()
+	test.Log("RECV closed gate yay :)))")
+	timeout := time.NewTimer(time.Second * 4)
+	select {
+	case <- sendDone:
+	case <- timeout.C:
+		test.Fatal("timed out waiting for sender goroutine to stop")
+	}
+}
+
+func TestGateCloseIdempotent(test *testing.T) {
+	gate := NewGate[int]()
+	err := gate.Close()
+	if err != nil { test.Fatal(err) }
+	err = gate.Close()
+	if !errors.Is(err, ErrAlreadyClosed) {
+		test.Fatal("wrong error: ", err)
+	}
+}

sync/locker.go

@@ -1,36 +0,0 @@
-package usync
-
-import "sync"
-
-// Locker guards access to a value
-type Locker[T any] struct {
-	value T
-	mutex sync.Mutex
-}
-
-// NewLocker creates a new locker with the specified value.
-func NewLocker[T any] (value T) Locker[T] {
-	return Locker[T] {
-		value: value,
-	}
-}
-
-// Set sets the value of the locker.
-func (this *Locker[T]) Set (value T) {
-	this.mutex.Lock()
-	defer this.mutex.Unlock()
-	this.value = value
-}
-
-// Borrow borrows the value from the locker, and returns a function that must
-// immediately be deferred, like this:
-//
-//   value, done := locker.Borrow()
-//   defer done()
-//
-// From the time Borrow is called to the time the done function is called, it is
-// safe to access the locked object from within the current goroutine.
-func (this *Locker[T]) Borrow () (T, func ()) {
-	this.mutex.Lock()
-	return this.value, this.mutex.Unlock
-}

sync/monitor.go

@@ -0,0 +1,103 @@
+package usync
+
+import "sync"
+
+// Monitor guards access to a value. It must not be copied after first use.
+type Monitor[T any] struct {
+	value T
+	mutex sync.Mutex
+}
+
+// NewMonitor creates a new Monitor with the specified value.
+func NewMonitor[T any] (value T) Monitor[T] {
+	return Monitor[T] {
+		value: value,
+	}
+}
+
+// Set sets the value of the Monitor.
+func (this *Monitor[T]) Set (value T) {
+	this.mutex.Lock()
+	defer this.mutex.Unlock()
+	this.value = value
+}
+
+// Borrow borrows the value from the Monitor, and returns a function that must
+// immediately be deferred, like this:
+//
+//   value, done := monitor.Borrow()
+//   defer done()
+//
+// From the time Borrow is called to the time the done function is called, it is
+// safe to access the locked object from within the current goroutine.
+func (this *Monitor[T]) Borrow () (T, func ()) {
+	this.mutex.Lock()
+	return this.value, this.mutex.Unlock
+}
+
+// BorrowReturn is like borrow, but returns a "done" function that takes in an
+// updated value. The intended use of this function is like this:
+//
+//   value, done := monitor.BorrowReturn()
+//   defer done(value)
+func (this *Monitor[T]) BorrowReturn () (T, func (T)) {
+	this.mutex.Lock()
+	return this.value, func (value T) {
+		defer this.mutex.Unlock()
+		this.value = value
+	}
+}
+
+// RWMonitor guards separate read/write access to a value.
+type RWMonitor[T any] struct {
+	value T
+	mutex sync.RWMutex
+}
+
+// NewRWMonitor creates a new Monitor with the specified value. It must not be
+// copied after first use.
+func NewRWMonitor[T any] (value T) RWMonitor[T] {
+	return RWMonitor[T] {
+		value: value,
+	}
+}
+
+// Set sets the value of the Monitor.
+func (this *RWMonitor[T]) Set (value T) {
+	this.mutex.Lock()
+	defer this.mutex.Unlock()
+	this.value = value
+}
+
+// Borrow borrows the value from the Monitor for write access, and returns a
+// function that must immediately be deferred, like this:
+//
+//   value, done := monitor.Borrow()
+//   defer done()
+//
+// From the time Borrow is called to the time the done function is called, it is
+// safe to access the locked object from within the current goroutine.
+func (this *RWMonitor[T]) Borrow () (T, func ()) {
+	this.mutex.Lock()
+	return this.value, this.mutex.Unlock
+}
+
+// BorrowReturn is like borrow, but returns a "done" function that takes in an
+// updated value. The intended use of this function is like this:
+//
+//   value, done := monitor.BorrowReturn()
+//   defer done(value)
+func (this *RWMonitor[T]) BorrowReturn () (T, func (T)) {
+	this.mutex.Lock()
+	return this.value, func (value T) {
+		defer this.mutex.Unlock()
+		this.value = value
+	}
+}
+
+// RBorrow is like Borrow, but returns the item for read access only. Do not
+// under any circumstances modify anything returned by this method.
+func (this *RWMonitor[T]) RBorrow () (T, func ()) {
+	this.mutex.Lock()
+	return this.value, this.mutex.Unlock
+}

sync/monitor_test.go

@@ -0,0 +1,90 @@
+package usync
+
+import "testing"
+import "math/rand"
+
+func TestMonitor (test *testing.T) {
+	mon := NewMonitor(9)
+	func () {
+		value, done := mon.Borrow()
+		defer done()
+		test.Log(value)
+		if value != 9 { test.Fatal("not equal") }
+	} ()
+	func () {
+		value, done := mon.BorrowReturn()
+		value += 3
+		defer done(value)
+	} ()
+	func () {
+		value, done := mon.Borrow()
+		defer done()
+		test.Log(value)
+		if value != 12 { test.Fatal("not equal") }
+	} ()
+	mon.Set(11)
+	func () {
+		value, done := mon.Borrow()
+		defer done()
+		test.Log(value)
+		if value != 11 { test.Fatal("not equal") }
+	} ()
+}
+
+func TestMonitorConcurrent (test *testing.T) {
+	mon := NewMonitor(map[int] int { })
+	for index := 0; index < 16; index ++ {
+		go func () {
+			for index := 0; index < 8192; index ++ {
+				func () {
+					value, done := mon.Borrow()
+					defer done()
+					value[rand.Int()] = rand.Int()
+				} ()
+			}
+		} ()
+	}
+}
+
+func TestRWMonitor (test *testing.T) {
+	mon := NewRWMonitor(9)
+	func () {
+		value, done := mon.Borrow()
+		defer done()
+		test.Log(value)
+		if value != 9 { test.Fatal("not equal") }
+	} ()
+	func () {
+		value, done := mon.BorrowReturn()
+		value += 3
+		defer done(value)
+	} ()
+	func () {
+		value, done := mon.RBorrow()
+		defer done()
+		test.Log(value)
+		if value != 12 { test.Fatal("not equal") }
+	} ()
+	mon.Set(11)
+	func () {
+		value, done := mon.RBorrow()
+		defer done()
+		test.Log(value)
+		if value != 11 { test.Fatal("not equal") }
+	} ()
+}
+
+func TestRWMonitorConcurrent (test *testing.T) {
+	mon := NewRWMonitor(map[int] int { })
+	for index := 0; index < 16; index ++ {
+		go func () {
+			for index := 0; index < 8192; index ++ {
+				func () {
+					value, done := mon.Borrow()
+					defer done()
+					value[rand.Int()] = rand.Int()
+				} ()
+			}
+		} ()
+	}
+}