Анализ исходного кода Redis — основные данные — skiplist (список пропуска)

Слишком долго читать версию

Таблица пропуска — это одна из низкоуровневых реализаций отсортированного набора zset, за исключением того, что она не имеет другого применения в Redis.

Высота слоя каждого узла таблицы пропуска представляет собой случайное число от 1 до 64.

Чем выше высота слоя, тем ниже вероятность появления, и вероятность того, что высота слоя равна i, равна $(1-p) * p^{i-1}, (p=1/4)$ .

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

Этот анализ основан наredisВерсия 5.0.0, файлы исходного кода, задействованные в этой статье: t_zset.c, server.h.

Что такое таблица прыжков

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

Все мы знаем, что для поиска в отсортированных массивах можно использовать бинарный поиск, чтобы уменьшить временную сложность до O(log n). Однако упорядоченный связанный список не может этого сделать, поскольку сложность получения элемента в упорядоченном связанном списке составляет O(n), а доступ к некоторым элементам нельзя пропустить за счет идеи дихотомии.

Например, если вы хотите найти элемент 50 на следующем рисунке, вы должны найти его в 5 -> 6 -> 10 -> 30 -> 49. Нельзя сказать, что центральный элемент 49 меньше 50, тогда начать поиск с правой стороны от центра и пропустить элемент 5. , 6, 10, 30 посещений.

Таблица прыжка извлекается в качестве индекса в узле, быстро найдет список упорядоченного списка. Это пространство (дополнительное степпер указатель) по существу для времени работы. Например, фига:

В это время элемент 50 поиска становится 5 -> 49, а промежуточные элементы 6, 10 и 30 пропускаются. На приведенном выше рисунке связанный список идеально разделен на две части за счет хранения указателей разной длины в первом узле, но фактический список пропуска аналогичен структуре следующего рисунка.В большинстве случаев предпочтение отдается не идеальной дихотомии. :

Таблица переходов использует случайный алгоритм (чем выше высота слоя, тем ниже вероятность) для определения высоты слоя, и одни и те же слои связаны указателями. Вероятность того, что высота слоя узла равна i в реализации Redis, равна $(1-p) * p^{i-1}$ .

Почему бы не использовать самую совершенную дихотомическую структуру?

Рассмотрим случай вставки узла. Когда узел вставляется посередине, структура пополам в это время будет нарушена, поэтому ее необходимо постоянно корректировать. Подумайте о сбалансированных деревьях, сложных операциях по перебалансировке красно-черных деревьев, а корректировка перебалансировки здесь — это нечто большее. Метод использования случайного алгоритма для выбора высоты слоя также может достигать средней сложности O(logN), и операция относительно упрощается.

Пространственная сложность таблицы пропуска (реализация Redis)

Связанные определения

// 层高最大值限制
#define ZSKIPLIST_MAXLEVEL 64 /* Should be enough for 2^64 elements */
// 层高是否继续增长的概率
#define ZSKIPLIST_P 0.25      /* Skiplist P = 1/4 */
// 跳表节点定义
typedef struct zskiplistNode {
    // 存储内容
    sds ele;
    // 分值,用于排序
    double score;
    // 后退指针
    struct zskiplistNode *backward;
    // 变长数组，记录层信息。层高越高跳过的节点越多(因为层高越高概率越低)
    struct zskiplistLevel {
        // 指向当前层下一个节点
        struct zskiplistNode *forward;
        // 当前节点与forward所指节点中间节点数
        unsigned long span;
    } level[];
} zskiplistNode;
// 跳表结构管理节点
typedef struct zskiplist {
    struct zskiplistNode *header, *tail;
    // 长度
    unsigned long length;
    // 跳表高度(所有节点最高层高)
    int level;
} zskiplist;

