<Spring> Coroutine Actor 이용해서 단일 서버 락 구현하기

📌 Actor

🔵 Actor란?

Actor 모델은 병렬 프로그래밍에서 상태를 가진 개체가 메시지를 주고받으며 비동기적으로 동작하는 개념이다.

즉, 자체적인 상태를 가지고 메시지를 수신하며 이를 처리하는 논리를 가지고 있는 실행 단위이다.

코루틴에서도 스레드 간 동기화를 지원하기 위해 이를 지원해 준다.

actor

상태의 액세스는 단일 스레드로 한정되며, actor와의 통신은 SendChannel을 통해 이뤄진다.

Channel을 통해 전달된 메시지는 하나씩 순서대로 처리되게 된다.

 

🔵 이걸로 어떻게 락을?

actor의 단일 스레드로 작동한다는 특징을 이용하면 될 것 같았다.

단일 스레드라서 작업을 하나씩 처리하게 되고, 이 때문에 락 작업에 있어서 동시성 처리가 가능하다.

또한 코루틴이라서 동시성 처리 + suspend가 가능하지 않을까 생각했다.

 

 

📌 그래서 어떻게 만들었나?

🔵 Pub / Sub

락을 만들려 할 때, 크게 2가지 선택지가 있다

1. 스핀락
2. pub / sub

코루틴을 사용하고 있는 상태에서 스핀락을 사용한다면, 코루틴의 suspend function의 장점을 사용할 수 없다고 판단했다.

이에 pub / sub으로 구현해야 했다.

 

그렇다면 락 획득을 요청하고, 락을 획득했을 때 락 획득을 어떻게 알릴 수 있을까?라는 고민을 해결해야 했다.

Actor는 SendChannel을 통해 메시지를 전달할 뿐, Actor쪽에서 Message Sender쪽으로 특정 응답을 줄 수 없다.

그렇다면 특정 요청에 해당하는 Channel을 Actor로 넘겨주고, Actor에서 락 관련 요청에 관한 응답을 전달한 채널을 통해 보내준다면, 락 획득에 대한 메세지를 받을 수 있을 것이라 생각했다.

 

또한 코루틴을 이용하기 때문에 non-blocking 하게 락에 대한 요청을 할 수 있을 것이라 생각했다.

 

🔵 LockActor

private enum class LockReturn {
    /**
     * 락 실행
     */
    PROCESS_LOCK,

    /**
     * 락 해제됨
     */
    UNLOCK,

    /**
     * 등록된 채널 삭제
     */
    DELETE_CHANNEL,

    /**
     * 락 큐가 안비었음
     */
    NOT_EMPTY_QUEUE,

    /**
     * 락 큐가 비었음
     */
    EMPTY_QUEUE,
    ;
}

private sealed class LockMsg {
    /** 락 획득 시도 */
    class TryLock(val channel: SendChannel<LockReturn>) : LockMsg()

    /** 락 해제 */
    class UnLock(val channel: SendChannel<LockReturn>) : LockMsg()

    /** 등록된 채널 지우기 */
    class DeleteChannel(val channel: SendChannel<LockReturn>) : LockMsg()

    /** 큐 비었는지 확인 */
    class CheckQueueEmpty(val channel: SendChannel<LockReturn>) : LockMsg()
}

@OptIn(ObsoleteCoroutinesApi::class)
private fun lockActor() = CoroutineScope(Dispatchers.IO).actor<LockMsg>(capacity = 1000) {
    // queue 맨 앞 == 락 설정
    val lockQueue = LinkedList<SendChannel<LockReturn>>()

    for (msg in channel) {
        when (msg) {
            is LockMsg.TryLock -> {
                // 큐에 채널 등록하기
                lockQueue.offer(msg.channel)

                // 만약 방금 등록한 채널이 큐의 맨 앞이라면 바로 실행
                if (lockQueue.peek() == msg.channel) {
                    msg.channel.send(LockReturn.PROCESS_LOCK)
                }
            }

            is LockMsg.UnLock -> {
                // 현재 락을 획득한 채널을 큐에서 삭제
                lockQueue.poll()

                // 다음 락 획득 대상 notify하기
                if (lockQueue.peek() != null) {
                    lockQueue.peek().send(LockReturn.PROCESS_LOCK)
                }

                // 락 해제 및 큐 삭제 완료 알리기
                msg.channel.send(LockReturn.UNLOCK)
            }

            is LockMsg.DeleteChannel -> {
                if (lockQueue.peek() == msg.channel) {
                    // 삭제하려는 채널이 큐의 맨 앞일 때, 큐에서 삭제하고 다음꺼 실행
                    lockQueue.poll()
                    if (lockQueue.peek() != null) {
                        lockQueue.peek().send(LockReturn.PROCESS_LOCK)
                    }
                } else {
                    // 삭제하려는 채널이 큐의 맨 앞이 아닐 때, 큐에서만 삭제
                    lockQueue.remove(msg.channel)
                }

                // 삭제 완료 처리 알리기
                msg.channel.send(LockReturn.DELETE_CHANNEL)
            }

            is LockMsg.CheckQueueEmpty -> {
                if (lockQueue.peek() == null) {
                    msg.channel.send(LockReturn.EMPTY_QUEUE)
                } else {
                    msg.channel.send(LockReturn.NOT_EMPTY_QUEUE)
                }
            }
        }
    }
}

 

