Commit ea35a81c authored by antirez's avatar antirez
Browse files

Threaded IO: 3rd version: use the mutex only to stop the thread.

parent 6f4f36c0
...@@ -2491,24 +2491,34 @@ void *IOThreadMain(void *myid) { ...@@ -2491,24 +2491,34 @@ void *IOThreadMain(void *myid) {
long id = (unsigned long)myid; long id = (unsigned long)myid;
while(1) { while(1) {
/* ... Wait for start ... */ /* Wait for start */
pthread_mutex_lock(&io_threads_mutex[id]); for (int j = 0; j < 1000000; j++) {
if (io_threads_pending[id]) { if (io_threads_pending[id] != 0) break;
if (tio_debug) printf("[%ld] %d to handle\n", id, (int)listLength(io_threads_list[id])); }
/* ... Process ... */ /* Give the main thread a chance to stop this thread. */
listIter li; if (io_threads_pending[id] == 0) {
listNode *ln; pthread_mutex_lock(&io_threads_mutex[id]);
listRewind(io_threads_list[id],&li); pthread_mutex_unlock(&io_threads_mutex[id]);
while((ln = listNext(&li))) { continue;
client *c = listNodeValue(ln);
writeToClient(c->fd,c,0);
io_threads_pending[id]--;
}
listEmpty(io_threads_list[id]);
} }
pthread_mutex_unlock(&io_threads_mutex[id]); serverAssert(io_threads_pending[id] != 0);
if (tio_debug) printf("[%ld] %d to handle\n", id, (int)listLength(io_threads_list[id]));
/* Process: note that the main thread will never touch our list
* before we drop the pending count to 0. */
listIter li;
listNode *ln;
listRewind(io_threads_list[id],&li);
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
writeToClient(c->fd,c,0);
}
listEmpty(io_threads_list[id]);
io_threads_pending[id] = 0;
if (tio_debug) printf("[%ld] Done\n", id); if (tio_debug) printf("[%ld] Done\n", id);
} }
} }
...@@ -2572,13 +2582,17 @@ int handleClientsWithPendingWritesUsingThreads(void) { ...@@ -2572,13 +2582,17 @@ int handleClientsWithPendingWritesUsingThreads(void) {
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; int target_id = item_id % server.io_threads_num;
pthread_mutex_lock(&io_threads_mutex[target_id]);
listAddNodeTail(io_threads_list[target_id],c); listAddNodeTail(io_threads_list[target_id],c);
io_threads_pending[target_id]++;
pthread_mutex_unlock(&io_threads_mutex[target_id]);
item_id++; item_id++;
} }
/* Give the start condition to the waiting threads, by setting the
* start condition atomic var. */
for (int j = 0; j < server.io_threads_num; j++) {
int count = listLength(io_threads_list[j]);
io_threads_pending[j] = count;
}
/* Wait for all threads to end their work. */ /* Wait for all threads to end their work. */
while(1) { while(1) {
unsigned long pending = 0; unsigned long pending = 0;
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment