前言

swoole_table 一個基于共享內(nèi)存和鎖實現(xiàn)的超高性能孔庭，并發(fā)數(shù)據(jù)結(jié)構(gòu)。用于解決多進程/多線程數(shù)據(jù)共享和同步加鎖問題材蛛。

swoole_table 的數(shù)據(jù)結(jié)構(gòu)

• swoole_table 實際上就是一個開鏈法實現(xiàn)的哈希表圆到，memory 是一個由哈希鍵與具體數(shù)據(jù)組成的數(shù)組，如果哈希沖突(不同的鍵值對應(yīng)同一個哈希)卑吭，那么就會從 pool 中分配出一個元素作為數(shù)組元素的鏈表尾
• size 是創(chuàng)建共享內(nèi)存表時設(shè)置的最大行數(shù)芽淡；conflict_proportion 是哈希沖突的最大比例，超過這個比例豆赏，共享內(nèi)存表就不允許再添加新的行元素挣菲；iterator 是內(nèi)存表的迭代器，可以利用它進行內(nèi)存表數(shù)據(jù)的瀏覽掷邦；columns 是內(nèi)存表的列元素集合白胀，由于內(nèi)存表的列元素也是一個 key-value 格式，因此也是一個哈希表 swHashMap 類型抚岗；mask 存放的是(最大行數(shù)-1)或杠，專門進行哈希值與數(shù)組 index 的轉(zhuǎn)化；item_size 是所有列元素的內(nèi)存大小總和宣蔚；

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 是內(nèi)存表的行元素向抢，其中 lock 是行鎖认境；active 代表該行是否被啟用；next 是哈希沖突的鏈表笋额；key 是該行的鍵值元暴，也就是哈希之前的原始鍵值;data 是真正的行數(shù)據(jù)，里面會加載各個列元素的值

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 是內(nèi)存表的單個列元素兄猩，name 是列的名字茉盏；type 是列的數(shù)據(jù)類型，可選參數(shù)為 swoole_table_type枢冤；index 說明當(dāng)前的列元素在表列中的位置鸠姨；size 是指列的數(shù)據(jù)類型占用的內(nèi)存大小

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 的構(gòu)造

• swoole_table->__construct(int $size, float $conflict_proportion = 0.2) 這個共享內(nèi)存表對象的創(chuàng)建對應(yīng)于下面這個函數(shù)
• 哈希沖突百分比設(shè)定為最小 0.2，最大 1
• 行數(shù)并不是嚴格按照用戶定義的數(shù)據(jù)而來淹真，如果 size 不是為 2 的 N 次方，如 1024核蘸、8192,65536 等，底層會自動調(diào)整為接近的一個數(shù)字，如果小于 1024 則默認成 1024袱吆，即 1024 是最小值
• 創(chuàng)建過程中各個成員變量的意義可見上一小節(jié)

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 添加新的列

• swoole_table->column(string $name, int $type, int $size = 0) 對應(yīng)下面的函數(shù)
• 列元素創(chuàng)建并初始化成功后婶希，會調(diào)用 swHashMap_add 函數(shù)將列元素添加到 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);
}

swoole_table 的創(chuàng)建