actor의 기능은 다음과 같다

1. 락 설정
2. 락 해제
3. 등록된 채널 삭제
4. 등록된 채널이 있는지 확인

actor는 락 획득 요청에 대한 관리와 현재 락을 획득한 요청에게 락 획득 메시지를 보내주는 역할을 한다.

subscriber의 채널을 큐에 넣어두고, 락 획득에 정보를 채널을 통해 메세지를 전달하는 일을 한다.

Lock 획득을 요청하면 채널을 queue에 넣어두고, Lock 해제를 요청하면 queue에서 본인 채널을 제거한 후 다른 채널에 락 획득 메세지를 전한다.

또한 에러 처리와 메모리 관리를 큐에 위치하는 채널 제거 및 큐에 등록된 채널이 있는지 확인하는 기능을 구현했다.

 

'락을 획득했다'의 기준은 queue의 맨 앞에 무엇이 있는가이다.

queue 맨 앞에 위치한 채널이 락을 소유했다고 여긴다.

lock을 획득하려 할 때 queue에 채널을 추가한 후 대기한다.

또한 lock을 해제할 때 맨 앞에 위치한 채널, 즉, 현재 락을 소유한 채널을 제거하고 다음 락 획득 대상에 메시지를 보내게 된다.

이러한 방식으로 락의 획득과 해제가 이뤄진다.

 

우선 actor와 락 서비스 간 통신은 채널을 기반으로 진행된다.

lockActor의 기능의 파리미터에 channel을 포함한다.

이 channel이 actor에서 락 서비스로 메시지를 보내는 역할을 한다.

그래서 actor와의 요청 처리는 모두 아래와 같은 방식으로 진행된다.

actor.send(LockMsg.TryLock(channel))
channel.receive()

 

🔵 락 처리

@Component
class SuspendableLockManager : LockManager {
    companion object {
        private const val WAIT_TIME = 3000L
        private const val LEASE_TIME = 3000L
    }

    private val actorMap = ConcurrentHashMap<String, SendChannel<LockMsg>>()

    override suspend fun <T> lock(key: String, block: suspend () -> T): T {
        // lock 관련 리턴 받을 채널
        Channel<LockReturn>().run {
            val channel = this

            // 락 설정
            val actor = tryLock(key, channel)

            try {
                // 로직 실행
                return withTimeout(LEASE_TIME) {
                    block()
                }
            } catch (e: TimeoutCancellationException) {
                // 락 보유 시간 에러 처리
                throw FailToExecuteException(ErrorCode.LOCK_TIMEOUT_ERROR)
            } catch (e: Exception) {
                // 나머지 에러 처리
                throw e
            } finally {
                // 락 해제
                releaseLock(actor, channel)

                // 큐가 빈 액터 삭제
                deleteEmptyQueueActor(channel, key)

                logger.info { actorMap }
            }
        }
    }

    private suspend fun tryLock(key: String, channel: Channel<LockReturn>): SendChannel<LockMsg> {
        val actor = actorMap.compute(key) { _, value ->
            val actor = value ?: lockActor()

            runBlocking(Dispatchers.Unconfined) {
                actor.send(LockMsg.TryLock(channel))
            }

            actor
        } ?: throw FailToExecuteException(ErrorCode.FAIL_TO_GET_LOCK)

        try {
            withTimeout(WAIT_TIME) {
                channel.receive()
            }
        } catch (e: TimeoutCancellationException) {
            // 락 획득 시간 에러 처리
            throw FailToExecuteException(ErrorCode.ACQUIRE_LOCK_TIMEOUT)
        } catch (e: Exception) {
            // 수신 채널 지우기
            actor.send(LockMsg.DeleteChannel(channel))
            channel.receive()

            throw FailToExecuteException(ErrorCode.FAIL_TO_EXECUTE_LOCK)
        }

        return actor
    }

    private suspend fun releaseLock(actor: SendChannel<LockMsg>, channel: Channel<LockReturn>) {
        actor.send(LockMsg.UnLock(channel))
        channel.receive()
    }

