Go — Advanced

// Generic function
func Map[T, U any](s []T, f func(T) U) []U {
    result := make([]U, len(s))
    for i, v := range s {
        result[i] = f(v)
    }
    return result
}

func Filter[T any](s []T, pred func(T) bool) []T {
    var out []T
    for _, v := range s {
        if pred(v) { out = append(out, v) }
    }
    return out
}

// Generic type
type Stack[T any] struct{ items []T }
func (s *Stack[T]) Push(v T) { s.items = append(s.items, v) }
func (s *Stack[T]) Pop() (T, bool) {
    if len(s.items) == 0 { var zero T; return zero, false }
    v := s.items[len(s.items)-1]
    s.items = s.items[:len(s.items)-1]
    return v, true
}

import "reflect"

func printFields(v any) {
    rv := reflect.ValueOf(v)
    rt := reflect.TypeOf(v)

    if rv.Kind() == reflect.Ptr {
        rv = rv.Elem()
        rt = rt.Elem()
    }

    for i := 0; i < rv.NumField(); i++ {
        field := rt.Field(i)
        value := rv.Field(i)
        tag := field.Tag.Get("json")
        fmt.Printf("Field: %s, Tag: %s, Value: %v\n",
            field.Name, tag, value)
    }
}

type Config struct {
    Host string `json:"host"`
    Port int    `json:"port"`
}
printFields(Config{"localhost", 8080})

type Reader interface { Read([]byte) (int, error) }
type Writer interface { Write([]byte) (int, error) }
type Closer interface { Close() error }

// Composed interfaces
type ReadWriter interface {
    Reader
    Writer
}
type ReadWriteCloser interface {
    Reader
    Writer
    Closer
}

// Accept minimum required capability
func compress(rw io.ReadWriter) error {
    // reads uncompressed, writes compressed to the same conn
}

// Custom composed interface for a specific domain
type Repository interface {
    Querier
    Mutator
    io.Closer
}

// File: signal_unix.go
//go:build linux || darwin || freebsd

package main

import (
    "os/signal"
    "syscall"
)

func listenForSignals() {
    ch := make(chan os.Signal, 1)
    signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT)
    <-ch
}

// File: signal_windows.go
//go:build windows

package main

func listenForSignals() {
    // Windows-specific implementation
}

// Custom tags: go build -tags integration
//go:build integration

import "unsafe"

// Zero-copy []byte to string conversion (read-only use only)
func bytesToString(b []byte) string {
    return *(*string)(unsafe.Pointer(&b))
}

// Inspect struct layout
type Point struct{ X, Y float32 }
p := Point{1.0, 2.0}
fmt.Println(unsafe.Sizeof(p))   // 8 bytes
fmt.Println(unsafe.Alignof(p))  // 4 bytes

// Access struct fields by offset
xPtr := (*float32)(unsafe.Pointer(&p))
yPtr := (*float32)(unsafe.Pointer(uintptr(unsafe.Pointer(&p)) + 4))
fmt.Println(*xPtr, *yPtr) // 1 2

package main

/*
#include <stdlib.h>
#include <string.h>

char* repeat(const char* s, int n) {
    int len = strlen(s);
    char* result = malloc(len * n + 1);
    for (int i = 0; i < n; i++) {
        memcpy(result + i * len, s, len);
    }
    result[len * n] = '\0';
    return result;
}
*/
import "C"
import "unsafe"

func Repeat(s string, n int) string {
    cs := C.CString(s)
    defer C.free(unsafe.Pointer(cs))
    result := C.repeat(cs, C.int(n))
    defer C.free(unsafe.Pointer(result))
    return C.GoString(result)
}

// Compile: go run pgo_setup.go
// Output:
//   hot path result: 333283335000
//
// Profile-Guided Optimization (PGO) is a build step, not code:
//   1. Collect a CPU profile from production:
//        curl localhost:6060/debug/pprof/profile?seconds=30 > cpu.pprof
//   2. Place it next to package main as `default.pgo`
//   3. `go build` picks it up automatically (Go 1.21+)
//

