sync/atomic: Actually call the load function

Locker is an interface type in the stdlib sync package.

README.md

@@ -1,7 +1,7 @@
-# goutil
+# go-util
 
-[![Go Reference](https://pkg.go.dev/badge/git.tebibyte.media/sashakoshka/goutil.svg)](https://pkg.go.dev/git.tebibyte.media/sashakoshka/goutil)
+[![Go Reference](https://pkg.go.dev/badge/git.tebibyte.media/sashakoshka/go-util.svg)](https://pkg.go.dev/git.tebibyte.media/sashakoshka/go-util)
 
-Goutil provides extensions to the Go standard library. This repository mimics
-its package/directory structure, prefixing all package names with 'u' to
+Util provides extensions to the Go standard library. This repository mimics its
+package/directory structure, prefixing all package names with 'u' to
 differentiate them from their standard library counterparts.

container/optional.go

@@ -1,31 +1,31 @@
 package ucontainer
 
-// Optional is an optional value.
+// Optional can either hold a value, or nothing.
 type Optional[T any] struct {
 	value T
 	exists bool
 }
 
-// Value returns the value and true if the value exists. If not, it returns the
-// last set value and false.
-func (this *Optional[T]) Value () (T, bool) {
-	return this.value, this.exists
+// O creates a new optional with the specified value.
+func O[T any] (value T) Optional[T] {
+	return Optional[T] {
+		value: value,
+		exists: true,
+	}
 }
 
-// Set sets the value.
-func (this *Optional[T]) Set (value T) {
-	this.value = value
-	this.exists = true
-}
-
-// Unset unsets the value.
-func (this *Optional[T]) Unset () {
-	var zero T
-	this.value = zero
-	this.exists = false
+// Void returns an optional with no value.
+func Void[T any] () Optional[T] {
+	return Optional[T] { }
 }
 
 // Exists returns if the value is currently set.
-func (this *Optional[T]) Exists () bool {
-	return this.exists
+func (optional Optional[T]) Exists () bool {
+	return optional.exists
+}
+
+// Value returns the value and true if the value exists. If not, it returns the
+// zero value and false.
+func (optional Optional[T]) Value () (T, bool) {
+	return optional.value, optional.exists
 }

container/set.go

@@ -3,6 +3,15 @@ package ucontainer
 // Set is a set of unique items, built on top of map.
 type Set[T comparable] map[T] struct { }
 
+// NewSet creates a new set that contains all specified items.
+func NewSet[T comparable] (items ...T) Set[T] {
+	set := make(Set[T])
+	for _, item := range items {
+		set.Add(item)
+	}
+	return set
+}
+
 // Empty returns true if there are no items in the set.
 func (set Set[T]) Empty () bool {
 	return set == nil || len(set) == 0

go.mod

@@ -1,3 +1,3 @@
-module git.tebibyte.media/sashakoshka/goutil
+module git.tebibyte.media/sashakoshka/go-util
 
-go 1.20.0
+go 1.22.0

image/color/color_test.go

@@ -1,6 +1,23 @@
 package ucolor
 
 import "testing"
+import "image/color"
+
+func TestTransparent (test *testing.T) {
+	if Transparent(color.NRGBA { A: 255 }) {
+		test.Fatal("false positive")
+	}
+	if !Transparent(color.NRGBA { A: 0 }) {
+		test.Fatal("false negative")
+	}
+}
+
+func TestToRGBA (test *testing.T) {
+	rgba := ToRGBA(color.NRGBA { R: 123, G: 100, B: 23, A: 230 })
+	if rgba != (color.RGBA { R: 111, G: 90, B: 20, A: 230 }) {
+		test.Fatalf("wrong value: %v", rgba)
+	}
+}
 
 func TestPremultiply (test *testing.T) {
 	r, g, b, a := Premultiply(0xFFFF, 0xFFFF, 0xFFFF, 0x8888)

sync/atomic/generic.go

@@ -0,0 +1,38 @@
+package uatomic
+
+import "sync/atomic"
+
+// Atom is a generic wrapper for atomic.Value.
+type Atom[T comparable] struct {
+	value atomic.Value
+}
+
+type atomWrapper[T comparable] struct { value T }
+
+// CompareAndSwap executes the compare-and-swap operation for the Atom.
+func (atom *Atom[T]) CompareAndSwap(old, neww T) (swapped bool) {
+	return atom.value.CompareAndSwap(
+		atomWrapper[T] { value: old },
+		atomWrapper[T] { value: neww })
+}
+
+func (atom *Atom[T]) Load() T {
+	return castSafe[T](atom.value.Load())
+}
+
+func (atom *Atom[T]) Store(val T) {
+	atom.value.Store(val)
+}
+
+func (atom *Atom[T]) Swap(neww T) T {
+	return castSafe[T](atom.value.Swap(neww))
+}
+
+func castSafe[T comparable](value any) T {
+	var zero T
+	if value == nil {
+		return zero
+	} else {
+		return value.(T)
+	}
+}

sync/doc.go

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

sync/gate.go

@@ -0,0 +1,85 @@
+package usync
+
+import "sync/atomic"
+
+// Error defines errors that this package can produce
+type Error string; const (
+	ErrAlreadyClosed Error = "AlreadyClosed"
+)
+
+// Error fullfills the error interface.
+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    atomic.Value
+	bufferSize int
+}
+
+// NewGate creates a new gate with no buffer.
+func NewGate[T any] () Gate[T] {
+	return NewBufferedGate[T](0)
+}
+
+// NewBufferedGate creates a new gate with a buffer.
+func NewBufferedGate[T any] (buffer int) Gate[T] {
+	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) (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 {
+	return this.load()
+}
+
+// Close closes the gate, unblocking any send or receive operations.
+func (this *Gate[T]) Close () error {
+	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.channel.Store(make(chan T, this.bufferSize))
+	return nil
+}
+
+// Open returns whether the gate is open.
+func (this *Gate[T]) Open () bool {
+	return this.load() != nil
+}
+
+// Len returns the amount of items in the channel.
+func (this *Gate[T]) Len () int {
+	return len(this.load())
+}
+
+// Cap returns the amount of items the channel can hold.
+func (this *Gate[T]) Cap () int {
+	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/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()
+		defer done(&value)
+		value += 3
+	} ()
+	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()
+		defer done(&value)
+		value += 3
+	} ()
+	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()
+				} ()
+			}
+		} ()
+	}
+}

sync/select.go

@@ -0,0 +1,85 @@
+package usync
+
+import "slices"
+import "context"
+import "reflect"
+
+// ChannelRecv is a type constraint that includes all channel types that can be
+// recieved from.
+type ChannelRecv[T any] interface {
+	~chan T | ~<- chan T
+}
+
+// A type-safe wrapper around reflect.Select. Taken from:
+// https://stackoverflow.com/questions/19992334
+func Select[T any, C ChannelRecv[T]] (ctx context.Context, channels ...C) (int, T, bool) {
+	var zero T
+
+	// add all channels as select cases
+	cases := make([]reflect.SelectCase, len(channels) + 1)
+	for i, ch := range channels {
+		cases[i] = reflect.SelectCase {
+			Dir:  reflect.SelectRecv,
+			Chan: reflect.ValueOf(ch),
+		}
+	}
+
+	// add ctx.Done() as another select case to stop listening when the
+	// context is closed
+	cases[len(channels)] = reflect.SelectCase {
+		Dir:  reflect.SelectRecv,
+		Chan: reflect.ValueOf(ctx.Done()),
+	}
+	
+	// read from the channel
+	chosen, value, ok := reflect.Select(cases)
+	if !ok {
+		if ctx.Err() != nil {
+			return -1, zero, false
+		}
+		return chosen, zero, false
+	}
+
+	// cast return value
+	if ret, ok := value.Interface().(T); ok {
+		return chosen, ret, true
+	}
+	return chosen, zero, false
+	
+}
+
+// Combine returns a channel that continuously returns the result of the select
+// until all input sources are exhauste, or the context is canceled.
+func Combine[T any, C ChannelRecv[T]] (ctx context.Context, channels ...C) <- chan T {
+	channel := make(chan T)
+	// our silly slection routine
+	go func () {
+		for {
+			if len(channels) < 2 {
+				// only the context is left, stop everything
+				close(channel)
+				return
+			}
+			// read new value
+			chosen, value, ok := Select(ctx, channels...)
+			if ok {
+				// we have a value
+				channel <- value
+			} else {
+				// a channel has been closed and we need to do
+				// something about it
+				if chosen == len(channels) - 1 {
+					// the context has expired, stop
+					// everything
+					close(channel)
+					return
+				} else {
+					// a normal channel has closed, remove
+					// it from the list
+					channels = slices.Delete(channels, chosen, chosen + 1)
+				}
+			}
+		}
+	} ()
+	return channel
+}

sync/select_test.go

@@ -0,0 +1,24 @@
+package usync
+
+import "time"
+import "testing"
+import "context"
+
+func TestSelect (test *testing.T) {
+	// https://stackoverflow.com/questions/19992334
+	c1 := make(chan int)
+	c2 := make(chan int)
+	c3 := make(chan int)
+	chs := []chan int { c1, c2, c3 }
+	go func () {
+		time.Sleep(time.Second)
+		c2 <- 42
+	} ()
+	ctx, done := context.WithTimeout(context.Background(), 5 * time.Second)
+	defer done()
+
+	chosen, val, ok := Select(ctx, chs...)
+	if !ok          { test.Fatal("not ok")                  }
+	if 1  != chosen { test.Fatal("expected 1, got", chosen) }
+	if 42 != val    { test.Fatal("expected 42, got", val)   }
+}