摘要：如果互斥鎖的持有者死亡了，或者持有這樣的互斥鎖的進程了互斥鎖所在的共享內存或者持有這樣的互斥鎖的進程執行了調用，則會解除鎖定該互斥鎖。互斥鎖的下一個持有者將獲取該互斥鎖并返回錯誤。

swoole_table 一個基于共享內存和鎖實現的超高性能，并發數據結構。用于解決多進程/多線程數據共享和同步加鎖問題。

swoole_table 實際上就是一個開鏈法實現的哈希表，memory 是一個由哈希鍵與具體數據組成的數組，如果哈希沖突(不同的鍵值對應同一個哈希)，那么就會從 pool 中分配出一個元素作為數組元素的鏈表尾

size 是創建共享內存表時設置的最大行數；conflict_proportion 是哈希沖突的最大比例，超過這個比例，共享內存表就不允許再添加新的行元素；iterator 是內存表的迭代器，可以利用它進行內存表數據的瀏覽；columns 是內存表的列元素集合，由于內存表的列元素也是一個 key-value 格式，因此也是一個哈希表 swHashMap 類型；mask 存放的是(最大行數-1)，專門進行哈希值與數組 index 的轉化；item_size 是所有列元素的內存大小總和；

typedef struct
{
    uint32_t absolute_index;
    uint32_t collision_index;
    swTableRow *row;
} swTable_iterator;

typedef struct
{
    swHashMap *columns;
    uint16_t column_num;
    swLock lock;
    size_t size;
    size_t mask;
    size_t item_size;
    size_t memory_size;
    float conflict_proportion;

    /**
     * total rows that in active state(shm)
     */
    sw_atomic_t row_num;

    swTableRow **rows;
    swMemoryPool *pool;

    swTable_iterator *iterator;

    void *memory;
} swTable;

swTableRow 是內存表的行元素，其中 lock 是行鎖；active 代表該行是否被啟用；next 是哈希沖突的鏈表；key 是該行的鍵值，也就是哈希之前的原始鍵值;data 是真正的行數據，里面會加載各個列元素的值

typedef struct _swTableRow
{
#if SW_TABLE_USE_SPINLOCK
    sw_atomic_t lock;
#else
    pthread_mutex_t lock;
#endif
    /**
     * 1:used, 0:empty
     */
    uint8_t active;
    /**
     * next slot
     */
    struct _swTableRow *next;
    /**
     * Hash Key
     */
    char key[SW_TABLE_KEY_SIZE];
    char data[0];
} swTableRow;

swTableColumn 是內存表的單個列元素，name 是列的名字；type 是列的數據類型，可選參數為 swoole_table_type；index 說明當前的列元素在表列中的位置；size 是指列的數據類型占用的內存大小

enum swoole_table_type
{
    SW_TABLE_INT = 1,
    SW_TABLE_INT8,
    SW_TABLE_INT16,
    SW_TABLE_INT32,
#ifdef __x86_64__
    SW_TABLE_INT64,
#endif
    SW_TABLE_FLOAT,
    SW_TABLE_STRING,
};

typedef struct
{
   uint8_t type;
   uint32_t size;
   swString* name;
   uint16_t index;
} swTableColumn;

swoole_table->__construct(int $size, float $conflict_proportion = 0.2) 這個共享內存表對象的創建對應于下面這個函數

哈希沖突百分比設定為最小 0.2，最大 1

行數并不是嚴格按照用戶定義的數據而來，如果 size 不是為 2 的 N 次方，如 1024、8192,65536 等，底層會自動調整為接近的一個數字，如果小于 1024 則默認成 1024，即 1024 是最小值

創建過程中各個成員變量的意義可見上一小節

swTable* swTable_new(uint32_t rows_size, float conflict_proportion)
{
    if (rows_size >= 0x80000000)
    {
        rows_size = 0x80000000;
    }
    else
    {
        uint32_t i = 10;
        while ((1U << i) < rows_size)
        {
            i++;
        }
        rows_size = 1 << i;
    }

    if (conflict_proportion > 1.0)
    {
        conflict_proportion = 1.0;
    }
    else if (conflict_proportion < SW_TABLE_CONFLICT_PROPORTION)
    {
        conflict_proportion = SW_TABLE_CONFLICT_PROPORTION;
    }

    swTable *table = SwooleG.memory_pool->alloc(SwooleG.memory_pool, sizeof(swTable));
    if (table == NULL)
    {
        return NULL;
    }
    if (swMutex_create(&table->lock, 1) < 0)
    {
        swWarn("mutex create failed.");
        return NULL;
    }
    table->iterator = sw_malloc(sizeof(swTable_iterator));
    if (!table->iterator)
    {
        swWarn("malloc failed.");
        return NULL;
    }
    table->columns = swHashMap_new(SW_HASHMAP_INIT_BUCKET_N, (swHashMap_dtor)swTableColumn_free);
    if (!table->columns)
    {
        return NULL;
    }

    table->size = rows_size;
    table->mask = rows_size - 1;
    table->conflict_proportion = conflict_proportion;

    bzero(table->iterator, sizeof(swTable_iterator));
    table->memory = NULL;
    return table;
}

