Sync Atomic #

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

func main() {
    // atomic.Int64 — a counter safe for concurrent access
    var counter atomic.Int64

    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter.Add(1)
        }()
    }
    wg.Wait()
    fmt.Println("Counter:", counter.Load()) // always 1000

    // atomic.Bool — a boolean flag
    var shutdown atomic.Bool
    shutdown.Store(false)

    go func() {
        // Simulate a shutdown signal
        shutdown.Store(true)
    }()

    // Swap — set a new value, return the old value
    previous := shutdown.Swap(true)
    fmt.Println("Previous:", previous)

    // CompareAndSwap — only change if the current value matches
    succeeded := shutdown.CompareAndSwap(true, false)
    fmt.Println("CAS succeeded:", succeeded)
    fmt.Println("Now:", shutdown.Load())

    // atomic.Pointer[T] — a safe pointer
    type Config struct {
        Host string
        Port int
    }

    var configPtr atomic.Pointer[Config]
    configPtr.Store(&Config{"localhost", 8080})

    cfg := configPtr.Load()
    fmt.Printf("Config: %s:%d\n", cfg.Host, cfg.Port)
}

// Integers
var i32 atomic.Int32
var i64 atomic.Int64
var u32 atomic.Uint32
var u64 atomic.Uint64
var uptr atomic.Uintptr

// Boolean
var b atomic.Bool

// Generic pointers
var p atomic.Pointer[MyStruct]

// The same methods for all:
// .Load() T          — read the current value
// .Store(val T)      — store a new value
// .Swap(val T) T     — store val, return the old value
// .CompareAndSwap(old, new T) bool — change only if == old
// .Add(delta T) T    — add delta, return the new value (numeric only)

import "sync/atomic"

// AddInt64 — add a delta atomically, return the new value
var counter int64
atomic.AddInt64(&counter, 1)   // counter++
atomic.AddInt64(&counter, -1)  // counter--
atomic.AddInt64(&counter, 10)  // counter += 10

// LoadInt64 — read the value atomically
val := atomic.LoadInt64(&counter)

// StoreInt64 — store a value atomically
atomic.StoreInt64(&counter, 0) // reset

// SwapInt64 — store and return the old value
previous := atomic.SwapInt64(&counter, 100)

// CompareAndSwapInt64 (CAS) — change only if equal to old
succeeded := atomic.CompareAndSwapInt64(&counter, 100, 200)
// counter changes to 200 only if it's currently == 100

// 32-bit versions
var c32 int32
atomic.AddInt32(&c32, 1)
atomic.LoadInt32(&c32)
atomic.StoreInt32(&c32, 0)
atomic.CompareAndSwapInt32(&c32, 0, 1)

// Unsigned
var uc uint64
atomic.AddUint64(&uc, 1)
atomic.LoadUint64(&uc)

import "sync/atomic"

// atomic.Value for hot-reloading configuration
type Config struct {
    Host    string
    Port    int
    Debug   bool
    MaxConn int
}

var globalConfig atomic.Value

func init() {
    // Store the initial config
    globalConfig.Store(&Config{
        Host:    "localhost",
        Port:    8080,
        Debug:   false,
        MaxConn: 100,
    })
}

// Read the config — very fast, no locks
func getConfig() *Config {
    return globalConfig.Load().(*Config)
}

// Update the config — usually from a background goroutine
func updateConfig(cfg *Config) {
    globalConfig.Store(cfg)
}

// Hot-reload from a file
func watchConfig(path string, ctx context.Context) {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            cfg, err := loadConfigFromFile(path)
            if err != nil {
                log.Printf("failed to reload config: %v", err)
                continue
            }
            updateConfig(cfg)
            log.Println("config reloaded successfully")
        }
    }
}

// Usage in a handler — very efficient
func apiHandler(w http.ResponseWriter, r *http.Request) {
    cfg := getConfig() // read the config without a lock
    if cfg.Debug {
        log.Printf("request: %s %s", r.Method, r.URL.Path)
    }
    // ...
}

var v atomic.Value

// RULE 1: the stored type must be consistent
v.Store(42)        // store an int
v.Store("string")  // PANIC! the type changed from int to string

// CORRECT: always store the same type
v.Store(42)
v.Store(100) // OK — both ints