    private suspend fun deleteEmptyQueueActor(channel: Channel<LockReturn>, key: String) {
        actorMap.computeIfPresent(key) { _, value ->
            val rtn = runBlocking(Dispatchers.Unconfined) {
                value.send(LockMsg.CheckQueueEmpty(channel))
                channel.receive()
            }

            if (rtn == LockReturn.EMPTY_QUEUE) {
                null
            } else {
                value
            }
        }
    }
}

코드는 크게 4개 부분으로 나뉜다.

1. 락 획득
2. 락 해제
3. actor 삭제

자세한 설명은 후술 하겠다.

 

🔵 락 획득

락의 범위를 최소화하고 싶었다.

그래서 키를 기준으로 actor를 생성해서 사용했다.

이때, actorMap의 조회 및 업데이트에 있어서 동시성을 보장할 필요가 있었다.

그래서 ConcurrentHashMap의 compute를 이용했다.

키가 다르다면 다른 actor를 사용하므로 서로 다른 키를 이용한 락이 서로에게 영향을 줄 수 없게 돼서, 락의 범위를 해당 키를 이용하는 로직으로 한정할 수 있었다.
또한 compute 연산에서 actor에 lock 요청 메시지를 보내도록 했다.

물론 ConcurrentHashMap에서 actor를 가져올 때, compute를 통한 연산을 하기 때문에 서로 간 영향이 있지만, compute 내부 로직에서 락 획득에 대해 대기하지 않기 때문에, 연산 속도에 있어서 큰 영향은 없을 거라 생각한다.

 

이 요청을 받은 actor는 락 획득 관련 채널을 queue에 넣는다.

이 과정이 다 수행되더라도 channel.receive()가 처리되지 않기 때문에, 락을 요청한 코루틴은 suspend 된 상태이다.

 

만약 다른 락이 unlock 하면서 이 채널에 락 획득한다면, actor에서 채널을 통해 메시지를 보내게 되고, 이는 해당 코루틴이 다시 실행됨을 의미한다.

그러므로 락을 얻은 상태에서 주어진 로직을 진행할 수 있게 된다.

 

락 획득 과정에서 에러가 발생하거나, 락 획득에 오랜 시간이 소요될 경우, 에러처리가 필요했다.

먼저 락 획득에 오랜 시간이 소요되는 경우는 withTimeout()으로 처리했다.

또한 획득 과정에서의 에러는 actor의 queue에서 채널을 제거하는 방식으로 처리했다.

이를 통해 actor가 에러가 발생한 요청에 관해 처리하지 않도록 했다.

 

🔵 락 해제

락 해제는 비교적 간단하다.

락을 현재 보유한 대상 즉, queue의 맨 앞 채널을 제거한 후, queue의 맨 앞에 위치한 채널에 락 획득 메시지를 보낸다.

 

🔵 Actor 삭제

이 부분이 없어도 잘 작동할 것이다.

하지만 우리는 서버 메모리 또한 고려해야 한다.

처리해야 하는 요청이 없더라도 해당 요청에 대한 키의 액터가 생성되어 있다면, 이는 삭제되지 않고 남아서 메모리만 차지하게 된다.

그래서 OOM을 막기 위해 사용하지 않는 액터를 지워야 했다.

 

이때 actor를 조회하는 부분과의 락도 고려해야 했다. 

actor를 삭제하는 로직과 actor를 조회, 큐에 채널을 등록하는 로직이 동시에 작동한다면, 에러가 발생할 것이 분명했다.

그래서 이 두 작업에 대한 동시성을 고려하기 위해 ConcurrentHashmap의 compute를 이용했다.

 

이를 통해 사용하지 않는 actor를 지워줌으로써 불필요한 메모리 낭비를 방지했다.

 

🔵 사용

suspend fun vote(user: AuthUser, id: Long, request: CreateVoteHistoryRequest) {
    lockManager.lock(LockKey.getVoteKey(id)) {
        when (request.isCancel) {
            true -> cancelVote(user.uid, id, request.optionId)
            false -> castVote(user.uid, id, request.optionId)
        }
    }
}

lock 함수 파라미터로 서비스 로직을 넘겨주면, 해당 로직에 관해 락을 적용하여 동작하게 된다.

키 생성에는 LockKey라는 클래스를 따로 만들어서 처리했다.

 

 

📌 마무리

일단 코드 머지는 되었으나, 개선해야 할 부분은 남았다 생각한다.

아직은 어딘지 모르겠지만, 데드락이 발생할 것 같은 불안함(?)이 있다.

일단 다 적용하진 않았고, 사용하면서 안정성이 검증된다면, 락이 필요한 곳에 점차 적용해 나갈 생각이다.