## 6.2 線程安全資源管理器 PHP中專門為解決線程安全的問題抽象出了一個線程安全資源管理器(Thread Safe Resource Mananger, TSRM)，實現原理比較簡單：既然共用資源這么困難那么就干脆不共用，各線程不再共享同一份全局變量，而是各復制一份，使用數據時各線程各取自己的副本，互不干擾。 ### 6.2.1 基本實現 TSRM核心思想就是為不同的線程分配獨立的內存空間，如果一個資源會被多線程使用，那么首先需要預先向TSRM注冊資源，然后TSRM為這個資源分配一個唯一的編號，并把這種資源的大小、初始化函數等保存到一個`tsrm_resource_type`結構中，各線程只能通過TSRM分配的那個編號訪問這個資源；然后當線程拿著這個編號獲取資源時TSRM如果發現是第一次請求，則會根據注冊時的資源大小分配一塊內存，然后調用初始化函數進行初始化，并把這塊資源保存下來供這個線程后續使用。 TSRM中通過兩個結構分別保存資源信息以及具體的資源：tsrm_resource_type、tsrm_tls_entry，前者是用來記錄資源大小、初始化函數等信息的，具體分配資源內存時會用到，而后者用來保存各線程所擁有的全部資源： ```c struct _tsrm_tls_entry { void **storage; //資源數組 int count; //擁有的資源數:storage數組大小 THREAD_T thread_id; //所屬線程id tsrm_tls_entry *next; }; typedef struct { size_t size; //資源的大小 ts_allocate_ctor ctor; //初始化函數 ts_allocate_dtor dtor; int done; } tsrm_resource_type; ``` 每個線程擁有一個`tsrm_tls_entry`結構，當前線程的所有資源保存在storage數組中，下標就是各資源的id。 另外所有線程的`tsrm_tls_entry`結構通過一個數組保存：tsrm_tls_table，這是個全局變量，所以操作這個變量時需要加鎖。這個值在TSRM初始化時按照預設置的線程數分配，每個線程的tsrm_tls_entry結構在這個數組中的位置是根據線程id與預設置的線程數(tsrm_tls_table_size)取模得到的，也就是說有可能多個線程保存在tsrm_tls_table同一位置，所以tsrm_tls_entry是個鏈表，查找資源時首先根據:`線程id % tsrm_tls_table_size`得到一個tsrm_tls_entry，然后開始遍歷鏈表比較thread_id確定是否是當前線程的。 #### 6.2.1.1 初始化 在使用TSRM之前需要主動開啟，一般這個步驟在sapi啟動時執行，主要工作就是分配tsrm_tls_table、resource_types_table內存以及創建線程互斥鎖，下面具體看下TSRM初始化的過程(以pthread為例)： ```c TSRM_API int tsrm_startup(int expected_threads, int expected_resources, int debug_level, char *debug_filename) { pthread_key_create( &tls_key, 0 ); //分配tsrm_tls_table tsrm_tls_table_size = expected_threads; tsrm_tls_table = (tsrm_tls_entry **) calloc(tsrm_tls_table_size, sizeof(tsrm_tls_entry *)); ... //初始化資源的遞增id，注冊資源時就是用的這個值 id_count=0; //分配資源類型數組：resource_types_table resource_types_table_size = expected_resources; resource_types_table = (tsrm_resource_type *) calloc(resource_types_table_size, sizeof(tsrm_resource_type)); ... //創建鎖 tsmm_mutex = tsrm_mutex_alloc(); } ``` #### 6.2.1.2 資源注冊 初始化完成各模塊就可以各自進行資源注冊了，注冊后TSRM會給注冊的資源分配唯一id，之后對此資源的操作只能依據此id，接下來我們以EG為例具體看下其注冊過程。 ```c #ifdef ZTS ZEND_API int executor_globals_id; #endif int zend_startup(zend_utility_functions *utility_functions, char **extensions) { ... #ifdef ZTS ts_allocate_id(&executor_globals_id, sizeof(zend_executor_globals), (ts_allocate_ctor) executor_globals_ctor, (ts_allocate_dtor) executor_globals_dtor); executor_globals = ts_resource(executor_globals_id); ... #endif } ``` 資源注冊調用`ts_allocate_id()`完成，此函數有4個參數有，第一個就是定義的資源id指針，注冊之后會把分配的id寫到這里，第二個是資源類型的大小，EG資源的結構是`zend_executor_globals`，所以這個值就是sizeof(zend_executor_globals)，后面兩個分別是資源的初始化函數以及銷毀函數，因為TSRM并不關心資源的具體類型，分配資源時它只按照size大小分配內存，然后回調各資源自己定義的ctor進行初始化。 ```c TSRM_API ts_rsrc_id ts_allocate_id(ts_rsrc_id *rsrc_id, size_t size, ts_allocate_ctor ctor, ts_allocate_dtor dtor) { //加鎖，保證各線程串行調用此函數 tsrm_mutex_lock(tsmm_mutex); //分配id，即id_count當前值，然后把id_count加1 *rsrc_id = TSRM_SHUFFLE_RSRC_ID(id_count++); //檢查resource_types_table數組當前大小是否已滿 if (resource_types_table_size < id_count) { //需要對resource_types_table擴容 resource_types_table = (tsrm_resource_type *) realloc(resource_types_table, sizeof(tsrm_resource_type)*id_count); ... //把數組大小修改新的大小 resource_types_table_size = id_count; } //將新注冊的資源插入resource_types_table數組，下標就是分配的資源id resource_types_table[TSRM_UNSHUFFLE_RSRC_ID(*rsrc_id)].size = size; resource_types_table[TSRM_UNSHUFFLE_RSRC_ID(*rsrc_id)].ctor = ctor; resource_types_table[TSRM_UNSHUFFLE_RSRC_ID(*rsrc_id)].dtor = dtor; resource_types_table[TSRM_UNSHUFFLE_RSRC_ID(*rsrc_id)].done = 0; ... } ``` 到這里并沒有結束，所有的資源并不是統一時機注冊的，所以注冊一個新資源時可能有線程已經分配先前注冊的資源了，因此需要對各線程的storage數組進行擴容，否則storage將沒有空間容納新的資源。擴容的過程比較簡單：遍歷各線程的tsrm_tls_entry，檢查storage當時是否有空閑空間，有的話跳過，沒有的話則擴展。 ```c for (i=0; i<tsrm_tls_table_size; i++) { tsrm_tls_entry *p = tsrm_tls_table[i]; //tsrm_tls_table[i]可能保存著多個線程，需要遍歷鏈表 while (p) { if (p->count < id_count) { int j; //將storage擴容 p->storage = (void *) realloc(p->storage, sizeof(void *)*id_count); //分配并初始化新注冊的資源，實際這里只會執行一次，不清楚為什么用循環 //另外這里不分配內存也可以，可以放到使用時再去分配 for (j=p->count; j<id_count; j++) { p->storage[j] = (void *) malloc(resource_types_table[j].size); if (resource_types_table[j].ctor) { //回調初始化函數進行初始化 resource_types_table[j].ctor(p->storage[j]); } } p->count = id_count; } p = p->next; } } ``` 最后將鎖釋放，完成注冊。 #### 6.2.1.3 獲取資源 資源的id在注冊后需要保存下來，根據id可以通過`ts_resource()`獲取到對應資源的值，比如EG，這里暫不考慮EG宏展開的結果，只分析最底層的根據資源id獲取資源的操作。 ```c zend_executor_globals *executor_globals; executor_globals = ts_resource(executor_globals_id); ``` 這樣獲取的`executor_globals`值就是各線程分離的了，對它的操作將不會再影響其它線程。根據資源id獲取當前線程資源的過程：首先是根據線程id哈希得到當前線程的tsrm_tls_entry在tsrm_tls_table哪個槽中，然后開始遍歷比較id，直到找到當前線程的tsrm_tls_entry，這個查找過程是需要加鎖的，最后根據資源id從storage中對應位置取出資源的地址，這個時候如果發現當前線程還沒有創建此資源則會從resource_types_table根據資源id取出資源注冊時的大小、初始化函數，然后分配內存、調用初始化函數進行初始化并插入所屬線程的storage中。 ```c TSRM_API void *ts_resource_ex(ts_rsrc_id id, THREAD_T *th_id) { THREAD_T thread_id; int hash_value; tsrm_tls_entry *thread_resources; //step 1：獲取線程id if (!th_id) { //獲取當前線程通過specific data保存的tsrm_tls_entry，暫時忽略 thread_resources = tsrm_tls_get(); if(thread_resources){ //找到線程的tsrm_tls_entry了 TSRM_SAFE_RETURN_RSRC(thread_resources->storage, id, thread_resources->count); //直接返回 } //pthread_self()，當前線程id thread_id = tsrm_thread_id(); }else{ thread_id = *th_id; } //step 2:查找線程tsrm_tls_entry tsrm_mutex_lock(tsmm_mutex); //加鎖 //實際就是thread_id % tsrm_tls_table_size hash_value = THREAD_HASH_OF(thread_id, tsrm_tls_table_size); //鏈表頭部 thread_resources = tsrm_tls_table[hash_value]; if (!thread_resources) { //當前線程第一次使用資源還未分配：先分配tsrm_tls_entry allocate_new_resource(&tsrm_tls_table[hash_value], thread_id); //分配完再次調用，這時候將走到下面的分支 return ts_resource_ex(id, &thread_id); }else{ //遍歷查找當前線程的tsrm_tls_entry do { //找到了 if (thread_resources->thread_id == thread_id) { break; } if (thread_resources->next) { thread_resources = thread_resources->next; } else { //遍歷到最后也沒找到，與上面的一致，先分配再查找 allocate_new_resource(&thread_resources->next, thread_id); return ts_resource_ex(id, &thread_id); } } while (thread_resources); } //解鎖 tsrm_mutex_unlock(tsmm_mutex); //step 3：返回資源 TSRM_SAFE_RETURN_RSRC(thread_resources->storage, id, thread_resources->count); } ``` 首先是獲取線程id，如果沒有傳的話就是當前線程，然后在tsrm_tls_table中查找當前線程的tsrm_tls_entry，不存在則表示當前線程第一次使用資源，則需要調用`allocate_new_resource()`為當前線程分配tsrm_tls_entry，并插入tsrm_tls_table，這個過程還會為當前已注冊的所有資源分配內存： ```c static void allocate_new_resource(tsrm_tls_entry **thread_resources_ptr, THREAD_T thread_id) { (*thread_resources_ptr) = (tsrm_tls_entry *) malloc(sizeof(tsrm_tls_entry)); (*thread_resources_ptr)->storage = NULL; //根據已注冊資源數分配storage數組大小，注意這里并不是分配為各資源分配空間 if (id_count > 0) { (*thread_resources_ptr)->storage = (void **) malloc(sizeof(void *)*id_count); } (*thread_resources_ptr)->count = id_count; (*thread_resources_ptr)->thread_id = thread_id; //將當前線程的tsrm_tls_entry保存到線程本地存儲(Thread Local Storage, TLS) tsrm_tls_set(*thread_resources_ptr); //為全部資源分配空間 for (i=0; i<id_count; i++) { ... (*thread_resources_ptr)->storage[i] = (void *) malloc(resource_types_table[i].size); ... } ... } ``` 這里還用到了一個多線程中經常用到的一個東西：線程本地存儲(Thread Local Storage, TLS)，在創建完當前線程的tsrm_tls_entry后會把這個值保存到當前線程的TLS中(即：tsrm_tls_set(*thread_resources_ptr)操作)，這樣在`ts_resource()`中就可以通過`tsrm_tls_get()`直接取到了，節省加鎖檢索的時間。 > __線程本地存儲(Thread Local Storage, TLS):__ 我們知道在一個進程中，所有線程是共享同一個地址空間的。所以，如果一個變量是全局的或者是靜態的，那么所有線程訪問的是同一份，如果某一個線程對其進行了修改，也就會影響到其他所有的線程。不過我們可能并不希望這樣，所以更多的推薦用基于堆棧的自動變量或函數參數來訪問數據，因為基于堆棧的變量總是和特定的線程相聯系的。TLS在各平臺下實現方式不同，主要分為兩類：靜態TLS、動態TLS，pthread中pthread_setspecific()、pthread_getspecific()的實現就可以認為是動態TLS的實現。 比如tsrm_tls_table_size初始化時設置為了2，當前有2個thread：thread 1、thread 2，假如注冊了CG、EG兩個資源，則存儲結構如下圖：  ### 6.2.2 Native-TLS 上一節我們介紹了資源的注冊以及根據資源id獲取資源的方法，那么PHP內核每次使用對應的資源時難道都需要調用`ts_resource()`嗎？如果是這樣的話那么多次在使用EG時實際都會調一次這個方法，相當于我們需要調用一個函數來獲取一個變量，這在性能上是不可接受的，那么有什么辦法解決呢？ `ts_resource()`最核心的操作就是根據線程id獲取各線程對應的storage數組，這也是最耗時的部分，至于接下來根據資源id從storage數組讀取資源就是普通的內存讀取了，這并不影響性能，所以解決上面那個問題的關鍵就在于 __盡可能的減少線程storage的檢索__ 。這一節我們來分析下PHP是如果解決這個問題的，在介紹PHP7實現方式之前我們先看下PHP5.x的處理方式。 PHP5的解決方式非常簡單，我們還是以EG為例，EG在內核中隨處可見，不是要減少對各線程storage的檢索次數嗎，那么我就只要檢索過一次就把已獲取的storage指針傳給接下來調用的函數用，其它函數再一級級往下傳，這樣一來各函數如果發現storage通過參數傳進來了就直接用，無需再檢索了，也就是通過層層傳遞的方式減少解決這個問題的。這樣以來豈不是每個函數都得帶這么一個參數？調用別的函數也得把這個值帶上？是的。即使這個函數自己不用它也得需要這個值，因為有可能調用別的函數的時候其它函數會用。 如果你對PHP5有所了解的話一定經常看到這兩個宏：TSRMLS_DC、TSRMLS_CC，這兩個宏就是用來傳遞storage指針的，TSRMLS_DC用在定義函數的參數中，實際上它就是一個普通的參數定義，TSRMLS_CC用在調用函數時，它就是一個普通的變量值，我們看下它的展開結果： ```c #define TSRMLS_DC , void ***tsrm_ls #define TSRMLS_CC , tsrm_ls ``` 它的用法是第一個檢索到storage的函數把它的指針傳遞給了下面的函數，參數是tsrm_ls，后面的函數直接根據接收的參數使用獲取再傳給其它函數，當然也可以不傳，那樣的話就得重新調用ts_resource()獲取了。現在我們再看下EG宏展開的結果： ```c # define EG(v) TSRMG(executor_globals_id, zend_executor_globals *, v) #define TSRMG(id, type, element) (((type) (*((void ***) tsrm_ls))[TSRM_UNSHUFFLE_RSRC_ID(id)])->element) ``` 比如：`EG(function_table) => (((zend_executor_globals *) (*((void ***) tsrm_ls))[executor_globals_id-1])->function_table)`，這樣我們在傳了tsrm_ls的函數中就可能讀取內存使用了。 PHP5的這種處理方式簡單但是很不優雅，不管你用不用TSRM都不得不在函數中加上那兩個宏，而且很容易遺漏。后來Anatol Belski在PHP的rfc提交了一種新的處理方式：[https://wiki.php.net/rfc/native-tls](https://wiki.php.net/rfc/native-tls)，新的處理方式最終在PHP7版本得以實現，通過靜態TLS將各線程的storage保存在全局變量中，各函數中使用時直接讀取即可。 linux下這種全局變量通過加上`__thread`定義，這樣各線程更新這個變量就不會沖突了，實際這是gcc提供的，詳細的內容這里不再展開，有興趣的可以再查下詳細的資料。舉個例子： ```c #include <stdio.h> #include <stdlib.h> #include <pthread.h> #include <unistd.h> __thread int num = 0; void* worker(void* arg){ while(1){ printf("thread:%d\n", num); sleep(1); } } int main(void) { pthread_t tid; int ret; if ((ret = pthread_create(&tid, NULL, worker, NULL)) != 0){ return 1; } while(1){ num = 4; printf("main:%d\n", num); sleep(1); } return 0; } ``` 這個例子有兩個線程，其中主線程修改了全局變量num，但是并沒有影響另外一個線程。 PHP7中用于緩存各線程storage的全局變量定義在`Zend/zend.c`: ```c #ifdef ZTS //這些都是全局變量 ZEND_API int compiler_globals_id; ZEND_API int executor_globals_id; static HashTable *global_function_table = NULL; static HashTable *global_class_table = NULL; static HashTable *global_constants_table = NULL; static HashTable *global_auto_globals_table = NULL; static HashTable *global_persistent_list = NULL; ZEND_TSRMLS_CACHE_DEFINE() //=>TSRM_TLS void *TSRMLS_CACHE = NULL; 展開后： __thread void *_tsrm_ls_cache = NULL; _tsrm_ls_cache就是各線程storage的地址 #endif ``` 比如EG： ```c # define EG(v) ZEND_TSRMG(executor_globals_id, zend_executor_globals *, v) #define ZEND_TSRMG TSRMG_STATIC #define TSRMG_STATIC(id, type, element) (TSRMG_BULK_STATIC(id, type)->element) #define TSRMG_BULK_STATIC(id, type) ((type) (*((void ***) TSRMLS_CACHE))[TSRM_UNSHUFFLE_RSRC_ID(id)]) ``` EG(xxx)最終展開：((zend_executor_globals *) (*((void ***) _tsrm_ls_cache))[executor_globals_id-1]->xxx)。