Привет, Хабр! Меня зовут Георгий Семёнов, в VK я занимаюсь разработкой в команде инфраструктуры рекомендательных систем, а в Университете ИТМО начинаю свой аспирантский путь в области децентрализованных коллаборативных сред.

В этой вводной статье я попытаюсь спекулятивно определить CRDT-типы, которые сегодня выступают передовым подходом для создания коллаборативных приложений реального времени. Я намеренно не коснусь многообразия теоретических и практических результатов в области и попробую принять отстраненную позицию, чтобы представить некое «интуиционистское» построение реплицируемых типов с исполнимыми иллюстрациями на Scala.

Коллаборативные приложения

В 1980-x в связи с появлением персональных компьютеров с графическим интерфейсом и мышкой появились перспективы автоматизации бизнеса с помощью внедрения офисных приложений в повседневную жизнь. С тех пор организация совместной работы в электронных средах — цифровой коллаборации — изучается несколькими научными областями — Computer-supported cooperative work (CSCW) и Human-computer interaction (HCI).

Для цифровой коллаборации можно предложить два ключевых основания для классификации:

1. Синхронность и асинхронность — происходит ли взаимодействие между пользователями в реальном времени. В то время, как существуют строго синхронные (например, видеозвонки) или строго асинхронные (например, работа с кодом) типы коллаборации, в зрелых цифровых средах они склонны перетекать друг в друга. Например, запись живого видеозвонка далее становится артефактом встречи, а работа над текстовым документом в реальном времени может прерываться для неколлаборативных правок.

2. Оффлайн и онлайн — происходит ли взаимодействие между пользователями «вживую» или через сетевое подключение. Например, в переговорной с участниками оффлайн и онлайн мы держим в голове необходимость говорить с нужной для микрофона громкостью и смотреть в камеру при обращении к собеседнику «за экраном». Ещё пример: в офисе для проверки кода, часто проще остаться на рабочих местах, поскольку — казалось бы, парадоксально — какие-то синхронные процессы удобнее выполнять онлайн.

Нам, как разработчикам, целесообразно иметь в виду описанные выше модальности, чтобы правильно проектировать коллаборативные приложения с поддержкой различных типов коллаборации.

О коллаборативных приложениях мы вынуждены рассуждать в терминах распределённых систем и систем управления базами данных, поскольку коллаборация — это применение конкурентных пользовательских транзакций к какому-то «оракулу» — глобальному источнику правды о том, что есть консолидированное состояние данных.

Невозможность сходу придумать универсальное идеальное коллаборативное программное решение постулирует классическая для распределённых систем CAP-теорема, предложенная Эриком Брюером в 2000 году:

В распределённой системе возможно обеспечить не более двух свойств из следующей тройки:

• С — Consistency. Система обеспечивает согласованность данных во всех её вычислительных узлах. Иными словами, от узлов системы без этого свойства в один момент времени можно получить противоречащие ответы.

• A — Availability. Система гарантирует наличие ответа за обозначенное время. В противном случае системе позволяется «виснуть» (Timeout) или «лежать» (Denial of Service).

• P — Partition Tolerance. Система продолжает функционировать при частичном отказе её вычислительных узлов. Без этого свойства система перестанет функционировать, например, если отвалится диск, память или сеть.

Утверждение, как известно, не является формальным, и поэтому любой разговор вокруг него сводится к «маханию руками». К описанию цифровой коллаборации оно вообще не очень подходит, поскольку мы немедленно сталкиваемся с подменой понятий «вычислительный узел» и «коллаборативный актор». Но попробую спекулятивно проиллюстрировать свойства CAP-теоремы на некоторых режимах коллаборации:

• С — Consistency. Все участники видеозвонка должны наблюдать за интерактивной доской в реальном времени и видеть одно и то же, иначе смысл коммуникации пропадёт.

• A — Availability. При работе с документом при кратковременном отключении сети содержание документа и локально применённые изменения не должны потеряться.

• P — Partition Tolerance. При параллельной работе с кодом участники могут работать автономно с последующем слиянием своих изменений через PR/MR.