int zslRandomLevel(void) {
    // 计算当前插入元素层高的随机函数
    int level = 1;
    // (random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF) 概率为1/4
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

Высота слоя равна 1, а вероятность равна 1-p (без ввода while)

Вероятность того, что высота слоя равна 2, равна p (перейти, пока) * (1 - p) (не вводить, пока)

Вероятность того, что высота слоя равна 3, равна p(продвижение, пока) * p(продвижение, пока) * (1 - p)(не вводите, пока)

...

Вероятность того, что высота слоя равна n, равна $(1-p) * p^{n-1}$

высокие ожидания E = 1/(1-p)

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

В реализации redis p=1/4, а высота слоя ожидается примерно равной 1,33, поэтому средняя высота слоя узла примерно равна 1,33, что является константой, так что пространственная сложность узла таблица переходов - O(n).

Операции, связанные с таблицей переходов (реализация Redis)

Создать таблицу пропуска

zskiplistNode *zslCreateNode(int level, double score, int ele) {
    zskiplistNode *zn =
        malloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
    zn->score = score;
    zn->ele = ele;
    return zn;
}

/* Create a new skiplist. */
zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;

    zsl = malloc(sizeof(*zsl));
    zsl->level = 1;
    zsl->length = 0;
    // 头节点层高为64(层高的最大限制)
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
    }
    zsl->header->backward = NULL;
    zsl->tail = NULL;
    return zsl;
}

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

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

вставить узел

int zslRandomLevel(void) {
    // 计算当前插入元素层高的随机函数
    int level = 1;
    // (random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF) 概率为1/4
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
    // update存放需要更新的节点
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;

    serverAssert(!isnan(score));
    x = zsl->header;
    // 第一步，收集需要更新的节点与步长信息
    for (i = zsl->level-1; i >= 0; i--) {
        /* store rank that is crossed to reach the insert position */
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // score可以重复，重复时使用ele大小进行排序
        while (x->level[i].forward &&
                (x->level[i].forward->score < score ||
                    (x->level[i].forward->score == score &&
                    sdscmp(x->level[i].forward->ele,ele) < 0)))
        {
            rank[i] += x->level[i].span;
            x = x->level[i].forward;
        }
        update[i] = x;
    }
    // 第二步， 获取随机层高，补全需要更新的节点
    level = zslRandomLevel();
    if (level > zsl->level) {
        for (i = zsl->level; i < level; i++) {
            rank[i] = 0;
            update[i] = zsl->header;
            update[i]->level[i].span = zsl->length;
        }
        zsl->level = level;
    }
    // 第三步，创建并分层插入节点，同时更新同层前一节点步长信息
    x = zslCreateNode(level,score,ele);
    for (i = 0; i < level; i++) {
        x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x;

        /* update span covered by update[i] as x is inserted here */
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }
    // 第四步，更新新增节点未涉及层节点的步长信息，以及跳表相关信息
    /* increment span for untouched levels */
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }

    x->backward = (update[0] == zsl->header) ? NULL : update[0];
    if (x->level[0].forward)
        x->level[0].forward->backward = x;
    else
        zsl->tail = x;
    zsl->length++;
    return x;
}

Вставка узла делится на четыре этапа (например, просмотр во время еды):

Предположим теперь, что мне нужно вставить элемент 80 и получить случайный уровень 5 (для всех рассмотренных случаев).

Соберите информацию об узле и шаге, которую необходимо обновить.
- После того, как новый узел вставлен, затронутые узлы каждого уровня будут сохранены в массиве обновлений, а update[i] — это затронутый узел i + 1-го слоя (красный прямоугольник — это узел, который может быть затронут в примере ).
- Сохраните количество узлов между головным узлом каждого слоя и затронутыми узлами в массиве рангов, а ранг[i] — это количество узлов между головным узлом и слоем i + 1, которые будут затронуты (ранг равен [6 , 5, 3], 3]).

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

Создавайте и вставляйте узлы иерархически и одновременно обновляйте информацию о размере шага предыдущего узла на том же уровне.
- Создайте узел, а затем выполните вставку узла на каждом уровне в соответствии с высотой слоя текущего узла (так же, как простая вставка связанного списка).
- Обновите информацию о размере шага предыдущего узла каждого слоя (узла, соответствующего update[i]) и его собственного узла.
Обновите информацию о размере шага вновь добавленных узлов, которые не включают узлы слоя, а также связанную информацию таблицы пропуска и самого узла.
- Если высота слоя текущего узла ниже высоты списка пропуска, то информация о размере шага узлов, ранжированных после текущего узла в тех слоях, которые выше, чем высота слоя текущего узла, требует +1 (поскольку вставка между ним и его предыдущий узел новые элементы).
- Обновите длину прыжка и текущий узел с задней частью первого слоя следующего узла (заднюю защитную иглу можно понимать только как базовый связанный список).

