Commit 99de8f95 authored by Geoff Simmons's avatar Geoff Simmons

trackrdrd: - no longer using SPMCQ_StopWorker() it was too aggressive

	- more carefully counting the running and exited workers
parent 301c31b5
......@@ -689,12 +689,10 @@ OSL_Track(void *priv, enum VSL_tag_e tag, unsigned fd, unsigned len,
if (term && htbl.open == 0)
return 1;
if (wrk_running < config.nworkers) {
wrk_running = WRK_Running();
if (wrk_running < config.nworkers)
LOG_Log(LOG_WARNING, "%d of %d workers running", wrk_running,
config.nworkers);
}
wrk_running = WRK_Running();
if (wrk_running < config.nworkers)
LOG_Log(LOG_WARNING, "%d of %d workers running", wrk_running,
config.nworkers);
/* spec != 'c' */
if ((spec & VSL_S_CLIENT) == 0)
......
......@@ -152,12 +152,3 @@ SPMCQ_NeedWorker(int running)
return spmcq_len() > spmcq_wrk_len_ratio(running - spmcq_datawaiter,
running);
}
bool
SPMCQ_StopWorker(int running)
{
if (running == 0)
return false;
return spmcq_len() < spmcq_wrk_len_ratio(running - spmcq_datawaiter - 1,
running);
}
......@@ -184,7 +184,6 @@ void SPMCQ_Enq(dataentry *ptr);
dataentry *SPMCQ_Deq(void);
void SPMCQ_Drain(void);
bool SPMCQ_NeedWorker(int running);
bool SPMCQ_StopWorker(int running);
#define spmcq_wait(what) \
do { \
......
......@@ -191,7 +191,9 @@ static void
wrk->id, err);
wrk->status = EXIT_FAILURE;
wrk->state = WRK_EXITED;
AZ(pthread_mutex_lock(&running_lock));
exited++;
AZ(pthread_mutex_unlock(&running_lock));
pthread_exit((void *) wrk);
}
......@@ -212,8 +214,7 @@ static void
if (amq_worker == NULL)
break;
if (!SPMCQ_StopWorker(running))
continue;
continue;
}
/* return space before sleeping */
......@@ -223,9 +224,8 @@ static void
}
/*
* Queue is empty or we should backoff
*
* wait until data are available, or quit is signaled.
* Queue is empty, wait until data are available, or quit is
* signaled.
*
* Grab the CV lock, which also constitutes an implicit memory
* barrier
......@@ -233,11 +233,9 @@ static void
AZ(pthread_mutex_lock(&spmcq_datawaiter_lock));
/*
* run is guaranteed to be fresh here
*
* also re-check the stop condition under the lock
*/
SPMCQ_Drain();
if (run && ((! entry) || SPMCQ_StopWorker(running))) {
if (run) {
wrk->waits++;
spmcq_datawaiter++;
wrk->state = WRK_WAITING;
......@@ -354,7 +352,9 @@ WRK_Restart(void)
CHECK_OBJ_NOTNULL(thread_data[i].wrk_data, WORKER_DATA_MAGIC);
wrk = thread_data[i].wrk_data;
if (wrk->state == WRK_EXITED) {
AZ(pthread_mutex_lock(&running_lock));
exited--;
AZ(pthread_mutex_unlock(&running_lock));
AZ(pthread_detach(thread_data[i].worker));
wrk->deqs = wrk->waits = wrk->sends = wrk->fails = wrk->reconnects
= 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