자주 사용하는 Java Application 수준에서의 Locking 기법

이 글에서는 개인적으로 자주 사용하는 동시성 제어를 위한 Java 의 Locking 기법에 대해 설명합니다.

 

StampedLock

StampedLock 은 ReentrantReadWriteLock 과 달리 재진입이 불가능하며 낙관적 읽기가 가능합니다.

 

ReentrantReadWriteLock 은 하나의 thread 에서 동일한 lock 을 여러번 잡을 수 있게 설계되어있습니다.

private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
private final Lock readLock = rwLock.readLock();
private final Lock writeLock = rwLock.writeLock();

private int value = 0;

/**
 * 쓰기 락 재진입 예시
 * outerWrite()에서 writeLock을 잡고, innerWrite() 안에서 또 writeLock을 획득
 */
public void outerWrite() {
    writeLock.lock();
    try {
        value++;
        // 내부 메서드에서 다시 writeLock 획득 -> 재진입 가능
        innerWrite();
    } finally {
        writeLock.unlock();
    }
}

private void innerWrite() {
    writeLock.lock();
    try {
        value++;
    } finally {
        writeLock.unlock();
    }
}

 

위 코드와 같이 잡은 Lock 만큼 Unlock 을 호출한다면 Deadlock 의 위험 없이 얼마든지 locking 할 수 있게 지원합니다.

 

하지만 Application Level 에서 이렇게 복잡 다단한 Locking 은 전혀 긍정적이지 않아 보입니다.

Idle Time 을 줄여나가는 것이 종국에는 Application 의 숙제라고 생각하고 있는데요.

Appliction 에 올라간, Memory 에 올라가있는, 빠른 접근이 가능한 데이터 하나를 조작하기 위해

여러개의 Locking 을 하는 것은 굉장한 과소비라는 기분이 듭니다.

 

StampedLock 은 읽기 락과 쓰기 락의 구분은 가능하지만 재진입이 불가능합니다.

public void increment() {
    long stamp = stampedLock.writeLock(); // 배타적 락 획득
    try {
        counter++;
        increment2();
    } finally {
        stampedLock.unlockWrite(stamp);
    }
}

public void increment2() {
    long stamp = stampedLock.writeLock(); // 배타적 락 획득
    try {
        System.out.println("is valid " + stampedLock.validate(stamp));
    } finally {
        stampedLock.unlockWrite(stamp);
    }
}

이 경우 즉시 Deadlock 에 걸리며 Application 은 먹통이 됩니다.

increment 에서 lock 을 해제하기 전에 increment2 에서 lock 을 위한 대기를 하고 있기 때문입니다.

 

그럼 이런 개복치같은 Lock 을 저는 왜 쓰고 있을까요?

 

public int optimisticReadValue() {
    long stamp = stampedLock.tryOptimisticRead(); // 현재 Lock 의 상태(stamp) 반환
    int current = value; // 값을 낙관적으로 읽는다, lock 이 변경되지 않았으므로 유효한 값

    // validate: 읽는 도중에 다른 스레드가 쓰기 락을 획득했으면 false
    if (!stampedLock.validate(stamp)) {
        // 낙관적 읽기가 실패하면 정식 readLock 획득하여 다시 읽음
        stamp = stampedLock.readLock();
        try {
            current = value;
        } finally {
            stampedLock.unlockRead(stamp);
        }
    }
    return current;
}

낙관적 읽기가 가능합니다.

Application 에서 I/O Event 가 아닌 다른 Idle Time 을 발생시키고 싶지 않을 때 유용하게 사용하고 있습니다.

 

위의 코드를 순서에 맞게 설명하자면 아래와 같습니다.

1. tryOptimisticRead() : 낙관적 읽기를 시작했다는 상태 + 스탬프 획득 (stamp 의 값을 사용하여 lock 의 유효성을 판단)

2. value 는 걍 읽은거... (실제로 락을 잡거나 하는 영향도가 없음)

3. validate(stamp) : 이 스탬프가 여전히 유효한지 (다른 스레드에서 lock 을 잡았는지 여부)

3-1. 깨졌다면 readLock 을 획득하여 다시 읽는다.

3-2. 깨지지 않았다면 충돌이 없다고 판단하고 낙관적 읽기가 성공한 것으로 판단한다.

 

이처럼 데이터의 쓰기가 종종 있으나 대체로 lock 을 얻지 않고, 성능에 문제 없이, 동시성을 제어하고 싶을 때 사용합니다.

그렇다면 아예 Lock 을 잡지 않아보면 더 나은 성능을 얻을 수 있지 않을까요?

 

Lock-Free, CAS(Compare And Set)

CAS 는 아예 락을 없애고 충돌 시 재시도하는 방식을 쓰는 방식을 얘기합니다.

하드웨어 수준에서 제공하는 원자적 명령어를 사용해서 값을 읽고, 예상값과 비교하고, 일치하면 새 값을 교체합니다.

java 의 Concurrent Package 에서 제공하고 있습니다.

 

public class LockFreeCounter {
    private final AtomicInteger counter = new AtomicInteger(0);

    public void increment() {
        // 내부적으로 compareAndSet 루프를 돌며 값 업데이트
        counter.incrementAndGet();
    }

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

