CRDT

CRDT?

Conflict-free Replicated Data Type

분산 컴퓨팅 환경에서 네트워크로 연결된 노드 간에 복제되어서 관리되는 데이터를 동기화하기 위해 사용하는 데이터 구조이다.

CRDT 는 아래 3가지 주요 특성을 가진다.

각 노드는 로컬에서 독립적으로 데이터 업데이트가 가능하다.

데이터 타입에 내장된 로직으로 다른 복제본과의 불일치를 해결한다.

특정 시점에 각 노드의 복제본이 서로 다를 수 있지만 최종적으로는 일관성을 보장한다.

즉, CRDT 는 각 서버에 복제된 상태의 일관성을 보장하기 위해 하나의 Primary Source (예를들어 Database) 를 사용하지 않는다.

CRDT 의 장점

분산 시스템에서는 네트워크가 중간에 걸쳐있어서 데이터 동기화에 어려움이 있다.

전통적으로는 락 (혹은 락-ish 한 어떤 것) 을 사용해서 해결하지만 이는 성능 저하의 원인이 되고 단일 실패지점이 되게 된다.

CRDT 는 위와같은 문제점을 해결하면서 데이터의 일관성을 보장한다.

CRDT Types

CRDT 는 크게 두가지 타입으로 분류한다.

Operation Based CRDT

update 작업을 전파한다.

변경 사항만을 전파하므로 네트워크 대역폭 사용이 효율적이다.

통신 protocol 이 엄격해야 한다.

작업은 한번만 전달되어야 한다.
작업이 누락되지 않아야 한다.
casual ordering 이 고려되어야 한다.

State Based CRDT

설계와 구현이 상대적으로 단순하다.

전체 상태를 주기적으로 전파한다.

각 노드는 현재 상태를 다른 노드와 주기적으로 공유한다.

복제본간 발생한 이격을 병합한다.

ORSWOT

여러가지 CRDT 구현들이 있는데 그중 ORSWOT 을 간단히 알아보자. 정식 명칭은 Observed-Remove Set Without Tombstones 이다.

ORSWOT 은 Phoenix Tracker 에서 사용하는 자료구조이다. Phoenix Tracker 는 서버와 실시간으로 연결된 소켓에 대한 정보를 제공하는 추상화된 Module 이다. OR-Set 이라는 자료구조를 개선해서 삭제된 요소를 추적하는 데이터를 최소화 한 자료구조이다.

특징

개별 노드마다 버전 별도로 관리된다.

Set 의 각 Element 는 노드의 버전정보와 함께 저장된다.

병합 시 각 노드의 버전을 참고해서 상태를 도출한다.


class ORSWOT:
    def __init__(self):
        # 전체 시스템의 버전 시계
        self.clock = {
            "node1": 0,  # 노드별 카운터
            "node2": 0,
            # ... 다른 노드들
        }
        
        # 실제 데이터를 저장하는 공간
        self.entries = {
            # key: 실제 데이터
            # value: 해당 데이터의 버전 정보
            "A": {"node1": 1},  # A는 node1의 버전 1에서 추가됨
            "B": {"node2": 3}   # B는 node2의 버전 3에서 추가됨
        }
        
        # 나중에 처리해야 할 삭제 작업들
        self.deferred = {
            # 미래의 버전 시계: 그 시점에 삭제될 항목들
            "{"node1": 5}": {"C", "D"}  # 노드1의 버전이 5가 되면 C와 D를 삭제
        }

    def add(self, item, node_id):
        # 1. 노드의 카운터 증가
        self.clock[node_id] += 1
        
        # 2. 항목 추가
        self.entries[item] = {
            node_id: self.clock[node_id]
        }
    
    def remove(self, item, context):
        if item in self.entries:
            # 현재 시점보다 이전 버전의 항목이면 바로 삭제
            if self._is_obsolete(self.entries[item]):
                del self.entries[item]
            else:
                # 나중에 삭제하기 위해 deferred에 추가
                self.deferred[context].add(item)
    
    def merge(self, other_orswot):
        # 1. entries 병합
        for item, version in other_orswot.entries.items():
            if item not in self.entries:
                # 우리가 모르는 새로운 항목
                self.entries[item] = version
            else:
                # 양쪽 다 있는 항목은 버전 비교 필요
                self._merge_versions(item, version)
        
        # 2. 시계 병합
        for node, counter in other_orswot.clock.items():
            self.clock[node] = max(
                self.clock.get(node, 0),
                counter
            )
        
        # 3. 지연된 삭제 작업 병합
        self._merge_deferred(other_orswot.deferred)

위는 ORSWOT 을 간단하게 의사코드로 표현한 버전이다.