Semaphore Isolation: Lightweight Concurrency Control

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

import java.util.concurrent.Semaphore;
 
public class BasicSemaphoreBulkhead {
    // Allow max 20 concurrent requests to payment service
    private final Semaphore paymentSemaphore = new Semaphore(20);
    
    public PaymentResult processPayment(PaymentRequest request) {
        // Try to acquire permit without blocking
        if (!paymentSemaphore.tryAcquire()) {
            // No permits available - fail fast
            throw new BulkheadFullException(
                "Payment bulkhead full: " + paymentSemaphore.availablePermits() + 
                " of 20 permits available"
            );
        }
        
        try {
            // Permit acquired - proceed with call
            return paymentService.process(request);
        } finally {
            // Always release permit, even if call fails
            paymentSemaphore.release();
        }
    }
    
    // With timeout option
    public PaymentResult processPaymentWithWait(PaymentRequest request) 
            throws InterruptedException {
        // Wait up to 100ms for a permit
        if (!paymentSemaphore.tryAcquire(100, TimeUnit.MILLISECONDS)) {
            throw new BulkheadFullException("Payment bulkhead full after 100ms wait");
        }
        
        try {
            return paymentService.process(request);
        } finally {
            paymentSemaphore.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

SEMAPHORE ISOLATION (permits=3)
                                
Initial State: 3 permits available
┌─────────────────────────────────────────────┐
│  Semaphore: ● ● ●  (3/3 available)         │
│  In-flight: 0                               │
└─────────────────────────────────────────────┘
 
Request 1 arrives: acquire() → success
┌─────────────────────────────────────────────┐
│  Semaphore: ○ ● ●  (2/3 available)         │
│  In-flight: [R1]                            │
└─────────────────────────────────────────────┘
 
Requests 2,3 arrive: both succeed
┌─────────────────────────────────────────────┐
│  Semaphore: ○ ○ ○  (0/3 available)         │
│  In-flight: [R1] [R2] [R3]                  │
└─────────────────────────────────────────────┘
 
Request 4 arrives: tryAcquire() → REJECTED!
┌─────────────────────────────────────────────┐
│  Semaphore: ○ ○ ○  (0/3 available)         │
│  In-flight: [R1] [R2] [R3]                  │
│  R4 → BulkheadFullException (fast fail)     │
└─────────────────────────────────────────────┘
 
Request 1 completes: release()
┌─────────────────────────────────────────────┐
│  Semaphore: ● ○ ○  (1/3 available)         │
│  In-flight: [R2] [R3]                       │
│  Next request can proceed                   │
└─────────────────────────────────────────────┘

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

import io.github.resilience4j.bulkhead.Bulkhead;
import io.github.resilience4j.bulkhead.BulkheadConfig;
import io.github.resilience4j.bulkhead.BulkheadRegistry;
 
// Configuration
BulkheadConfig config = BulkheadConfig.custom()
    .maxConcurrentCalls(25)               // Max concurrent executions
    .maxWaitDuration(Duration.ofMillis(50)) // Max wait time for permit
    .writableStackTraceEnabled(true)      // Include stack trace in exceptions
    .build();
 
// Create registry (manages multiple bulkheads)
BulkheadRegistry registry = BulkheadRegistry.of(config);
 
// Create named bulkhead
Bulkhead paymentBulkhead = registry.bulkhead("payment");
Bulkhead inventoryBulkhead = registry.bulkhead("inventory");
 
// Decorate a supplier
Supplier<PaymentResult> decoratedSupplier = Bulkhead.decorateSupplier(
    paymentBulkhead,
    () -> paymentService.process(request)
);
 
// Execute with fallback
Try<PaymentResult> result = Try.ofSupplier(decoratedSupplier)
    .recover(BulkheadFullException.class, ex -> {
        metrics.recordBulkheadRejection("payment");
        return PaymentResult.queued("High load - payment queued");
    });
 
// Functional composition with other patterns
CheckedFunction0<PaymentResult> decorated = 
    Decorators.ofSupplier(() -> paymentService.process(request))
        .withBulkhead(paymentBulkhead)
        .withCircuitBreaker(paymentCircuitBreaker)
        .withRetry(paymentRetry)
        .decorate();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

resilience4j:
  bulkhead:
    instances:
      payment:
        maxConcurrentCalls: 25
        maxWaitDuration: 50ms
        writableStackTraceEnabled: true
      inventory:
        maxConcurrentCalls: 50
        maxWaitDuration: 100ms
      recommendations:
        maxConcurrentCalls: 15    # Less critical, smaller limit
        maxWaitDuration: 0ms      # Fail immediately if full
        
  # Thread pool bulkhead for comparison
  thread-pool-bulkhead:
    instances:
      payment-heavy:
        maxThreadPoolSize: 30
        coreThreadPoolSize: 15
        queueCapacity: 50
        keepAliveDuration: 60s

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

@Service
public class PaymentService {
    
    @Bulkhead(name = "payment", fallbackMethod = "paymentFallback")
    public PaymentResult processPayment(PaymentRequest request) {
        return externalPaymentGateway.charge(request);
    }
    
    // Fallback method - same signature + Throwable
    private PaymentResult paymentFallback(PaymentRequest request, Throwable t) {
        if (t instanceof BulkheadFullException) {
            log.warn("Payment bulkhead full, queuing: {}", request.getId());
            return PaymentResult.queued(request.getId());
        }
        log.error("Payment failed", t);
        return PaymentResult.failed("Payment temporarily unavailable");
    }
}

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

CHOOSING BETWEEN THREAD POOL AND SEMAPHORE ISOLATION
 
Start: What type of dependency are you protecting?
│
├─► Fast dependency (< 50ms typical response)
│   │
│   ├─► Reliable? (> 99.9% success rate)
│   │   └─► USE SEMAPHORE
│   │       - Low overhead
│   │       - Preserves thread context
│   │       - Fast fail is acceptable
│   │
│   └─► Occasionally slow/unreliable?
│       └─► USE SEMAPHORE + TIMEOUT
│           - Set aggressive timeout (100-200ms)
│           - Prevent context thread blocking
│
├─► Slow dependency (> 100ms typical)
│   │
│   └─► USE THREAD POOL
│       - Isolation protects caller thread
│       - Slow calls don't block Tomcat workers
│       - Timeouts can interrupt stuck threads
│
├─► External third-party API
│   │
│   └─► USE THREAD POOL (usually)
│       - Unpredictable latency
│       - Need timeout + interrupt capability
│       - Want queue for burst handling
│
└─► Memory-constrained environment
    │
    └─► USE SEMAPHORE
        - Cannot afford thread memory overhead
        - Accept trade-off of weaker isolation
        - Pair with aggressive timeouts

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

@Configuration
public class BulkheadConfiguration {
    
    // Thread pool for critical, potentially slow dependencies
    @Bean
    public ThreadPoolBulkhead paymentBulkhead() {
        return ThreadPoolBulkhead.of("payment",
            ThreadPoolBulkheadConfig.custom()
                .maxThreadPoolSize(30)
                .coreThreadPoolSize(15)
                .queueCapacity(50)
                .build()
        );
    }
    
    // Semaphore for fast, internal services
    @Bean
    public Bulkhead userServiceBulkhead() {
        return Bulkhead.of("user-service",
            BulkheadConfig.custom()
                .maxConcurrentCalls(100)
                .maxWaitDuration(Duration.ZERO)  // Fail immediately
                .build()
        );
    }
    
    // Semaphore for non-critical dependencies
    @Bean
    public Bulkhead recommendationsBulkhead() {
        return Bulkhead.of("recommendations",
            BulkheadConfig.custom()
                .maxConcurrentCalls(20)
                .maxWaitDuration(Duration.ofMillis(10))
                .build()
        );
    }
}
 
@Service
public class OrderService {
    
    public OrderResult createOrder(OrderRequest request) {
        // Critical: Thread pool isolation
        CompletableFuture<PaymentResult> payment = 
            paymentBulkhead.executeCompletionStage(() -> 
                paymentService.charge(request.getPayment())
            ).toCompletableFuture();
        
        // Fast internal: Semaphore isolation
        User user = Bulkhead.decorateSupplier(userServiceBulkhead,
            () -> userService.getUser(request.getUserId())
        ).get();
        
        // Non-critical: Semaphore with fallback
        List<Product> recommendations = Try.ofSupplier(
            Bulkhead.decorateSupplier(recommendationsBulkhead,
                () -> recService.getRecommendations(user.getId())
            )
        ).getOrElse(Collections.emptyList());  // Empty if bulkhead full
        
        return buildOrder(payment.join(), user, recommendations);
    }
}

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

// Access bulkhead metrics
Bulkhead.Metrics metrics = paymentBulkhead.getMetrics();
 
// Available metrics
int available = metrics.getAvailableConcurrentCalls();  // Free permits
int max = metrics.getMaxAllowedConcurrentCalls();       // Total permits
 
// Calculate utilization
double utilization = 1.0 - ((double) available / max);
 
// With Micrometer integration
@Bean
public MeterBinder bulkheadMetrics(BulkheadRegistry registry) {
    return new BulkheadMetrics(registry);
}
 
// Exposed metrics in Prometheus format:
// resilience4j_bulkhead_available_concurrent_calls{name="payment"} 20
// resilience4j_bulkhead_max_allowed_concurrent_calls{name="payment"} 25
 
// Custom metrics for rejections
registry.bulkhead("payment").getEventPublisher()
    .onCallRejected(event -> {
        meterRegistry.counter("bulkhead.rejected",
            "name", event.getBulkheadName()
        ).increment();
        log.warn("Bulkhead {} rejected call", event.getBulkheadName());
    })
    .onCallPermitted(event -> {
        meterRegistry.counter("bulkhead.permitted",
            "name", event.getBulkheadName()
        ).increment();
    })
    .onCallFinished(event -> {
        meterRegistry.timer("bulkhead.duration",
            "name", event.getBulkheadName()
        ).record(event.getWaitDuration());
    });

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

// ❌ WRONG: Permit leak on exception
public Result process(Request r) {
    semaphore.acquire();
    return service.call(r);  // If this throws, permit is never released!
}
 
// ✅ CORRECT: Always release in finally
public Result process(Request r) {
    semaphore.acquire();
    try {
        return service.call(r);
    } finally {
        semaphore.release();
    }
}
 
// ❌ WRONG: Blocking indefinitely
if (semaphore.acquire()) {  // Blocks forever if no permit
 
// ✅ CORRECT: Use tryAcquire with timeout or no wait
if (semaphore.tryAcquire(50, TimeUnit.MILLISECONDS)) {  // Bounded wait
if (semaphore.tryAcquire()) {  // Immediate fail if none available
 
// ❌ WRONG: No fallback for rejection
return Bulkhead.decorateSupplier(bulkhead, () -> service.call(r)).get();
// BulkheadFullException propagates as-is
 
// ✅ CORRECT: Handle rejection gracefully
return Try.ofSupplier(Bulkhead.decorateSupplier(bulkhead, () -> service.call(r)))
    .recover(BulkheadFullException.class, ex -> getDefaultResult())
    .get();