Guarded Suspension Pattern #

// ANTI-PATTERN: busy waiting — CPU fully consumed just to wait
for len(queue) == 0 {
    // spin loop — wasting CPU cycles for nothing
    time.Sleep(1 * time.Millisecond) // even with a sleep, this is still polling
}
process(queue[0])

// CORRECT: guarded suspension — CPU free for other goroutines
mu.Lock()
for len(queue) == 0 {
    cond.Wait() // release the mutex, suspend, wait for a signal — no CPU wasted
}
item := queue[0]
mu.Unlock()
process(item)

package main

import (
	"fmt"
	"sync"
	"time"
)

// BoundedQueue is a queue with a maximum capacity
// Producers wait when full, consumers wait when empty
type BoundedQueue struct {
	items    []int
	capacity int
	mu       sync.Mutex
	notEmpty *sync.Cond // signal: there is a new item (for consumers)
	notFull  *sync.Cond // signal: there is a free slot (for producers)
}

func NewBoundedQueue(capacity int) *BoundedQueue {
	q := &BoundedQueue{
		items:    make([]int, 0, capacity),
		capacity: capacity,
	}
	q.notEmpty = sync.NewCond(&q.mu)
	q.notFull = sync.NewCond(&q.mu)
	return q
}

// Enqueue adds an item — waits if the queue is full
func (q *BoundedQueue) Enqueue(item int) {
	q.mu.Lock()
	defer q.mu.Unlock()

	// GUARDED SUSPENSION: wait until there is a free slot
	// MUST use 'for', not 'if' — the reason is explained below
	for len(q.items) >= q.capacity {
		fmt.Printf("[Producer] Queue full (%d/%d), waiting...\n", len(q.items), q.capacity)
		q.notFull.Wait() // release the mutex, suspend, wait for the notFull signal
	}

	q.items = append(q.items, item)
	fmt.Printf("[Producer] Enqueue %d (size: %d/%d)\n", item, len(q.items), q.capacity)
	q.notEmpty.Signal() // wake one waiting consumer
}

// Dequeue takes an item — waits if the queue is empty
func (q *BoundedQueue) Dequeue() int {
	q.mu.Lock()
	defer q.mu.Unlock()

	// GUARDED SUSPENSION: wait until there is an item
	for len(q.items) == 0 {
		fmt.Println("[Consumer] Queue empty, waiting...")
		q.notEmpty.Wait() // release the mutex, suspend, wait for the notEmpty signal
	}

	item := q.items[0]
	q.items = q.items[1:]
	fmt.Printf("[Consumer] Dequeue %d (size: %d/%d)\n", item, len(q.items), q.capacity)
	q.notFull.Signal() // wake one waiting producer
	return item
}

// Size returns the current number of items
func (q *BoundedQueue) Size() int {
	q.mu.Lock()
	defer q.mu.Unlock()
	return len(q.items)
}

func main() {
	queue := NewBoundedQueue(3) // queue with capacity 3
	var wg sync.WaitGroup

	// Producer: produces 6 items (more than the capacity)
	wg.Add(1)
	go func() {
		defer wg.Done()
		for i := 1; i <= 6; i++ {
			queue.Enqueue(i)
			time.Sleep(100 * time.Millisecond)
		}
	}()

	// Consumer: processes slower than the producer
	wg.Add(1)
	go func() {
		defer wg.Done()
		for i := 0; i < 6; i++ {
			time.Sleep(300 * time.Millisecond) // consumer is slower
			item := queue.Dequeue()
			_ = item
		}
	}()

	wg.Wait()
	fmt.Println("All items processed")
}

// ANTI-PATTERN: using 'if' before Wait()
func (q *Queue) DequeueWrong() int {
	q.mu.Lock()
	defer q.mu.Unlock()

	if len(q.items) == 0 { // ✗ 'if' only checks once
		q.cond.Wait()
		// After waking up, continue directly — without re-checking!
		// Problem: the goroutine can be woken by a spurious wakeup
		// or by Broadcast() even though the condition is not satisfied
	}

	return q.items[0] // ✗ can panic if the queue is still empty!
}

// CORRECT: using 'for' before Wait()
func (q *Queue) DequeueCorrect() int {
	q.mu.Lock()
	defer q.mu.Unlock()

	for len(q.items) == 0 { // ✓ 'for' always re-checks after waking
		q.cond.Wait()
		// After waking up, the loop returns to the condition
		// If the condition is still unmet → Wait() again
	}

	return q.items[0] // ✓ guaranteed the queue is not empty here
}

// Signal: wake ONE waiting goroutine
// Use when only one goroutine can take advantage of the change
q.cond.Signal()