swoole_table->column(string $name, int $type, int $size = 0) 對應下面的函數

列元素創建并初始化成功后，會調用 swHashMap_add 函數將列元素添加到 table->columns 中

int swTableColumn_add(swTable *table, char *name, int len, int type, int size)
{
    swTableColumn *col = sw_malloc(sizeof(swTableColumn));
    if (!col)
    {
        return SW_ERR;
    }
    col->name = swString_dup(name, len);
    if (!col->name)
    {
        sw_free(col);
        return SW_ERR;
    }
    switch(type)
    {
    case SW_TABLE_INT:
        switch(size)
        {
        case 1:
            col->size = 1;
            col->type = SW_TABLE_INT8;
            break;
        case 2:
            col->size = 2;
            col->type = SW_TABLE_INT16;
            break;
#ifdef __x86_64__
        case 8:
            col->size = 8;
            col->type = SW_TABLE_INT64;
            break;
#endif
        default:
            col->size = 4;
            col->type = SW_TABLE_INT32;
            break;
        }
        break;
    case SW_TABLE_FLOAT:
        col->size = sizeof(double);
        col->type = SW_TABLE_FLOAT;
        break;
    case SW_TABLE_STRING:
        col->size = size + sizeof(swTable_string_length_t);
        col->type = SW_TABLE_STRING;
        break;
    default:
        swWarn("unkown column type.");
        swTableColumn_free(col);
        return SW_ERR;
    }
    col->index = table->item_size;
    table->item_size += col->size;
    table->column_num ++;
    return swHashMap_add(table->columns, name, len, col);
}

通過 swTable_get_memory_size 函數計算整個共享內存表需要的內存總數，這個內存總數包含了哈希沖突需要的多余的內存占用。

申請了 memory_size 后，會將首地址賦值給 table->rows；值得注意的是 table->rows 是 swTableRow ** 類型，后面還要通過循環給各個行元素賦值首地址

為了降低行鎖的時間消耗，設置行鎖為 PTHREAD_PRIO_INHERIT，提高行鎖的優先級(如果更高優先級的線程因 thrd1 所擁有的一個或多個互斥鎖而被阻塞,而這些互斥鎖是用 PTHREAD_PRIO_INHERIT 初始化的,則 thrd1 的運行優先級為優先級 pri1 和優先級 pri2 中優先級較高的那一個,如果沒有優先級繼承，底優先級的線程可能會在很長一段時間內都得不到調度，而這會導致等待低優先級線程鎖持有的鎖的高優先級線程也等待很長時間（因為低優先級線程無法運行，因而就無法釋放鎖，所以高優先級線程只能繼續阻塞在鎖上）。使用優先級繼承可以短時間的提高低優先級線程的優先級，從而使它可以盡快得到調度，然后釋放鎖。低優先級線程在釋放鎖后就會恢復自己的優先級。)

PTHREAD_MUTEX_ROBUST_NP： 如果互斥鎖的持有者“死亡”了，或者持有這樣的互斥鎖的進程 unmap 了互斥鎖所在的共享內存或者持有這樣的互斥鎖的進程執行了 exec 調用，則會解除鎖定該互斥鎖。互斥鎖的下一個持有者將獲取該互斥鎖,并返回錯誤 EOWNWERDEAD。

table->rows 創建成功之后，就要對哈希沖突的行元素分配地址空間。可以看到，哈希沖突的行元素首地址為 memory += row_memory_size * table->size，并且利用已有的內存構建 FixedPool 隨機內存池，row_memory_size 作為內存池內部元素的大小