• 通過 swTable_get_memory_size 函數(shù)計算整個共享內(nèi)存表需要的內(nèi)存總數(shù)，這個內(nèi)存總數(shù)包含了哈希沖突需要的多余的內(nèi)存占用蓬衡。
• 申請了 memory_size 后饲趋，會將首地址賦值給 table->rows；值得注意的是 table->rows 是 swTableRow ** 類型撤蟆，后面還要通過循環(huán)給各個行元素賦值首地址
• 為了降低行鎖的時間消耗奕塑，設(shè)置行鎖為 PTHREAD_PRIO_INHERIT，提高行鎖的優(yōu)先級(如果更高優(yōu)先級的線程因 thrd1 所擁有的一個或多個互斥鎖而被阻塞,而這些互斥鎖是用 PTHREAD_PRIO_INHERIT 初始化的,則 thrd1 的運行優(yōu)先級為優(yōu)先級 pri1 和優(yōu)先級 pri2 中優(yōu)先級較高的那一個,如果沒有優(yōu)先級繼承家肯，底優(yōu)先級的線程可能會在很長一段時間內(nèi)都得不到調(diào)度龄砰，而這會導(dǎo)致等待低優(yōu)先級線程鎖持有的鎖的高優(yōu)先級線程也等待很長時間（因為低優(yōu)先級線程無法運行，因而就無法釋放鎖，所以高優(yōu)先級線程只能繼續(xù)阻塞在鎖上）换棚。使用優(yōu)先級繼承可以短時間的提高低優(yōu)先級線程的優(yōu)先級式镐，從而使它可以盡快得到調(diào)度，然后釋放鎖固蚤。低優(yōu)先級線程在釋放鎖后就會恢復(fù)自己的優(yōu)先級娘汞。)
• PTHREAD_MUTEX_ROBUST_NP： 如果互斥鎖的持有者“死亡”了，或者持有這樣的互斥鎖的進程 unmap 了互斥鎖所在的共享內(nèi)存或者持有這樣的互斥鎖的進程執(zhí)行了 exec 調(diào)用夕玩，則會解除鎖定該互斥鎖你弦。互斥鎖的下一個持有者將獲取該互斥鎖,并返回錯誤 EOWNWERDEAD燎孟。
• table->rows 創(chuàng)建成功之后禽作，就要對哈希沖突的行元素分配地址空間】常可以看到旷偿，哈希沖突的行元素首地址為 memory += row_memory_size * table->size，并且利用已有的內(nèi)存構(gòu)建 FixedPool 隨機內(nèi)存池爆侣，row_memory_size 作為內(nèi)存池內(nèi)部元素的大小

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;
}

• 計算整個共享內(nèi)存表的內(nèi)存大衅汲獭：
  (內(nèi)存表行數(shù)+哈希沖突行數(shù))*(行元素大小+各個列元素大小總和)+哈希沖突內(nèi)存池頭部大小+行元素指針大小*內(nèi)存表行數(shù)
• 比較難以理解的是最后那個 行元素大小*內(nèi)存表行數(shù)，這個其實是在創(chuàng)建 table->rows[table->size] 這個指針數(shù)組兔仰，我們之前說過 table->rows 是個二維數(shù)組尘喝，這個數(shù)組里面存放的是多個 swTableRow* 類型的數(shù)據(jù)，例如 table->rows[0]等斋陪，table->rows[0] 等才是存放各個行元素首地址的地方，如果沒有這個指針數(shù)組置吓，那么每次去取行元素都要計算行元素的首地址无虚，效率沒有這么快。

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;
}

swoole_table 添加新的數(shù)據(jù)

• 共享內(nèi)存表添加新的元素需要調(diào)用三個函數(shù)衍锚，分別是 swTableRow_set 設(shè)置行的 key 值友题、swTableColumn_get 獲取列元素，swTableRow_set_value 函數(shù)根據(jù)列的數(shù)據(jù)類型為 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 函數(shù)

• 我們先來看 swTableRow_set 函數(shù)度宦，從下面的代碼來看，這個函數(shù)主要的作用就是判斷新添加的 key 是否造成了哈希沖突告匠，如果沒有沖突(row->active=0)戈抄，那么直接 table->row_num 自增，設(shè)置 row->key 就可以了后专。
• 如果發(fā)生哈希沖突划鸽，那么就要循環(huán)當(dāng)前行元素的鏈表，直到（1）找到相同的 key 值，說明并不是真的發(fā)生了哈希沖突裸诽，而是用戶要修改已有的行數(shù)據(jù)嫂用，那么就直接跳出函數(shù)，然后更改 row->data 的值丈冬；(2) 沒有找到相同的 key 值嘱函，說明的確遇到了哈希沖突，不同的 key 值對應(yīng)了相同的哈希值埂蕊，此時已經(jīng)循環(huán)到達鏈表的末尾往弓，需要從內(nèi)存池中構(gòu)建出一個 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 這個函數(shù)是怎能計算哈希值的——我們發(fā)現(xiàn)哈希函數(shù)有兩種:
  • swoole_hash_php 是 php 的經(jīng)典哈希函數(shù)粒梦，也就是 time33/DJB 算法
  • swoole_hash_austin 是 MurmurHash 哈希算法亮航，廣泛應(yīng)用在 redis、Memcached 等算法中