// Broadcast: wake ALL waiting goroutines
// Use when the change is relevant to every waiter
q.cond.Broadcast()

// Example of using Broadcast for shutdown
type WorkerPool struct {
	mu       sync.Mutex
	cond     *sync.Cond
	jobs     []Job
	shutdown bool
}

func (p *WorkerPool) Shutdown() {
	p.mu.Lock()
	p.shutdown = true
	p.mu.Unlock()
	p.cond.Broadcast() // wake ALL workers so each one can exit
}

func (p *WorkerPool) worker() {
	p.mu.Lock()
	defer p.mu.Unlock()

	for {
		// Guard condition: there is work OR shutdown was requested
		for len(p.jobs) == 0 && !p.shutdown {
			p.cond.Wait()
		}

		if p.shutdown && len(p.jobs) == 0 {
			return // exit cleanly
		}

		job := p.jobs[0]
		p.jobs = p.jobs[1:]
		p.mu.Unlock()   // release the lock while processing
		job.Execute()
		p.mu.Lock()     // re-acquire the lock for the next iteration
	}
}

package main

import (
	"context"
	"fmt"
	"time"
)

// GuardedQueue uses a channel as the guarded suspension mechanism
type GuardedQueue struct {
	ch chan int
}

func NewGuardedQueue(capacity int) *GuardedQueue {
	return &GuardedQueue{
		ch: make(chan int, capacity), // buffered = queue capacity
	}
}

