Memcached源码分析之thread.c

  1. /*
  2. * 文件开头先啰嗦几句：
  3. *
  4. * thread.c文件代表的是线程模块。但是你会看到这个模块里面有很多其它方法，
  5. 例如关于item的各种操作函数，item_alloc，item_remove，item_link等等。
  6. 我们有个items模块，这些不都是items模块要做的事情吗？为什么thread模块也有？
  7. 你仔细看会发现，thread里面的这种函数，例如item_remove，items模块里面
  8. 都会有一个对应的do_item_remove函数，而thread中的item_remove仅仅是调用
  9. items模块中的do_item_remove，唯一多出来的就是thread在do_item_remove前后
  10. 加了加锁和解锁的操作。
  11. 其实这是很好的一种设计。
  12. 1）因为像"删除item"这样的一个逻辑都是由某个线程，而且这里是工作线程执行，
  13. 所以这是一个线程层面的事情。就是说是“某个工作线程去删除item”这样一件事。
  14. 2）更重要的是原子性及一致性问题，某个item数据，很有可能同时多个线程在修改，
  15. 那么需要加锁，那么锁最应该加在哪个地方？既然问题是线程引起的，那么负责
  16. 解决的无疑是线程模块。
  17. 3）所以这里像这种函数，thread此时相当于是items的外壳，起调控作用，在线程层面
  18. 开放给外部调用，同时在内部加锁。而items模块里面定义的do_xxx函数都不需要多
  19. 加考虑，无条件执行对item进行修改，而由外部被调用方来控制。相信很多需要加锁
  20. 的项目都会面临这样的问题：锁应该加在哪一层？可以参考memcached这样的设计。
  21. *
  22. */
  23. #include "memcached.h"
  24. #include <assert.h>
  25. #include <stdio.h>
  26. #include <errno.h>
  27. #include <stdlib.h>
  28. #include <errno.h>
  29. #include <string.h>
  30. #include <pthread.h>
  31. #ifdef __sun
  32. #include <atomic.h>
  33. #endif
  34. #define ITEMS_PER_ALLOC 64
  35. /**
  36. 下面这个CQ_ITEM结构体：
  37. 可以这么理解，主线程accept连接后，把client fd
  38. 分发到worker线程的同时会顺带一些与此client连接相关的信息，
  39. 而CQ_ITEM是包装了这些信息的一个对象，有点"参数对象"的概念。
  40. 记住这货是主线程那边丢过来的。
  41. CQ_ITEM中的CQ虽然是connection queue的缩写，
  42. 它与memcached.h中定义的conn结构体是完全不一样的概念，
  43. 但worker线程会利用这个CQ_ITEM对象去初始化conn对象
  44. */
  45. typedef struct conn_queue_item CQ_ITEM;
  46. struct conn_queue_item {
  47. int sfd;
  48. enum conn_states init_state;
  49. int event_flags;
  50. int read_buffer_size;
  51. enum network_transport transport;
  52. CQ_ITEM *next;
  53. };
  54. /*
  55. 上面的CQ_ITEM的队列对象，每个worker线程对象都保存着这样一个队列，处理
  56. 主线程那边分发过来的连接请求时用到。
  57. */
  58. typedef struct conn_queue CQ;
  59. struct conn_queue {
  60. CQ_ITEM *head;
  61. CQ_ITEM *tail;
  62. pthread_mutex_t lock;
  63. };
  64. //下面是各种锁
  65. /**
  66. 个人认为这个锁用于锁住全局数量不变的对象，例如slabclass，LRU链表等等
  67. 区别于item锁，由于item对象是动态增长的，数量非常多，
  68. item锁是用hash的方式分配一张大大的item锁表来控制锁的粒度
  69. */
  70. pthread_mutex_t cache_lock;
  71. pthread_mutex_t conn_lock = PTHREAD_MUTEX_INITIALIZER; //连接锁
  72. #if !defined(HAVE_GCC_ATOMICS) && !defined(__sun)
  73. pthread_mutex_t atomics_mutex = PTHREAD_MUTEX_INITIALIZER;
  74. #endif
  75. static pthread_mutex_t stats_lock; //统计锁
  76. static CQ_ITEM *cqi_freelist;
  77. static pthread_mutex_t cqi_freelist_lock;
  78. static pthread_mutex_t *item_locks; //item锁
  79. static uint32_t item_lock_count; //item锁总数
  80. static unsigned int item_lock_hashpower; //item锁的hash表 指数，锁总数为2的item_lock_hashpower个，见下面的hashsize
  81. #define hashsize(n) ((unsigned long int)1<<(n))
  82. #define hashmask(n) (hashsize(n)-1)
  83. static pthread_mutex_t item_global_lock;
  84. static pthread_key_t item_lock_type_key;
  85. static LIBEVENT_DISPATCHER_THREAD dispatcher_thread;
  86. static LIBEVENT_THREAD *threads;
  87. static int init_count = 0; //有多少个worker线程已经被初始化
  88. static pthread_mutex_t init_lock; //初始化锁
  89. static pthread_cond_t init_cond; //初始化条件变量
  90. static void thread_libevent_process(int fd, short which, void *arg);
  91. //引用计数加1
  92. unsigned short refcount_incr(unsigned short *refcount) {
  93. #ifdef HAVE_GCC_ATOMICS
  94. return __sync_add_and_fetch(refcount, 1);
  95. #elif defined(__sun)
  96. return atomic_inc_ushort_nv(refcount);
  97. #else
  98. unsigned short res;
  99. mutex_lock(&atomics_mutex);
  100. (*refcount)++;
  101. res = *refcount;
  102. mutex_unlock(&atomics_mutex);
  103. return res;
  104. #endif
  105. }
  106. //引用计数减1
  107. unsigned short refcount_decr(unsigned short *refcount) {
  108. #ifdef HAVE_GCC_ATOMICS
  109. return __sync_sub_and_fetch(refcount, 1);
  110. #elif defined(__sun)
  111. return atomic_dec_ushort_nv(refcount);
  112. #else
  113. unsigned short res;
  114. mutex_lock(&atomics_mutex);
  115. (*refcount)--;
  116. res = *refcount;
  117. mutex_unlock(&atomics_mutex);
  118. return res;
  119. #endif
  120. }
  121. void item_lock_global(void) {
  122. mutex_lock(&item_global_lock);
  123. }
  124. void item_unlock_global(void) {
  125. mutex_unlock(&item_global_lock);
  126. }
  127. void item_lock(uint32_t hv) {
  128. uint8_t *lock_type = pthread_getspecific(item_lock_type_key);
  129. if (likely(*lock_type == ITEM_LOCK_GRANULAR)) {
  130. mutex_lock(&item_locks[hv & hashmask(item_lock_hashpower)]);
  131. } else {
  132. mutex_lock(&item_global_lock);
  133. }
  134. }
  135. void *item_trylock(uint32_t hv) {
  136. pthread_mutex_t *lock = &item_locks[hv & hashmask(item_lock_hashpower)];
  137. if (pthread_mutex_trylock(lock) == 0) {
  138. return lock;
  139. }
  140. return NULL;
  141. }
  142. void item_trylock_unlock(void *lock) {
  143. mutex_unlock((pthread_mutex_t *) lock);
  144. }
  145. void item_unlock(uint32_t hv) {
  146. uint8_t *lock_type = pthread_getspecific(item_lock_type_key);
  147. if (likely(*lock_type == ITEM_LOCK_GRANULAR)) {
  148. mutex_unlock(&item_locks[hv & hashmask(item_lock_hashpower)]);
  149. } else {
  150. mutex_unlock(&item_global_lock);
  151. }
  152. }
  153. static void wait_for_thread_registration(int nthreads) {
  154. while (init_count < nthreads) {
  155. pthread_cond_wait(&init_cond, &init_lock); //主线程利用条件变量等待所有worker线程启动完毕
  156. }
  157. }
  158. //worker线程注册函数，主要是统计worker线程完成初始化个数。
  159. static void register_thread_initialized(void) {
  160. pthread_mutex_lock(&init_lock);
  161. init_count++;
  162. pthread_cond_signal(&init_cond);
  163. pthread_mutex_unlock(&init_lock);
  164. }
  165. //item锁的粒度有几种，这里是切换类型
  166. void switch_item_lock_type(enum item_lock_types type) {
  167. char buf[1];
  168. int i;
  169. switch (type) {
  170. case ITEM_LOCK_GRANULAR:
  171. buf[0] = ‘l‘;
  172. break;
  173. case ITEM_LOCK_GLOBAL:
  174. buf[0] = ‘g‘;
  175. break;
  176. default:
  177. fprintf(stderr, "Unknown lock type: %d\n", type);
  178. assert(1 == 0);
  179. break;
  180. }
  181. pthread_mutex_lock(&init_lock);
  182. init_count = 0;
  183. for (i = 0; i < settings.num_threads; i++) {
  184. if (write(threads[i].notify_send_fd, buf, 1) != 1) {
  185. perror("Failed writing to notify pipe");
  186. /* TODO: This is a fatal problem. Can it ever happen temporarily? */
  187. }
  188. }
  189. wait_for_thread_registration(settings.num_threads);
  190. pthread_mutex_unlock(&init_lock);
  191. }
  192. /*
  193. * Initializes a connection queue.
  194. 初始化一个CQ对象，CQ结构体和CQ_ITEM结构体的作用见它们定义处。
  195. */
  196. static void cq_init(CQ *cq) {
  197. pthread_mutex_init(&cq->lock, NULL);
  198. cq->head = NULL;
  199. cq->tail = NULL;
  200. }
  201. /**
  202. 从worker线程的CQ队列里面pop出一个CQ_ITEM对象
  203. */
  204. static CQ_ITEM *cq_pop(CQ *cq) {
  205. CQ_ITEM *item;
  206. pthread_mutex_lock(&cq->lock);
  207. item = cq->head;
  208. if (NULL != item) {
  209. cq->head = item->next;
  210. if (NULL == cq->head)
  211. cq->tail = NULL;
  212. }
  213. pthread_mutex_unlock(&cq->lock);
  214. return item;
  215. }
  216. /**
  217. push一个CQ_ITEM对象到worker线程的CQ队列中
  218. */
  219. static void cq_push(CQ *cq, CQ_ITEM *item) {
  220. item->next = NULL;
  221. pthread_mutex_lock(&cq->lock);
  222. if (NULL == cq->tail)
  223. cq->head = item;
  224. else
  225. cq->tail->next = item;
  226. cq->tail = item;
  227. pthread_mutex_unlock(&cq->lock);
  228. }
  229. /*
  230. * Returns a fresh connection queue item.
  231. 分配一个CQ_ITEM对象
  232. */
  233. static CQ_ITEM *cqi_new(void) {
  234. CQ_ITEM *item = NULL;
  235. pthread_mutex_lock(&cqi_freelist_lock);
  236. if (cqi_freelist) {
  237. item = cqi_freelist;
  238. cqi_freelist = item->next;
  239. }
  240. pthread_mutex_unlock(&cqi_freelist_lock);
  241. if (NULL == item) {
  242. int i;
  243. /* Allocate a bunch of items at once to reduce fragmentation */
  244. item = malloc(sizeof(CQ_ITEM) * ITEMS_PER_ALLOC);
  245. if (NULL == item) {
  246. STATS_LOCK();
  247. stats.malloc_fails++;
  248. STATS_UNLOCK();
  249. return NULL;
  250. }
  251. for (i = 2; i < ITEMS_PER_ALLOC; i++)
  252. item[i - 1].next = &item[i];
  253. pthread_mutex_lock(&cqi_freelist_lock);
  254. item[ITEMS_PER_ALLOC - 1].next = cqi_freelist;
  255. cqi_freelist = &item[1];
  256. pthread_mutex_unlock(&cqi_freelist_lock);
  257. }
  258. return item;
  259. }
  260. /*
  261. * Frees a connection queue item (adds it to the freelist.)
  262. */
  263. static void cqi_free(CQ_ITEM *item) {
  264. pthread_mutex_lock(&cqi_freelist_lock);
  265. item->next = cqi_freelist;
  266. cqi_freelist = item;
  267. pthread_mutex_unlock(&cqi_freelist_lock);
  268. }
  269. /*
  270. 创建并启动worker线程，在thread_init主线程初始化时调用
  271. */
  272. static void create_worker(void *(*func)(void *), void *arg) {
  273. pthread_t thread;
  274. pthread_attr_t attr;
  275. int ret;
  276. pthread_attr_init(&attr);
  277. if ((ret = pthread_create(&thread, &attr, func, arg)) != 0) {
  278. fprintf(stderr, "Can‘t create thread: %s\n",
  279. strerror(ret));
  280. exit(1);
  281. }
  282. }
  283. void accept_new_conns(const bool do_accept) {
  284. pthread_mutex_lock(&conn_lock);
  285. do_accept_new_conns(do_accept);
  286. pthread_mutex_unlock(&conn_lock);
  287. }
  288. /****************************** LIBEVENT THREADS *****************************/
  289. /*
  290. * 装备worker线程，worker线程的event_base在此设置
  291. */
  292. static void setup_thread(LIBEVENT_THREAD *me) {
  293. me->base = event_init(); //为每个worker线程分配自己的event_base
  294. if (! me->base) {
  295. fprintf(stderr, "Can‘t allocate event base\n");
  296. exit(1);
  297. }
  298. /* Listen for notifications from other threads */
  299. event_set(&me->notify_event, me->notify_receive_fd,
  300. EV_READ | EV_PERSIST, thread_libevent_process, me); //监听管道接收fd，这里即监听
  301. //来自主线程的消息，事件处理函数为thread_libevent_process
  302. event_base_set(me->base, &me->notify_event);
  303. if (event_add(&me->notify_event, 0) == -1) {
  304. fprintf(stderr, "Can‘t monitor libevent notify pipe\n");
  305. exit(1);
  306. }
  307. me->new_conn_queue = malloc(sizeof(struct conn_queue)); //CQ_ITEM队列
  308. if (me->new_conn_queue == NULL) {
  309. perror("Failed to allocate memory for connection queue");
  310. exit(EXIT_FAILURE);
  311. }
  312. cq_init(me->new_conn_queue); //初始化CQ_ITEM对象队列
  313. if (pthread_mutex_init(&me->stats.mutex, NULL) != 0) {
  314. perror("Failed to initialize mutex");
  315. exit(EXIT_FAILURE);
  316. }
  317. me->suffix_cache = cache_create("suffix", SUFFIX_SIZE, sizeof(char*),
  318. NULL, NULL);
  319. if (me->suffix_cache == NULL) {
  320. fprintf(stderr, "Failed to create suffix cache\n");
  321. exit(EXIT_FAILURE);
  322. }
  323. }
  324. /*
  325. * 这里主要是让worker线程进入event_base_loop
  326. */
  327. static void *worker_libevent(void *arg) {
  328. LIBEVENT_THREAD *me = arg;
  329. /* Any per-thread setup can happen here; thread_init() will block until
  330. * all threads have finished initializing.
  331. */
  332. /* set an indexable thread-specific memory item for the lock type.
  333. * this could be unnecessary if we pass the conn *c struct through
  334. * all item_lock calls...
  335. */
  336. me->item_lock_type = ITEM_LOCK_GRANULAR;
  337. pthread_setspecific(item_lock_type_key, &me->item_lock_type);
  338. //每一个worker线程进入loop，全局init_count++操作，
  339. //见thread_init函数后面几行代码和wait_for_thread_registration函数，
  340. //主线程通过init_count来确认所有线程都启动完毕。
  341. register_thread_initialized();
  342. event_base_loop(me->base, 0);
  343. return NULL;
  344. }
  345. //主线程分发client fd给worker线程后，同时往管道写入buf，唤醒worker线程调用此函数
  346. static void thread_libevent_process(int fd, short which, void *arg) {
  347. LIBEVENT_THREAD *me = arg;
  348. CQ_ITEM *item;
  349. char buf[1];
  350. if (read(fd, buf, 1) != 1)
  351. if (settings.verbose > 0)
  352. fprintf(stderr, "Can‘t read from libevent pipe\n");
  353. switch (buf[0]) {
  354. case ‘c‘:
  355. item = cq_pop(me->new_conn_queue); //取出主线程丢过来的CQ_ITEM
  356. if (NULL != item) {
  357. /*
  358. worker线程创建 conn连接对象，注意由主线程丢过来的CQ_ITEM的init_state为conn_new_cmd （TCP情况下）
  359. */
  360. conn *c = conn_new(item->sfd, item->init_state, item->event_flags,
  361. item->read_buffer_size, item->transport, me->base);
  362. if (c == NULL) {
  363. if (IS_UDP(item->transport)) {
  364. fprintf(stderr, "Can‘t listen for events on UDP socket\n");
  365. exit(1);
  366. } else {
  367. if (settings.verbose > 0) {
  368. fprintf(stderr, "Can‘t listen for events on fd %d\n",
  369. item->sfd);
  370. }
  371. close(item->sfd);
  372. }
  373. } else {
  374. c->thread = me; //设置监听连接的线程为当前worker线程
  375. }
  376. cqi_free(item);
  377. }
  378. break;
  379. /* we were told to flip the lock type and report in */
  380. case ‘l‘:
  381. me->item_lock_type = ITEM_LOCK_GRANULAR;
  382. register_thread_initialized();
  383. break;
  384. case ‘g‘:
  385. me->item_lock_type = ITEM_LOCK_GLOBAL;
  386. register_thread_initialized();
  387. break;
  388. }
  389. }
  390. void dispatch_conn_new(int sfd, enum conn_states init_state, int event_flags,
  391. int read_buffer_size, enum network_transport transport) {
  392. /**
  393. 这下面有一个CQ_ITEM结构体，可以这么理解，主线程accept连接后，把client fd
  394. 分发到worker线程的同时会顺带一些与此client连接相关的信息，例如dispatch_conn_new的形参上面列的，
  395. 而CQ_ITEM是包装了这些信息的一个对象。
  396. CQ_ITEM中的CQ是connection queue的缩写，但它与conn结构体是完全不一样的概念，CQ_ITEM仅仅是把client连接相关的信息
  397. 打包成一个对象而已。
  398. */
  399. CQ_ITEM *item = cqi_new();
  400. char buf[1];
  401. if (item == NULL) {
  402. close(sfd);
  403. /* given that malloc failed this may also fail, but let‘s try */
  404. fprintf(stderr, "Failed to allocate memory for connection object\n");
  405. return ;
  406. }
  407. int tid = (last_thread + 1) % settings.num_threads;
  408. LIBEVENT_THREAD *thread = threads + tid; //通过简单的轮叫方式选择处理当前client fd的worker线程
  409. last_thread = tid;
  410. //初始化CQ_ITEM对象，即把信息包装
  411. item->sfd = sfd;
  412. item->init_state = init_state;
  413. item->event_flags = event_flags;
  414. item->read_buffer_size = read_buffer_size;
  415. item->transport = transport;
  416. cq_push(thread->new_conn_queue, item); //每个worker线程保存着所有被分发给自己的CQ_ITEM，即new_conn_queue
  417. MEMCACHED_CONN_DISPATCH(sfd, thread->thread_id);
  418. /*
  419. 主线程向处理当前client fd的worker线程管道中简单写进一个‘c‘字符，
  420. 由于每个worker线程都监听了管道的receive_fd，于是相应的worker进程收到事件通知，
  421. 触发注册的handler，即thread_libevent_process
  422. */
  423. buf[0] = ‘c‘;
  424. if (write(thread->notify_send_fd, buf, 1) != 1) {
  425. perror("Writing to thread notify pipe");
  426. }
  427. }
  428. int is_listen_thread() {
  429. return pthread_self() == dispatcher_thread.thread_id;
  430. }
  431. /********************************* ITEM ACCESS *******************************/
  432. /**
  433. 下面是一堆关于item操作的函数，具体逻辑代码都放在items::do_xxx相应的地方
  434. 就像本文件开头说的，这里主要是加了锁而已
  435. */
  436. /*
  437. * Allocates a new item.
  438. 分配item空间
  439. */
  440. item *item_alloc(char *key, size_t nkey, int flags, rel_time_t exptime, int nbytes) {
  441. item *it;
  442. /* do_item_alloc handles its own locks */
  443. /**
  444. 这里比较特殊，与其它item_xxx函数不一样，这里把锁放在do_item_alloc里面做了。
  445. 个人猜测是因为do_item_alloc这个逻辑实在有点复杂，甚至加解锁有可能在某个if条件下要发
  446. 生，加解锁和逻辑本身代码耦合，所以外部不好加锁。因此把锁交给do_item_alloc内部进行考虑。
  447. */
  448. it = do_item_alloc(key, nkey, flags, exptime, nbytes, 0);
  449. return it;
  450. }
  451. /*
  452. * Returns an item if it hasn‘t been marked as expired,
  453. * lazy-expiring as needed.
  454. 取得item，上面这里有句英文注释，说返回不超时的item，因为memcached并没有做实时或者定时把
  455. 超时item清掉的逻辑，而是用了延迟超时。就是当要用这个item的时候，再来针对这个item做超时处理
  456. */
  457. item *item_get(const char *key, const size_t nkey) {
  458. item *it;
  459. uint32_t hv;
  460. hv = hash(key, nkey);
  461. item_lock(hv);
  462. it = do_item_get(key, nkey, hv);
  463. item_unlock(hv);
  464. return it;
  465. }
  466. item *item_touch(const char *key, size_t nkey, uint32_t exptime) {
  467. item *it;
  468. uint32_t hv;
  469. hv = hash(key, nkey);
  470. item_lock(hv);
  471. it = do_item_touch(key, nkey, exptime, hv);
  472. item_unlock(hv);
  473. return it;
  474. }
  475. /*
  476. * Links an item into the LRU and hashtable.
  477. */
  478. int item_link(item *item) {
  479. int ret;
  480. uint32_t hv;
  481. hv = hash(ITEM_key(item), item->nkey);
  482. item_lock(hv);
  483. ret = do_item_link(item, hv);
  484. item_unlock(hv);
  485. return ret;
  486. }
  487. void item_remove(item *item) {
  488. uint32_t hv;
  489. hv = hash(ITEM_key(item), item->nkey);
  490. item_lock(hv);
  491. do_item_remove(item);
  492. item_unlock(hv);
  493. }
  494. int item_replace(item *old_it, item *new_it, const uint32_t hv) {
  495. return do_item_replace(old_it, new_it, hv);
  496. }
  497. /*
  498. * Unlinks an item from the LRU and hashtable.
  499. * 见items::item_unlink
  500. */
  501. void item_unlink(item *item) {
  502. uint32_t hv;
  503. hv = hash(ITEM_key(item), item->nkey);
  504. item_lock(hv);
  505. do_item_unlink(item, hv);
  506. item_unlock(hv);
  507. }
  508. /**
  509. 主要作用是重置在最近使用链表中的位置，更新最近使用时间，见items::do_item_update
  510. */
  511. void item_update(item *item) {
  512. uint32_t hv;
  513. hv = hash(ITEM_key(item), item->nkey);
  514. item_lock(hv);
  515. do_item_update(item);
  516. item_unlock(hv);
  517. }
  518. enum delta_result_type add_delta(conn *c, const char *key,
  519. const size_t nkey, int incr,
  520. const int64_t delta, char *buf,
  521. uint64_t *cas) {
  522. enum delta_result_type ret;
  523. uint32_t hv;
  524. hv = hash(key, nkey);
  525. item_lock(hv);
  526. ret = do_add_delta(c, key, nkey, incr, delta, buf, cas, hv);
  527. item_unlock(hv);
  528. return ret;
  529. }
  530. /*
  531. * Stores an item in the cache (high level, obeys set/add/replace semantics)
  532. * 保存item信息，主要是调用items::do_store_item，但由于是多线程，所以需求加锁
  533. * store_item是线程上的操作，所以写在thread模块，在此对外开放，而内部加锁。
  534. * 除了store_item函数，其它关于item的操作均如此。
  535. */
  536. enum store_item_type store_item(item *item, int comm, conn* c) {
  537. enum store_item_type ret;
  538. uint32_t hv;
  539. hv = hash(ITEM_key(item), item->nkey); //锁住item
  540. item_lock(hv);
  541. ret = do_store_item(item, comm, c, hv);
  542. item_unlock(hv);
  543. return ret;
  544. }
  545. void item_flush_expired() {
  546. mutex_lock(&cache_lock);
  547. do_item_flush_expired();
  548. mutex_unlock(&cache_lock);
  549. }
  550. char *item_cachedump(unsigned int slabs_clsid, unsigned int limit, unsigned int *bytes) {
  551. char *ret;
  552. mutex_lock(&cache_lock);
  553. ret = do_item_cachedump(slabs_clsid, limit, bytes);
  554. mutex_unlock(&cache_lock);
  555. return ret;
  556. }
  557. void item_stats(ADD_STAT add_stats, void *c) {
  558. mutex_lock(&cache_lock);
  559. do_item_stats(add_stats, c);
  560. mutex_unlock(&cache_lock);
  561. }
  562. void item_stats_totals(ADD_STAT add_stats, void *c) {
  563. mutex_lock(&cache_lock);
  564. do_item_stats_totals(add_stats, c);
  565. mutex_unlock(&cache_lock);
  566. }
  567. void item_stats_sizes(ADD_STAT add_stats, void *c) {
  568. mutex_lock(&cache_lock);
  569. do_item_stats_sizes(add_stats, c);
  570. mutex_unlock(&cache_lock);
  571. }
  572. /******************************* GLOBAL STATS ******************************/
  573. void STATS_LOCK() {
  574. pthread_mutex_lock(&stats_lock);
  575. }
  576. void STATS_UNLOCK() {
  577. pthread_mutex_unlock(&stats_lock);
  578. }
  579. void threadlocal_stats_reset(void) {
  580. int ii, sid;
  581. for (ii = 0; ii < settings.num_threads; ++ii) {
  582. pthread_mutex_lock(&threads[ii].stats.mutex);
  583. threads[ii].stats.get_cmds = 0;
  584. threads[ii].stats.get_misses = 0;
  585. threads[ii].stats.touch_cmds = 0;
  586. threads[ii].stats.touch_misses = 0;
  587. threads[ii].stats.delete_misses = 0;
  588. threads[ii].stats.incr_misses = 0;
  589. threads[ii].stats.decr_misses = 0;
  590. threads[ii].stats.cas_misses = 0;
  591. threads[ii].stats.bytes_read = 0;
  592. threads[ii].stats.bytes_written = 0;
  593. threads[ii].stats.flush_cmds = 0;
  594. threads[ii].stats.conn_yields = 0;
  595. threads[ii].stats.auth_cmds = 0;
  596. threads[ii].stats.auth_errors = 0;
  597. for(sid = 0; sid < MAX_NUMBER_OF_SLAB_CLASSES; sid++) {
  598. threads[ii].stats.slab_stats[sid].set_cmds = 0;
  599. threads[ii].stats.slab_stats[sid].get_hits = 0;
  600. threads[ii].stats.slab_stats[sid].touch_hits = 0;
  601. threads[ii].stats.slab_stats[sid].delete_hits = 0;
  602. threads[ii].stats.slab_stats[sid].incr_hits = 0;
  603. threads[ii].stats.slab_stats[sid].decr_hits = 0;
  604. threads[ii].stats.slab_stats[sid].cas_hits = 0;
  605. threads[ii].stats.slab_stats[sid].cas_badval = 0;
  606. }
  607. pthread_mutex_unlock(&threads[ii].stats.mutex);
  608. }
  609. }
  610. void threadlocal_stats_aggregate(struct thread_stats *stats) {
  611. int ii, sid;
  612. /* The struct has a mutex, but we can safely set the whole thing
  613. * to zero since it is unused when aggregating. */
  614. memset(stats, 0, sizeof(*stats));
  615. for (ii = 0; ii < settings.num_threads; ++ii) {
  616. pthread_mutex_lock(&threads[ii].stats.mutex);
  617. stats->get_cmds += threads[ii].stats.get_cmds;
  618. stats->get_misses += threads[ii].stats.get_misses;
  619. stats->touch_cmds += threads[ii].stats.touch_cmds;
  620. stats->touch_misses += threads[ii].stats.touch_misses;
  621. stats->delete_misses += threads[ii].stats.delete_misses;
  622. stats->decr_misses += threads[ii].stats.decr_misses;
  623. stats->incr_misses += threads[ii].stats.incr_misses;
  624. stats->cas_misses += threads[ii].stats.cas_misses;
  625. stats->bytes_read += threads[ii].stats.bytes_read;
  626. stats->bytes_written += threads[ii].stats.bytes_written;
  627. stats->flush_cmds += threads[ii].stats.flush_cmds;
  628. stats->conn_yields += threads[ii].stats.conn_yields;
  629. stats->auth_cmds += threads[ii].stats.auth_cmds;
  630. stats->auth_errors += threads[ii].stats.auth_errors;
  631. for (sid = 0; sid < MAX_NUMBER_OF_SLAB_CLASSES; sid++) {
  632. stats->slab_stats[sid].set_cmds +=
  633. threads[ii].stats.slab_stats[sid].set_cmds;
  634. stats->slab_stats[sid].get_hits +=
  635. threads[ii].stats.slab_stats[sid].get_hits;
  636. stats->slab_stats[sid].touch_hits +=
  637. threads[ii].stats.slab_stats[sid].touch_hits;
  638. stats->slab_stats[sid].delete_hits +=
  639. threads[ii].stats.slab_stats[sid].delete_hits;
  640. stats->slab_stats[sid].decr_hits +=
  641. threads[ii].stats.slab_stats[sid].decr_hits;
  642. stats->slab_stats[sid].incr_hits +=
  643. threads[ii].stats.slab_stats[sid].incr_hits;
  644. stats->slab_stats[sid].cas_hits +=
  645. threads[ii].stats.slab_stats[sid].cas_hits;
  646. stats->slab_stats[sid].cas_badval +=
  647. threads[ii].stats.slab_stats[sid].cas_badval;
  648. }
  649. pthread_mutex_unlock(&threads[ii].stats.mutex);
  650. }
  651. }
  652. void slab_stats_aggregate(struct thread_stats *stats, struct slab_stats *out) {
  653. int sid;
  654. out->set_cmds = 0;
  655. out->get_hits = 0;
  656. out->touch_hits = 0;
  657. out->delete_hits = 0;
  658. out->incr_hits = 0;
  659. out->decr_hits = 0;
  660. out->cas_hits = 0;
  661. out->cas_badval = 0;
  662. for (sid = 0; sid < MAX_NUMBER_OF_SLAB_CLASSES; sid++) {
  663. out->set_cmds += stats->slab_stats[sid].set_cmds;
  664. out->get_hits += stats->slab_stats[sid].get_hits;
  665. out->touch_hits += stats->slab_stats[sid].touch_hits;
  666. out->delete_hits += stats->slab_stats[sid].delete_hits;
  667. out->decr_hits += stats->slab_stats[sid].decr_hits;
  668. out->incr_hits += stats->slab_stats[sid].incr_hits;
  669. out->cas_hits += stats->slab_stats[sid].cas_hits;
  670. out->cas_badval += stats->slab_stats[sid].cas_badval;
  671. }
  672. }
  673. //初始化主线程
  674. void thread_init(int nthreads, struct event_base *main_base) {
  675. int i;
  676. int power;
  677. pthread_mutex_init(&cache_lock, NULL);
  678. pthread_mutex_init(&stats_lock, NULL);
  679. pthread_mutex_init(&init_lock, NULL);
  680. pthread_cond_init(&init_cond, NULL);
  681. pthread_mutex_init(&cqi_freelist_lock, NULL);
  682. cqi_freelist = NULL;
  683. /* Want a wide lock table, but don‘t waste memory */
  684. /**
  685. 初始化item lock
  686. */
  687. //调配item锁的数量
  688. //之所以需要锁是因为线程之间的并发，所以item锁的数量当然是根据线程的个数进行调配了。
  689. if (nthreads < 3) {
  690. power = 10; //这个power是指数
  691. } else if (nthreads < 4) {
  692. power = 11;
  693. } else if (nthreads < 5) {
  694. power = 12;
  695. } else {
  696. /* 8192 buckets, and central locks don‘t scale much past 5 threads */
  697. power = 13;
  698. }
  699. item_lock_count = hashsize(power);
  700. item_lock_hashpower = power;
  701. item_locks = calloc(item_lock_count, sizeof(pthread_mutex_t));
  702. if (! item_locks) {
  703. perror("Can‘t allocate item locks");
  704. exit(1);
  705. }
  706. for (i = 0; i < item_lock_count; i++) {
  707. pthread_mutex_init(&item_locks[i], NULL);
  708. }
  709. pthread_key_create(&item_lock_type_key, NULL);
  710. pthread_mutex_init(&item_global_lock, NULL);
  711. //_mark2_1
  712. threads = calloc(nthreads, sizeof(LIBEVENT_THREAD)); //创建worker线程对象
  713. if (! threads) {
  714. perror("Can‘t allocate thread descriptors");
  715. exit(1);
  716. }
  717. //_mark2_3
  718. dispatcher_thread.base = main_base; //设置主线程对象的event_base
  719. dispatcher_thread.thread_id = pthread_self(); //设置主线程对象pid
  720. //_mark2_5
  721. for (i = 0; i < nthreads; i++) { //为每个worker线程创建与主线程通信的管道
  722. int fds[2];
  723. if (pipe(fds)) {
  724. perror("Can‘t create notify pipe");
  725. exit(1);
  726. }
  727. threads[i].notify_receive_fd = fds[0]; //worker线程管道接收fd
  728. threads[i].notify_send_fd = fds[1]; //worker线程管道写入fd
  729. //_mark2_6
  730. setup_thread(&threads[i]); //装载 worker线程
  731. /* Reserve three fds for the libevent base, and two for the pipe */
  732. stats.reserved_fds += 5;
  733. }
  734. /* Create threads after we‘ve done all the libevent setup. */
  735. for (i = 0; i < nthreads; i++) {
  736. //_mark2_7
  737. create_worker(worker_libevent, &threads[i]); //启动worker线程，见worker_libevent
  738. }
  739. /* Wait for all the threads to set themselves up before returning. */
  740. pthread_mutex_lock(&init_lock);
  741. wait_for_thread_registration(nthreads); //等待所有worker线程启动完毕
  742. pthread_mutex_unlock(&init_lock);
  743. }
