Locks and Synchronization

/**
 * Get a distributed reentrant lock
 * @param name - unique name of the lock
 * @return RLock instance
 */
public RLock getLock(String name);
public RLock getLock(CommonOptions options);

/**
 * Get a spin lock that uses spinning instead of blocking
 * @param name - unique name of the spin lock
 * @return RLock instance configured for spinning
 */
public RLock getSpinLock(String name);
public RLock getSpinLock(String name, LockOptions.BackOff backOff);

/**
 * Get a fair lock ensuring FIFO order of lock acquisition
 * @param name - unique name of the fair lock
 * @return RLock instance with fair ordering
 */
public RLock getFairLock(String name);
public RLock getFairLock(CommonOptions options);

/**
 * Get a read-write lock for reader-writer synchronization
 * @param name - unique name of the read-write lock
 * @return RReadWriteLock instance with separate read and write locks
 */
public RReadWriteLock getReadWriteLock(String name);
public RReadWriteLock getReadWriteLock(CommonOptions options);

/**
 * Get a fenced lock with fencing tokens for ordering
 * @param name - unique name of the fenced lock
 * @return RFencedLock instance with fencing token support
 */
public RFencedLock getFencedLock(String name);
public RFencedLock getFencedLock(CommonOptions options);

/**
 * Create a multi-lock that acquires multiple locks atomically
 * @param locks - array of locks to acquire together
 * @return RLock instance representing the multi-lock
 */
public RLock getMultiLock(RLock... locks);
public RLock getMultiLock(String group, Collection<Object> values);

// Main distributed lock interface
public interface RLock extends Lock, RLockAsync {
    void lock();
    void lock(long leaseTime, TimeUnit unit);
    void lockInterruptibly() throws InterruptedException;
    void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException;
    
    boolean tryLock();
    boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException;
    boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;
    
    void unlock();
    Condition newCondition();
    
    // Redisson-specific methods
    void forceUnlock();
    boolean isLocked();
    boolean isHeldByThread(long threadId);
    boolean isHeldByCurrentThread();
    int getHoldCount();
    long remainTimeToLive();
    String getName();
}

// Read-write lock interface
public interface RReadWriteLock extends ReadWriteLock {
    RLock readLock();
    RLock writeLock();
    String getName();
}

// Fenced lock with fencing tokens
public interface RFencedLock extends RLock {
    Long lockAndGetToken();
    Long lockAndGetToken(long leaseTime, TimeUnit unit);
    boolean tryLockAndGetToken();
    boolean tryLockAndGetToken(long waitTime, TimeUnit unit) throws InterruptedException;
    boolean tryLockAndGetToken(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;
    
    Long getToken();
}

import org.redisson.api.*;
import java.util.concurrent.TimeUnit;

// Basic lock usage
RLock lock = redisson.getLock("myLock");
lock.lock();
try {
    // Critical section - only one thread can execute this across all instances
    System.out.println("Protected code execution");
} finally {
    lock.unlock(); // Always unlock in finally block
}

// Lock with automatic expiration (lease time)
RLock lockWithLease = redisson.getLock("timedLock");
lockWithLease.lock(10, TimeUnit.SECONDS); // Auto-unlock after 10 seconds
try {
    // Protected code with timeout
} finally {
    if (lockWithLease.isHeldByCurrentThread()) {
        lockWithLease.unlock();
    }
}

// Try lock with timeout
RLock tryLock = redisson.getLock("conditionalLock");
if (tryLock.tryLock(5, TimeUnit.SECONDS)) {
    try {
        // Got the lock within 5 seconds
    } finally {
        tryLock.unlock();
    }
} else {
    System.out.println("Could not acquire lock within timeout");
}

// Fair lock for FIFO ordering
RLock fairLock = redisson.getFairLock("fairQueue");
fairLock.lock();
try {
    // Threads acquire this lock in the order they requested it
} finally {
    fairLock.unlock();
}

// Read-write lock for reader-writer scenarios
RReadWriteLock rwLock = redisson.getReadWriteLock("dataLock");

// Reader threads
RLock readLock = rwLock.readLock();
readLock.lock();
try {
    // Multiple readers can access simultaneously
    String data = readSharedData();
} finally {
    readLock.unlock();
}

// Writer thread
RLock writeLock = rwLock.writeLock();
writeLock.lock();
try {
    // Exclusive access for writing
    writeSharedData("new data");
} finally {
    writeLock.unlock();
}

// Multi-lock for atomic operations across multiple resources
RLock lock1 = redisson.getLock("resource1");
RLock lock2 = redisson.getLock("resource2");
RLock multiLock = redisson.getMultiLock(lock1, lock2);
multiLock.lock();
try {
    // Both locks acquired atomically
} finally {
    multiLock.unlock();
}

// Fenced lock with tokens
RFencedLock fencedLock = redisson.getFencedLock("fencedResource");
Long token = fencedLock.lockAndGetToken();
try {
    // Use token for ordering operations
    processWithToken(token);
} finally {
    fencedLock.unlock();
}

/**
 * Get a distributed semaphore
 * @param name - unique name of the semaphore
 * @return RSemaphore instance
 */
public RSemaphore getSemaphore(String name);
public RSemaphore getSemaphore(CommonOptions options);

/**
 * Get a permit-expirable semaphore where permits can expire
 * @param name - unique name of the permit expirable semaphore
 * @return RPermitExpirableSemaphore instance
 */
public RPermitExpirableSemaphore getPermitExpirableSemaphore(String name);
public RPermitExpirableSemaphore getPermitExpirableSemaphore(CommonOptions options);

// Standard distributed semaphore
public interface RSemaphore extends RObject, RExpirable {
    void acquire() throws InterruptedException;
    void acquire(int permits) throws InterruptedException;
    void acquireUninterruptibly();
    void acquireUninterruptibly(int permits);
    