При проектировании коллаборативного пользовательского опыта мы всегда ищем баланс между целостностью (Consistency) и доступностью (Availability) данных, с которыми работаем. Давайте рассмотрим, как этот компромисс реализуется на практике.

Пессимистическая репликация

При коллаборативной работе мы избыточно копируем данные для всех коллаборативных акторов — это называется репликацией.

Пессимистическая репликация — режим репликации, при котором объект блокируется на запись для одного пользователя. «Пессимизм» подхода заключается в принципиальном отказе от конкурентных модификаций данных, что заведомо ограничивает возможность коллаборации. Здесь мы обеспечиваем свойство строгой согласованности (strong consistency), жертвуя доступностью.

Схема записи с блокировкой выглядит следующим образом: централизованно или с помощью консенсуса блокируют запись, загружают центральную ревизию данных, модифицируют их, актуализируют центральную ревизию и снимают блокировку.

Достоинства подхода:

• Последовательная история изменений с естественным тотальным порядком.

• Интерпретируемость коллаборативного процесса.

• Отсутствие необходимости объединять конкурентные правки.

• Удобство для простых атомарных и сложных слабоструктурированных данных.

Недостатки:

• Низкая зрелость коллаборации из-за невозможности работать синхронно.

• Использование блокировки в распределённом контуре.

Пессимистический подход широко применяется в СУБД, в системах управления версиями, в облачных хранилищах и приложениях с «толстыми» клиентами (например, в CAD-системах).

Двумя ключевыми режимами синхронизации здесь выступают:

• полная репликация: копирование всей ревизии данных (например, rsync);

• инкрементальная репликация: распространение патчей — минимальных разностных преобразований над данными (например, diff, или иначе 2-way merge).

Оптимистическая репликация

Оптимистическая репликация — режим репликации, при котором пользователям разрешается работать с данными параллельно. «Оптимизм» заключается в том, что у нас получится консолидировать все конкурентные изменения с помощью некоторой процедуры слияния (merge).

В рамках этого подхода мы обеспечиваем базовую доступность данных (base availability), а от классической согласованности переходим к согласованности в конечном счёте (eventual consistency). Дополнительно проговорив, что мы разрешаем гибкое состояние (soft state) для фоновой синхронизации между репликами данных, мы получили набор требований, который известен как BASE.

Итак, оптимистический подход фактически разрешает свободную децентрализованную коллаборативную работу с данными, предполагая их дальнейшую синхронизацию. Приведу несколько причин, почему это остаётся весьма нетривиальным:

1. Разрешение конфликтов. При совместной работе невозможно избежать противоречащих друг другу правок — конфликтов. К ним относится широкий класс ситуаций, среди которых:

   1. Изменения в атомах. Невозможно семантически корректно определить результат параллельной работы над полем, типу которого соответствуют неделимые, «атомарные» значения.

   2. Конкурентные вставки и удаления в списках. Для разрешения такого конфликта недостаточно просто упорядочить операции. Требуется вручную убедиться, что конечное преобразование удовлетворяет пользовательским ожиданиям всех коллаборативных акторов.

   3. Ссылочные и алгебраические ограничения. Например, не ясно, как поддержать ограничения кардинальности в коллекциях или ограничения ссылочной целостности.

2. Сборка мусора. В децентрализованном контуре создаются и удаляются экземпляры сущностей, для полного удаления которых требуется полный консенсус. То же самое — уменьшать и стабилизировать — хотелось бы делать с экземплярами операций, иначе объём метаданных будет стремительно расти.

На данный момент не существует целостного подхода к решению сформулированных выше проблем, однако сильным фундаментом сейчас остаётся идея журнала операций. Как минимум, избежать конфликтов при слиянии данных позволяют следующие подходы:

1. Тотальное упорядочивание операций. Каждой операции с данными присваивается временная метка: хронологическая (unix timestamp) или логическая (часы Лампорта). Все операции линейно упорядочиваются по времени и применяются в этом порядке.

2. Каузальное упорядочивание операций. Векторные часы позволяют задать частичный порядок операций, который аналогичен ветвлениям в git. Причинно-следственные отношения образуют полурешётку, которую можно представить в виде диаграммы Хассе. Таким образом, любую операцию относительно любой другой можно охарактеризовать как произошедшую до, после или конкурентно.

