Chapitre 14 : Concurrency Patterns

Durée estimée : 5 séances de 3h Objectifs : Maîtriser les patterns de concurrence (Active Object, Reactor, Proactor, Thread Pool), les patterns Go (fan-in, fan-out, pipeline), et les patterns modernes (Virtual Threads, async/await).

---

Partie 1 : Fondamentaux de la Concurrence

1.1 Concurrence vs Parallélisme

Concurrence : Composition de tâches qui peuvent s'exécuter en ordre entrelacé (single-core possible). Parallélisme : Exécution simultanée réelle (multi-core nécessaire).

"Concurrency is about dealing with lots of things at once. Parallelism is about doing lots of things at once." — Rob Pike

1.2 Problèmes classiques

• Race conditions : Accès concurrent non synchronisé à des données partagées
• Deadlock : Deux threads s'attendent mutuellement
• Starvation : Un thread n'accède jamais à la ressource
• Livelock : Les threads changent d'état sans progresser
• Thread overhead : Création/destruction coûteuse

---

Partie 2 : Active Object Pattern

2.1 Principe

Active Object découple l'appel de méthode de son exécution. Chaque objet a son propre thread de contrôle et une file d'attente de requêtes.

2.2 Structure

Diagramme en cours de génération...

2.3 Implémentation

class Future<T> {
    private result: T | null = null;
    private error: Error | null = null;
    private callbacks: Array<(value: T) => void> = [];
    private errorCallbacks: Array<(error: Error) => void> = [];

    resolve(value: T): void {
        this.result = value;
        this.callbacks.forEach(cb => cb(value));
    }

    reject(error: Error): void {
        this.error = error;
        this.errorCallbacks.forEach(cb => cb(error));
    }

    then(callback: (value: T) => void): Future<T> {
        if (this.result !== null) {
            callback(this.result);
        } else {
            this.callbacks.push(callback);
        }
        return this;
    }

    catch(callback: (error: Error) => void): Future<T> {
        if (this.error !== null) {
            callback(this.error);
        } else {
            this.errorCallbacks.push(callback);
        }
        return this;
    }

    get(): T | null { return this.result; }
    isDone(): boolean { return this.result !== null || this.error !== null; }
}

// Active Object
class ActiveObject {
    private queue: Array<() => void> = [];
    private running = false;

    constructor() {
        this.startWorker();
    }

    private startWorker(): void {
        this.running = true;
        this.processQueue();
    }

    private async processQueue(): Promise<void> {
        while (this.running) {
            if (this.queue.length > 0) {
                const task = this.queue.shift()!;
                try {
                    task();
                } catch (err) {
                    console.error('Task failed:', err);
                }
            } else {
                await this.sleep(10);
            }
        }
    }

    private sleep(ms: number): Promise<void> {
        return new Promise(resolve => setTimeout(resolve, ms));
    }

    enqueue<T>(task: () => T): Future<T> {
        const future = new Future<T>();
        this.queue.push(() => {
            try {
                future.resolve(task());
            } catch (err) {
                future.reject(err as Error);
            }
        });
        return future;
    }

    stop(): void {
        this.running = false;
    }
}

// Usage
const ao = new ActiveObject();

const future1 = ao.enqueue(() => {
    console.log('Task 1 running');
    return 42;
});

const future2 = ao.enqueue(() => {
    console.log('Task 2 running');
    return 'Hello';
});

future1.then(value => console.log('Result 1:', value));
future2.then(value => console.log('Result 2:', value));

---

Partie 3 : Reactor Pattern

3.1 Principe

Reactor est un pattern de démultiplexage d'événements qui dispatche les événements entrants vers les handlers appropriés.

3.2 Structure

Diagramme en cours de génération...

3.3 Implémentation (Event Loop)

type EventType = 'read' | 'write' | 'error' | 'timer';

interface Event {
    type: EventType;
    data?: any;
    source?: string;
}

interface EventHandler {
    handle(event: Event): Promise<void>;
    canHandle(event: Event): boolean;
}

class Reactor {
    private handlers: Map<EventType, EventHandler[]> = new Map();
    private eventQueue: Event[] = [];
    private running = false;

    registerHandler(eventType: EventType, handler: EventHandler): void {
        if (!this.handlers.has(eventType)) {
            this.handlers.set(eventType, []);
        }
        this.handlers.get(eventType)!.push(handler);
    }

    async dispatch(event: Event): Promise<void> {
        const handlers = this.handlers.get(event.type) || [];
        for (const handler of handlers) {
            if (handler.canHandle(event)) {
                await handler.handle(event);
                return;
            }
        }
    }

    enqueue(event: Event): void {
        this.eventQueue.push(event);
    }