int swTable_create(swTable *table)
{
    size_t memory_size = swTable_get_memory_size(table);
    size_t row_memory_size = sizeof(swTableRow) + table->item_size;

    void *memory = sw_shm_malloc(memory_size);
    if (memory == NULL)
    {
        return SW_ERR;
    }

    table->memory_size = memory_size;
    table->memory = memory;

    table->rows = memory;
    memory += table->size * sizeof(swTableRow *);
    memory_size -= table->size * sizeof(swTableRow *);

#if SW_TABLE_USE_SPINLOCK == 0
    pthread_mutexattr_t attr;
    pthread_mutexattr_init(&attr);
    pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
    pthread_mutexattr_setprotocol(&attr, PTHREAD_PRIO_INHERIT);
    pthread_mutexattr_setrobust_np(&attr, PTHREAD_MUTEX_ROBUST_NP);
#endif

    int i;
    for (i = 0; i < table->size; i++)
    {
        table->rows[i] = memory + (row_memory_size * i);
        memset(table->rows[i], 0, sizeof(swTableRow));
#if SW_TABLE_USE_SPINLOCK == 0
        pthread_mutex_init(&table->rows[i]->lock, &attr);
#endif
    }

    memory += row_memory_size * table->size;
    memory_size -= row_memory_size * table->size;
    table->pool = swFixedPool_new2(row_memory_size, memory, memory_size);

    return SW_OK;
}

計算整個共享內存表的內存大小：

(內存表行數+哈希沖突行數)*(行元素大小+各個列元素大小總和)+哈希沖突內存池頭部大小+行元素指針大小*內存表行數

比較難以理解的是最后那個 行元素大小*內存表行數，這個其實是在創建 table->rows[table->size] 這個指針數組，我們之前說過 table->rows 是個二維數組，這個數組里面存放的是多個 swTableRow* 類型的數據，例如 table->rows[0]等，table->rows[0] 等才是存放各個行元素首地址的地方，如果沒有這個指針數組，那么每次去取行元素都要計算行元素的首地址，效率沒有這么快。

size_t swTable_get_memory_size(swTable *table)
{
    /**
     * table size + conflict size
     */
    size_t row_num = table->size * (1 + table->conflict_proportion);

    /*
     * header + data
     */
    size_t row_memory_size = sizeof(swTableRow) + table->item_size;

    /**
     * row data & header
     */
    size_t memory_size = row_num * row_memory_size;

    /**
     * memory pool for conflict rows
     */
    memory_size += sizeof(swMemoryPool) + sizeof(swFixedPool) + ((row_num - table->size) * sizeof(swFixedPool_slice));

    /**
     * for iterator, Iterate through all the elements
     */
    memory_size += table->size * sizeof(swTableRow *);

    return memory_size;
}

共享內存表添加新的元素需要調用三個函數，分別是 swTableRow_set 設置行的 key 值、swTableColumn_get 獲取列元素，swTableRow_set_value 函數根據列的數據類型為 row->data 賦值，流程如下：

static PHP_METHOD(swoole_table, set)
{
    zval *array;
    char *key;
    zend_size_t keylen;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "sa", &key, &keylen, &array) == FAILURE)
    {
        RETURN_FALSE;
    }

    swTable *table = swoole_get_object(getThis());

    swTableRow *_rowlock = NULL;
    swTableRow *row = swTableRow_set(table, key, keylen, &_rowlock);

    swTableColumn *col;
    zval *v;
    char *k;
    uint32_t klen;
    int ktype;
    HashTable *_ht = Z_ARRVAL_P(array);

    SW_HASHTABLE_FOREACH_START2(_ht, k, klen, ktype, v)
    {
        col = swTableColumn_get(table, k, klen);

        else if (col->type == SW_TABLE_STRING)
        {
            convert_to_string(v);
            swTableRow_set_value(row, col, Z_STRVAL_P(v), Z_STRLEN_P(v));
        }
        else if (col->type == SW_TABLE_FLOAT)
        {
            convert_to_double(v);
            swTableRow_set_value(row, col, &Z_DVAL_P(v), 0);
        }
        else
        {
            convert_to_long(v);
            swTableRow_set_value(row, col, &Z_LVAL_P(v), 0);
        }
    }
    swTableRow_unlock(_rowlock);
}

我們先來看 swTableRow_set 函數，從下面的代碼來看，這個函數主要的作用就是判斷新添加的 key 是否造成了哈希沖突，如果沒有沖突(row->active=0)，那么直接 table->row_num 自增，設置 row->key 就可以了。