// RULE 2: can't Store a nil interface
var cfg *Config = nil
v.Store(cfg) // PANIC! can't Store nil

// CORRECT: store a pointer to a zero value
v.Store(&Config{}) // OK — a valid pointer

// RULE 3: Load before the first Store returns nil
var v2 atomic.Value
result := v2.Load() // nil — never stored
if result != nil {
    cfg := result.(*Config)
    _ = cfg
}

// Swap — store and return the old value (atomic)
old := v.Swap(200)
fmt.Println(old) // 100 (the previous value)

// CompareAndSwap — change only if equal to old
succeeded := v.CompareAndSwap(200, 300)
fmt.Println(succeeded) // true

// Performance comparison (benchmark illustration):
// atomic.AddInt64: ~5 ns/op
// Mutex.Lock + value++ + Mutex.Unlock: ~25 ns/op
// (~5x faster for single operations)

// ANTI-PATTERN: atomic for operations that must be consistent together
var total int64
var count int64

// This is NOT safe! total and count can be inconsistent
// (another goroutine could read between the two Adds)
atomic.AddInt64(&total, price)
atomic.AddInt64(&count, 1)
// average := total/count could be inconsistent!

// CORRECT: Mutex for several values that must stay consistent together
var mu sync.Mutex
var total2 int64
var count2 int64

mu.Lock()
total2 += price
count2++
mu.Unlock()
// total2 and count2 are always consistent

// CORRECT: atomic for single independent values
var requestCount atomic.Int64
var errorCount atomic.Int64

// Both are independent — atomic is fine
requestCount.Add(1)
if err != nil {
    errorCount.Add(1)
}

// The CAS loop pattern — for operations needing atomic read-modify-write
func incrementSafe(counter *int64) {
    for {
        old := atomic.LoadInt64(counter)
        new := old + 1
        if atomic.CompareAndSwapInt64(counter, old, new) {
            return // succeeded
        }
        // failed — the value changed between Load and CAS — repeat
    }
}

// With the new types — much cleaner
var counter atomic.Int64
counter.Add(1) // Add already does the CAS loop internally

// CAS for a state machine — only allow valid transitions
type State int32

const (
    StateIdle    State = 0
    StateRunning State = 1
    StateStopped State = 2
)

var state atomic.Int32

func start() bool {
    // Can only start if currently Idle
    return state.CompareAndSwap(int32(StateIdle), int32(StateRunning))
}

func stop() bool {
    // Can only stop if currently Running
    return state.CompareAndSwap(int32(StateRunning), int32(StateStopped))
}

// Usage
if start() {
    fmt.Println("started successfully")
} else {
    fmt.Println("can't start — not in the Idle state")
}

// Metric counters for monitoring — frequently updated, frequently read
type Metrics struct {
    RequestTotal  atomic.Int64
    RequestError  atomic.Int64
    BytesReceived atomic.Int64
    BytesSent     atomic.Int64
    ActiveConns   atomic.Int64
}

var metrics Metrics

func handlerWithMetrics(w http.ResponseWriter, r *http.Request) {
    metrics.RequestTotal.Add(1)
    metrics.ActiveConns.Add(1)
    defer metrics.ActiveConns.Add(-1)

    // Count the received bytes
    body, err := io.ReadAll(r.Body)
    if err != nil {
        metrics.RequestError.Add(1)
        http.Error(w, "error", 500)
        return
    }
    metrics.BytesReceived.Add(int64(len(body)))

    // Process and send the response
    response := []byte(`{"status":"ok"}`)
    w.Write(response)
    metrics.BytesSent.Add(int64(len(response)))
}

// An endpoint to expose the metrics
func metricsHandler(w http.ResponseWriter, r *http.Request) {
    fmt.Fprintf(w, "request_total %d\n", metrics.RequestTotal.Load())
    fmt.Fprintf(w, "request_error %d\n", metrics.RequestError.Load())
    fmt.Fprintf(w, "bytes_received %d\n", metrics.BytesReceived.Load())
    fmt.Fprintf(w, "bytes_sent %d\n", metrics.BytesSent.Load())
    fmt.Fprintf(w, "active_connections %d\n", metrics.ActiveConns.Load())
}

// Approach 1: sync.Once — one-time initialization
var (
    dbOnce     sync.Once
    dbInstance *sql.DB
)