时间： 2024-10-22 00:51:59

Memcached源码分析之thread.c的相关文章

Memcached源码分析

作者:Calix,转载请注明出处:http://calixwu.com 最近研究了一下memcached的源码,在这里系统总结了一下笔记和理解,写了几 篇源码分析和大家分享,整个系列分为“结构篇”和“源码篇”,建议先从结构篇开始看起,要特别说明的是我本次分析的是memcached1.4.20的版 本,不同版本会有所差异,另外,文章均为本人的个人理解,如果解析得不好或者错误的地方敬请指正. 好了,不啰嗦了,下面是导航: [结构篇] Memcached源码分析之从SET命令开始说起 Memcache

memcached源码分析-----item过期失效处理以及LRU爬虫

memcached源码分析-----item过期失效处理以及LRU爬虫,memcached-----item 转载请注明出处:http://blog.csdn.net/luotuo44/article/details/42963793 温馨提示:本文用到了一些可以在启动memcached设置的全局变量.关于这些全局变量的含义可以参考<memcached启动参数详解>.对于这些全局变量,处理方式就像<如何阅读memcached源代码>所说的那样直接取其默认值. 过期失效处理: 一个i

Linux c 开发 - Memcached源码分析之命令解析（2）

前言 从我们上一章<Linux c 开发 - Memcached源码分析之基于Libevent的网络模型>我们基本了解了Memcached的网络模型.这一章节,我们需要详细解读Memcached的命令解析. 我们回顾上一章发现Memcached会分成主线程和N个工作线程.主线程主要用于监听accpet客户端的Socket连接,而工作线程主要用于接管具体的客户端连接. 主线程和工作线程之间主要通过基于Libevent的pipe的读写事件来监听,当有连接练上来的时候,主线程会将连接交个某一个工作线

