Слишком долго читать версию
- Упакованный список — это линейная структура данных, реализованная для экономии места и по существу представляющая собой массив байтов.
- Элементы упакованного списка могут быть целыми числами или строками.
- Сжатые списки используются в быстрых списках, объектах списка и хеш-объектах.
- Добавление списка сжатия (средняя сложность O(n)) и удаление узлов (средняя сложность O(n)) может инициировать обновление цепочки (средняя сложность O(n^2)), поскольку вероятность срабатывания невысока, поэтому влияют на производительность.
- Поскольку данные хранилища узла могут быть строками, а сопоставление строк имеет сложность O(n), средняя сложность узла поиска в сжатом списке составляет O(n^2).
Этот анализ основан наredisВерсия 5.0.0, исходные файлы, задействованные в этой статье, — ziplist.c, ziplist.h.
Что такое сжатый список
/* Create a new empty ziplist. */
unsigned char *ziplistNew(void) {
unsigned int bytes = ZIPLIST_HEADER_SIZE+1;
unsigned char *zl = zmalloc(bytes);
// ...
return zl;
}
Как видно из приведенного выше кода создания, сжатый список по сути представляет собой массив байтов, представляющий собой линейную структуру данных, реализованную Redis с использованием массивов байтов, а его элементами могут быть целые числа или строки. Он используется в быстрых списках, объектах списков и хеш-объектах и обычно используется, когда элементов мало и число байтов элементов невелико.
为什么要实现压缩列表?
Потому что для того, чтобы максимально экономить память. Представьте, что теперь есть два элемента: целое число 1 и строка 'abc', два узла. Сами данные занимают всего 8 байт (на 64-битной машине). Если вы используете односвязный список, вам нужно добавить еще два обратных указателя, занимающих 24 байта, из которых данные узла связанного списка занимают 16 байтов, что в два раза превышает размер самих данных. В этом примере хранилище сжатого списка требует, чтобы каждый узел записывал только длину предыдущего узла (1 байт) и информацию о кодировании самого узла (1 байт) (конкретные правила будут объяснены позже), занимая в общей сложности 10 байт.
Состав сжатого списка
#define ZIP_END 255 /* Special "end of ziplist" entry. */
#define ZIPLIST_BYTES(zl) (*((uint32_t*)(zl)))
#define ZIPLIST_TAIL_OFFSET(zl) (*((uint32_t*)((zl)+sizeof(uint32_t))))
#define ZIPLIST_LENGTH(zl) (*((uint16_t*)((zl)+sizeof(uint32_t)*2)))
#define ZIPLIST_HEADER_SIZE (sizeof(uint32_t)*2+sizeof(uint16_t))
#define ZIPLIST_END_SIZE (sizeof(uint8_t))
/* Create a new empty ziplist. */
unsigned char *ziplistNew(void) {
unsigned int bytes = ZIPLIST_HEADER_SIZE+1;
unsigned char *zl = zmalloc(bytes);
// 压缩列表总字节长度
ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);
// 尾部节点字节距离
ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);
// 压缩列表节点个数
ZIPLIST_LENGTH(zl) = 0;
// 255特殊结尾值
zl[bytes-1] = ZIP_END;
return zl;
}
Сжатый список состоит из общей длины в байтах (4 байта), смещения хвостового узла (4 байта), количества узлов (2 байта), узла и специального терминатора (1 байт) со значением 255. , по списку Начальный адрес смещен назад на байты смещения хвостового узла, и информацию о хвостовом узле можно получить с временной сложностью O(1).
Информация самого сжатого списка занимает всего 11 байт, в то время как связанному списку требуется всего 16 байт для хранения указателя начала и конца, поэтому более экономично использовать сжатый список для случая небольшого объема данных. (несколько узлов и небольшие узлы).
intrev32ifbe函数为大小端转换函数,统一转换为小端存储。为什么要进行转换?
Поскольку в операции сжатия списка задействовано много битовых операций, если он не будет унифицирован, возникнет путаница. Все последующие битовые операции выполняются с прямым порядком байтов.Большие и маленькие точки знаний, нажмите здесь, чтобы понять
Состав узлов сжатого списка
typedef struct zlentry {
// 前一节点长度信息的长度
unsigned int prevrawlensize;
// 前一节点长度
unsigned int prevrawlen;
// 当前节点长度信息长度
unsigned int lensize;
// 当前节点长度
unsigned int len;
// 当前节点头部信息长度
unsigned int headersize;
// 当前节点数据编码
unsigned char encoding;
unsigned char *p;
} zlentry;
void zipEntry(unsigned char *p, zlentry *e) {
// 前一节点长度信息解析
ZIP_DECODE_PREVLEN(p, e->prevrawlensize, e->prevrawlen);
// 当前节点数据长度与编码信息解析
ZIP_DECODE_LENGTH(p + e->prevrawlensize, e->encoding, e->lensize, e->len);
e->headersize = e->prevrawlensize + e->lensize;
e->p = p;
}
Как упоминалось ранее, сжатый список по сути представляет собой массив байтов, а redis определяет структуру zlentry для удобства работы и вычислений. При выполнении операций вычисления информация, содержащаяся в байтовом массиве, разбирается в структуру zlentry по правилам, что удобно для последующих вычислений. С точки зрения хранения узел сжатого списка разделен на три части:
- информация о предыдущей длине узла
- Текущая длина данных узла и информация о кодировании
- Данные узла
На следующем рисунке представлена схема состава каждой части узла сжатого списка и поля длины структуры zlentry:
информация о предыдущей длине узла
#define ZIP_BIG_PREVLEN 254
#define ZIP_DECODE_PREVLENSIZE(ptr, prevlensize) do { \
if ((ptr)[0] < ZIP_BIG_PREVLEN) { \
(prevlensize) = 1; \
} else { \
(prevlensize) = 5; \
} \
} while(0);
#define ZIP_DECODE_PREVLEN(ptr, prevlensize, prevlen) do { \
ZIP_DECODE_PREVLENSIZE(ptr, prevlensize); \
if ((prevlensize) == 1) { \
(prevlen) = (ptr)[0]; \
} else if ((prevlensize) == 5) { \
assert(sizeof((prevlen)) == 4); \
memcpy(&(prevlen), ((char*)(ptr)) + 1, 4); \
memrev32ifbe(&prevlen); \
} \
} while(0);
Когда длина предыдущего узла меньше 254, используйте 1 байт для сохранения информации о длине предыдущего узла.
Когда длина предыдущего узла превышает 254, используйте 5 байтов для сохранения информации о длине предыдущего узла. Первый байт имеет фиксированное значение 254, а следующие 4 байта используются для хранения информации о длине.
宏定义为什么要写成do {} while(0);
Написано do {} while(0); чтобы гарантировать, что макрос всегда будет вести себя одинаково, независимо от того, как в вызывающем коде используются точки с запятой и фигурные скобки. Пример ошибки следующий:
#define TEST() test1(); test2();
// 此时test2必然被执行,不符合预期
if (a > 0)
TEST();
#define TEST() { test1(); test2(); }
// 此时就会出现语法错误
if (a > 0)
TEST();
Текущая длина данных узла и информация о кодировании
/* Different encoding/length possibilities */
#define ZIP_STR_MASK 0xc0 /* 11000000 */
#define ZIP_STR_06B (0 << 6) /* 00000000 */
#define ZIP_STR_14B (1 << 6) /* 01000000 */
#define ZIP_STR_32B (2 << 6) /* 10000000 */
#define ZIP_INT_16B (0xc0 | 0<<4) /* 11000000 */
#define ZIP_INT_32B (0xc0 | 1<<4) /* 11010000 */
#define ZIP_INT_64B (0xc0 | 2<<4) /* 11100000 */
#define ZIP_INT_24B (0xc0 | 3<<4) /* 11110000 */
#define ZIP_INT_8B 0xfe /* 11111110 */
/* 4 bit integer immediate encoding |1111xxxx| with xxxx between
* 0001 and 1101. */
#define ZIP_INT_IMM_MASK 0x0f /* 00001111 */
#define ZIP_INT_IMM_MIN 0xf1 /* 11110001 */
#define ZIP_INT_IMM_MAX 0xfd /* 11111101 */
#define ZIP_ENTRY_ENCODING(ptr, encoding) do { \
(encoding) = (ptr[0]); \
if ((encoding) < ZIP_STR_MASK) (encoding) &= ZIP_STR_MASK; \
} while(0)
#define ZIP_DECODE_LENGTH(ptr, encoding, lensize, len) do { \
ZIP_ENTRY_ENCODING((ptr), (encoding)); \
if ((encoding) < ZIP_STR_MASK) { \
if ((encoding) == ZIP_STR_06B) { \
(lensize) = 1; \
(len) = (ptr)[0] & 0x3f; \
} else if ((encoding) == ZIP_STR_14B) { \
(lensize) = 2; \
(len) = (((ptr)[0] & 0x3f) << 8) | (ptr)[1]; \
} else if ((encoding) == ZIP_STR_32B) { \
(lensize) = 5; \
(len) = ((ptr)[1] << 24) | \
((ptr)[2] << 16) | \
((ptr)[3] << 8) | \
((ptr)[4]); \
} else { \
panic("Invalid string encoding 0x%02X", (encoding)); \
} \
} else { \
(lensize) = 1; \
(len) = zipIntSize(encoding); \
} \
} while(0);
unsigned int zipIntSize(unsigned char encoding) {
switch(encoding) {
case ZIP_INT_8B: return 1;
case ZIP_INT_16B: return 2;
case ZIP_INT_24B: return 3;
case ZIP_INT_32B: return 4;
case ZIP_INT_64B: return 8;
}
if (encoding >= ZIP_INT_IMM_MIN && encoding <= ZIP_INT_IMM_MAX)
return 0; /* 4 bit immediate */
panic("Invalid integer encoding 0x%02X", encoding);
return 0;
}
int zipTryEncoding(unsigned char *entry, unsigned int entrylen, long long *v, unsigned char *encoding) {
long long value;
if (entrylen >= 32 || entrylen == 0) return 0;
if (string2ll((char*)entry,entrylen,&value)) {
if (value >= 0 && value <= 12) {
*encoding = ZIP_INT_IMM_MIN+value;
// ...
return 1;
}
return 0;
}
Кодировка и соответствующая длина данных показаны в следующей таблице:
кодирование | длина информации о длине | Длина данных |
---|---|---|
ZIP_STR_06B(00bbbbbb) (первые два бита маркируются кодом, последние 6 бит хранятся в длине) | 1 байт | байтовый массив длины |
ZIP_STR_14B (01bbbbbb xxxxxxxx) (первые 2 бита закодированы, а последние 14 бит — длина) | 2 байта | байтовый массив длины |
ZIP_STR_06B(10______ aaaaaaaa bbbbbbbb cccccccc) (6 бит остаются пустыми после первых 2 бит кодирования тега, а последние 24 бита — это длина) | 5 байт | байтовый массив длины |
ZIP_INT_16B(11000000) | 2 байта | int16_t (короткий) целочисленный тип |
ZIP_INT_32B(11010000) | 4 байта | int32_t (int) тип целочисленный |
ZIP_INT_64B(11100000) | 8 байт | int64_t (длинный длинный) целочисленный тип |
ZIP_INT_24B(11110000) | 3 байта | 3-байтовое длинное целое со знаком |
ZIP_INT_8B(11111110) | 1 байт | 1-байтовое длинное целое со знаком |
1111 xxxx | 4 бит | Целое число без знака от 0 до 12 (в настоящее время узел не имеет части контента, а данные хранятся в части кодирования) |
В соответствии с типом и размером данных сопоставьте различные коды, идеи и ранее проанализированныебазовый тип сдсТак же, как и использование информации заголовка в 5, все это делается для максимизации пространства сжатия.
Операции, связанные со сжатым списком
узел обновления
#define ZIPLIST_HEAD 0
#define ZIPLIST_TAIL 1
#define ZIPLIST_HEADER_SIZE (sizeof(uint32_t)*2+sizeof(uint16_t))
#define ZIPLIST_ENTRY_HEAD(zl) ((zl)+ZIPLIST_HEADER_SIZE)
#define ZIPLIST_ENTRY_END(zl) ((zl)+intrev32ifbe(ZIPLIST_BYTES(zl))-1)
unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where) {
unsigned char *p;
// 头部或尾部插入节点
p = (where == ZIPLIST_HEAD) ? ZIPLIST_ENTRY_HEAD(zl) : ZIPLIST_ENTRY_END(zl);
return __ziplistInsert(zl,p,s,slen);
}
unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) {
/// 插入节点至p指针指向元素前方
return __ziplistInsert(zl,p,s,slen);
}
unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) {
size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), reqlen;
unsigned int prevlensize, prevlen = 0;
size_t offset;
int nextdiff = 0;
unsigned char encoding = 0;
long long value = 123456789;
/* Find out prevlen for the entry that is inserted. */
if (p[0] != ZIP_END) {
// 插入位置不在尾部
ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);
} else {
// 插入位置在尾部
unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);
// 尾节点存在取长度,不存在长度为0(列表为空时不存在)
if (ptail[0] != ZIP_END) {
prevlen = zipRawEntryLength(ptail);
}
}
// s指针指向新增节点数据 slen为数据长度
// 确定数据编码。数据长度,为整数时返回对应固定长度,为字符串使用slen
if (zipTryEncoding(s,slen,&value,&encoding)) {
reqlen = zipIntSize(encoding);
} else {
reqlen = slen;
}
// 加上前一节点长度信息的长度
reqlen += zipStorePrevEntryLength(NULL,prevlen);
// 加上编码与长度信息的长度
// 此时reqlen为新加入节点的整体长度
reqlen += zipStoreEntryEncoding(NULL,encoding,slen);
int forcelarge = 0;
//
nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;
// 修复bug,详细分析见:https://segmentfault.com/a/1190000018878466?utm_source=tag-newest
if (nextdiff == -4 && reqlen < 4) {
nextdiff = 0;
forcelarge = 1;
}
offset = p-zl;
// 调整内存大小
zl = ziplistResize(zl,curlen+reqlen+nextdiff);
p = zl+offset;
// 非空列表插入
if (p[0] != ZIP_END) {
// 将p节点后移(没有移动p节点前一节点长度信息),留出当前节点位置
memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);
// 写入p节点前一节点长度信息(要插入节点的长度)
if (forcelarge)
zipStorePrevEntryLengthLarge(p+reqlen,reqlen);
else
zipStorePrevEntryLength(p+reqlen,reqlen);
// 更新尾节点偏移量
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);
zipEntry(p+reqlen, &tail);
if (p[reqlen+tail.headersize+tail.len] != ZIP_END) {
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
}
} else {
// 空列表插入,只更新尾节点偏移量
ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl);
}
// 连锁更新
if (nextdiff != 0) {
offset = p-zl;
zl = __ziplistCascadeUpdate(zl,p+reqlen);
p = zl+offset;
}
// 写入前一节点长度信息
p += zipStorePrevEntryLength(p,prevlen);
// 写入节点编码与长度信息
p += zipStoreEntryEncoding(p,encoding,slen);
// 写入数据
if (ZIP_IS_STR(encoding)) {
memcpy(p,s,slen);
} else {
zipSaveInteger(p,value,encoding);
}
// 增加列表长度
ZIPLIST_INCR_LENGTH(zl,1);
return zl;
}
обновление цепочки
unsigned char *__ziplistCascadeUpdate(unsigned char *zl, unsigned char *p) {
size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), rawlen, rawlensize;
size_t offset, noffset, extra;
unsigned char *np;
zlentry cur, next;
while (p[0] != ZIP_END) {
// 解析当前节点信息
zipEntry(p, &cur);
// 当前节点总长
rawlen = cur.headersize + cur.len;
// 保存当前节点长度信息所需长度
rawlensize = zipStorePrevEntryLength(NULL,rawlen);
// 列表末尾,停止遍历
if (p[rawlen] == ZIP_END) break;
// 解析下一节点信息
zipEntry(p+rawlen, &next);
/* Abort when "prevlen" has not changed. */
if (next.prevrawlen == rawlen) break;
if (next.prevrawlensize < rawlensize) {
/* The "prevlen" field of "next" needs more bytes to hold
* the raw length of "cur". */
offset = p-zl;
// 下一节点因 前一节点长度信息 字段长度变更引发的自身长度变化大小
extra = rawlensize-next.prevrawlensize;
// 内存重新分配
zl = ziplistResize(zl,curlen+extra);
p = zl+offset;
/* Current pointer and offset for next element. */
np = p+rawlen;
noffset = np-zl;
// 如果下一节点不是尾节点,则需要更新 尾部节点偏移量
if ((zl+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))) != np) {
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+extra);
}
/* Move the tail to the back. */
memmove(np+rawlensize,
np+next.prevrawlensize,
curlen-noffset-next.prevrawlensize-1);
zipStorePrevEntryLength(np,rawlen);
p += rawlen;
curlen += extra;
} else {
// 如果 next节点原本的 前一节点长度信息 字段长度可以容纳新插入节点的长度信息,则直接写入并退出遍历
if (next.prevrawlensize > rawlensize) {
/* This would result in shrinking, which we want to avoid.
* So, set "rawlen" in the available bytes. */
zipStorePrevEntryLengthLarge(p+rawlen,rawlen);
} else {
zipStorePrevEntryLength(p+rawlen,rawlen);
}
/* Stop here, as the raw length of "next" has not changed. */
break;
}
}
return zl;
}
Ранее мы видели, что каждое поле узла сжатого списка является переменным, и если длина предыдущего узла изменится, это может привести к изменению длины следующего узла (если длина предыдущего узла до этого была меньше 254, а больше 254 после изменения, тогда байты, занимаемые информацией о длине предыдущего узла следующего узла, изменятся с 1 байта на 5 байтов). Рассмотрим предельный случай: длина последующих узлов вставленного узла находится в диапазоне от 250 до 253. В это время вставка узла, длина которого превышает 254, вызовет обновление цепочки, как показано на следующем рисунке:
Каждая строка на рисунке представляет собой перераспределение памяти, а красным прямоугольником показаны байты каждого расширения. В настоящее время временная сложность вставки узла составляет O(n^2) (n раз операций перераспределения пространства, наихудшая сложность каждого перераспределения пространства составляет O(n) ).
连锁更新最坏情况下带来的性能消耗是灾难性的,为什么还可以放心的使用?
- Условие срабатывания (имеется ровно несколько последовательных узлов длиной от 250 до 253) делает вероятность срабатывания очень низкой.
- Сценарий применения сжатого списка (количество узлов невелико и длина данных узла мала) и случай, когда длина составляет от 250 до 253, является низким, так что даже если есть обновление цепочки, есть несколько узлов, которые необходимо обновить.
Таким образом, средняя сложность вставки узла в сжатый список составляет O(n).
найти узел
unsigned char *ziplistFind(unsigned char *p, unsigned char *vstr, unsigned int vlen, unsigned int skip) {
int skipcnt = 0;
unsigned char vencoding = 0;
long long vll = 0;
// 遍历压缩列表 skip为查找前跳过skip个节点
while (p[0] != ZIP_END) {
unsigned int prevlensize, encoding, lensize, len;
unsigned char *q;
ZIP_DECODE_PREVLENSIZE(p, prevlensize);
ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
q = p + prevlensize + lensize;
if (skipcnt == 0) {
/* Compare current entry with specified entry */
if (ZIP_IS_STR(encoding)) {
// 数据为字符串,判断字符串相等复杂度为O(n)
if (len == vlen && memcmp(q, vstr, vlen) == 0) {
return p;
}
} else {
if (vencoding == 0) {
// 首次比对时,对传入值进行解码
if (!zipTryEncoding(vstr, vlen, &vll, &vencoding)) {
/* If the entry can't be encoded we set it to
* UCHAR_MAX so that we don't retry again the next
* time. */
vencoding = UCHAR_MAX;
}
/* Must be non-zero by now */
assert(vencoding);
}
/* Compare current entry with specified entry, do it only
* if vencoding != UCHAR_MAX because if there is no encoding
* possible for the field it can't be a valid integer. */
if (vencoding != UCHAR_MAX) {
long long ll = zipLoadInteger(q, encoding);
if (ll == vll) {
return p;
}
}
}
/* Reset skip count */
skipcnt = skip;
} else {
/* Skip entry */
skipcnt--;
}
/* Move to next entry */
p = q + len;
}
return NULL;
}
Поскольку в узлах есть строки, а сложность сравнения строк на равенство составляет O(n), сложность сжатия списка для поиска узлов составляет O(n^2).
удалить узел
unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p) {
size_t offset = *p-zl;
zl = __ziplistDelete(zl,*p,1);
/* Store pointer to current element in p, because ziplistDelete will
* do a realloc which might result in a different "zl"-pointer.
* When the delete direction is back to front, we might delete the last
* entry and end up with "p" pointing to ZIP_END, so check this. */
*p = zl+offset;
return zl;
}
/* Delete a range of entries from the ziplist. */
unsigned char *ziplistDeleteRange(unsigned char *zl, int index, unsigned int num) {
unsigned char *p = ziplistIndex(zl,index);
return (p == NULL) ? zl : __ziplistDelete(zl,p,num);
}
unsigned char *__ziplistDelete(unsigned char *zl, unsigned char *p, unsigned int num) {
unsigned int i, totlen, deleted = 0;
size_t offset;
int nextdiff = 0;
zlentry first, tail;
// 删除的首个节点
zipEntry(p, &first);
for (i = 0; p[0] != ZIP_END && i < num; i++) {
// 跳过所有要删除的节点
p += zipRawEntryLength(p);
deleted++;
}
// first.p指向第一个要被删除节点的首地址
// p指向的是最后一个删除节点的下一节点的首地址, totlen大于0表示需要删除节点
totlen = p-first.p; /* Bytes taken by the element(s) to delete. */
if (totlen > 0) {
if (p[0] != ZIP_END) {
// 获取p节点 【前一节点长度信息】 字段与首个被删除节点 【前一个节点长度信息】 字段的差值
// zipPrevLenByteDiff 的返回值有三种可能:
// 1)新旧两个节点的【前一个节点长度信息】字段长度 相等,返回 0
// 2)新节点【前一个节点长度信息】字段长度 > 旧节点【前一个节点长度信息】字段长度,返回 5 - 1 = 4
// 3)旧节点【前一个节点长度信息】字段长度 > 新节点【前一个节点长度信息】字段长度,返回 1 - 5 = -4
nextdiff = zipPrevLenByteDiff(p,first.prevrawlen);
// 根据差值,向前向后偏移 p指针
p -= nextdiff;
// 将首个被删除节点【前一节点长度信息】写入p指针指向的节点
zipStorePrevEntryLength(p,first.prevrawlen);
// 更新尾节点偏移量
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))-totlen);
/* When the tail contains more than one entry, we need to take
* "nextdiff" in account as well. Otherwise, a change in the
* size of prevlen doesn't have an effect on the *tail* offset. */
zipEntry(p, &tail);
/* 如果p节点不是尾节点, 则尾节点偏移量需要加上nextdiff的变更量
因为尾节点偏移量是指列表首地址到尾节点首地址的距离
p节点的 【前一节点长度信息】 字段的长度变化只影响它字段之后的信息地址。
p节点为尾节点时,为节点首地址在【前一节点长度信息】字段前边,所以不受影响。*/
if (p[tail.headersize+tail.len] != ZIP_END) {
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
}
/* Move tail to the front of the ziplist */
memmove(first.p,p,
intrev32ifbe(ZIPLIST_BYTES(zl))-(p-zl)-1);
} else {
/* The entire tail was deleted. No need to move memory. */
// 一直删除到尾节点,不需要变更中间节点,只需要调整下尾节点偏移量
ZIPLIST_TAIL_OFFSET(zl) =
intrev32ifbe((first.p-zl)-first.prevrawlen);
}
/* Resize and update length */
offset = first.p-zl;
// 重新分配内存大小
zl = ziplistResize(zl, intrev32ifbe(ZIPLIST_BYTES(zl))-totlen+nextdiff);
// 减少节点长度
ZIPLIST_INCR_LENGTH(zl,-deleted);
p = zl+offset;
// 如果最后一个被删除节点的下一节点的【前一个节点长度信息】字段长度 需要变更,则可能会触发连锁更新
if (nextdiff != 0)
zl = __ziplistCascadeUpdate(zl,p);
}
return zl;
}
Поскольку могут запускаться каскадные обновления, операции удаления в наихудшем случае имеют сложность O(n^2) и среднюю сложность O(n).