資訊專欄INFORMATION COLUMN
摘要：如果互斥鎖的持有者死亡了，或者持有這樣的互斥鎖的進(jìn)程了互斥鎖所在的共享內(nèi)存或者持有這樣的互斥鎖的進(jìn)程執(zhí)行了調(diào)用，則會解除鎖定該互斥鎖?；コ怄i的下一個持有者將獲取該互斥鎖并返回錯誤。
前言
swoole_table 一個基于共享內(nèi)存和鎖實(shí)現(xiàn)的超高性能，并發(fā)數(shù)據(jù)結(jié)構(gòu)。用于解決多進(jìn)程/多線程數(shù)據(jù)共享和同步加鎖問題。
swoole_table 的數(shù)據(jù)結(jié)構(gòu)swoole_table 實(shí)際上就是一個開鏈法實(shí)現(xiàn)的哈希表，memory 是一個由哈希鍵與具體數(shù)據(jù)組成的數(shù)組，如果哈希沖突(不同的鍵值對應(yīng)同一個哈希)，那么就會從 pool 中分配出一個元素作為數(shù)組元素的鏈表尾
size 是創(chuàng)建共享內(nèi)存表時設(shè)置的最大行數(shù)；conflict_proportion 是哈希沖突的最大比例，超過這個比例，共享內(nèi)存表就不允許再添加新的行元素；iterator 是內(nèi)存表的迭代器，可以利用它進(jìn)行內(nèi)存表數(shù)據(jù)的瀏覽；columns 是內(nèi)存表的列元素集合，由于內(nèi)存表的列元素也是一個 key-value 格式，因此也是一個哈希表 swHashMap 類型；mask 存放的是(最大行數(shù)-1)，專門進(jìn)行哈希值與數(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ù)并不是嚴(yán)格按照用戶定義的數(shù)據(jù)而來，如果 size 不是為 2 的 N 次方，如 1024、8192,65536 等，底層會自動調(diào)整為接近的一個數(shù)字，如果小于 1024 則默認(rèn)成 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ùn)行優(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ùn)行，因而就無法釋放鎖，所以高優(yōu)先級線程只能繼續(xù)阻塞在鎖上）。使用優(yōu)先級繼承可以短時間的提高低優(yōu)先級線程的優(yōu)先級，從而使它可以盡快得到調(diào)度，然后釋放鎖。低優(yōu)先級線程在釋放鎖后就會恢復(fù)自己的優(yōu)先級。)
PTHREAD_MUTEX_ROBUST_NP： 如果互斥鎖的持有者“死亡”了，或者持有這樣的互斥鎖的進(jìn)程 unmap 了互斥鎖所在的共享內(nèi)存或者持有這樣的互斥鎖的進(jìn)程執(zhí)行了 exec 調(diào)用，則會解除鎖定該互斥鎖?；コ怄i的下一個持有者將獲取該互斥鎖,并返回錯誤 EOWNWERDEAD。
table->rows 創(chuàng)建成功之后，就要對哈希沖突的行元素分配地址空間?？梢钥吹剑_突的行元素首地址為 memory += row_memory_size * table->size，并且利用已有的內(nèi)存構(gòu)建 FixedPool 隨機(jī)內(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)存大?。?/p>
(內(nèi)存表行數(shù)+哈希沖突行數(shù))*(行元素大小+各個列元素大小總和)+哈希沖突內(nèi)存池頭部大小+行元素指針大小*內(nèi)存表行數(shù)
比較難以理解的是最后那個 行元素大小*內(nèi)存表行數(shù)，這個其實(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)到達(dá)鏈表的末尾，需要從內(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 進(jìn)行了邏輯與計算，目的是得到一個小于等于 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 實(shí)際上指的是 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é)點(diǎn)的方法一致，減小共享表行數(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 類實(shí)現(xiàn)了迭代器，可以使用 foreach 進(jìn)行遍歷。
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 進(jìn)行初始化的時候，為這個類繼承了 spl_iterator 這個接口，我們知道，對繼承了這個接口的類進(jìn)行 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 這個正則表達(dá)式庫的依賴，本人非常疑惑，希望有人能夠解疑。
rewindstatic 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 就是將迭代器向前進(jìn)行一步，其中 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 就是迭代器向前進(jìn)一步：
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);
}
文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請注明本文地址：http://systransis.cn/yun/29223.html
摘要：受限于的實(shí)現(xiàn)，程序無法使用多線程進(jìn)行編程開發(fā)。比如實(shí)現(xiàn)一個聊天室程序，用戶在進(jìn)程中處理，用戶在進(jìn)程中處理，和如果在同一個，這個在多線程環(huán)境中直接用表示，和加到對應(yīng)的中即可。想要解決這個問題，必須實(shí)現(xiàn)一個基于共享內(nèi)存的數(shù)據(jù)結(jié)構(gòu)。 Swoole項目從 2012 年推出到現(xiàn)在已經(jīng)有 5 年的歷史，現(xiàn)在越來越多的互聯(lián)網(wǎng)企業(yè)使用Swoole來開發(fā)各類后臺應(yīng)用。受限于 PHP 的ZendVM實(shí)現(xiàn)，...
摘要：從入門到放棄三一進(jìn)程子進(jìn)程創(chuàng)建成功后要執(zhí)行的函數(shù)重定向子進(jìn)程的標(biāo)準(zhǔn)輸入和輸出。默認(rèn)為阻塞讀取。是否創(chuàng)建管道，啟用后，此選項將忽略用戶參數(shù)，強(qiáng)制為。 swoole——從入門到放棄（三） 一、進(jìn)程 swoole_process SwooleProcess swoole_process::__construct(callable $function, $redirect_stdin...
摘要：從入門到放棄三一進(jìn)程子進(jìn)程創(chuàng)建成功后要執(zhí)行的函數(shù)重定向子進(jìn)程的標(biāo)準(zhǔn)輸入和輸出。默認(rèn)為阻塞讀取。是否創(chuàng)建管道，啟用后，此選項將忽略用戶參數(shù)，強(qiáng)制為。 swoole——從入門到放棄（三） 一、進(jìn)程 swoole_process SwooleProcess swoole_process::__construct(callable $function, $redirect_stdin...
摘要：前言中為了更好的進(jìn)行內(nèi)存管理，減少頻繁分配釋放內(nèi)存空間造成的損耗和內(nèi)存碎片，程序設(shè)計并實(shí)現(xiàn)了三種不同功能的內(nèi)存池，和。比較特殊的是單鏈表內(nèi)存池的內(nèi)存只能增加不會減少。 前言 Swoole 中為了更好的進(jìn)行內(nèi)存管理，減少頻繁分配釋放內(nèi)存空間造成的損耗和內(nèi)存碎片，程序設(shè)計并實(shí)現(xiàn)了三種不同功能的內(nèi)存池：FixedPool，RingBuffer 和 MemoryGlobal。 其中 Memor...
摘要：前言我們知道，由于沒有多線程模型，所以更多的使用多進(jìn)程模型，因此代碼相對來說更加簡潔，減少了各種線程鎖的阻塞與同步，但是也帶來了新的問題數(shù)據(jù)同步。相比多線程之前可以直接共享進(jìn)程的內(nèi)存，進(jìn)程之間數(shù)據(jù)的相互同步依賴于共享內(nèi)存。 前言 我們知道，由于 PHP 沒有多線程模型，所以 swoole 更多的使用多進(jìn)程模型，因此代碼相對來說更加簡潔，減少了各種線程鎖的阻塞與同步，但是也帶來了新的問題...
閱讀 1763·2023-04-25 16:28
閱讀 697·2021-11-23 09:51
閱讀 1478·2019-08-30 15:54
閱讀 1164·2019-08-30 15:53
閱讀 2838·2019-08-30 15:53
閱讀 3431·2019-08-30 15:43
閱讀 3269·2019-08-30 11:18
閱讀 3292·2019-08-26 10:25