    async start(): Promise<void> {
        this.running = true;
        while (this.running) {
            while (this.eventQueue.length > 0) {
                const event = this.eventQueue.shift()!;
                await this.dispatch(event);
            }
            await this.sleep(1); // Yield
        }
    }

    stop(): void { this.running = false; }

    private sleep(ms: number): Promise<void> {
        return new Promise(resolve => setTimeout(resolve, ms));
    }
}

// Example: HTTP Server with Reactor
class HttpReadHandler implements EventHandler {
    canHandle(event: Event): boolean {
        return event.type === 'read' && event.source === 'http';
    }

    async handle(event: Event): Promise<void> {
        console.log(`Processing HTTP request: ${event.data}`);
        // Parse request, generate response
    }
}

class WebSocketHandler implements EventHandler {
    canHandle(event: Event): boolean {
        return event.type === 'read' && event.source === 'websocket';
    }

    async handle(event: Event): Promise<void> {
        console.log(`Processing WebSocket message: ${event.data}`);
    }
}

---

Partie 4 : Proactor Pattern

4.1 Principe

Proactor (async completion) est similaire à Reactor mais les opérations sont initiées et les handlers sont appelés à la complétion (asynchrone), plutôt qu'à la disponibilité des événements.

4.2 Reactor vs Proactor

---

Partie 5 : Thread Pool Pattern

5.1 Principe

Thread Pool maintient un ensemble de threads réutilisables qui exécutent des tâches depuis une file d'attente.

5.2 Implémentation

type Task = () => Promise<void>;

class ThreadPool {
    private workers: Worker[] = [];
    private taskQueue: Task[] = [];
    private running = true;

    constructor(size: number) {
        for (let i = 0; i < size; i++) {
            this.workers.push(new Worker(i, this));
        }
    }

    submit(task: Task): void {
        this.taskQueue.push(task);
    }

    getNextTask(): Task | null {
        return this.taskQueue.shift() || null;
    }

    async shutdown(): Promise<void> {
        this.running = false;
        await Promise.all(this.workers.map(w => w.stop()));
    }

    get pendingTasks(): number { return this.taskQueue.length; }
}

class Worker {
    private running = true;

    constructor(
        private id: number,
        private pool: ThreadPool
    ) {
        this.start();
    }

    private async start(): Promise<void> {
        while (this.running) {
            const task = this.pool.getNextTask();
            if (task) {
                try {
                    console.log(`Worker ${this.id} executing task`);
                    await task();
                } catch (err) {
                    console.error(`Worker ${this.id} failed:`, err);
                }
            } else {
                await this.sleep(10);
            }
        }
    }

    async stop(): Promise<void> {
        this.running = false;
    }

    private sleep(ms: number): Promise<void> {
        return new Promise(resolve => setTimeout(resolve, ms));
    }
}

// Usage
const pool = new ThreadPool(4);

for (let i = 0; i < 10; i++) {
    pool.submit(async () => {
        await new Promise(resolve => setTimeout(resolve, 100));
        console.log(`Task ${i} completed`);
    });
}

5.3 Thread Pool en Java

import java.util.concurrent.*;