3. Переупорядочивание операций. Существуют операциональные базисы, для которых можно поменять местами операции с заменой их аргументов. Такой подход носит название операциональные преобразования (OT = operational tranformations) и, например, используется в Google Docs для коллаборативной работы над текстом.
   
   

   

4. Допущение конфликтов данных:

   1. Приоритеты. Конфликтующие операции разрешаются в соответствии с некоторым бизнес-правилом. Например, изменения от актора с более широким ролевым доступом приоритизируются, а другие операции утрачивают свою силу.

   2. Семантическая реконсиляция. К типу применяется нечёткая процедура слияния, результатом которой является максимальное подмножество, консолидирующее бесконфликтный набор операций, а также набор конфликтов для ручного разрешения пользователями.

Достоинства подхода:

• Базовая доступность данных для чтения и модификации.

• Поддержка централизованной-децентрализованной и синхронной-асинхронной коллаборации.

• Возможность git-подобной интерпретируемой «ветвистости».

Недостатки: необходимость fine-grained метода слияния данных, ревизии которых могут конфликтовать.

Примечательно, что децентрализованное приложение оптимистической репликации можно «пессимизировать», добавив распределённые блокировки, но не наоборот.  Поэтому наиболее общим случаем является вариант с оптимистической репликацией.

Получается, что оптимистическая репликация данных не выглядит очень оптимистичной: для её реализации требуется найти баланс между корректностью алгоритмов слияния и их выразительностью применительно к бизнес-логике целевого приложения. Но куда уж деться, это единственный путь навстречу коллаборации в реальном времени.

Синхронизация данных

Обычно в компьютерных науках под «синхронизацией» понимают организацию работы нескольких потоков в общей памяти с помощью мьютексов, семафоров, барьеров и т. п. В коллаборативных архитектурах же рассматривают синхронизацию данных как механизм обеспечения согласованности между узлами.

Существуют некоторые простые случаи, когда синхронизация данных становится тривиальной:

1. Данные с областью значений в полурешётке. Более общий случай, когда для любых ревизий можно определить их наименьшую верхнюю границу (least upper bound), что и является функцией слияния. Формально, полурешётка — полугруппа с идемпотентной и коммутативной операцией. Её можно построить как подвешенное дерево, в котором разрешили перекрёстные рёбра.
   
   

   

2. Работа с независимыми данными. Декомпозиция структурированных данных позволяет работать с ними изолированно. На самом деле, это частный случай прошлого, в котором область значений полурешётки — это декартово произведение областей значений независимых данных.

Давайте рассмотрим примеры синхронизации данных без использования дополнительных метаданных.

Синхронизация в постановке «2-way merge». Пусть у нас есть две ревизии данных A и B, требуется определить функцию слияния для получения M — их консолидированного значения. Функция должна быть коммутативной, идемпотентной и ассоциативной.

Для типов с отображением их домена на полурешетку всё просто: такой функцией является минимальная верхняя граница (least upper bound). Например, для множеств со значениями-атомами или для ассоциативных массивов с запретом на правки по «чужим» ключам.

В общем случае двустороннее слияние определить нельзя — например, для ассоциативных массивов могут возникнуть конфликты значений по ключам, а корректную семантику слияния на строках определить совсем трудно.

Если при слиянии данных мы можем воспользоваться знаниями о наименьшем общем предке (иными словами, LCA — least common ancestor), к которому применяли конкурентные модификации, то получится более интересный случай — трёхстороннее слияние.

Синхронизация в постановке «3-way merge». Пусть у нас есть две ревизии данных A и B, являющиеся модификациями общей ревизии M. Аналогично, требуется определить функцию слияния.

Фактически мы извлекаем операциональные разности A/M и B/M, а затем ищем в постановке «2-way merge» результирующую операциональную разность, которую применим к M для получения требуемого результата. Такая постановка для слияния текстовых данных на основе алгоритма Майерса реализована в diff3 и применяется в git. Впрочем, без конфликтов обойтись не получится, например, при конкурентных вставках или удалениях по индексу в списке.

Можно заметить, что обычный «2-way merge» — это 2-way merge на данных, а «3-way merge» — 2-way merge на операциях.