如果發生哈希沖突，那么就要循環當前行元素的鏈表，直到（1）找到相同的 key 值，說明并不是真的發生了哈希沖突，而是用戶要修改已有的行數據，那么就直接跳出函數，然后更改 row->data 的值；(2) 沒有找到相同的 key 值，說明的確遇到了哈希沖突，不同的 key 值對應了相同的哈希值，此時已經循環到達鏈表的末尾，需要從內存池中構建出一個 swTableRow 行元素，放到鏈表的尾部

swTableRow* swTableRow_set(swTable *table, char *key, int keylen, swTableRow **rowlock)
{
    if (keylen > SW_TABLE_KEY_SIZE)
    {
        keylen = SW_TABLE_KEY_SIZE;
    }

    swTableRow *row = swTable_hash(table, key, keylen);
    *rowlock = row;
    swTableRow_lock(row);

#ifdef SW_TABLE_DEBUG
    int _conflict_level = 0;
#endif

    if (row->active)
    {
        for (;;)
        {
            if (strncmp(row->key, key, keylen) == 0)
            {
                break;
            }
            else if (row->next == NULL)
            {
                table->lock.lock(&table->lock);
                swTableRow *new_row = table->pool->alloc(table->pool, 0);

#ifdef SW_TABLE_DEBUG
                conflict_count ++;
                if (_conflict_level > conflict_max_level)
                {
                    conflict_max_level = _conflict_level;
                }

#endif
                table->lock.unlock(&table->lock);

                if (!new_row)
                {
                    return NULL;
                }
                //add row_num
                bzero(new_row, sizeof(swTableRow));
                sw_atomic_fetch_add(&(table->row_num), 1);
                row->next = new_row;
                row = new_row;
                break;
            }
            else
            {
                row = row->next;
#ifdef SW_TABLE_DEBUG
                _conflict_level++;
#endif
            }
        }
    }
    else
    {
#ifdef SW_TABLE_DEBUG
        insert_count ++;
#endif
        sw_atomic_fetch_add(&(table->row_num), 1);
    }

    memcpy(row->key, key, keylen);
    row->active = 1;
    return row;
}

那么接下來我們看代碼中 swTable_hash 這個函數是怎能計算哈希值的——我們發現哈希函數有兩種:

swoole_hash_php 是 php 的經典哈希函數，也就是 time33/DJB 算法

swoole_hash_austin 是 MurmurHash 哈希算法，廣泛應用在 redis、Memcached 等算法中

哈希計算之后，我們發現哈希值又與 table->mask 進行了邏輯與計算，目的是得到一個小于等于 table->mask(rows_size - 1) 的數字，作為行元素的 index

static sw_inline swTableRow* swTable_hash(swTable *table, char *key, int keylen)
{
#ifdef SW_TABLE_USE_PHP_HASH
    uint64_t hashv = swoole_hash_php(key, keylen);
#else
    uint64_t hashv = swoole_hash_austin(key, keylen);
#endif
    uint64_t index = hashv & table->mask;
    assert(index < table->size);
    return table->rows[index];
}

我們接下來看行鎖的加鎖函數：

若是普通的互斥鎖，那么就直接使用 pthread_mutex_lock 即可，如果不是互斥鎖，程序實現了一個自旋鎖

若是自旋鎖，就調用 swoole 自定義的自旋鎖加鎖

static sw_inline void swTableRow_lock(swTableRow *row)
{
#if SW_TABLE_USE_SPINLOCK
    sw_spinlock(&row->lock);
#else
    pthread_mutex_lock(&row->lock);
#endif
}

從多個列元素組成的 hashMap 中根據 column_key 快速找到對應的列元素

static sw_inline swTableColumn* swTableColumn_get(swTable *table, char *column_key, int keylen)
{
    return swHashMap_find(table->columns, column_key, keylen);
}

根據取出的列元素數據的類型，為 row->data 對應的位置上賦值，值得注意的是 default 實際上指的是 SW_TABLE_STRING 類型，這時會先儲存字符串長度，再存儲字符串值：