func GetDB() *sql.DB {
    dbOnce.Do(func() {
        db, err := sql.Open("postgres", os.Getenv("DATABASE_URL"))
        if err != nil {
            panic(err)
        }
        dbInstance = db
    })
    return dbInstance
}

// Approach 2: atomic.Pointer — updatable (for hot-reload)
var dbPtr atomic.Pointer[sql.DB]

func GetDBv2() *sql.DB {
    return dbPtr.Load()
}

func SetDB(db *sql.DB) {
    dbPtr.Store(db)
}

// Useful for testing — can swap the DB with a test DB
func TestHandler(t *testing.T) {
    testDB := setupTestDB(t)
    SetDB(testDB)
    defer SetDB(productionDB()) // restore after the test
    // ...
}

// A simple rate limiter using atomic
type RateLimiter struct {
    limit   int64
    window  time.Duration
    count   atomic.Int64
    resetAt atomic.Int64 // Unix nano
}

func NewRateLimiter(limit int, window time.Duration) *RateLimiter {
    rl := &RateLimiter{
        limit:  int64(limit),
        window: window,
    }
    rl.resetAt.Store(time.Now().Add(window).UnixNano())
    return rl
}

func (rl *RateLimiter) Allow() bool {
    now := time.Now().UnixNano()
    resetAt := rl.resetAt.Load()

    // Reset the window if it has passed
    if now >= resetAt {
        newResetAt := time.Now().Add(rl.window).UnixNano()
        if rl.resetAt.CompareAndSwap(resetAt, newResetAt) {
            rl.count.Store(0) // reset the count
        }
    }

    // Increment and check the limit
    current := rl.count.Add(1)
    return current <= rl.limit
}

// Usage
limiter := NewRateLimiter(100, time.Second) // 100 req/second

func rateLimitedHandler(w http.ResponseWriter, r *http.Request) {
    if !limiter.Allow() {
        http.Error(w, "rate limit exceeded", http.StatusTooManyRequests)
        return
    }
    // process the request
}

// A simple cache with atomic — no lock needed when reading
type AtomicCache[K comparable, V any] struct {
    mu   sync.Mutex      // only for writes
    data atomic.Pointer[map[K]V]
}

func NewAtomicCache[K comparable, V any]() *AtomicCache[K, V] {
    c := &AtomicCache[K, V]{}
    m := make(map[K]V)
    c.data.Store(&m)
    return c
}

// Get — no lock needed, very fast
func (c *AtomicCache[K, V]) Get(key K) (V, bool) {
    m := c.data.Load()
    v, ok := (*m)[key]
    return v, ok
}

// Set — needs a lock for writes, but creates a new copy so Get stays lock-free
func (c *AtomicCache[K, V]) Set(key K, val V) {
    c.mu.Lock()
    defer c.mu.Unlock()

    // Make a copy of the existing map
    old := c.data.Load()
    newMap := make(map[K]V, len(*old)+1)
    for k, v := range *old {
        newMap[k] = v
    }
    newMap[key] = val

    // Atomic swap — readers in flight get either the old or the new map,
    // but both are consistent (no half-updated map)
    c.data.Store(&newMap)
}

// Usage
cache := NewAtomicCache[string, *User]()

// Thousands of goroutines can Get simultaneously without locks
go func() {
    user, ok := cache.Get("budi")
    if ok {
        fmt.Println(user.Name)
    }
}()

// Occasional writes
cache.Set("budi", &User{Name: "Budi"})

// ANTI-PATTERN: two atomic ops that should be one unit
var success atomic.Int64
var failure atomic.Int64

// This is NOT atomic as one unit!
// Another goroutine could read success and failure between these two operations
success.Add(1)
// ← another goroutine could snapshot here: success+1, failure+0 (inconsistent!)
failure.Add(-1)

// CORRECT: use a Mutex for operations that must be consistent
var mu sync.Mutex
var success2, failure2 int64

mu.Lock()
success2++
failure2--
mu.Unlock()

// ANTI-PATTERN: attempting a wrong lock-free implementation
var flag atomic.Bool
var data []byte

// Goroutine A:
flag.Store(true)
data = append(data, 1) // this is NOT guaranteed to be visible after flag.Store!

// Goroutine B:
if flag.Load() {
    fmt.Println(data) // might see empty data!
}
// Memory ordering is more complex than this — use a channel or Mutex!