Scheduler and Proxy: Advanced Patterns

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118

/**
 * Priority-Based Scheduler
 * Processes high-priority requests before low-priority ones
 */
 
enum Priority {
    CRITICAL = 0,  // Lowest value = highest priority
    HIGH = 1,
    NORMAL = 2,
    LOW = 3,
    BACKGROUND = 4
}
 
interface PrioritizedRequest {
    priority: Priority;
    request: MethodRequest<unknown>;
    enqueueTime: number;  // For fairness within priority level
}
 
class PriorityActivationQueue {
    private readonly queues: Map<Priority, Array<MethodRequest<unknown>>> = new Map();
    private readonly lock = new Mutex();
    private readonly notEmpty = new ConditionVariable();
    private isShutdown = false;
    
    constructor() {
        // Initialize queues for each priority level
        for (const priority of Object.values(Priority)) {
            if (typeof priority === 'number') {
                this.queues.set(priority, []);
            }
        }
    }
    
    async enqueue(request: MethodRequest<unknown>, priority: Priority = Priority.NORMAL): Promise<void> {
        await this.lock.acquire();
        try {
            if (this.isShutdown) {
                throw new Error("Queue is shutdown");
            }
            
            this.queues.get(priority)!.push(request);
            this.notEmpty.signal();
        } finally {
            this.lock.release();
        }
    }
    
    async dequeue(): Promise<MethodRequest<unknown> | null> {
        await this.lock.acquire();
        try {
            while (this.isEmpty() && !this.isShutdown) {
                await this.notEmpty.wait(this.lock);
            }
            
            if (this.isEmpty()) {
                return null;
            }
            
            // Process highest priority first (lowest numeric value)
            for (const priority of [
                Priority.CRITICAL,
                Priority.HIGH,
                Priority.NORMAL,
                Priority.LOW,
                Priority.BACKGROUND
            ]) {
                const queue = this.queues.get(priority)!;
                if (queue.length > 0) {
                    return queue.shift()!;
                }
            }
            
            return null;
        } finally {
            this.lock.release();
        }
    }
    
    private isEmpty(): boolean {
        for (const queue of this.queues.values()) {
            if (queue.length > 0) return false;
        }
        return true;
    }
}
 
/**
 * Priority-Aware Proxy that allows clients to specify priority
 */
class PriorityImageProcessorProxy implements ImageProcessor {
    private readonly queue: PriorityActivationQueue;
    
    processImage(imageId: string, data: Buffer): Promise<ProcessedImage> {
        return this.processImageWithPriority(imageId, data, Priority.NORMAL);
    }
    
    processImageWithPriority(
        imageId: string, 
        data: Buffer, 
        priority: Priority
    ): Promise<ProcessedImage> {
        const future = new Future<ProcessedImage>();
        const request = new ProcessImageRequest(imageId, data, future);
        this.queue.enqueue(request, priority);
        return future.toPromise();
    }
    
    // Example: Express lane for thumbnail generation
    processImageThumbnail(imageId: string, data: Buffer): Promise<ProcessedImage> {
        return this.processImageWithPriority(imageId, data, Priority.HIGH);
    }
    