найти узел

/* Find the rank for an element by both score and key.
 * Returns 0 when the element cannot be found, rank otherwise.
 * Note that the rank is 1-based due to the span of zsl->header to the
 * first element. */
unsigned long zslGetRank(zskiplist *zsl, double score, sds ele) {
    zskiplistNode *x;
    unsigned long rank = 0;
    int i;

    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward &&
            (x->level[i].forward->score < score ||
                (x->level[i].forward->score == score &&
                sdscmp(x->level[i].forward->ele,ele) <= 0))) {
            rank += x->level[i].span;
            x = x->level[i].forward;
        }

        /* x might be equal to zsl->header, so test if obj is non-NULL */
        if (x->ele && sdscmp(x->ele,ele) == 0) {
            return rank;
        }
    }
    return 0;
}

/* Finds an element by its rank. The rank argument needs to be 1-based. */
zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) {
    zskiplistNode *x;
    unsigned long traversed = 0;
    int i;

    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward && (traversed + x->level[i].span) <= rank)
        {
            traversed += x->level[i].span;
            x = x->level[i].forward;
        }
        if (traversed == rank) {
            return x;
        }
    }
    return NULL;
}

В реализации Redis таблица переходов и dict совместно реализуют zset.Словарь достигает сложности O(1) для получения элемента, соответствующего счету.Таблица переходов используется для обработки связанных операций интервальных запросов.В то же время, поскольку счет можно повторить, таблицу прыжков не нужно получать через ele.score (проверка через дикт) и получать ele через score (вроде бы такого требования нет).

Есть два общих требования к запросу:

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

Общий процесс заключается в запросе в соответствии с маршрутом поиска из левого верхнего угла в правый нижний (как показано красной линией на рисунке выше).

удалить узел

/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
    int i;
    for (i = 0; i < zsl->level; i++) {
        // 被删除节点在第i层有节点，则update[i]为被删除节点的前一个节点
        if (update[i]->level[i].forward == x) {
            // 步长 = 原步长 + 被删除节点步长 - 1（被删除节点）
            update[i]->level[i].span += x->level[i].span - 1;
            // 指针越过被删除节点
            update[i]->level[i].forward = x->level[i].forward;
        } else {
            // 被删除节点在第i层无节点，则 步长 = 原步长 - 1(被删除节点)
            update[i]->level[i].span -= 1;
        }
    }
    if (x->level[0].forward) {
        // 更新被删除节点下一节点的后退指针
        x->level[0].forward->backward = x->backward;
    } else {
        zsl->tail = x->backward;
    }
    while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
        zsl->level--;
    zsl->length--;
}

int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    int i;

    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward &&
                (x->level[i].forward->score < score ||
                    (x->level[i].forward->score == score &&
                     sdscmp(x->level[i].forward->ele,ele) < 0)))
        {
            x = x->level[i].forward;
        }
        update[i] = x;
    }
    /* We may have multiple elements with the same score, what we need
     * is to find the element with both the right score and object. */
    x = x->level[0].forward;
    if (x && score == x->score && sdscmp(x->ele,ele) == 0) {
        zslDeleteNode(zsl, x, update);
        if (!node)
            zslFreeNode(x);
        else
            *node = x;
        return 1;
    }
    return 0; /* not found */
}

Удаление узла аналогично добавлению узла и делится на три шага:

Соберите узлы, которые необходимо обновить.
Удалите связанный список слоя, в котором находится узел, чтобы удалить узел (так же, как простой связанный список, чтобы удалить узел), и обновите информацию о размере шага предыдущего узла (узел хранится в update[i]).
Обновление высоты и длина прыжка.