• 哈希計算之后匀们，我們發(fā)現(xiàn)哈希值又與 table->mask 進行了邏輯與計算缴淋，目的是得到一個小于等于 table->mask(rows_size - 1) 的數(shù)字，作為行元素的 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];
}

• 我們接下來看行鎖的加鎖函數(shù)：
  • 若是普通的互斥鎖泄朴，那么就直接使用 pthread_mutex_lock 即可重抖，如果不是互斥鎖，程序?qū)崿F(xiàn)了一個自旋鎖
  • 若是自旋鎖祖灰，就調(diào)用 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
}

swTableColumn_get 函數(shù)

從多個列元素組成的 hashMap 中根據(jù) column_key 快速找到對應(yīng)的列元素

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

swTableRow_set_value 函數(shù)

根據(jù)取出的列元素數(shù)據(jù)的類型钟沛，為 row->data 對應(yīng)的位置上賦值，值得注意的是 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;
    }
}

swoole_table 獲取數(shù)據(jù)

• 根據(jù)鍵值獲取行元素需要調(diào)用三個函數(shù)：swTableRow_get 獲取行對象元素，如果只取特定字段三妈，那么會調(diào)用 php_swoole_table_get_field_value畜埋，如果需要去全部字段，那么會調(diào)用 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 函數(shù)

利用 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 函數(shù)

首先通過 swHashMap_find 函數(shù)根據(jù) 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;
        }
    }
}

php_swoole_table_row2array 函數(shù)

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

#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;
            }
        }
    }
}

swoole_table->incr 字段值自增

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);
}

swoole_table->incr 字段值自減

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);
}

swoole_table->del 列表數(shù)據(jù)的刪除

共享內(nèi)存表的數(shù)據(jù)刪除稍微有些復(fù)雜分為以下幾個情況：

• 要刪除的行元素沒有哈希沖突的鏈表
  • 如果鍵值一致辽旋，那么利用 bzero 初始化該行元素硬鞍，減小共享表行數(shù)
  • 如果鍵值不一致，說明并沒有這行數(shù)據(jù)，那么直接解鎖返回
• 要刪除的行元素存在哈希沖突的鏈表固该，那么就要循環(huán)鏈表來找出鍵值一致的行元素
  • 如果遍歷鏈表都沒有找到锅减，那么直接解鎖返回
  • 如果發(fā)現(xiàn)是鏈表的頭元素，那么將鏈表的第二個元素的數(shù)據(jù)賦值給頭元素伐坏，然后從內(nèi)存池中釋放鏈表的第二個元素怔匣，減小共享表行數(shù)
  • 如果是鏈表的中間元素，那么和普通刪除鏈表節(jié)點的方法一致桦沉，減小共享表行數(shù)

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->del 列表數(shù)據(jù)的遍歷

swoole_table 類實現(xiàn)了迭代器每瞒，可以使用 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浓利，不會觸發(fā)原始的對象成員變量的遍歷，而是會調(diào)用 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

關(guān)于為什么要 PCRE 這個正則表達式庫的依賴贷掖，本人非常疑惑，希望有人能夠解疑渴语。

rewind

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 函數(shù)就是將數(shù)據(jù)迭代器返回到開始的位置苹威，對于 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

current 方法很簡單，取出當(dāng)前迭代器的行元素阿蝶，再轉(zhuǎn)化為 php 數(shù)組即可

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;
}

key

取出當(dāng)前迭代器的鍵值：

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

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);
}

valid

驗證當(dāng)前行元素是否為空：

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);
}