Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
5be3a15a
Commit
5be3a15a
authored
Jan 13, 2020
by
antirez
Browse files
Setting N I/O threads should mean N-1 additional + 1 main thread.
parent
cbabf779
Changes
1
Show whitespace changes
Inline
Side-by-side
src/networking.c
View file @
5be3a15a
...
@@ -2658,9 +2658,9 @@ int io_threads_active; /* Are the threads currently spinning waiting I/O? */
...
@@ -2658,9 +2658,9 @@ int io_threads_active; /* Are the threads currently spinning waiting I/O? */
int io_threads_op; /* IO_THREADS_OP_WRITE or IO_THREADS_OP_READ. */
int io_threads_op; /* IO_THREADS_OP_WRITE or IO_THREADS_OP_READ. */
/* This is the list of clients each thread will serve when threaded I/O is
/* This is the list of clients each thread will serve when threaded I/O is
* used. We spawn
N
threads
, and the N+1 list is used for the clients that
* used. We spawn
io_
threads
_num-1 threads, since one is the main thread
*
are processed by the main thread itself (this is why ther is "+1")
. */
*
itself
. */
list
*
io_threads_list
[
IO_THREADS_MAX_NUM
+
1
];
list *io_threads_list[IO_THREADS_MAX_NUM];
void *IOThreadMain(void *myid) {
void *IOThreadMain(void *myid) {
/* The ID is the thread number (from 0 to server.iothreads_num-1), and is
/* The ID is the thread number (from 0 to server.iothreads_num-1), and is
...
@@ -2720,12 +2720,16 @@ void initThreadedIO(void) {
...
@@ -2720,12 +2720,16 @@ void initThreadedIO(void) {
exit(1);
exit(1);
}
}
/* Spawn the I/O threads. */
/* Spawn
and initialize
the I/O threads. */
for (int i = 0; i < server.io_threads_num; i++) {
for (int i = 0; i < server.io_threads_num; i++) {
/* Things we do for all the threads including the main thread. */
io_threads_list[i] = listCreate();
if (i == 0) continue; /* Thread 0 is the main thread. */
/* Things we do only for the additional threads. */
pthread_t tid;
pthread_t tid;
pthread_mutex_init(&io_threads_mutex[i],NULL);
pthread_mutex_init(&io_threads_mutex[i],NULL);
io_threads_pending[i] = 0;
io_threads_pending[i] = 0;
io_threads_list
[
i
]
=
listCreate
();
pthread_mutex_lock(&io_threads_mutex[i]); /* Thread will be stopped. */
pthread_mutex_lock(&io_threads_mutex[i]); /* Thread will be stopped. */
if (pthread_create(&tid,NULL,IOThreadMain,(void*)(long)i) != 0) {
if (pthread_create(&tid,NULL,IOThreadMain,(void*)(long)i) != 0) {
serverLog(LL_WARNING,"Fatal: Can't initialize IO thread.");
serverLog(LL_WARNING,"Fatal: Can't initialize IO thread.");
...
@@ -2733,14 +2737,13 @@ void initThreadedIO(void) {
...
@@ -2733,14 +2737,13 @@ void initThreadedIO(void) {
}
}
io_threads[i] = tid;
io_threads[i] = tid;
}
}
io_threads_list
[
server
.
io_threads_num
]
=
listCreate
();
/* For main thread */
}
}
void startThreadedIO(void) {
void startThreadedIO(void) {
if (tio_debug) { printf("S"); fflush(stdout); }
if (tio_debug) { printf("S"); fflush(stdout); }
if (tio_debug) printf("--- STARTING THREADED IO ---\n");
if (tio_debug) printf("--- STARTING THREADED IO ---\n");
serverAssert(io_threads_active == 0);
serverAssert(io_threads_active == 0);
for
(
int
j
=
0
;
j
<
server
.
io_threads_num
;
j
++
)
for (int j =
1
; j < server.io_threads_num; j++)
pthread_mutex_unlock(&io_threads_mutex[j]);
pthread_mutex_unlock(&io_threads_mutex[j]);
io_threads_active = 1;
io_threads_active = 1;
}
}
...
@@ -2754,7 +2757,7 @@ void stopThreadedIO(void) {
...
@@ -2754,7 +2757,7 @@ void stopThreadedIO(void) {
(int) listLength(server.clients_pending_read),
(int) listLength(server.clients_pending_read),
(int) listLength(server.clients_pending_write));
(int) listLength(server.clients_pending_write));
serverAssert(io_threads_active == 1);
serverAssert(io_threads_active == 1);
for
(
int
j
=
0
;
j
<
server
.
io_threads_num
;
j
++
)
for (int j =
1
; j < server.io_threads_num; j++)
pthread_mutex_lock(&io_threads_mutex[j]);
pthread_mutex_lock(&io_threads_mutex[j]);
io_threads_active = 0;
io_threads_active = 0;
}
}
...
@@ -2805,7 +2808,7 @@ int handleClientsWithPendingWritesUsingThreads(void) {
...
@@ -2805,7 +2808,7 @@ int handleClientsWithPendingWritesUsingThreads(void) {
while((ln = listNext(&li))) {
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
client *c = listNodeValue(ln);
c->flags &= ~CLIENT_PENDING_WRITE;
c->flags &= ~CLIENT_PENDING_WRITE;
int
target_id
=
item_id
%
(
server
.
io_threads_num
+
1
)
;
int target_id = item_id % server.io_threads_num;
listAddNodeTail(io_threads_list[target_id],c);
listAddNodeTail(io_threads_list[target_id],c);
item_id++;
item_id++;
}
}
...
@@ -2813,23 +2816,23 @@ int handleClientsWithPendingWritesUsingThreads(void) {
...
@@ -2813,23 +2816,23 @@ int handleClientsWithPendingWritesUsingThreads(void) {
/* Give the start condition to the waiting threads, by setting the
/* Give the start condition to the waiting threads, by setting the
* start condition atomic var. */
* start condition atomic var. */
io_threads_op = IO_THREADS_OP_WRITE;
io_threads_op = IO_THREADS_OP_WRITE;
for
(
int
j
=
0
;
j
<
server
.
io_threads_num
;
j
++
)
{
for (int j =
1
; j < server.io_threads_num; j++) {
int count = listLength(io_threads_list[j]);
int count = listLength(io_threads_list[j]);
io_threads_pending[j] = count;
io_threads_pending[j] = count;
}
}
/* Also use the main thread to process a sli
d
e of clients. */
/* Also use the main thread to process a sli
c
e of clients. */
listRewind
(
io_threads_list
[
server
.
io_threads_num
],
&
li
);
listRewind(io_threads_list[
0
],&li);
while((ln = listNext(&li))) {
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
client *c = listNodeValue(ln);
writeToClient(c,0);
writeToClient(c,0);
}
}
listEmpty
(
io_threads_list
[
server
.
io_threads_num
]);
listEmpty(io_threads_list[
0
]);
/* Wait for all threads to end their work. */
/* Wait for all
the other
threads to end their work. */
while(1) {
while(1) {
unsigned long pending = 0;
unsigned long pending = 0;
for
(
int
j
=
0
;
j
<
server
.
io_threads_num
;
j
++
)
for (int j =
1
; j < server.io_threads_num; j++)
pending += io_threads_pending[j];
pending += io_threads_pending[j];
if (pending == 0) break;
if (pending == 0) break;
}
}
...
@@ -2890,7 +2893,7 @@ int handleClientsWithPendingReadsUsingThreads(void) {
...
@@ -2890,7 +2893,7 @@ int handleClientsWithPendingReadsUsingThreads(void) {
int item_id = 0;
int item_id = 0;
while((ln = listNext(&li))) {
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
client *c = listNodeValue(ln);
int
target_id
=
item_id
%
(
server
.
io_threads_num
+
1
)
;
int target_id = item_id % server.io_threads_num;
listAddNodeTail(io_threads_list[target_id],c);
listAddNodeTail(io_threads_list[target_id],c);
item_id++;
item_id++;
}
}
...
@@ -2898,23 +2901,23 @@ int handleClientsWithPendingReadsUsingThreads(void) {
...
@@ -2898,23 +2901,23 @@ int handleClientsWithPendingReadsUsingThreads(void) {
/* Give the start condition to the waiting threads, by setting the
/* Give the start condition to the waiting threads, by setting the
* start condition atomic var. */
* start condition atomic var. */
io_threads_op = IO_THREADS_OP_READ;
io_threads_op = IO_THREADS_OP_READ;
for
(
int
j
=
0
;
j
<
server
.
io_threads_num
;
j
++
)
{
for (int j =
1
; j < server.io_threads_num; j++) {
int count = listLength(io_threads_list[j]);
int count = listLength(io_threads_list[j]);
io_threads_pending[j] = count;
io_threads_pending[j] = count;
}
}
/* Also use the main thread to process a sli
d
e of clients. */
/* Also use the main thread to process a sli
c
e of clients. */
listRewind
(
io_threads_list
[
server
.
io_threads_num
],
&
li
);
listRewind(io_threads_list[
0
],&li);
while((ln = listNext(&li))) {
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
client *c = listNodeValue(ln);
readQueryFromClient(c->conn);
readQueryFromClient(c->conn);
}
}
listEmpty
(
io_threads_list
[
server
.
io_threads_num
]);
listEmpty(io_threads_list[
0
]);
/* Wait for all threads to end their work. */
/* Wait for all
the other
threads to end their work. */
while(1) {
while(1) {
unsigned long pending = 0;
unsigned long pending = 0;
for
(
int
j
=
0
;
j
<
server
.
io_threads_num
;
j
++
)
for (int j =
1
; j < server.io_threads_num; j++)
pending += io_threads_pending[j];
pending += io_threads_pending[j];
if (pending == 0) break;
if (pending == 0) break;
}
}
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment