В предыдущей статье — часть 1, обзорная — я рассказал о том, зачем нужны распределенные структуры данных (далее — РСД) и разобрал несколько вариантов, предлагаемых распределенным кешем Apache Ignite.

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

Итак:

Начнем с того, что, как минимум в случае с Apache Ignite, РСД не реализованы с нуля, а являются надстройкой над распределенным кешем.

Итак, перейдем к подробностям реализации.

Для обеспечения работы РСД используются два кеша: один Replicated и один Partitioned.

Replicated-кеш — в данном случае это системный кеш, (ignite-sys-cache) отвечающий, в том числе, за хранение информации об РСД, зарегистрированных в системе.

Partitioned-кеш (ignite-atomics-sys-cache) хранит данные, необходимые для работы РСД, и их состояние.

Итак, большинство РСД создается следующим образом:

1. Транзакция стартует.
2. В кеше ignite-sys-cache, по ключу DATA_STRUCTURES_KEY, берется Map<Имя_РСД, DataStructureInfo> (при необходимости создается), и в нее добавляется новый элемент с описанием, например, IgniteAtomicReference.
3. В кеш ignite-atomics-sys-cache, по ключу из добавленного ранее DataStructureInfo добавляется элемент, отвечающий за состояние РСД.
4. Транзакция коммитится.

При первом запросе на создание РСД создается новый экземпляр, а последующие запросы получают уже ранее созданный.

IgniteAtomicReference и IgniteAtomicLong (краткая вводная)

Третий шаг инициализации для обоих типов сводится к добавлению в ignite-atomics-sys-cache объекта типа GridCacheAtomicReferenceValue или GridCacheAtomicLongValue.

Оба класса содержат одно единственное поле val.

Соответственно, любое изменение IgniteAtomicReference:

// Изменим значение, если текущее соответствует ожидаемому.
ref.compareAndSet(expVal, newVal);

… это запуск EntryProcessor со следующим кодом метода process:

Boolean process(MutableEntry> e, Object... args) {
    GridCacheAtomicReferenceValue val = e.getValue();

    T curVal = val.get();

    // Переменные expVal и newVal  — параметры метода
    // ref.compareAndSet(expVal, newVal);
    if (F.eq(expVal, curVal)) {
        e.setValue(new GridCacheAtomicReferenceValue(newVal));

        return true;
    }

    return false;
}

IgniteAtomicLong является дефакто расширением IgniteAtomicReference, поэтому и его метод compareAndSet реализован аналогичным образом.

Метод incrementAndGet не имеет проверок на ожидаемое значение, а просто добавляет единицу.

Long process(MutableEntry e, Object... args) {
    GridCacheAtomicLongValue val = e.getValue();

    long newVal = val.get() + 1;

    e.setValue(new GridCacheAtomicLongValue(newVal));

    return newVal;
}

IgniteAtomicSequence (краткая вводная)

При создании каждого экземпляра IgniteAtomicSequence...

// Создадим или получим ранее созданный IgniteAtomicSequence.
final IgniteAtomicSequence seq = ignite.atomicSequence("seqName", 0, true);

… ему выделяется пул идентификаторов.