//go:generate stringer -type=Status
//go:generate mockgen -source=service.go -destination=mock_service.go

package main

type Status int

const (
    Active Status = iota
    Inactive
    Pending
)

// After go:generate runs, a status_string.go file is created with:
// func (s Status) String() string { return [Active Inactive Pending][s] }

func main() {
    s := Active
    fmt.Println(s) // Active  (uses generated String() method)
}

// BROKEN — data race, no happens-before relationship
var ready bool
var data string
func setup() { data = "loaded"; ready = true }   // writes may be reordered
func wait()  { for !ready {}; print(data) }       // may see "" or never exit

// CORRECT — the channel send/receive synchronizes both writes
var done = make(chan struct{})
func setupOK() {
    data = "loaded"
    close(done) // send happens-before receive
}
func waitOK() {
    <-done       // after this, data's write is guaranteed visible
    print(data)  // always "loaded"
}

// Inspect with: go build -gcflags='-m' escape_analysis.go

// Does NOT escape — value stays on the stack
func sumLocal() int {
    x := [3]int{1, 2, 3} // "does not escape"
    return x[0] + x[1] + x[2]
}

// ESCAPES — the pointer outlives the function, so n moves to the heap
func newCounter() *int {
    n := 0    // "moved to heap: n"
    return &n
}

// ESCAPES — captured by a returned closure
func adder() func(int) int {
    sum := 0           // "moved to heap: sum"
    return func(x int) int { sum += x; return sum }
}

type User struct {
    ID    int    `json:"id" db:"user_id"`
    Name  string `json:"name" db:"user_name" validate:"required,min=2"`
    Email string `json:"email,omitempty" db:"email" validate:"email"`
    pass  string `json:"-"` // unexported + ignored by json
}

// Reading tags via reflection
t := reflect.TypeOf(User{})
field, _ := t.FieldByName("Name")
field.Tag.Get("json")     // "name"
field.Tag.Get("db")       // "user_name"
field.Tag.Get("validate") // "required,min=2"

// Lookup distinguishes "absent" from "empty"
value, ok := field.Tag.Lookup("xml") // "", false

class SingletonMeta(type):
    """Metaclass that ensures only one instance per class."""
    _instances: dict = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

class Config(metaclass=SingletonMeta):
    def __init__(self):
        self.debug = False
        self.version = "1.0"

a = Config()
b = Config()
print(a is b)       # True — same instance
a.debug = True
print(b.debug)      # True — same object

# __init_subclass__ — simpler registration pattern
class Plugin:
    _registry: dict[str, type] = {}

    def __init_subclass__(cls, name: str = "", **kwargs):
        super().__init_subclass__(**kwargs)
        if name:
            Plugin._registry[name] = cls

class AuthPlugin(Plugin, name="auth"):
    def run(self): return "auth"

print(Plugin._registry)  # {'auth': <class 'AuthPlugin'>}

class Typed:
    """Descriptor that enforces a type constraint."""
    def __set_name__(self, owner, name: str) -> None:
        self._name = f"_{name}"
        self._public = name

    def __init__(self, expected_type: type):
        self._type = expected_type

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self  # Class-level access returns the descriptor itself
        return getattr(obj, self._name, None)

    def __set__(self, obj, value) -> None:
        if not isinstance(value, self._type):
            raise TypeError(f"{self._public} must be {self._type.__name__}, got {type(value).__name__}")
        setattr(obj, self._name, value)

class Person:
    name = Typed(str)
    age  = Typed(int)

    def __init__(self, name: str, age: int):
        self.name = name
        self.age  = age

p = Person("Alice", 30)
print(p.name, p.age)  # Alice 30
try:
    p.age = "thirty"
except TypeError as e:
    print(e)  # age must be int, got str

from abc import ABC, abstractmethod
from typing import Iterator