CRDT

CRDT (conflict-free replicated data types) — это оптимистически реплицируемые типы данных с механизмом синхронизации, обладающим формальным свойством строгой конечной согласованности (strong eventual consistency = SEC).

Давайте разберёмся с определением — что означает «отсутствие конфликтов», заложенное в нём? Существует два типа ситуаций, которые называют конфликтами:

1. Конфликт данных (data conflict). Это ситуации, для которых функция синхронизации в постановке 2-way merge на данных или операциях не определена. Важно, что слияние должно быть функционально «чистым», то есть использование алгоритмов консенсуса недопустимо. Чтобы избежать конфликта данных, достаточно доопределить функцию для конфликтных случаев на основе детерминированного разрешения конфликта данных, например, на основе упорядочивания операций (рассматривалось выше) или детерминированного выбора приоритетной операции (в этом случае потеряем одну из ветвей модификации).
   
   Конфликт данных — это, если угодно, расходящееся распределённое вычисление (⊥). Если предложить теоретико-типовую метафору, слияние — это процедура, по формализму и «синтаксической» сути родственная унификации типовых термов в λ-исчислении.
   
   Так вот, в CRDT-типах формально гарантируется отсутствие конфликтов данных. Иными словами, функция синхронизации в CRDT тотальна.

2. Конфликт пользовательских намерений (user intention conflict). Это аномалии коллаборативной работы, при которых утрачивается или искажается семантика примененной пользователем модификации по сравнению с той, что подразумевается в однопользовательском контексте. Несмотря на то, что абстрактные типы данных обладают понятным интерфейсом (например, примитивные значения или списки), композиция конкурентных операций может приводить к неожиданным результатам.
   
   Например, при коллаборативной работе с кодом как с текстом могут происходить дублирования include-ов или import-ов. Несмотря на то, что бесконфликтное слияние привело к синтаксически корректному результату, изначальная семантика модификации как добавления в точности одного include-утверждения была нарушена. Таким образом, произошел конфликт пользовательского намерения.
   
   Итак, CRDT-типы (впрочем, как и другие подходы) не смогут избежать конфликтов пользовательских намерений. Однако выбор правильного CRDT-типа для конкретной ситуации может помочь избежать или минимизировать подобные аномалии.

Осталось разобраться с определением строгой конечной согласованности (SEC), которую гарантируют CRDT-типы, — корректные реплики объекта со свойством конечной согласованности с одинаковыми наборами доставленных обновлений обладают эквивалентным состоянием. Иными словами, в присутствии сетевого прогресса компоненты сетевой связности являются классами эквивалентности по ревизиям данных, к которым они относятся.

Свойство SEC является центральным, поскольку оно гарантирует принципиальную невозможность рассинхронизации реплик данных с помощью запрета конфликтов данных.

Итак, давайте определим общую модель распределённой коллаборации на CRDT-типах. Существует набор реплик с динамической сетевой связностью, в каждой из которых располагается ревизия данных. Модификации к репликам применяют локально, а движок синхронизации распространяет изменения на другие реплики. Механизм синхронизации может быть децентрализованным (peer-to-peer), централизованным (c master-ареной) или гибридным (например, централизованная иерархия компонентов peer-to-peer).

CRDT-значение обладает адресом (identity) и внутренним состоянием (payload), из которого извлекают целевое значение. Таким образом, интерфейс CRDT-типа на Scala можно представить так:

trait CRDT[S] {
   def id() : RIdentity
   def apply(): S
}

Основные виды CRDT-типов:

1. регистр — атомарное значение, которое рассматривается как единое целое;

2. счётчик — структура с операциональным базисом из инкремента или декремента;

3. множество — типы для неупорядоченных коллекций;

4. список или текст — типы для коллаборативной работы над текстом. 

В зависимости от интерфейса синхронизации выделяют два класса CRDT-типов — CvRDT и CmRDT.

CvRDT

CvRDT (convergent replicated data type, или state-based CRDT) — это CRDT с механизмом «естественной» синхронизации значений на полурешётках. Слово convergent в аббревиатуре можно интерпретировать неверно: это не значит, что другие классы CRDT не удовлетворяют свойству сходимости, просто CvRDT сходятся на основе слияния состояний.

