Deep Dive Concurrency

do {
    acquire(); // 락을 획득 : entry section
        critical section // 임계 구역
    release(); // 락을 반환 : exit section
        remainder section // 나머지 구역
} while(true);

acquire() {
    while(!available)
        ; /* busy wait */
    available = false;
}

release() {
    available =  true;
}

@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    long time = unit.toMillis(waitTime);
    long current = System.currentTimeMillis();
    long threadId = Thread.currentThread().getId();
    Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
    
    // lock acquired
    if (ttl == null) {
        return true; // 락 획득 성공
    }
    
    time -= System.currentTimeMillis() - current;
    if (time <= 0) {
        acquireFailed(waitTime, unit, threadId);
        return false;
    }
    
    current = System.currentTimeMillis();
    
    // 락이 해제될 때 알림을 받기 위해 구독
    CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
    try {
        subscribeFuture.get(time, TimeUnit.MILLISECONDS);
    } catch (TimeoutException e) {
        if (!subscribeFuture.completeExceptionally(new RedisTimeoutException(
                "Unable to acquire subscription lock after " + time + "ms. " +
                        "Try to increase 'subscriptionsPerConnection' and/or 'subscriptionConnectionPoolSize' parameters."))) {
            subscribeFuture.whenComplete((res, ex) -> {
                if (ex == null) {
                    unsubscribe(res, threadId);
                }
            });
        }
        acquireFailed(waitTime, unit, threadId);
        return false;
    } catch (ExecutionException e) {
        acquireFailed(waitTime, unit, threadId);
        return false;
    }

    try {
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
    
        // 락 획득 재시도 루프
        while (true) {
            long currentTime = System.currentTimeMillis();
            ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
            
            // lock acquired
            if (ttl == null) {
                return true; // 락 획득 성공
            }

            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }

            // Semaphore 를 사용한 대기
            // waiting for message
            currentTime = System.currentTimeMillis();
            if (ttl >= 0 && ttl < time) {
                // Semaphore(getLatch())를 사용해 다른 스레드가 락을 해제할 때까지 대기
                commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
            } else {
                commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
            }

            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }
        }
    } finally {
        unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);
    }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
}

long time = unit.toMillis(waitTime);
long current = System.currentTimeMillis();
long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
    return true;
}

time -= System.currentTimeMillis() - current;
if (time <= 0) {
    acquireFailed(waitTime, unit, threadId);
    return false;
}

current = System.currentTimeMillis();
CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);

/**
 * @param channelName threadId
 */
public CompletableFuture<E> subscribe(String entryName, String channelName) {
    AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));
    CompletableFuture<E> newPromise = new CompletableFuture<>();

    semaphore.acquire().thenAccept(c -> {
        if (newPromise.isDone()) {
            semaphore.release();
            return;
        }

        E entry = entries.get(entryName);
        if (entry != null) {
            entry.acquire(); 
            semaphore.release();
            entry.getPromise().whenComplete((r, e) -> {
                if (e != null) {
                    newPromise.completeExceptionally(e);
                    return;
                }
                newPromise.complete(r);
            });
            return;
        }

        E value = createEntry(newPromise);
        value.acquire();

        E oldValue = entries.putIfAbsent(entryName, value);
        if (oldValue != null) {
            oldValue.acquire();
            semaphore.release();
            oldValue.getPromise().whenComplete((r, e) -> {
                if (e != null) {
                    newPromise.completeExceptionally(e);
                    return;
                }
                newPromise.complete(r);
            });
            return;
        }

        RedisPubSubListener<Object> listener = createListener(channelName, value);
        CompletableFuture<PubSubConnectionEntry> s = service.subscribeNoTimeout(LongCodec.INSTANCE, channelName, semaphore, listener);
        newPromise.whenComplete((r, e) -> {
            if (e != null) {
                s.completeExceptionally(e);
            }
        });
        s.whenComplete((r, e) -> {
            if (e != null) {
                entries.remove(entryName);
                value.getPromise().completeExceptionally(e);
                return;
            }
            value.getPromise().complete(value);
        });

    });

    return newPromise;
}

public class RedissonLockEntry implements PubSubEntry<RedissonLockEntry> {

    // volatile 키워드를 사용하여 멀티스레드 환경에서 counter 변수의 가시성을 보장
    private volatile int counter;  // 락 획득 횟수를 추적, Reentrancy
    
    // Semaphore 를 통해 락 획득/해제 시의 동기화 처리
    // 락의 획득 가능 여부를 제어하는 세마포어
    // 다른 스레드의 접근을 블록하는데 사용됨
    // 해당 값을 통해 락을 획득할 수 있는지 알 수 있음
    private final Semaphore latch;  // 동기화를 위한 세마포어
    
    private final CompletableFuture<RedissonLockEntry> promise;  // 비동기 작업 완료를 처리
    
    // ConcurrentLinkedQueue 를 사용하여 thread-safe 한 리스너 관리
    private final ConcurrentLinkedQueue<Runnable> listeners;  // 이벤트 리스너 목록
    
    // permits 이 0으로 초기화된다는 것은 "처음에는 아무도 리소스를 획득할 수 없는 상태"를 의미
    // 락을 처음 생성할 때는 아무도 접근할 수 없게 하기 위함
    // 락이 해제될 때만 다른 스레드가 깨어나서 락 획득을 시도하도록 제어
    // 불필요한 락 획득 시도를 방지하고 효율적인 대기 구현
    public RedissonLockEntry(CompletableFuture<RedissonLockEntry> promise) {
        super();
        this.latch = new Semaphore(0);
        this.promise = promise;
    }