// Enqueue sends an item — blocks if full (built-in guarded suspension)
func (q *GuardedQueue) Enqueue(ctx context.Context, item int) error {
	select {
	case q.ch <- item: // ✓ guarded suspension: wait if the channel is full
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// Dequeue takes an item — blocks if empty (built-in guarded suspension)
func (q *GuardedQueue) Dequeue(ctx context.Context) (int, error) {
	select {
	case item := <-q.ch: // ✓ guarded suspension: wait if the channel is empty
		return item, nil
	case <-ctx.Done():
		return 0, ctx.Err()
	}
}

// Close closes the queue — waiting consumers will receive the zero value
func (q *GuardedQueue) Close() {
	close(q.ch)
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	queue := NewGuardedQueue(3)
	done := make(chan struct{})

	// Producer
	go func() {
		for i := 1; i <= 8; i++ {
			if err := queue.Enqueue(ctx, i); err != nil {
				fmt.Println("[Producer] Context cancelled:", err)
				return
			}
			fmt.Printf("[Producer] Enqueued: %d\n", i)
			time.Sleep(150 * time.Millisecond)
		}
		queue.Close()
	}()

	// Consumer
	go func() {
		defer close(done)
		for {
			item, err := queue.Dequeue(ctx)
			if err != nil {
				fmt.Println("[Consumer] Context cancelled:", err)
				return
			}
			fmt.Printf("[Consumer] Processing: %d\n", item)
			time.Sleep(400 * time.Millisecond)
		}
	}()

	<-done
}

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
)

// Connection represents a database connection
type Connection struct {
	id   int
	busy bool
}

func (c *Connection) Execute(query string) string {
	time.Sleep(200 * time.Millisecond) // simulated query
	return fmt.Sprintf("result from conn-%d: %s", c.id, query)
}

// ConnectionPool is a pool with guarded suspension
type ConnectionPool struct {
	connections []*Connection
	available   []*Connection
	mu          sync.Mutex
	cond        *sync.Cond
}

func NewConnectionPool(size int) *ConnectionPool {
	pool := &ConnectionPool{}
	pool.cond = sync.NewCond(&pool.mu)

	for i := 0; i < size; i++ {
		conn := &Connection{id: i + 1}
		pool.connections = append(pool.connections, conn)
		pool.available = append(pool.available, conn)
	}
	return pool
}

// Acquire gets a connection — waits if all are in use
func (p *ConnectionPool) Acquire(ctx context.Context) (*Connection, error) {
	p.mu.Lock()
	defer p.mu.Unlock()

	// GUARDED SUSPENSION: wait until a connection is available
	for len(p.available) == 0 {
		// Check the context inside the loop — no infinite wait if ctx is cancelled
		select {
		case <-ctx.Done():
			return nil, errors.New("timeout waiting for connection")
		default:
		}
		fmt.Println("[Pool] All connections busy, waiting...")
		p.cond.Wait()
	}

	// Take the first available connection
	conn := p.available[0]
	p.available = p.available[1:]
	conn.busy = true
	fmt.Printf("[Pool] Connection %d handed out (remaining: %d)\n", conn.id, len(p.available))
	return conn, nil
}

// Release returns a connection to the pool and wakes a waiter
func (p *ConnectionPool) Release(conn *Connection) {
	p.mu.Lock()
	defer p.mu.Unlock()

	conn.busy = false
	p.available = append(p.available, conn)
	fmt.Printf("[Pool] Connection %d returned (available: %d)\n", conn.id, len(p.available))
	p.cond.Signal() // wake one waiting goroutine
}

func main() {
	pool := NewConnectionPool(2) // only 2 connections

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var wg sync.WaitGroup
	queries := []string{"SELECT users", "SELECT orders", "SELECT products", "SELECT inventory"}

	for i, query := range queries {
		wg.Add(1)
		go func(id int, q string) {
			defer wg.Done()

			conn, err := pool.Acquire(ctx)
			if err != nil {
				fmt.Printf("[Worker %d] Failed to get a connection: %v\n", id, err)
				return
			}
			defer pool.Release(conn)

			result := conn.Execute(q)
			fmt.Printf("[Worker %d] %s\n", id, result)
		}(i+1, query)
	}

	wg.Wait()
}

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

type TimedQueue struct {
	items []int
	mu    sync.Mutex
	cond  *sync.Cond
}

func NewTimedQueue() *TimedQueue {
	q := &TimedQueue{}
	q.cond = sync.NewCond(&q.mu)
	return q
}

func (q *TimedQueue) Enqueue(item int) {
	q.mu.Lock()
	defer q.mu.Unlock()
	q.items = append(q.items, item)
	q.cond.Signal()
}

// DequeueWithContext waits for an item with a deadline using a context
func (q *TimedQueue) DequeueWithContext(ctx context.Context) (int, error) {
	// Run a watcher goroutine to cancel Wait() when the context finishes
	done := make(chan struct{})
	go func() {
		select {
		case <-ctx.Done():
			q.cond.Broadcast() // wake all waiters so they can check ctx.Done()
		case <-done:
		}
	}()
	defer close(done)

	q.mu.Lock()
	defer q.mu.Unlock()

	for len(q.items) == 0 {
		// Check ctx after every wakeup
		select {
		case <-ctx.Done():
			return 0, ctx.Err()
		default:
		}
		q.cond.Wait()
	}

	item := q.items[0]
	q.items = q.items[1:]
	return item, nil
}

func main() {
	queue := NewTimedQueue()

	// Consumer with a 1-second timeout
	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
	defer cancel()

	go func() {
		time.Sleep(2 * time.Second) // too late — past the deadline
		queue.Enqueue(42)
	}()

	item, err := queue.DequeueWithContext(ctx)
	if err != nil {
		fmt.Println("Timeout:", err)
	} else {
		fmt.Println("Received:", item)
	}
}

// ✗ Busy waiting — wasted CPU
for len(queue) == 0 {
    time.Sleep(1 * time.Millisecond) // polling = CPU waste
}

// ✓ Guarded suspension — CPU free while waiting
mu.Lock()
for len(queue) == 0 {
    cond.Wait() // sleeps, does not poll
}
mu.Unlock()

// ✗ Using 'if' before Wait() — can panic on a spurious wakeup
func (q *Queue) BadDequeue() int {
    q.mu.Lock()
    defer q.mu.Unlock()
    if len(q.items) == 0 { // ✗ checks only once
        q.cond.Wait()
    }
    return q.items[0] // ✗ can panic!
}

// ✓ Always use 'for' before Wait()
func (q *Queue) GoodDequeue() int {
    q.mu.Lock()
    defer q.mu.Unlock()
    for len(q.items) == 0 { // ✓ re-checks after waking
        q.cond.Wait()
    }
    return q.items[0] // ✓ safe
}

// ✗ Forgetting to call Signal/Broadcast — waiters wait forever
func (q *Queue) BrokenEnqueue(item int) {
    q.mu.Lock()
    defer q.mu.Unlock()
    q.items = append(q.items, item)
    // ✗ no Signal() — consumers sleep forever
}

// ✓ Always Signal/Broadcast after the condition changes
func (q *Queue) CorrectEnqueue(item int) {
    q.mu.Lock()
    defer q.mu.Unlock()
    q.items = append(q.items, item)
    q.cond.Signal() // ✓ wake the waiting consumer
}

// ✗ Wait() outside the mutex — panic: sync: unlock of unlocked mutex
func (q *Queue) PanicDequeue() int {
    for len(q.items) == 0 {
        q.cond.Wait() // ✗ must be inside the mutex lock!
    }
    return q.items[0]
}

// ✓ Wait() must always be inside the mutex lock
func (q *Queue) SafeDequeue() int {
    q.mu.Lock()
    defer q.mu.Unlock()
    for len(q.items) == 0 {
        q.cond.Wait() // ✓ inside the lock — safe
    }
    return q.items[0]
}