static sw_inline void swTableRow_set_value(swTableRow *row, swTableColumn * col, void *value, int vlen)
{
    int8_t _i8;
    int16_t _i16;
    int32_t _i32;
#ifdef __x86_64__
    int64_t _i64;
#endif
    switch(col->type)
    {
    case SW_TABLE_INT8:
        _i8 = *(int8_t *) value;
        memcpy(row->data + col->index, &_i8, 1);
        break;
    case SW_TABLE_INT16:
        _i16 =  *(int16_t *) value;
        memcpy(row->data + col->index, &_i16, 2);
        break;
    case SW_TABLE_INT32:
        _i32 =  *(int32_t *) value;
        memcpy(row->data + col->index, &_i32, 4);
        break;
#ifdef __x86_64__
    case SW_TABLE_INT64:
        _i64 =  *(int64_t *) value;
        memcpy(row->data + col->index, &_i64, 8);
        break;
#endif
    case SW_TABLE_FLOAT:
        memcpy(row->data + col->index, value, sizeof(double));
        break;
    default:
        if (vlen > (col->size - sizeof(swTable_string_length_t)))
        {
            swWarn("[key=%s,field=%s]string value is too long.", row->key, col->name->str);
            vlen = col->size - sizeof(swTable_string_length_t);
        }
        memcpy(row->data + col->index, &vlen, sizeof(swTable_string_length_t));
        memcpy(row->data + col->index + sizeof(swTable_string_length_t), value, vlen);
        break;
    }
}

根據鍵值獲取行元素需要調用三個函數：swTableRow_get 獲取行對象元素，如果只取特定字段，那么會調用 php_swoole_table_get_field_value，如果需要去全部字段，那么會調用 php_swoole_table_row2array：

static PHP_METHOD(swoole_table, get)
{
    char *key;
    zend_size_t keylen;

    char *field = NULL;
    zend_size_t field_len = 0;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s|s", &key, &keylen, &field, &field_len) == FAILURE)
    {
        RETURN_FALSE;
    }

    swTableRow *_rowlock = NULL;
    swTable *table = swoole_get_object(getThis());

    swTableRow *row = swTableRow_get(table, key, keylen, &_rowlock);
    if (field && field_len > 0)
    {
        php_swoole_table_get_field_value(table, row, return_value, field, (uint16_t) field_len);
    }
    else
    {
        php_swoole_table_row2array(table, row, return_value);
    }
    swTableRow_unlock(_rowlock);
}

swTableRow_get 函數

利用 key 計算出行元素的 index 值，遇到存在哈希鏈表的情況，要不斷對比 key 的值，直到找到完全相等的鍵值返回：

swTableRow* swTableRow_get(swTable *table, char *key, int keylen, swTableRow** rowlock)
{
    if (keylen > SW_TABLE_KEY_SIZE)
    {
        keylen = SW_TABLE_KEY_SIZE;
    }

    swTableRow *row = swTable_hash(table, key, keylen);
    *rowlock = row;
    swTableRow_lock(row);

    for (;;)
    {
        if (strncmp(row->key, key, keylen) == 0)
        {
            if (!row->active)
            {
                row = NULL;
            }
            break;
        }
        else if (row->next == NULL)
        {
            row = NULL;
            break;
        }
        else
        {
            row = row->next;
        }
    }

    return row;
}

php_swoole_table_get_field_value 函數

首先通過 swHashMap_find 函數根據 field 確定字段類型，如果是字符串，需要先獲取字符串的長度：

static inline void php_swoole_table_get_field_value(swTable *table, swTableRow *row, zval *return_value, char *field, uint16_t field_len)
{
    swTable_string_length_t vlen = 0;
    double dval = 0;
    int64_t lval = 0;

    swTableColumn *col = swHashMap_find(table->columns, field, field_len);
    
    if (col->type == SW_TABLE_STRING)
    {
        memcpy(&vlen, row->data + col->index, sizeof(swTable_string_length_t));
        SW_ZVAL_STRINGL(return_value, row->data + col->index + sizeof(swTable_string_length_t), vlen, 1);
    }
    else if (col->type == SW_TABLE_FLOAT)
    {
        memcpy(&dval, row->data + col->index, sizeof(dval));
        ZVAL_DOUBLE(return_value, dval);
    }
    else
    {
        switch (col->type)
        {
        case SW_TABLE_INT8:
            memcpy(&lval, row->data + col->index, 1);
            ZVAL_LONG(return_value, (int8_t) lval);
            break;
        case SW_TABLE_INT16:
            memcpy(&lval, row->data + col->index, 2);
            ZVAL_LONG(return_value, (int16_t) lval);
            break;
        case SW_TABLE_INT32:
            memcpy(&lval, row->data + col->index, 4);
            ZVAL_LONG(return_value, (int32_t) lval);
            break;
        default:
            memcpy(&lval, row->data + col->index, 8);
            ZVAL_LONG(return_value, lval);
            break;
        }
    }
}