    // 현재 락 획득 횟수를 반환
    public int acquired() {
        return counter;
    }
    
    // 락 카운터를 증가
    public void acquire() {
        counter++;
    }

    // 락 카운터를 감소시키고 현재 값을 반환
    public int release() {
        return --counter;
    }

    // 락 작업의 완료를 비동기적으로 처리하기 위한 CompletableFuture 를 반환
    public CompletableFuture<RedissonLockEntry> getPromise() {
        return promise;
    }

    // 락 상태 변경 시 실행될 리스너를 추가
    public void addListener(Runnable listener) {
        listeners.add(listener);
    }

    public boolean removeListener(Runnable listener) {
        return listeners.remove(listener);
    }

    public ConcurrentLinkedQueue<Runnable> getListeners() {
        return listeners;
    }

    public Semaphore getLatch() {
        return latch;
    }

    @Override
    public String toString() {
        return "RedissonLockEntry{" +
                "counter=" + counter +
                '}';
    }
}

public class RecursiveExample {
    private RedissonLockEntry lockEntry;

    public void recursiveMethod(int depth) {
        lockEntry.acquire();  // 첫 호출: counter = 1
        try {
            if (depth > 0) {
                recursiveMethod(depth - 1);  // 재귀 호출: counter = 2, 3, ...
            }
        } finally {
            lockEntry.release();  // 각 단계마다 counter 감소
        }
    }
}

public void unsubscribe(E entry, String entryName, String channelName) {
    ChannelName cn = new ChannelName(channelName);
    AsyncSemaphore semaphore = service.getSemaphore(cn);
    semaphore.acquire().thenAccept(c -> {
        if (entry.release() == 0) {
            entries.remove(entryName);
            service.unsubscribeLocked(cn)
                    .whenComplete((r, e) -> {
                        semaphore.release();
                    });
        } else {
            semaphore.release();
        }
    });
}

public class AsyncSemaphore {

    private final AtomicInteger counter;
    private final Queue<CompletableFuture<Void>> listeners = new ConcurrentLinkedQueue<>();

    public AsyncSemaphore(int permits) {
        counter = new AtomicInteger(permits);
    }
    
    public int queueSize() {
        return listeners.size();
    }
    
    public void removeListeners() {
        listeners.clear();
    }

    public CompletableFuture<Void> acquire() {
        CompletableFuture<Void> future = new CompletableFuture<>();
        listeners.add(future);
        tryRun();
        return future;
    }

    private void tryRun() {
        while (true) {
            if (counter.decrementAndGet() >= 0) {
                CompletableFuture<Void> future = listeners.poll();
                if (future == null) {
                    counter.incrementAndGet();
                    return;
                }

                if (future.complete(null)) {
                    return;
                }
            }

            if (counter.incrementAndGet() <= 0) {
                return;
            }
        }
    }

    public int getCounter() {
        return counter.get();
    }

    public void release() {
        counter.incrementAndGet();
        tryRun();
    }

    @Override
    public String toString() {
        return "value:" + counter + ":queue:" + queueSize();
    }
}

CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
try {
    subscribeFuture.get(time, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
    if (!subscribeFuture.completeExceptionally(new RedisTimeoutException(
            "Unable to acquire subscription lock after " + time + "ms. " +
                    "Try to increase 'subscriptionsPerConnection' and/or 'subscriptionConnectionPoolSize' parameters."))) {
        subscribeFuture.whenComplete((res, ex) -> {
            if (ex == null) {
                unsubscribe(res, threadId);
            }
        });
    }
    acquireFailed(waitTime, unit, threadId);
    return false;
} catch (ExecutionException e) {
    acquireFailed(waitTime, unit, threadId);
    return false;
}

try {
    time -= System.currentTimeMillis() - current;
    if (time <= 0) {
        acquireFailed(waitTime, unit, threadId);
        return false;
    }
    
    while (true) {
        long currentTime = System.currentTimeMillis();
        ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return true;
        }
    
        time -= System.currentTimeMillis() - currentTime;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
    
        // waiting for message
        // 여기서 Semaphore tryAcquire 를 통해서 latch 를 증가시킴을 알 수 있다. 
        currentTime = System.currentTimeMillis();
        if (ttl >= 0 && ttl < time) {
            // 주어진 ttl 동안 permit 을 기다림
            commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
        } else {
            commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
        }
    
        time -= System.currentTimeMillis() - currentTime;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
    }
} finally {
    unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);
}

@Override
public boolean isHeldByCurrentThread() {
    return isHeldByThread(Thread.currentThread().getId());
}

Semaphore semaphore = new Semaphore(1); // permits = 1

// Thread A가 permit 보유 중
// Thread B, C, D가 대기 중...

// Thread E가 새로 도착
boolean acquired = semaphore.tryAcquire(5, TimeUnit.SECONDS);

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    return evalWriteSyncedAsync(getRawName(), LongCodec.INSTANCE, command,
            "if ((redis.call('exists', KEYS[1]) == 0) " +
                        "or (redis.call('hexists', KEYS[1], ARGV[2]) == 1)) then " +
                    "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                    "return nil; " +
                "end; " +
                "return redis.call('pttl', KEYS[1]);",
            Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}