oysteikt/mpd
0
0
Fork 0
Code Issues Pull Requests Releases Activity

db/upnp/WorkQueue: use array instead of std::list

Browse Source 
Reduce bloat further.
This commit is contained in:
Max Kellermann
2014-01-14 11:23:13 +01:00
parent 73fd98b82e
commit c6d8f6da70
1 changed files with 19 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/db/upnp/WorkQueue.hxx
View File

@@ -59,7 +59,8 @@ class WorkQueue {
unsigned n_workers_exited; unsigned n_workers_exited;
bool ok; bool ok;
std::list<pthread_t> threads; unsigned n_threads;
pthread_t *threads;
// Synchronization // Synchronization
std::queue<T> queue; std::queue<T> queue;
@@ -81,6 +82,7 @@ public:
:name(_name), high(hi), low(lo), :name(_name), high(hi), low(lo),
n_workers_exited(0), n_workers_exited(0),
ok(true), ok(true),
n_threads(0), threads(nullptr),
n_clients_waiting(0), n_workers_waiting(0) n_clients_waiting(0), n_workers_waiting(0)
{ {
} }
@@ -101,15 +103,19 @@ public:
{ {
const ScopeLock protect(mutex); const ScopeLock protect(mutex);
assert(nworkers > 0);
assert(n_threads == 0);
assert(threads == nullptr);
threads = new pthread_t[n_threads];
for (int i = 0; i < nworkers; i++) { for (int i = 0; i < nworkers; i++) {
int err; int err;
pthread_t thr; if ((err = pthread_create(&threads[n_threads++], 0, workproc, arg))) {
if ((err = pthread_create(&thr, 0, workproc, arg))) {
LOGERR(("WorkQueue:%s: pthread_create failed, err %d\n", LOGERR(("WorkQueue:%s: pthread_create failed, err %d\n",
name.c_str(), err)); name.c_str(), err));
return false; return false;
} }
threads.push_back(thr);
} }
return true; return true;
} }
@@ -178,7 +184,7 @@ public:
// We're done when the queue is empty AND all workers are back // We're done when the queue is empty AND all workers are back
// waiting for a task. // waiting for a task.
while (IsOK() && (queue.size() > 0 || while (IsOK() && (queue.size() > 0 ||
n_workers_waiting != threads.size())) { n_workers_waiting != n_threads)) {
n_clients_waiting++; n_clients_waiting++;
client_cond.wait(mutex); client_cond.wait(mutex);
n_clients_waiting--; n_clients_waiting--;
@@ -197,13 +203,9 @@ public:
{ {
const ScopeLock protect(mutex); const ScopeLock protect(mutex);
if (threads.empty())
// Already called ?
return;
// Wait for all worker threads to have called workerExit() // Wait for all worker threads to have called workerExit()
ok = false; ok = false;
while (n_workers_exited < threads.size()) { while (n_workers_exited < n_threads) {
worker_cond.broadcast(); worker_cond.broadcast();
n_clients_waiting++; n_clients_waiting++;
client_cond.wait(mutex); client_cond.wait(mutex);
@@ -212,13 +214,15 @@ public:
// Perform the thread joins and compute overall status // Perform the thread joins and compute overall status
// Workers return (void*)1 if ok // Workers return (void*)1 if ok
while (!threads.empty()) { for (unsigned i = 0; i < n_threads; ++i) {
void *status; void *status;
auto thread = threads.front(); pthread_join(threads[i], &status);
pthread_join(thread, &status);
threads.pop_front();
} }
delete[] threads;
threads = nullptr;
n_threads = 0;
// Reset to start state. // Reset to start state.
n_workers_exited = n_clients_waiting = n_workers_waiting = 0; n_workers_exited = n_clients_waiting = n_workers_waiting = 0;
ok = true; ok = true;
@@ -283,7 +287,7 @@ public:
private: private:
bool IsOK() bool IsOK()
{ {
return ok && n_workers_exited == 0 && !threads.empty(); return ok && n_workers_exited == 0 && n_threads > 0;
} }
}; };