class DataSource(ABC):
    @abstractmethod
    def connect(self) -> None: ...

    @abstractmethod
    def read_batch(self, size: int) -> list[dict]: ...

    @abstractmethod
    def close(self) -> None: ...

    def read_all(self, batch_size: int = 100) -> Iterator[list[dict]]:
        """Template method — uses abstract read_batch."""
        self.connect()
        try:
            while batch := self.read_batch(batch_size):
                yield batch
        finally:
            self.close()

class MemorySource(DataSource):
    def __init__(self, data: list[dict]):
        self._data = data
        self._pos = 0

    def connect(self) -> None: pass
    def close(self) -> None:   pass

    def read_batch(self, size: int) -> list[dict]:
        batch = self._data[self._pos:self._pos + size]
        self._pos += size
        return batch

src = MemorySource([{"id": i} for i in range(5)])
for batch in src.read_all(batch_size=2):
    print(batch)

from typing import Protocol, runtime_checkable

@runtime_checkable
class Drawable(Protocol):
    def draw(self) -> str: ...
    def area(self) -> float: ...

class Circle:
    def __init__(self, r: float): self.r = r
    def draw(self) -> str:  return f"Circle(r={self.r})"
    def area(self) -> float: return 3.14159 * self.r ** 2

class Rectangle:
    def __init__(self, w: float, h: float): self.w, self.h = w, h
    def draw(self) -> str:  return f"Rect({self.w}x{self.h})"
    def area(self) -> float: return self.w * self.h

def render(shape: Drawable) -> None:
    print(f"{shape.draw()} — area: {shape.area():.2f}")

shapes: list[Drawable] = [Circle(5), Rectangle(3, 4)]
for s in shapes:
    render(s)

print(isinstance(Circle(1), Drawable))    # True (runtime_checkable)

import sys

class PointDict:
    def __init__(self, x, y, z):
        self.x, self.y, self.z = x, y, z

class PointSlots:
    __slots__ = ("x", "y", "z")
    def __init__(self, x, y, z):
        self.x, self.y, self.z = x, y, z

pd = PointDict(1.0, 2.0, 3.0)
ps = PointSlots(1.0, 2.0, 3.0)

print(f"PointDict  size: {sys.getsizeof(pd)} bytes + {sys.getsizeof(pd.__dict__)} dict")
print(f"PointSlots size: {sys.getsizeof(ps)} bytes (no __dict__)")
print(f"Has __dict__: PointDict={hasattr(pd, '__dict__')}, PointSlots={hasattr(ps, '__dict__')}")

# Slots with inheritance (child must also declare slots)
class Point4D(PointSlots):
    __slots__ = ("w",)
    def __init__(self, x, y, z, w):
        super().__init__(x, y, z)
        self.w = w

import gc
import weakref
import sys

class Node:
    def __init__(self, value):
        self.value = value
        self.next: "Node | None" = None

# Reference cycle — would leak without cycle GC
a = Node("A")
b = Node("B")
a.next = b
b.next = a  # Cycle!

print(f"GC before collect: {gc.get_count()}")  # (counts objects in each gen)
del a, b
collected = gc.collect()
print(f"Cycle GC collected {collected} objects")

# weakref — cache that doesn't prevent GC
class ExpensiveObject:
    def __init__(self, key): self.key = key

cache: weakref.WeakValueDictionary = weakref.WeakValueDictionary()
obj = ExpensiveObject("config")
cache["config"] = obj
print(cache.get("config"))  # <ExpensiveObject ...>
del obj
gc.collect()
print(cache.get("config"))  # None — auto-evicted

import dis
import marshal

def factorial(n: int) -> int:
    result = 1
    for i in range(2, n + 1):
        result *= i
    return result

print("=== Bytecode for factorial ===")
dis.dis(factorial)
print(f"\nConstants: {factorial.__code__.co_consts}")
print(f"Locals:    {factorial.__code__.co_varnames}")
print(f"Stack size:{factorial.__code__.co_stacksize}")