Определим интерфейс CvRDT на основе функции слияния, возвращающей иммутабельное состояние как результат объединения с другим экземпляром типа:

trait CvRDT[T <: CvRDT[T, S], S] extends CRDT[S] {
   def merge(other: T): T
}

Легко заметить, что операция слияния в CvRDT соответствует слиянию в постановке 2-way merge на данных. Воспользуемся идеей полурешётки и построим простейшие типы CvRDT.

Примеры CvRDT

Max-Register — регистр, принимающий максимальное из значений своих реплик:

case class Max[T : Ordering] (
  value: T,
  objectId: RIdentity = RIdentity.empty
)(
  implicit ordering: Ordering[T]
) extends CvRDT[Max[T], T] {

  override def id(): RIdentity = objectId

  override def merge(other: Max[T]): Max[T] = Max(
    ordering.max(value, other.value),
    objectId
  )(ordering)

  override def apply(): T = value
}

В коде показан пример с типом, в котором задан тотальный порядок — Ordering, однако в общем случае для CvRDT будет достаточно лишь частичного порядка — PartialOrdering.

G-Set (Grow-only set) — множество, в которое можно лишь добавлять элементы:

case class GSet[T](items: Set[T], objectId: RIdentity) extends CvRDT[GSet[T], Set[T]] {
  override def id(): RIdentity = objectId
  override def merge(other: GSet[T]): GSet[T] = GSet(items.union(other.items), objectId)
  override def apply(): Set[T] = items
}

В булеане множества (множестве всех подмножеств) существует отношение частичного порядка, поэтому можно выразительно проиллюстрировать функцию слияния с помощью диаграммы Хассе.

G-Counter (Grow-only counter) — монотонный счётчик с операцией инкремента. В CvRDT-реализации мы воспользуемся трюком с разделением счётчика по независимым шардам, которые будут соответствовать вкладу каждой реплики в общее дело. Иными словами, мы воспользуемся ассоциативным массивом из Max-регистров, в каждый из которых будет писать строго один пользователь. Итоговое значение счётчика определяется как сумма всех Max-регистров:

case class GCounter[T](
  shards: Map[ReplicaId, Max[T]] = Map.empty,
  objectId: RIdentity = RIdentity.empty
)(
  implicit monoid: Monoid[T]
) extends CvRDT[GCounter[T], T] {

  override def id(): RIdentity = objectId

  override def apply(): T = shards.values
	.foldLeft(monoid.empty)
	((r, shard) => monoid.combine(r, shard()))

  override def merge(other: GCounter[T]): GCounter[T] = {
    val replicaIds = shards.keySet.union(other.shards.keySet)

    val mergedShards: Map[ReplicaId, Max[T]] = replicaIds.map { replicaId =>
      val left = shards.get(replicaId)
      val right = other.shards.get(replicaId)

      val merged = (left, right) match {
        case (Some(l), Some(r)) => l.merge(r)
        case (Some(l), None) => l
        case (None, Some(r)) => r
        case (None, None) => throw new IllegalStateException()
      }

      replicaId -> merged
    }.toMap

    GCounter(mergedShards, objectId)
  }
}

Если снова изобразить на диаграмме Хассе, как происходит слияние, то можно увидеть что-то очень похожее на векторные часы. Действительно, G-Counter иллюстрирует идею попытки сведения монотонной временной метки с частичным порядком к монотонному тотально упорядоченному значению.

Регистры

Если ранее для построения CvRDT мы пользовались «естественной сходимостью» типов или трюком с разделением данных, то в общем случае так сделать не получится. Тогда на помощь приходят регистры — «коробочки» для хранения произвольных значений.

LWW-Register (Last-Write-Wins Register) — регистр, значение в котором определяется хронологически последней записью, то есть побеждает тот пользователь, который записал последним. 