Memcached源码分析之线程模型

作者:Calix 一)模型分析 memcached到底是如何处理我们的网络连接的? memcached通过epoll(使用libevent,下面具体再讲)实现异步的服务器,但仍然使用多线程,主要有两种线程,分别是“主线程”和“worker线程”,一个主线程,多个worker线程. 主线程负责监听网络连接,并且accept连接.当监听到连接时,accept后,连接成功,把相应的client fd丢给其中一个worker线程.worker线程接收主线程丢过来的client fd,加入到自己的epol

Memcached源码分析之内存管理

先再说明一下,我本次分析的memcached版本是1.4.20,有些旧的版本关于内存管理的机制和数据结构与1.4.20有一定的差异(本文中会提到). 一)模型分析在开始解剖memcached关于内存管理的源代码之前,先宏观上分析一下memcached内存管理的模型是怎样子的: 提个建议,我觉得memcached内存管理的模型与我们平时做作业的作业本“画格子给我们往格子里面写字”的逻辑很像,一本本作业本就是我们的内存空间,而我们往里写的字就是我们要存下来的数据,所以分析的时候可以想像一下用方格作业

Memcached源码分析之从SET命令开始说起

作者:Calix 如果直接把memcached的源码从main函数开始说,恐怕会有点头大,所以这里以一句经典的“SET”命令简单地开个头,算是回忆一下memcached的作用,后面的结构篇中关于命令解析部分主要也是围绕着SET命令展开分析,相信把一句SET命令背后做的事情都搞清楚,那么memcached大部分源码都了解得七七八八了. 那么,回忆一下,set命令做了个什么事情? 无非就是把一个value set到某个key上面,保存在内存当中. 再细化一下: 1)memcached是一个缓存服务器