# .pyc file format note
import py_compile, pathlib, importlib.util
import tempfile, os
with tempfile.NamedTemporaryFile(suffix=".py", mode="w", delete=False) as f:
    f.write("x = 1 + 2\n")
    tmp = f.name
py_compile.compile(tmp, doraise=True)
print(f"\nCompiled to: {importlib.util.cache_from_source(tmp)}")
os.unlink(tmp)

import importlib
import sys

# Dynamic import
def load_plugin(module_name: str):
    try:
        module = importlib.import_module(module_name)
        print(f"Loaded: {module.__name__}")
        return module
    except ImportError as e:
        print(f"Failed to load {module_name}: {e}")
        return None

# sys.modules cache
import json
print(f"json cached: {'json' in sys.modules}")   # True after first import
cached = sys.modules["json"]
print(f"Same object: {cached is json}")            # True

# __all__ controls wildcard imports
print(f"json.__all__ (first 5): {json.__all__[:5] if hasattr(json, '__all__') else 'not defined'}")

# Reload a module (rarely needed, but useful in REPL development)
import math
original_pi = math.pi
importlib.reload(math)
print(f"After reload: math.pi = {math.pi}")

from typing import TypeGuard, Union
from dataclasses import dataclass

@dataclass
class Cat:
    name: str
    indoor: bool

@dataclass
class Dog:
    name: str
    breed: str

Animal = Union[Cat, Dog]

def is_cat(animal: Animal) -> TypeGuard[Cat]:
    return isinstance(animal, Cat)

def describe(animal: Animal) -> str:
    if is_cat(animal):
        # Type checker knows animal is Cat here
        return f"Cat {animal.name}, indoor={animal.indoor}"
    else:
        # Type checker knows animal is Dog here
        return f"Dog {animal.name}, breed={animal.breed}"

animals: list[Animal] = [Cat("Whiskers", True), Dog("Rex", "Labrador")]
for a in animals:
    print(describe(a))

import cProfile
import pstats
import io
import timeit

def slow_sum(n: int) -> int:
    """Intentionally slow version using string conversion."""
    total = 0
    for i in range(n):
        total += int(str(i))  # Silly but profiling shows it
    return total

def fast_sum(n: int) -> int:
    return sum(range(n))

# Profile slow_sum
pr = cProfile.Profile()
pr.enable()
result = slow_sum(50_000)
pr.disable()

buf = io.StringIO()
ps = pstats.Stats(pr, stream=buf).sort_stats("cumulative")
ps.print_stats(5)  # Top 5 functions
print(buf.getvalue())

# Quick benchmark
slow_t = timeit.timeit("slow_sum(1000)", globals={"slow_sum": slow_sum}, number=100)
fast_t = timeit.timeit("fast_sum(1000)", globals={"fast_sum": fast_sum}, number=100)
print(f"slow: {slow_t:.3f}s  fast: {fast_t:.3f}s  speedup: {slow_t/fast_t:.1f}x")

import sys

def demonstrate_threading_landscape():
    """Shows the evolution of Python parallelism options."""
    version = sys.version_info

    options = {
        "threading (I/O-bound)": "GIL released on I/O — works today",
        "multiprocessing":        "True parallelism via processes — works today",
        "asyncio":                "Cooperative concurrency — works today",
        "C extensions (NumPy)":   "Release GIL explicitly — works today",
    }

    if version >= (3, 12):
        options["Per-interpreter GIL"] = "PEP 684 — experimental (3.12+)"
    if version >= (3, 13):
        options["Free-threaded (no GIL)"] = "PEP 703 — use python3.13t build"

    for name, status in options.items():
        print(f"  {name:35} → {status}")

print(f"Python {sys.version_info.major}.{sys.version_info.minor} parallelism options:")
demonstrate_threading_landscape()

# Check if running in free-threaded mode
if hasattr(sys, "_is_gil_enabled"):
    print(f"\nGIL enabled: {sys._is_gil_enabled()}")
    if not sys._is_gil_enabled():
        print("  Running in FREE-THREADED mode — true CPU parallelism available!")