// Начинаем транзакцию
try (GridNearTxLocal tx = CU.txStartInternal(ctx, seqView, PESSIMISTIC, REPEATABLE_READ)) {
GridCacheAtomicSequenceValue seqVal = cast(dsView.get(key), GridCacheAtomicSequenceValue.class);

// Нижняя граница локального пула идентификаторов
locCntr = seqVal.get();

// Верхняя граница
upBound = locCntr + off;

seqVal.set(upBound + 1);

// Обновляем экземпляр GridCacheAtomicSequenceValue в кеше
dsView.put(key, seqVal);

// Завершаем транзакцию
tx.commit();

Соответственно, вызов...

seq.incrementAndGet(); 

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

При достижении границы происходит выделение нового пула идентификаторов, аналогично тому, как это происходит при создании нового экземпляра IgniteAtomicSequence.

IgniteCountDownLatch (краткая вводная)

Декремент счетчика:

latch.countDown();

… реализуется следующим образом:

 // Начинаем транзакцию
 try (GridNearTxLocal tx = CU.txStartInternal(ctx, latchView, PESSIMISTIC, REPEATABLE_READ)) {
    GridCacheCountDownLatchValue latchVal = latchView.get(key);

    int retVal;

    if (val > 0) {
        // Декрементируем значение
        retVal = latchVal.get() - val;

        if (retVal < 0)
            retVal = 0;
    }
    else
        retVal = 0;

    latchVal.set(retVal);

    // Сохраняем значение
    latchView.put(key, latchVal);

    // Завершаем транзакцию
    tx.commit();

    return retVal;
}

Ожидание декрементации счетчика до 0...

latch.await();

… реализуется через механизм Continuous Queries, то есть при каждом изменении GridCacheCountDownLatchValue в кеше все экземпляры IgniteCountDownLatch уведомляются об этих изменениях.

Каждый экземпляр IgniteCountDownLatch имеет локальный:

/** Internal latch (transient). */
private CountDownLatch internalLatch;

Каждое уведомление декрементирует internalLatch до актуального значения. Поэтому latch.await() реализуется очень просто:

if (internalLatch.getCount() > 0)
    internalLatch.await();

IgniteSemaphore (краткая вводная)

Получение разрешения...

semaphore.acquire();

… происходит следующим образом:

// Пока разрешение не будет получено
for (;;) {
    int expVal = getState();

    int newVal = expVal - acquires;

    try (GridNearTxLocal tx = CU.txStartInternal(ctx, semView, PESSIMISTIC, REPEATABLE_READ)) {
        GridCacheSemaphoreState val = semView.get(key);

        boolean retVal = val.getCount() == expVal;

        if (retVal) {
            // Сохраняем информацию о получивших разрешения.
            // В случае выхода из строя какой-либо node,
            // захваченные ею разрешения будут возвращены.
            {
                UUID nodeID = ctx.localNodeId();

                Map map = val.getWaiters();

                int waitingCnt = expVal - newVal;

                if (map.containsKey(nodeID))
                    waitingCnt += map.get(nodeID);

                map.put(nodeID, waitingCnt);

                val.setWaiters(map);
            }

            // Устанавливаем новое значение
            val.setCount(newVal);

            semView.put(key, val);

            tx.commit();
        }

        return retVal;
    }
}

Возврат разрешения...

semaphore.release();

… происходит аналогичным образом, за исключением того, что новое значение больше текущего.

int newVal = cur + releases;

IgniteQueue (краткая вводная)

В отличие от остальных РСД, IgniteQueue не использует ignite-atomics-sys-cache. Используемый кеш описывается через параметр colCfg.

// Создадим или получим ранее созданный IgniteQueue.
IgniteQueue queue = ignite.queue("queueName", 0, colCfg);

В зависимости от указанного Atomicity Mode (TRANSACTIONAL, ATOMIC) можно получить разные варианты IgniteQueue.

queue = new GridCacheQueueProxy(cctx, cctx.atomic() ? 
    new GridAtomicCacheQueueImpl<>(name, hdr, cctx) : 
    new GridTransactionalCacheQueueImpl<>(name, hdr, cctx));

В обоих случаях состояние IgniteQueue контролируется с помощью:

class GridCacheQueueHeader{
   private long head;
   private long tail;
   private int cap;
... 

Для добавления элемента используется AddProcessor...

Long process(MutableEntry e, Object... args) {
    GridCacheQueueHeader hdr = e.getValue();

    boolean rmvd = queueRemoved(hdr, id);

    if (rmvd || !spaceAvailable(hdr, size))
        return rmvd ? QUEUE_REMOVED_IDX : null;

    GridCacheQueueHeader newHdr = new GridCacheQueueHeader(hdr.id(),
        hdr.capacity(),
        hdr.collocated(),
        hdr.head(),
        hdr.tail() + size, // Выделяем место под элемент
        hdr.removedIndexes());

    e.setValue(newHdr);

    return hdr.tail();
}

… который, по сути, просто перемещает указатель на хвост очереди.

После этого...

// По ключу, сформированному на основе
// нового hdr.tail()
QueueItemKey key = itemKey(idx);

… в очередь добавляется новый элемент:

cache.getAndPut(key, item);

Удаление элемента происходит аналогично, но указатель меняется не на tail, а на head...

GridCacheQueueHeader newHdr = new GridCacheQueueHeader(hdr.id(),
    hdr.capacity(),
    hdr.collocated(),
    hdr.head() + 1, // Двигаем указатель на голову
    hdr.tail(),
    null);

… и элемент удаляется.

Long idx = transformHeader(new PollProcessor(id));

QueueItemKey key = itemKey(idx);

T data = (T)cache.getAndRemove(key);

Разница между GridAtomicCacheQueueImpl и GridTransactionalCacheQueueImpl состоит в том, что:

• GridAtomicCacheQueueImpl при добавлении элемента сначала атомарно инкрементирует hdr.tail(), а потом уже добавляет по полученному индексу элемент в кеш.

  
  
• GridTransactionalCacheQueueImpl делает оба действия в рамках одной транзакции.

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

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

Эта проблема решается таймаутом ожидания значения.

long stop = U.currentTimeMillis() + RETRY_TIMEOUT;

while (U.currentTimeMillis() < stop) {
    data = (T)cache.getAndRemove(key);

    if (data != null)
        return data;
}

Вместо заключения

Я хотел бы еще раз отметить, что распределенный кеш — это, по сути, ConcurrentHashMap в рамках множества компьютеров, объединенных в кластер.

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

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

https://habrahabr.ru/post/328368/