與上一個函數相比，這個函數僅僅是換成了利用 swHashMap_each 遍歷列元素，然后利用列元素取值的過程，取值之后，還有利用 add_assoc_stringl_ex 等 zend 的 API, 將值不斷轉化為 PHP 數組：

#define sw_add_assoc_string                   add_assoc_string
#define sw_add_assoc_stringl_ex               add_assoc_stringl_ex
#define sw_add_assoc_stringl                  add_assoc_stringl
#define sw_add_assoc_double_ex                add_assoc_double_ex
#define sw_add_assoc_long_ex                  add_assoc_long_ex
#define sw_add_next_index_stringl             add_next_index_stringl

static inline void php_swoole_table_row2array(swTable *table, swTableRow *row, zval *return_value)
{
    array_init(return_value);

    swTableColumn *col = NULL;
    swTable_string_length_t vlen = 0;
    double dval = 0;
    int64_t lval = 0;
    char *k;

    while(1)
    {
        col = swHashMap_each(table->columns, &k);
        if (col == NULL)
        {
            break;
        }
        if (col->type == SW_TABLE_STRING)
        {
            memcpy(&vlen, row->data + col->index, sizeof(swTable_string_length_t));
            sw_add_assoc_stringl_ex(return_value, col->name->str, col->name->length + 1, row->data + col->index + sizeof(swTable_string_length_t), vlen, 1);
        }
        else if (col->type == SW_TABLE_FLOAT)
        {
            memcpy(&dval, row->data + col->index, sizeof(dval));
            sw_add_assoc_double_ex(return_value, col->name->str, col->name->length + 1, dval);
        }
        else
        {
            switch (col->type)
            {
            case SW_TABLE_INT8:
                memcpy(&lval, row->data + col->index, 1);
                sw_add_assoc_long_ex(return_value, col->name->str, col->name->length + 1, (int8_t) lval);
                break;
            case SW_TABLE_INT16:
                memcpy(&lval, row->data + col->index, 2);
                sw_add_assoc_long_ex(return_value, col->name->str, col->name->length + 1, (int16_t) lval);
                break;
            case SW_TABLE_INT32:
                memcpy(&lval, row->data + col->index, 4);
                sw_add_assoc_long_ex(return_value, col->name->str, col->name->length + 1, (int32_t) lval);
                break;
            default:
                memcpy(&lval, row->data + col->index, 8);
                sw_add_assoc_long_ex(return_value, col->name->str, col->name->length + 1, lval);
                break;
            }
        }
    }
}

static PHP_METHOD(swoole_table, incr)
{
    char *key;
    zend_size_t key_len;
    char *col;
    zend_size_t col_len;
    zval* incrby = NULL;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "ss|z", &key, &key_len, &col, &col_len, &incrby) == FAILURE)
    {
        RETURN_FALSE;
    }

    swTableRow *_rowlock = NULL;
    swTable *table = swoole_get_object(getThis());

    swTableRow *row = swTableRow_set(table, key, key_len, &_rowlock);

    swTableColumn *column;
    column = swTableColumn_get(table, col, col_len);
    if (column->type == SW_TABLE_STRING)
    {
        swTableRow_unlock(_rowlock);
        swoole_php_fatal_error(E_WARNING, "can"t execute "incr" on a string type column.");
        RETURN_FALSE;
    }
    else if (column->type == SW_TABLE_FLOAT)
    {
        double set_value = 0;
        memcpy(&set_value, row->data + column->index, sizeof(set_value));
        if (incrby)
        {
            convert_to_double(incrby);
            set_value += Z_DVAL_P(incrby);
        }
        else
        {
            set_value += 1;
        }
        swTableRow_set_value(row, column, &set_value, 0);
        RETVAL_DOUBLE(set_value);
    }
    else
    {
        int64_t set_value = 0;
        memcpy(&set_value, row->data + column->index, column->size);
        if (incrby)
        {
            convert_to_long(incrby);
            set_value += Z_LVAL_P(incrby);
        }
        else
        {
            set_value += 1;
        }
        swTableRow_set_value(row, column, &set_value, 0);
        RETVAL_LONG(set_value);
    }
    swTableRow_unlock(_rowlock);
}