    // Example: Background batch processing
    processImageBatch(imageId: string, data: Buffer): Promise<ProcessedImage> {
        return this.processImageWithPriority(imageId, data, Priority.BACKGROUND);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

/**
 * Thread Pool Scheduler
 * Dispatches requests to a pool of worker threads for parallel execution
 */
 
class ThreadPoolScheduler {
    private readonly activationQueue: ActivationQueue;
    private readonly servantPool: ServantPool;
    private readonly workerCount: number;
    private readonly workers: Worker[] = [];
    private isRunning = false;
    
    constructor(
        activationQueue: ActivationQueue, 
        servantFactory: () => ImageProcessorServant,
        workerCount: number = 4
    ) {
        this.activationQueue = activationQueue;
        this.workerCount = workerCount;
        
        // Create a pool of Servants - each worker gets its own
        this.servantPool = new ServantPool(servantFactory, workerCount);
    }
    
    start(): void {
        if (this.isRunning) return;
        
        this.isRunning = true;
        
        // Spawn multiple worker threads, each running the same processing loop
        for (let i = 0; i < this.workerCount; i++) {
            const worker = this.createWorker(i);
            this.workers.push(worker);
        }
        
        console.log(`[ThreadPoolScheduler] Started with ${this.workerCount} workers`);
    }
    
    private createWorker(workerId: number): Worker {
        // In reality, this would be a true thread (Web Worker, Worker Thread, etc.)
        // Here we simulate with an async loop
        const workerLoop = async () => {
            // Each worker gets its own Servant instance
            const servant = this.servantPool.acquire();
            
            try {
                while (this.isRunning) {
                    const request = await this.activationQueue.dequeue();
                    
                    if (request === null) {
                        break;  // Shutdown signal
                    }
                    
                    // Workers process in parallel - each on its own Servant
                    try {
                        if (request.guard(servant)) {
                            request.call(servant);
                        } else {
                            // Re-queue if guard fails
                            await this.activationQueue.enqueue(request);
                        }
                    } catch (error) {
                        console.error(`[Worker-${workerId}] Error:`, error);
                    }
                }
            } finally {
                this.servantPool.release(servant);
            }
            
            console.log(`[Worker-${workerId}] Stopped`);
        };
        
        // Start the worker
        workerLoop();
        return { id: workerId } as Worker;
    }
    
    async stop(): Promise<void> {
        this.isRunning = false;
        await this.activationQueue.shutdown();
        console.log("[ThreadPoolScheduler] Shutdown complete");
    }
}
 
/**
 * Servant Pool - manages a pool of Servant instances
 * Each worker thread gets its own Servant to avoid shared state
 */
class ServantPool {
    private readonly servants: Array<{ servant: ImageProcessorServant; inUse: boolean }>;
    private readonly lock = new Mutex();
    private readonly available = new ConditionVariable();
    
    constructor(factory: () => ImageProcessorServant, poolSize: number) {
        this.servants = [];
        for (let i = 0; i < poolSize; i++) {
            this.servants.push({ servant: factory(), inUse: false });
        }
    }
    
    async acquire(): Promise<ImageProcessorServant> {
        await this.lock.acquire();
        try {
            while (true) {
                const available = this.servants.find(s => !s.inUse);
                if (available) {
                    available.inUse = true;
                    return available.servant;
                }
                await this.available.wait(this.lock);
            }
        } finally {
            this.lock.release();
        }
    }
    
    async release(servant: ImageProcessorServant): Promise<void> {
        await this.lock.acquire();
        try {
            const entry = this.servants.find(s => s.servant === servant);
            if (entry) {
                entry.inUse = false;
                this.available.signal();
            }
        } finally {
            this.lock.release();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174

/**
 * Cancellation Token - shared between client and request
 */
class CancellationToken {
    private _isCancelled = false;
    private readonly callbacks: Array<() => void> = [];
    
    get isCancelled(): boolean {
        return this._isCancelled;
    }
    
    cancel(): void {
        if (this._isCancelled) return;
        
        this._isCancelled = true;
        
        // Notify all registered callbacks
        for (const callback of this.callbacks) {
            try { callback(); } catch (e) { /* ignore */ }
        }
        this.callbacks.length = 0;
    }
    
    onCancel(callback: () => void): void {
        if (this._isCancelled) {
            callback();
        } else {
            this.callbacks.push(callback);
        }
    }
    
    throwIfCancelled(): void {
        if (this._isCancelled) {
            throw new CancellationError("Operation was cancelled");
        }
    }
}
 
class CancellationError extends Error {
    constructor(message: string) {
        super(message);
        this.name = 'CancellationError';
    }
}
 
/**
 * Cancellable Method Request
 */
abstract class CancellableMethodRequest<T> extends MethodRequest<T> {
    protected readonly cancellationToken: CancellationToken;
    
    constructor(future: Future<T>, cancellationToken: CancellationToken) {
        super(future);
        this.cancellationToken = cancellationToken;
        
        // If cancelled, complete the future with cancellation error
        cancellationToken.onCancel(() => {
            this.future.reject(new CancellationError("Request was cancelled"));
        });
    }
    
    call(servant: ImageProcessorServant): void {
        // Check cancellation before starting
        if (this.cancellationToken.isCancelled) {
            this.future.reject(new CancellationError("Request was cancelled"));
            return;
        }
        
        try {
            // Execute with cancellation token available
            const result = this.executeWithCancellation(servant);
            
            // Check cancellation after completion (result may be stale)
            if (!this.cancellationToken.isCancelled) {
                this.future.resolve(result);
            }
        } catch (error) {
            if (error instanceof CancellationError) {
                this.future.reject(error);
            } else {
                this.future.reject(error as Error);
            }
        }
    }
    
    abstract executeWithCancellation(servant: ImageProcessorServant): T;
}
 
/**
 * Cancellable implementation checking periodically
 */
class CancellableProcessImageRequest extends CancellableMethodRequest<ProcessedImage> {
    private readonly imageId: string;
    private readonly data: Buffer;
    
    constructor(
        imageId: string, 
        data: Buffer, 
        future: Future<ProcessedImage>,
        cancellationToken: CancellationToken
    ) {
        super(future, cancellationToken);
        this.imageId = imageId;
        this.data = data;
    }
    
    executeWithCancellation(servant: ImageProcessorServant): ProcessedImage {
        // Check before each expensive operation
        this.cancellationToken.throwIfCancelled();
        const decoded = servant.decodeImage(this.data);
        
        this.cancellationToken.throwIfCancelled();
        const filtered = servant.applyFilters(decoded);
        
        this.cancellationToken.throwIfCancelled();
        const resized = servant.resize(filtered);
        
        this.cancellationToken.throwIfCancelled();
        return servant.compress(resized);
    }
}
 
/**
 * Cancellable Proxy returning both Future and CancellationToken
 */
class CancellableImageProcessorProxy {
    private readonly queue: ActivationQueue;
    
    processImage(imageId: string, data: Buffer): CancellableOperation<ProcessedImage> {
        const future = new Future<ProcessedImage>();
        const cancellationToken = new CancellationToken();
        
        const request = new CancellableProcessImageRequest(
            imageId, data, future, cancellationToken
        );
        
        this.queue.enqueue(request);
        
        return {
            result: future.toPromise(),
            cancel: () => cancellationToken.cancel()
        };
    }
}
 
interface CancellableOperation<T> {
    result: Promise<T>;
    cancel: () => void;
}
 
// Client usage
async function clientWithCancellation() {
    const processor = new CancellableImageProcessorProxy();
    
    const operation = processor.processImage("img-1", imageBuffer);
    
    // Set up timeout
    const timeout = setTimeout(() => {
        console.log("Operation timed out, cancelling...");
        operation.cancel();
    }, 5000);
    
    try {
        const result = await operation.result;
        clearTimeout(timeout);
        console.log("Processing complete:", result);
    } catch (error) {
        if (error instanceof CancellationError) {
            console.log("Operation was cancelled");
        } else {
            throw error;
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

/**
 * Timeout-Aware Proxy
 * Wraps operations with timeout handling
 */
 
class TimeoutImageProcessorProxy {
    private readonly delegate: CancellableImageProcessorProxy;
    private readonly defaultTimeout: number;
    
    constructor(delegate: CancellableImageProcessorProxy, defaultTimeoutMs: number = 30000) {
        this.delegate = delegate;
        this.defaultTimeout = defaultTimeoutMs;
    }
    
    /**
     * Process with default timeout
     */
    processImage(imageId: string, data: Buffer): Promise<ProcessedImage> {
        return this.processImageWithTimeout(imageId, data, this.defaultTimeout);
    }
    
    /**
     * Process with custom timeout
     */
    async processImageWithTimeout(
        imageId: string, 
        data: Buffer, 
        timeoutMs: number
    ): Promise<ProcessedImage> {
        const operation = this.delegate.processImage(imageId, data);
        
        // Race between operation and timeout
        return Promise.race([
            operation.result,
            this.createTimeout<ProcessedImage>(timeoutMs, () => operation.cancel())
        ]);
    }
    
    private createTimeout<T>(ms: number, onTimeout: () => void): Promise<T> {
        return new Promise((_, reject) => {
            setTimeout(() => {
                onTimeout();
                reject(new TimeoutError(`Operation timed out after ${ms}ms`));
            }, ms);
        });
    }
}
 
class TimeoutError extends Error {
    constructor(message: string) {
        super(message);
        this.name = 'TimeoutError';
    }
}
 
/**
 * Advanced: Deadline-based timeout with queue time consideration
 */
class DeadlineAwareRequest<T> extends CancellableMethodRequest<T> {
    private readonly deadline: number;  // Absolute timestamp
    
    constructor(
        future: Future<T>,
        cancellationToken: CancellationToken,
        deadlineMs: number  // Relative timeout from now
    ) {
        super(future, cancellationToken);
        this.deadline = Date.now() + deadlineMs;
    }
    
    /**
     * Check if deadline has passed (including time spent in queue)
     */
    isExpired(): boolean {
        return Date.now() > this.deadline;
    }
    
    /**
     * Remaining time until deadline
     */
    remainingMs(): number {
        return Math.max(0, this.deadline - Date.now());
    }
}
 
/**
 * Scheduler that respects deadlines
 */
class DeadlineAwareScheduler {
    async processLoop(): Promise<void> {
        while (this.isRunning) {
            const request = await this.activationQueue.dequeue();
            if (request === null) break;
            
            // Check if request has already expired while in queue
            if (request instanceof DeadlineAwareRequest && request.isExpired()) {
                // Don't even start processing - it would be wasted work
                request.future.reject(new TimeoutError(
                    "Request expired while waiting in queue"
                ));
                continue;
            }
            
            // Execute with remaining deadline
            request.call(this.servant);
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167

/**
 * Pattern 1: Batching Proxy
 * Accumulates requests and submits them as a batch
 */
class BatchingImageProcessorProxy {
    private readonly queue: ActivationQueue;
    private pendingBatch: Array<{
        request: BatchableRequest;
        future: Future<ProcessedImage>;
    }> = [];
    private batchTimer: NodeJS.Timeout | null = null;
    private readonly batchSize: number;
    private readonly batchDelayMs: number;
    
    constructor(queue: ActivationQueue, batchSize = 10, batchDelayMs = 100) {
        this.queue = queue;
        this.batchSize = batchSize;
        this.batchDelayMs = batchDelayMs;
    }
    
    processImage(imageId: string, data: Buffer): Promise<ProcessedImage> {
        const future = new Future<ProcessedImage>();
        
        this.pendingBatch.push({
            request: { imageId, data },
            future
        });
        
        // Flush if batch is full
        if (this.pendingBatch.length >= this.batchSize) {
            this.flushBatch();
        } else if (!this.batchTimer) {
            // Start timer for partial batch
            this.batchTimer = setTimeout(() => this.flushBatch(), this.batchDelayMs);
        }
        
        return future.toPromise();
    }
    
    private flushBatch(): void {
        if (this.batchTimer) {
            clearTimeout(this.batchTimer);
            this.batchTimer = null;
        }
        
        if (this.pendingBatch.length === 0) return;
        
        const batch = this.pendingBatch;
        this.pendingBatch = [];
        
        // Submit as single batch request
        const batchRequest = new BatchProcessImageRequest(batch);
        this.queue.enqueue(batchRequest);
    }
}
 
/**
 * Pattern 2: Deduplicating Proxy
 * Reuses in-flight requests for identical operations
 */
class DeduplicatingImageProcessorProxy {
    private readonly queue: ActivationQueue;
    private readonly inFlight: Map<string, Promise<ProcessedImage>> = new Map();
    
    processImage(imageId: string, data: Buffer): Promise<ProcessedImage> {
        // Create deduplication key
        const key = this.createKey(imageId, data);
        
        // Check if identical request is in flight
        const existing = this.inFlight.get(key);
        if (existing) {
            console.log(`Deduplicating request for ${imageId}`);
            return existing;
        }
        
        // Create new request
        const future = new Future<ProcessedImage>();
        const request = new ProcessImageRequest(imageId, data, future);
        this.queue.enqueue(request);
        
        const promise = future.toPromise();
        
        // Track the in-flight request
        this.inFlight.set(key, promise);
        
        // Remove from tracking when complete
        promise.finally(() => {
            this.inFlight.delete(key);
        });
        
        return promise;
    }
    
    private createKey(imageId: string, data: Buffer): string {
        // Simple key based on imageId and data hash
        return `${imageId}:${this.hash(data)}`;
    }
    
    private hash(data: Buffer): string {
        // Use fast hash for deduplication
        return crypto.createHash('md5').update(data).digest('hex');
    }
}
 
/**
 * Pattern 3: Rate-Limiting Proxy
 * Controls request rate to prevent overwhelming the system
 */
class RateLimitingImageProcessorProxy {
    private readonly delegate: ImageProcessor;
    private readonly rateLimiter: RateLimiter;
    
    constructor(delegate: ImageProcessor, requestsPerSecond: number) {
        this.delegate = delegate;
        this.rateLimiter = new TokenBucketRateLimiter(requestsPerSecond);
    }
    
    async processImage(imageId: string, data: Buffer): Promise<ProcessedImage> {
        // Wait for rate limiter token
        await this.rateLimiter.acquire();
        
        try {
            return await this.delegate.processImage(imageId, data);
        } finally {
            // Token is consumed; new token will be added after refill interval
        }
    }
}
 
class TokenBucketRateLimiter {
    private tokens: number;
    private readonly maxTokens: number;
    private readonly refillInterval: number;
    private lastRefill: number;
    
    constructor(tokensPerSecond: number) {
        this.maxTokens = tokensPerSecond;
        this.tokens = tokensPerSecond;
        this.refillInterval = 1000 / tokensPerSecond;
        this.lastRefill = Date.now();
    }
    
    async acquire(): Promise<void> {
        while (true) {
            this.refill();
            
            if (this.tokens >= 1) {
                this.tokens -= 1;
                return;
            }
            
            // Wait for next refill
            await new Promise(resolve => 
                setTimeout(resolve, this.refillInterval)
            );
        }
    }
    
    private refill(): void {
        const now = Date.now();
        const elapsed = now - this.lastRefill;
        const newTokens = elapsed / this.refillInterval;
        
        this.tokens = Math.min(this.maxTokens, this.tokens + newTokens);
        this.lastRefill = now;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

/**
 * Instrumented Scheduler with comprehensive monitoring
 */
 
interface ActiveObjectMetrics {
    queueDepth: Gauge;
    queueWaitTime: Histogram;
    executionTime: Histogram;
    requestsCompleted: Counter;
    requestsFailed: Counter;
    requestsCancelled: Counter;
}
 
class InstrumentedScheduler {
    private readonly metrics: ActiveObjectMetrics;
    private readonly servant: ImageProcessorServant;
    private readonly activationQueue: ActivationQueue;
    
    async processLoop(): Promise<void> {
        while (this.isRunning) {
            // Update queue depth metric
            this.metrics.queueDepth.set(await this.activationQueue.size());
            
            const request = await this.activationQueue.dequeue();
            if (request === null) break;
            
            // Calculate queue wait time
            const queueWaitMs = Date.now() - request.enqueueTime;
            this.metrics.queueWaitTime.observe(queueWaitMs);
            
            // Execute and measure
            const executionStart = Date.now();
            try {
                if (request.guard(this.servant)) {
                    request.call(this.servant);
                    
                    const executionMs = Date.now() - executionStart;
                    this.metrics.executionTime.observe(executionMs);
                    this.metrics.requestsCompleted.inc();
                    
                    // Log slow requests
                    if (executionMs > 5000) {
                        console.warn(`Slow request: ${request.constructor.name} took ${executionMs}ms`);
                    }
                }
            } catch (error) {
                if (error instanceof CancellationError) {
                    this.metrics.requestsCancelled.inc();
                } else {
                    this.metrics.requestsFailed.inc();
                    console.error(`Request failed: ${error}`);
                }
            }
        }
    }
}
 
/**
 * Health check endpoint for the Active Object
 */
class ActiveObjectHealthCheck {
    private readonly activationQueue: ActivationQueue;
    private readonly maxQueueDepth: number;
    private readonly maxQueueWaitMs: number;
    
    async isHealthy(): Promise<HealthStatus> {
        const queueDepth = await this.activationQueue.size();
        const queueUtilization = queueDepth / this.maxQueueDepth;
        
        // Calculate average queue wait time from recent samples
        const avgWaitMs = this.getAverageQueueWaitMs();
        
        if (queueUtilization > 0.9) {
            return {
                status: 'UNHEALTHY',
                reason: `Queue nearly full: ${(queueUtilization * 100).toFixed(1)}% utilization`
            };
        }
        
        if (avgWaitMs > this.maxQueueWaitMs) {
            return {
                status: 'DEGRADED',
                reason: `High queue wait time: ${avgWaitMs}ms average`
            };
        }
        
        return {
            status: 'HEALTHY',
            metrics: {
                queueDepth,
                queueUtilization,
                avgQueueWaitMs: avgWaitMs
            }
        };
    }
}