public class ThreadPoolExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(4);
        
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            executor.submit(() -> {
                System.out.println("Task " + taskId + " running on " + 
                    Thread.currentThread().getName());
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        executor.shutdown();
        try {
            executor.awaitTermination(1, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

---

Partie 6 : Go Concurrency Patterns

6.1 Pipeline Pattern

package main

import "fmt"

// Stage 1: Generate numbers
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// Stage 2: Square numbers
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

// Stage 3: Filter even
func filterEven(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            if n%2 == 0 {
                out <- n
            }
        }
    }()
    return out
}

func main() {
    // Pipeline: generate -> square -> filterEven
    in := generate(1, 2, 3, 4, 5, 6)
    squared := square(in)
    filtered := filterEven(squared)

    for result := range filtered {
        fmt.Println("Result:", result) // 4, 16, 36
    }
}

6.2 Fan-out / Fan-in

package main

import (
    "fmt"
    "sync"
)

// Fan-out: distribute work across multiple workers
func fanOut(input <-chan int, workers int) []<-chan int {
    channels := make([]<-chan int, workers)
    for i := 0; i < workers; i++ {
        channels[i] = worker(i, input)
    }
    return channels
}

func worker(id int, input <-chan int) <-chan int {
    output := make(chan int)
    go func() {
        defer close(output)
        for n := range input {
            output <- n * 2
        }
    }()
    return output
}

// Fan-in: merge multiple channels into one
func fanIn(channels ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    out := make(chan int)

    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for n := range c {
                out <- n
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func main() {
    input := make(chan int)
    go func() {
        defer close(input)
        for i := 1; i <= 10; i++ {
            input <- i
        }
    }()

    // Fan-out to 3 workers
    channels := fanOut(input, 3)
    
    // Fan-in all results
    results := fanIn(channels...)

    for result := range results {
        fmt.Println("Result:", result)
    }
}

6.3 Worker Pool

package main

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

type Job struct {
    ID    int
    Data  string
}

type Result struct {
    JobID     int
    Output    string
    Processed bool
    Duration  time.Duration
}

func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        start := time.Now()
        fmt.Printf("Worker %d processing job %d\n", id, job.ID)
        
        // Simulate work
        time.Sleep(time.Millisecond * 100)
        
        results <- Result{
            JobID:     job.ID,
            Output:    fmt.Sprintf("Processed by worker %d: %s", id, job.Data),
            Processed: true,
            Duration:  time.Since(start),
        }
    }
}

func main() {
    const numJobs = 20
    const numWorkers = 5

    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)
    var wg sync.WaitGroup

    // Start workers
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    // Send jobs
    for j := 1; j <= numJobs; j++ {
        jobs <- Job{ID: j, Data: fmt.Sprintf("Task %d", j)}
    }
    close(jobs)

    // Wait for all workers to finish
    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect results
    for result := range results {
        fmt.Printf("Job %d: %s (took %v)\n", result.JobID, result.Output, result.Duration)
    }
}

---

Partie 7 : Virtual Threads (Project Loom)

7.1 Principe

Les Virtual Threads (Java 21+) sont des threads légers gérés par la JVM, permettant de créer des millions de threads sans overhead système.

7.2 Exemple Java

import java.util.concurrent.*;

public class VirtualThreadsExample {
    public static void main(String[] args) throws Exception {
        // Creating a virtual thread
        Thread vThread = Thread.startVirtualThread(() -> {
            System.out.println("Hello from virtual thread: " + 
                Thread.currentThread());
        });
        vThread.join();

        // Virtual thread executor
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var futures = new CompletableFuture<?>[10_000];
            
            for (int i = 0; i < futures.length; i++) {
                final int taskId = i;
                futures[i] = CompletableFuture.runAsync(() -> {
                    System.out.println("Task " + taskId + " on " + 
                        Thread.currentThread());
                }, executor);
            }
            
            CompletableFuture.allOf(futures).join();
        }
    }
}

7.3 Virtual Threads vs Platform Threads

---

Partie 8 : async/await Patterns

8.1 Puits de promesses

async function fetchUserData(userId: string) {
    const [profile, posts, friends] = await Promise.all([
        fetch(`/api/users/${userId}/profile`).then(r => r.json()),
        fetch(`/api/users/${userId}/posts`).then(r => r.json()),
        fetch(`/api/users/${userId}/friends`).then(r => r.json())
    ]);
    
    return { profile, posts, friends };
}

8.2 Cancellation Pattern

class CancellationToken {
    private cancelled = false;
    private callbacks: Array<() => void> = [];

    cancel(): void {
        this.cancelled = true;
        this.callbacks.forEach(cb => cb());
    }

    onCancellation(callback: () => void): void {
        if (this.cancelled) {
            callback();
        } else {
            this.callbacks.push(callback);
        }
    }

    isCancelled(): boolean { return this.cancelled; }
}

async function performOperation(token: CancellationToken): Promise<string> {
    return new Promise((resolve, reject) => {
        token.onCancellation(() => reject(new Error('Operation cancelled')));
        
        setTimeout(() => {
            if (!token.isCancelled()) {
                resolve('Operation completed');
            }
        }, 1000);
    });
}

---

Partie 9 : Patterns Additionnels

9.1 Balking Pattern

Empêche l'exécution d'une opération si l'objet n'est pas dans le bon état.

class Downloader {
    private downloading = false;

    async download(url: string): Promise<void> {
        if (this.downloading) return; // Balking
        this.downloading = true;
        try {
            const data = await fetch(url);
            console.log('Downloaded:', url);
        } finally {
            this.downloading = false;
        }
    }
}

9.2 Guarded Suspension

Suspend un thread jusqu'à ce qu'une condition soit satisfaite.

class GuardedQueue<T> {
    private queue: T[] = [];
    private resolvers: Array<(value: T) => void> = [];

    async get(): Promise<T> {
        if (this.queue.length > 0) {
            return this.queue.shift()!;
        }
        return new Promise(resolve => {
            this.resolvers.push(resolve);
        });
    }

    put(item: T): void {
        if (this.resolvers.length > 0) {
            const resolve = this.resolvers.shift()!;
            resolve(item);
        } else {
            this.queue.push(item);
        }
    }
}

---

Résumé

Ce chapitre conclut la série Modern-Design-Patterns.