static PHP_METHOD(swoole_table, decr)
{
    char *key;
    zend_size_t key_len;
    char *col;
    zend_size_t col_len;
    zval *decrby = NULL;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "ss|z", &key, &key_len, &col, &col_len, &decrby) == FAILURE)
    {
        RETURN_FALSE;
    }

    swTableRow *_rowlock = NULL;
    swTable *table = swoole_get_object(getThis());

    swTableRow *row = swTableRow_set(table, key, key_len, &_rowlock);

    swTableColumn *column;
    column = swTableColumn_get(table, col, col_len);
    if (column->type == SW_TABLE_STRING)
    {
        swTableRow_unlock(_rowlock);
        swoole_php_fatal_error(E_WARNING, "can"t execute "decr" on a string type column.");
        RETURN_FALSE;
    }
    else if (column->type == SW_TABLE_FLOAT)
    {
        double set_value = 0;
        memcpy(&set_value, row->data + column->index, sizeof(set_value));
        if (decrby)
        {
            convert_to_double(decrby);
            set_value -= Z_DVAL_P(decrby);
        }
        else
        {
            set_value -= 1;
        }
        swTableRow_set_value(row, column, &set_value, 0);
        RETVAL_DOUBLE(set_value);
    }
    else
    {
        int64_t set_value = 0;
        memcpy(&set_value, row->data + column->index, column->size);
        if (decrby)
        {
            convert_to_long(decrby);
            set_value -= Z_LVAL_P(decrby);
        }
        else
        {
            set_value -= 1;
        }
        swTableRow_set_value(row, column, &set_value, 0);
        RETVAL_LONG(set_value);
    }
    swTableRow_unlock(_rowlock);
}

共享內存表的數據刪除稍微有些復雜分為以下幾個情況：

要刪除的行元素沒有哈希沖突的鏈表

如果鍵值一致，那么利用 bzero 初始化該行元素，減小共享表行數

如果鍵值不一致，說明并沒有這行數據，那么直接解鎖返回

要刪除的行元素存在哈希沖突的鏈表，那么就要循環鏈表來找出鍵值一致的行元素

如果遍歷鏈表都沒有找到，那么直接解鎖返回

如果發現是鏈表的頭元素，那么將鏈表的第二個元素的數據賦值給頭元素，然后從內存池中釋放鏈表的第二個元素，減小共享表行數

如果是鏈表的中間元素，那么和普通刪除鏈表節點的方法一致，減小共享表行數

int swTableRow_del(swTable *table, char *key, int keylen)
{
    if (keylen > SW_TABLE_KEY_SIZE)
    {
        keylen = SW_TABLE_KEY_SIZE;
    }

    swTableRow *row = swTable_hash(table, key, keylen);
    //no exists
    if (!row->active)
    {
        return SW_ERR;
    }

    swTableRow_lock(row);
    if (row->next == NULL)
    {
        if (strncmp(row->key, key, keylen) == 0)
        {
            bzero(row, sizeof(swTableRow) + table->item_size);
            goto delete_element;
        }
        else
        {
            goto not_exists;
        }
    }
    else
    {
        swTableRow *tmp = row;
        swTableRow *prev = NULL;

        while (tmp)
        {
            if ((strncmp(tmp->key, key, keylen) == 0))
            {
                break;
            }
            prev = tmp;
            tmp = tmp->next;
        }

        if (tmp == NULL)
        {
            not_exists:
            swTableRow_unlock(row);
            return SW_ERR;
        }

        //when the deleting element is root, we should move the first element"s data to root,
        //and remove the element from the collision list.
        if (tmp == row)
        {
            tmp = tmp->next;
            row->next = tmp->next;
            memcpy(row->key, tmp->key, strlen(tmp->key));
            memcpy(row->data, tmp->data, table->item_size);
        }
        if (prev)
        {
            prev->next = tmp->next;
        }
        table->lock.lock(&table->lock);
        bzero(tmp, sizeof(swTableRow) + table->item_size);
        table->pool->free(table->pool, tmp);
        table->lock.unlock(&table->lock);
    }

    delete_element:
    sw_atomic_fetch_sub(&(table->row_num), 1);
    swTableRow_unlock(row);

    return SW_OK;
}

swoole_table 類實現了迭代器，可以使用 foreach 進行遍歷。

void swoole_table_init(int module_number TSRMLS_DC)
{
    #ifdef HAVE_PCRE
    zend_class_implements(swoole_table_class_entry_ptr TSRMLS_CC, 2, spl_ce_Iterator, spl_ce_Countable);
#endif
}