case class LWWRegister[T](
   value: T,
   timestamp: Instant = Instant.now(),
   objectId: RIdentity = RIdentity.empty
 ) extends CvRDT[LWWRegister[T], T] {

  override def id(): RIdentity = objectId

  def write(newValue: T, newTimestamp: Instant = Instant.now()): LWWRegister[T] = {
    if (newTimestamp.isAfter(timestamp)) {
      LWWRegister(newValue, newTimestamp)
    } else {
      this
    }
  }

  override def merge(other: LWWRegister[T]): LWWRegister[T] = {
    if (other.timestamp.isAfter(this.timestamp)) other else this
  }
}

Очевидным недостатком LWW-регистра является неявная перезапись значения. Эта аномалия способна приводить к конфликтам пользовательских намерений и плохо интерпретируется.

MV-Register (Multi-Value Register) — регистр, аккумулирующий все значения, полученные при параллельных модификациях. Он не позволяет пропустить конкурентное изменение и предлагает способ для ручного разрешения конфликта.

Здесь я опущу пример с реализацией MV-регистра и взамен расскажу, чем она будет отличаться от реализации LWW-регистра:

1. Переход к частичному порядку. Отказываясь от тотальной упорядоченности временных меток с помощью векторных часов, определим для операций отношение конкурентности. Дизъюнктивное замыкание пар конкурентных операций образует конфликтное множество.

2. Операция сужения домена конфликтующих значений. Фактически мы хотим позволить пользователям разрешать конфликты, однако противоречащие разрешения конфликтов сами образуют конфликт — это представляет проблему.

CmRDT

CmRDT (commutative replicated data type, или operation-based CRDT) — это CRDT с интерфейсом из коммутирующих операций, то есть их применение в любом порядке приводит к одному и тому же состоянию.

trait CmRDT[T <: CmRDT[T, Op, S], Op, S] extends CRDT[S] {
   def accept(op: Op): T
}

Фактически мы имеем дело с постановкой 2-way merge на операциях, в которых функция слияния — это теоретико-множественное объединение в силу коммутативности операций. Иными словами, если рассмотреть начальное состояние S, из которого конкурентно получены состояния S₁ = S ∘ O₁ и S₂ = S ∘ O₂, то результирующее состояние S’ = S ∘ O₁ ∘ O₂ = S ∘ O₂ ∘ O₁ = S ∘ ( O₁ ⋃ O₂ ).

Об эквивалентности CvRDT и CmRDT

Определим ранее рассмотренный Grow-only Counter как CmRDT-тип. Воспользуемся той же идеей с шардированием вклада различных пользователей, но здесь нам уже не потребуются Max-регистры.

case class GCounter[T](
  shards: Map[ReplicaId, T] = Map.empty,
  objectId: RIdentity = RIdentity.empty
)(
  implicit monoid : Monoid[T]
) extends CmRDT[GCounter[T], GCounter.Op[T], T] {

  override def id(): RIdentity = objectId

  override def apply(): T = shards.values
    .foldLeft(monoid.empty)
    ((r, shard) => monoid.combine(r, shard))

  override def accept(op: GCounter.Op[T]): GCounter[T] = op match {
      case Increment(add : T) => GCounter {
        shards.transform((replicaId, shard) => {
          if (replicaId == op.replica()) monoid.combine(shard, add)
          else shard
        })
      }
    }
  }

object GCounter {
  sealed trait Op[T] extends OpMetadata

  case class Increment[T](add: T)(implicit val ctx : RContext) extends Op[T] with WithIncrement[T] {
    override def replica(): ReplicaId = ctx.me()
    override def incCount(): T = add
  }
}

На самом деле, то, что мы смогли построить один и тот же тип данных как CvRDT и CmRDT, не является случайностью. Существует утверждение об эквивалентности двух этих классов типов [Shapiro et al. 2011; п. 3.2].

Действительно, мы можем тривиально эмулировать любой CvRDT как CmRDT с операцией merge. И наоборот: мы можем сериализовать любой экземпляр CmRDT в CvRDT — монотонно растущее множество операций (G-Set), которыми было порождено текущее состояние объекта.

Заключение

Мы определили CRDT-типы как подход к оптимистической репликации и реализовали несколько примеров. К современным исследовательским проблемам в области относятся формальная верификация моделей на CRDT-типах, построение реплицируемых списков («move-семантика» и проблема «перекрытий») и объектное моделирование в коллаборативных приложениях на основе CRDT.