| 3461 | } |
| 3462 | |
| 3463 | void *IOThreadMain(void *myid) { |
| 3464 | /* The ID is the thread number (from 0 to server.iothreads_num-1), and is |
| 3465 | * used by the thread to just manipulate a single sub-array of clients. */ |
| 3466 | long id = (unsigned long)myid; |
| 3467 | char thdname[16]; |
| 3468 | |
| 3469 | snprintf(thdname, sizeof(thdname), "io_thd_%ld", id); |
| 3470 | redis_set_thread_title(thdname); |
| 3471 | redisSetCpuAffinity(server.server_cpulist); |
| 3472 | makeThreadKillable(); |
| 3473 | |
| 3474 | while(1) { |
| 3475 | /* Wait for start */ |
| 3476 | for (int j = 0; j < 1000000; j++) { |
| 3477 | if (getIOPendingCount(id) != 0) break; |
| 3478 | } |
| 3479 | |
| 3480 | /* Give the main thread a chance to stop this thread. */ |
| 3481 | if (getIOPendingCount(id) == 0) { |
| 3482 | pthread_mutex_lock(&io_threads_mutex[id]); |
| 3483 | pthread_mutex_unlock(&io_threads_mutex[id]); |
| 3484 | continue; |
| 3485 | } |
| 3486 | |
| 3487 | serverAssert(getIOPendingCount(id) != 0); |
| 3488 | |
| 3489 | /* Process: note that the main thread will never touch our list |
| 3490 | * before we drop the pending count to 0. */ |
| 3491 | listIter li; |
| 3492 | listNode *ln; |
| 3493 | listRewind(io_threads_list[id],&li); |
| 3494 | while((ln = listNext(&li))) { |
| 3495 | client *c = listNodeValue(ln); |
| 3496 | if (io_threads_op == IO_THREADS_OP_WRITE) { |
| 3497 | writeToClient(c,0); |
| 3498 | } else if (io_threads_op == IO_THREADS_OP_READ) { |
| 3499 | readQueryFromClient(c->conn); |
| 3500 | } else { |
| 3501 | serverPanic("io_threads_op value is unknown"); |
| 3502 | } |
| 3503 | } |
| 3504 | listEmpty(io_threads_list[id]); |
| 3505 | setIOPendingCount(id, 0); |
| 3506 | } |
| 3507 | } |
| 3508 | |
| 3509 | /* Initialize the data structures needed for threaded I/O. */ |
| 3510 | void initThreadedIO(void) { |
nothing calls this directly
no test coverage detected