可以看到，swoole 在對 swoole_table 進行初始化的時候，為這個類繼承了 spl_iterator 這個接口，我們知道，對繼承了這個接口的類進行 foreach，不會觸發原始的對象成員變量的遍歷，而是會調用 spl_iterator 的 rewind、next 等方法：

#ifdef HAVE_PCRE
static PHP_METHOD(swoole_table, rewind);
static PHP_METHOD(swoole_table, next);
static PHP_METHOD(swoole_table, current);
static PHP_METHOD(swoole_table, key);
static PHP_METHOD(swoole_table, valid);
#endif

關于為什么要 PCRE 這個正則表達式庫的依賴，本人非常疑惑，希望有人能夠解疑。

static PHP_METHOD(swoole_table, rewind)
{
    swTable *table = swoole_get_object(getThis());
    if (!table->memory)
    {
        swoole_php_fatal_error(E_ERROR, "the swoole table does not exist.");
        RETURN_FALSE;
    }
    swTable_iterator_rewind(table);
    swTable_iterator_forward(table);
}

void swTable_iterator_rewind(swTable *table)
{
    bzero(table->iterator, sizeof(swTable_iterator));
}

rewind 函數就是將數據迭代器返回到開始的位置，對于 swTable 來說，就是將 absolute_index、collision_index、row 等重置為 0 即可。

swTable_iterator_forward 就是將迭代器向前進行一步，其中 absolute_index 類似于共享表的行索引，collision_index 類似于共享表的列索引。不同的是，對于沒有哈希沖突的行，列索引只有一個 0，對于哈希沖突的行，列索引就是開鏈法的鏈表索引：

static sw_inline swTableRow* swTable_iterator_get(swTable *table, uint32_t index)
{
    swTableRow *row = table->rows[index];
    return row->active ? row : NULL;
}

void swTable_iterator_forward(swTable *table)
{
    for (; table->iterator->absolute_index < table->size; table->iterator->absolute_index++)
    {
        swTableRow *row = swTable_iterator_get(table, table->iterator->absolute_index);
        if (row == NULL)
        {
            continue;
        }
        else if (row->next == NULL)
        {
            table->iterator->absolute_index++;
            table->iterator->row = row;
            return;
        }
        else
        {
            int i = 0;
            for (;; i++)
            {
                if (row == NULL)
                {
                    table->iterator->collision_index = 0;
                    break;
                }
                if (i == table->iterator->collision_index)
                {
                    table->iterator->collision_index++;
                    table->iterator->row = row;
                    return;
                }
                row = row->next;
            }
        }
    }
    table->iterator->row = NULL;
}

current 方法很簡單，取出當前迭代器的行元素，再轉化為 php 數組即可

static PHP_METHOD(swoole_table, current)
{
    swTable *table = swoole_get_object(getThis());
    if (!table->memory)
    {
        swoole_php_fatal_error(E_ERROR, "the swoole table does not exist.");
        RETURN_FALSE;
    }
    swTableRow *row = swTable_iterator_current(table);
    swTableRow_lock(row);
    php_swoole_table_row2array(table, row, return_value);
    swTableRow_unlock(row);
}

swTableRow* swTable_iterator_current(swTable *table)
{
    return table->iterator->row;
}

取出當前迭代器的鍵值：

static PHP_METHOD(swoole_table, key)
{
    swTable *table = swoole_get_object(getThis());
    if (!table->memory)
    {
        swoole_php_fatal_error(E_ERROR, "the swoole table does not exist.");
        RETURN_FALSE;
    }
    swTableRow *row = swTable_iterator_current(table);
    swTableRow_lock(row);
    SW_RETVAL_STRING(row->key, 1);
    swTableRow_unlock(row);
}

next 就是迭代器向前進一步：

static PHP_METHOD(swoole_table, next)
{
    swTable *table = swoole_get_object(getThis());
    if (!table->memory)
    {
        swoole_php_fatal_error(E_ERROR, "the swoole table does not exist.");
        RETURN_FALSE;
    }
    swTable_iterator_forward(table);
}

驗證當前行元素是否為空：

static PHP_METHOD(swoole_table, valid)
{
    swTable *table = swoole_get_object(getThis());
    if (!table->memory)
    {
        swoole_php_fatal_error(E_ERROR, "the swoole table does not exist.");
        RETURN_FALSE;
    }
    swTableRow *row = swTable_iterator_current(table);
    RETURN_BOOL(row != NULL);
}