    boolean tryAcquire();
    boolean tryAcquire(int permits);
    boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException;
    boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException;
    
    void release();
    void release(int permits);
    
    int availablePermits();
    int drainPermits();
    boolean hasQueuedThreads();
    int getQueueLength();
    
    boolean trySetPermits(int permits);
    void addPermits(int permits);
    boolean updateLeaseTime(String permitId, long leaseTime, TimeUnit unit);
}

// Semaphore with expirable permits
public interface RPermitExpirableSemaphore extends RObject, RExpirable {
    String acquire() throws InterruptedException;
    List<String> acquire(int permits) throws InterruptedException;
    String acquire(long leaseTime, TimeUnit unit) throws InterruptedException;
    List<String> acquire(int permits, long leaseTime, TimeUnit unit) throws InterruptedException;
    
    String tryAcquire();
    List<String> tryAcquire(int permits);
    String tryAcquire(long waitTime, TimeUnit unit) throws InterruptedException;
    String tryAcquire(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;
    List<String> tryAcquire(int permits, long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;
    
    boolean tryRelease(String permitId);
    int tryRelease(Collection<String> permitIds);
    void release(String permitId);
    void release(Collection<String> permitIds);
    
    int availablePermits();
    boolean trySetPermits(int permits);
    void addPermits(int permits);
    boolean updateLeaseTime(String permitId, long leaseTime, TimeUnit unit);
}

// Basic semaphore usage
RSemaphore semaphore = redisson.getSemaphore("connectionPool");
semaphore.trySetPermits(10); // Set maximum permits to 10

// Acquire permit
semaphore.acquire(); // Blocks until permit available
try {
    // Use limited resource (e.g., database connection)
    processWithLimitedResource();
} finally {
    semaphore.release(); // Return permit
}

// Try acquire with timeout
if (semaphore.tryAcquire(5, TimeUnit.SECONDS)) {
    try {
        // Got permit within 5 seconds
    } finally {
        semaphore.release();
    }
} else {
    System.out.println("No permits available");
}

// Acquire multiple permits
RSemaphore batchSemaphore = redisson.getSemaphore("batchProcessor");
batchSemaphore.acquire(3); // Acquire 3 permits for batch operation
try {
    // Process batch requiring 3 resources
} finally {
    batchSemaphore.release(3);
}

// Expirable permits semaphore
RPermitExpirableSemaphore expirableSemaphore = redisson.getPermitExpirableSemaphore("tempResources");
String permitId = expirableSemaphore.acquire(30, TimeUnit.SECONDS); // Permit expires in 30 seconds
if (permitId != null) {
    try {
        // Use resource with automatic cleanup
    } finally {
        expirableSemaphore.release(permitId);
    }
}

// Multiple permits with expiration
List<String> permitIds = expirableSemaphore.acquire(3, 60, TimeUnit.SECONDS);
try {
    // Use 3 resources that expire in 60 seconds
} finally {
    expirableSemaphore.release(permitIds);
}

/**
 * Get a distributed countdown latch
 * @param name - unique name of the countdown latch
 * @return RCountDownLatch instance
 */
public RCountDownLatch getCountDownLatch(String name);
public RCountDownLatch getCountDownLatch(CommonOptions options);

public interface RCountDownLatch extends RObject, RExpirable {
    void await() throws InterruptedException;
    boolean await(long timeout, TimeUnit unit) throws InterruptedException;
    
    void countDown();
    long getCount();
    boolean trySetCount(long count);
}

// Coordination example: wait for all workers to complete initialization
RCountDownLatch initLatch = redisson.getCountDownLatch("initLatch");
int workerCount = 5;
initLatch.trySetCount(workerCount);

// Worker threads
// Each worker does this after initialization:
// initLatch.countDown();

// Main thread waits for all workers
initLatch.await(); // Blocks until count reaches 0
System.out.println("All workers initialized, starting main processing");

// With timeout
RCountDownLatch processLatch = redisson.getCountDownLatch("processLatch");
processLatch.trySetCount(3);

if (processLatch.await(30, TimeUnit.SECONDS)) {
    System.out.println("All processes completed within timeout");
} else {
    System.out.println("Timeout waiting for processes");
}

// Racing start example
RCountDownLatch startLatch = redisson.getCountDownLatch("raceStart");
startLatch.trySetCount(1);

// All participants wait
startLatch.await();
System.out.println("Race started!");

// Coordinator starts the race
// startLatch.countDown(); // This releases all waiting threads

/**
 * Get atomic long for distributed atomic operations
 * @param name - unique name of the atomic long
 * @return RAtomicLong instance
 */
public RAtomicLong getAtomicLong(String name);

/**
 * Get atomic double for distributed atomic operations
 * @param name - unique name of the atomic double  
 * @return RAtomicDouble instance
 */
public RAtomicDouble getAtomicDouble(String name);

/**
 * Get long adder for high-throughput distributed counting
 * @param name - unique name of the long adder
 * @return RLongAdder instance
 */
public RLongAdder getLongAdder(String name);

/**
 * Get double adder for high-throughput distributed summing
 * @param name - unique name of the double adder
 * @return RDoubleAdder instance  
 */
public RDoubleAdder getDoubleAdder(String name);

// Distributed atomic long
public interface RAtomicLong extends RObject, RExpirable {
    long get();
    void set(long newValue);
    long getAndSet(long newValue);
    
    boolean compareAndSet(long expect, long update);
    long addAndGet(long delta);
    long getAndAdd(long delta);
    long incrementAndGet();
    long getAndIncrement();
    long decrementAndGet();
    long getAndDecrement();
    
    long updateAndGet(LongUnaryOperator updateFunction);
    long getAndUpdate(LongUnaryOperator updateFunction);
    long accumulateAndGet(long x, LongBinaryOperator accumulatorFunction);
    long getAndAccumulate(long x, LongBinaryOperator accumulatorFunction);
}

// Distributed atomic double
public interface RAtomicDouble extends RObject, RExpirable {
    double get();
    void set(double newValue);
    double getAndSet(double newValue);
    
    boolean compareAndSet(double expect, double update);
    double addAndGet(double delta);
    double getAndAdd(double delta);
}

// High-performance distributed counter
public interface RLongAdder extends RObject, RExpirable {
    void add(long x);
    void increment();
    void decrement();
    long sum();
    void reset();
    long sumThenReset();
}

// Distributed counter
RAtomicLong counter = redisson.getAtomicLong("globalCounter");
counter.set(0);

// Atomic operations across multiple JVMs
long newValue = counter.incrementAndGet(); // Thread-safe increment
long previousValue = counter.getAndAdd(5); // Add 5 and return previous value

// Compare and swap
long current = counter.get();
boolean success = counter.compareAndSet(current, current * 2);

// High-performance counting with RLongAdder
RLongAdder adder = redisson.getLongAdder("highThroughputCounter");
// Multiple threads can call this concurrently with high performance
adder.increment();
adder.add(10);

// Get sum when needed (more expensive operation)
long total = adder.sum();
System.out.println("Total count: " + total);

// Common options for most synchronization primitives
public class CommonOptions extends PlainOptions {
    private String name;
    private Codec codec;
    
    public CommonOptions name(String name);
    public CommonOptions codec(Codec codec);
}

// Lock-specific options
public class LockOptions {
    public enum BackOff {
        EXPONENTIAL_BACKOFF,
        LINEAR_BACKOFF, 
        CONSTANT_BACKOFF
    }
    
    public static class BackOffPolicy {
        private BackOff backOff;
        private long initialDelay;
        private long maxDelay;
        private double multiplier;
        
        // Configuration methods
        public BackOffPolicy backOff(BackOff backOff);
        public BackOffPolicy initialDelay(long initialDelay, TimeUnit timeUnit);
        public BackOffPolicy maxDelay(long maxDelay, TimeUnit timeUnit);  
        public BackOffPolicy multiplier(double multiplier);
    }
}

// Transaction options for coordinated operations
public class TransactionOptions {
    private long responseTimeout = 3000;
    private int retryAttempts = 3;
    private long retryInterval = 1500;
    private SyncMode syncMode = SyncMode.ACK;
    
    public TransactionOptions responseTimeout(long responseTimeout, TimeUnit unit);
    public TransactionOptions retryAttempts(int retryAttempts);  
    public TransactionOptions retryInterval(long retryInterval, TimeUnit unit);
    public TransactionOptions syncMode(SyncMode syncMode);
}