memcached源码分析-----安装、调试以及如何阅读memcached源码

        转载请注明出处:http://blog.csdn.net/luotuo44/article/details/42639131 安装: 安装memcached之前要先安装Libevent.现在假定Libevent安装在/usr/local/libevent目录了. 因为memcached安装后不像Libevent那样,有一堆头文件和库文件.安装后的memcached不是用来编程而直接用来运行的.所以不需要在/usr/local目录下专门为memcached建立一个目录.直接把mem

memcached源码分析-----memcached启动参数详解以及关键配置的默认值

转载请注明出处: http://blog.csdn.net/luotuo44/article/details/42672913 本文开启本系列博文的代码分析.本系列博文研究是memcached版本是1.4.21. 本文将给出memcached启动时各个参数的详细解释以及一些关键配置的默认值.以便在分析memcached源码的时候好随时查看.当然也方便使用memcached时可以随时查看各个参数的含义.<如何阅读memcached源码>说到memcached有很多全局变量(也就是关键配置),这些

Memcached源码分析——process_command函数解析

以下为个人笔记 /** * process_command 在memcached中是用来处理用户发送的命令的, * 包括get set,add,delete,replace,stats,flush_all等常用的和不常用的命令,全在这进行处理的. * 一共有两个参数,conn *c 和字符串指针 char *command : * 关于conn结构体就先不说了,反正它是整个memcached中最重要的结构体就行了,等以后研究明白了再说,先从整体上领会精神吧. * 这里我想说一下的是, memca