    public static void main(String[] args) throws InterruptedException {
        final int threadCount = 10;
        final int incrementsPerThread = 1000;

        LockFreeCounter counter = new LockFreeCounter();
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        CountDownLatch latch = new CountDownLatch(threadCount);

        for (int i = 0; i < threadCount; i++) {
            executor.submit(() -> {
                for (int j = 0; j < incrementsPerThread; j++) {
                    counter.increment();
                }
                latch.countDown();
            });
        }

        latch.await();
        executor.shutdown();

        System.out.println(threadCount * incrementsPerThread); // 10000
        System.out.println(counter.getValue()); // 10000
    }
}

위는 ConcurrentPackage 에 포함됨 AtomicInteger 를 사용한 동시성 테스트입니다.

 

10개의 스레드가 1000번의 +1 을 하도록 하였습니다.

따라서 10000 이 기대값이었고, 기대에 맞는 수행을 해주었습니다.

 

public class CASWithConcurrentHashMapExample {

    private final ConcurrentHashMap<String, MyObj> map = new ConcurrentHashMap<>();

    public CASWithConcurrentHashMapExample() {
        // 초기 데이터 삽입
        map.put("key", new MyObj());
    }

    // "key"에 해당하는 MyObj의 field를 +1 (CAS 기반)
    public void increment(String key) {
        map.compute(key, (k, oldValue) -> {
            if (oldValue == null) {
                oldValue = new MyObj();
            }
            oldValue.field++;
            return oldValue;
        });
    }

    public int get(String key) {
        MyObj obj = map.get(key);
        return (obj != null) ? obj.field : 0;
    }

    static class MyObj {
        int field;
    }

    public static void main(String[] args) throws InterruptedException {
        CASWithConcurrentHashMapExample example = new CASWithConcurrentHashMapExample();
        int threadCount = 10;
        int incrementsPerThread = 1000;

        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        CountDownLatch latch = new CountDownLatch(threadCount);

        for (int i = 0; i < threadCount; i++) {
            executor.submit(() -> {
                // 각 스레드마다 1000번 increment
                for (int j = 0; j < incrementsPerThread; j++) {
                    example.increment("key");
                }
                latch.countDown();
            });
        }

        latch.await();       
        executor.shutdown();

        int finalValue = example.get("key");
        System.out.println("Final value: " + finalValue); // 10000
    }
}

이번에는 같은 concurrent package 에 포함되어있는 ConcurrentHashMap 을 사용하였습니다.

increment 에서는 compute 를 사용하여 값을 1씩 증가시키는 동작을 하도록 하였습니다.

마찬가지로 10개의 쓰레드가 1000번을 수행하므로 기대값은 10000입니다.

 

신기한 것은, MyObj 가 그냥 POJO 임에도 동시성에는 문제가 없었습니다.

Read-Modify-Write 원자성은 어떤 값에 대해 읽고 -> 로직을 수행하고 -> 결과 반영 하는 과정을 한 번에 처리하는 것을 의미합니다.

compute 는 RMW 를 지원하고, ConcurrentHashMap 이 지원하는 RMW 메서드는 아래와 같습니다.

• compute(key, remappingFunction) / computeIfPresent / computeIfAbsent
  • 키가 존재하는지 확인 → remappingFunction을 통해 새 값 계산 → 최종 값 저장 과정을 하나의 원자적 연산으로 처리
• putIfAbsent(key, value)
  • 키가 맵에 없으면 새 값 삽입, 있으면 아무것도 안 함 을 원자적으로 보장
  • 다른 스레드가 같은 키에 대해서 동시에 putIfAbsent를 해도, 결국 한 스레드만 성공하고 나머지는 무시
• replace(key, oldValue, newValue)
  • 키의 현재 값이 oldValue와 일치하면 newValue로 교체를 원자적으로 수행
• remove(key, value)
  • 키의 현재 값이 특정 value와 동일하면 제거를 원자적으로 수행
• merge(key, value, remappingFunction)
  • 해당 키에 기존 값이 있으면 remappingFunction(oldValue, value) 결과로 교체, 없으면 value로 삽입을 원자적으로 처리

이런 특성을 통해 성능에 문제가 없으면서도, Lock 을 거의 걸지 않으면서도, 동시성을 제어할 수 있습니다.

하지만 Lock 을 아주 걸지 않는다고 보기는 어렵습니다.

CAS 실패시 Spin Lock 을 걸기도 하고 부분적으로 synchronized block 이 있기도 합니다.

자세한 내용은 아래의 글을 참고해주세요.

 

https://liltdevs.tistory.com/166

Java 코드로 보는 Lock Striping 과 ConcurrentHashMap, CAS (Compare-And-Swap)

 

제가 자주 사용하는 Java Application 에서의 동시성 제어 기법에 대해서 작성해봤습니다.

 

위와 같은 특성으로 어떤 영역을 한정하여 critical section 으로 만들고 싶을 때에는 StampedLock을,

 

어떤 Object 들을 자료구조에 넣고 안전하게 관리하고 싶을 때에는

ConcurrentHashMap 등의 Concurrent Package 를 주로 사용합니다.

혹은 단순한 카운터의 경우 AtomicInteger 를 사용합니다.

 

어떤 방식을 사용하더라도, 목표는 락 범위를 최소화하고 필요한 부분에만 Locking 하는 것으로 하고 있습니다.

Idle time 줄이기와 함께 데이터 무결성 보장 / 동시성 제어를 점점 중요하게 생각하게 되는 것 같습니다.

 

끝!