Hi -
This was a doozie. As the commit text hints, I haven't found a way of
testing this on an elfutils/tests snack-sized data set (except perhaps
by injecting long sleeps?), but the changes work on the larger
archives such as the debuginfod.systemtap.org set.
commit 34e67018914cf9ebbef07065965755b6554fd66e (HEAD -> master,
origin/fche/debuginfod-PR25394)
Author: Frank Ch. Eigler <f...@redhat.com>
Date: Sun Jan 19 20:33:32 2020 -0500
PR25394: debuginfod mutex between grooming and scanning
Extended the work-queue concept with "idlers" - other threads that
block on the work queue until it becomes empty (rather than normal
consumers that block on it until it becomes non-empty). Use this
facility for the groomer thread to avoid working at the same time as
the scanner threads. Use this for the fts traversal thread for
similar reasons. One user-visible effect: response to SIGUSR1 and
SIGUSR2 will wait until the work queue runs empty, but the man page
was unspecific so does not need changing.
It's not obvious how to test this with a tests/ dataset so small that
scanning takes negligible time, so the former races are very tight.
P.S. We also evaluated using sqlite level transactions to isolate the
scanner thread groups-of-operations from the groomer. These
experiments failed to produce a nominally concurrent debuginfod,
having triggered "database locked" type errors. So we remain
single-threaded (fully serialized) at the sqlite API level.
diff --git a/debuginfod/ChangeLog b/debuginfod/ChangeLog
index 35130b2a5d85..83b94a1108a6 100644
--- a/debuginfod/ChangeLog
+++ b/debuginfod/ChangeLog
@@ -1,3 +1,14 @@
+2020-01-19 Frank Ch. Eigler <f...@redhat.com>
+
+ * debuginfod.cxx (scanq): Rework to let groomer/fts threads
+ synchronize with an empty workqueue, and lock out workqueue
+ consumers.
+ (thread_groom): Adopt new scanq idle APIs to lock out scanners.
+ (thread_main_fts_source_paths): Adopt new scanq idler API to
+ avoid being restarted while scanners haven't even finished yet.
+ (thread_main_*): Increment thread_work_total metric only after
+ a work cycle is completed, not when it begins.
+
2020-01-18 Frank Ch. Eigler <f...@redhat.com>
* debuginfod.cxx (thread_main_scanner): Handle empty source_paths[].
diff --git a/debuginfod/debuginfod.cxx b/debuginfod/debuginfod.cxx
index bf680048d44d..b93e29ff117d 100644
--- a/debuginfod/debuginfod.cxx
+++ b/debuginfod/debuginfod.cxx
@@ -374,7 +374,7 @@ static struct argp argp =
static string db_path;
-static sqlite3 *db;
+static sqlite3 *db; // single connection, serialized across all our threads!
static unsigned verbose;
static volatile sig_atomic_t interrupted = 0;
static volatile sig_atomic_t sigusr1 = 0;
@@ -537,23 +537,25 @@ struct elfutils_exception: public reportable_exception
template <typename Payload>
class workq
{
- deque<Payload> q;
+ set<Payload> q; // eliminate duplicates
mutex mtx;
condition_variable cv;
bool dead;
+ unsigned idlers;
public:
- workq() { dead = false;}
+ workq() { dead = false; idlers = 0; }
~workq() {}
void push_back(const Payload& p)
{
unique_lock<mutex> lock(mtx);
- q.push_back(p);
+ q.insert (p);
set_metric("thread_work_pending","role","scan", q.size());
- cv.notify_one();
+ cv.notify_all();
}
+ // kill this workqueue, wake up all idlers / scanners
void nuke() {
unique_lock<mutex> lock(mtx);
// optional: q.clear();
@@ -561,28 +563,53 @@ class workq
cv.notify_all();
}
+ // block this scanner thread until there is work to do and no active
bool wait_front (Payload& p)
{
unique_lock<mutex> lock(mtx);
- while (q.size() == 0 && !dead)
+ while (!dead && (q.size() == 0 || idlers > 0))
cv.wait(lock);
if (dead)
return false;
else
{
- p = q.front();
- q.pop_front();
+ p = * q.begin();
+ q.erase (q.begin());
set_metric("thread_work_pending","role","scan", q.size());
+ if (q.size() == 0)
+ cv.notify_all(); // maybe wake up waiting idlers
return true;
}
}
+
+ // block this idler thread until there is no work to do
+ void wait_idle ()
+ {
+ unique_lock<mutex> lock(mtx);
+ cv.notify_all(); // maybe wake up waiting scanners
+ while (!dead && (q.size() != 0))
+ cv.wait(lock);
+ idlers ++;
+ }
+
+ void done_idle ()
+ {
+ unique_lock<mutex> lock(mtx);
+ idlers --;
+ cv.notify_all(); // maybe wake up waiting scanners, but probably not
(shutting down)
+ }
};
typedef struct stat stat_t;
typedef pair<string,stat_t> scan_payload;
+inline bool operator< (const scan_payload& a, const scan_payload& b)
+{
+ return a.first < b.first; // don't bother compare the stat fields
+}
static workq<scan_payload> scanq; // just a single one
-// producer: thread_main_fts_source_paths()
+// producer & idler: thread_main_fts_source_paths()
// consumer: thread_main_scanner()
+// idler: thread_main_groom()
@@ -2309,7 +2336,6 @@ thread_main_scanner (void* arg)
bool gotone = scanq.wait_front(p);
add_metric("thread_busy", "role", "scan", 1);
if (! gotone) continue; // or break
- inc_metric("thread_work_total", "role","scan");
try
{
@@ -2347,6 +2373,8 @@ thread_main_scanner (void* arg)
{
e.report(cerr);
}
+
+ inc_metric("thread_work_total", "role","scan");
}
add_metric("thread_busy", "role", "scan", -1);
@@ -2448,41 +2476,44 @@ thread_main_fts_source_paths (void* arg)
{
(void) arg; // ignore; we operate on global data
- unsigned rescan_timer = 0;
sig_atomic_t forced_rescan_count = 0;
- set_metric("thread_timer_max", "role","traverse", rescan_s);
set_metric("thread_tid", "role","traverse", tid());
add_metric("thread_count", "role", "traverse", 1);
+
+ time_t last_rescan = 0;
+
while (! interrupted)
{
- set_metric("thread_timer", "role","traverse", rescan_timer);
- // set_metric("thread_forced_total", "role","traverse",
forced_rescan_count);
- if (rescan_s && rescan_timer > rescan_s)
- rescan_timer = 0;
+ sleep (1);
+ scanq.wait_idle(); // don't start a new traversal while scanners haven't
finished the job
+ scanq.done_idle(); // release the hounds
+ if (interrupted) break;
+
+ time_t now = time(NULL);
+ bool rescan_now = false;
+ if (last_rescan == 0) // at least one initial rescan is documented even
for -t0
+ rescan_now = true;
+ if (rescan_s > 0 && now > last_rescan + rescan_s)
+ rescan_now = true;
if (sigusr1 != forced_rescan_count)
{
forced_rescan_count = sigusr1;
- rescan_timer = 0;
+ rescan_now = true;
}
- if (rescan_timer == 0)
+ if (rescan_now)
try
{
set_metric("thread_busy", "role","traverse", 1);
- inc_metric("thread_work_total", "role","traverse");
scan_source_paths();
+ inc_metric("thread_work_total", "role","traverse");
set_metric("thread_busy", "role","traverse", 0);
}
catch (const reportable_exception& e)
{
e.report(cerr);
}
- sleep (1);
- rescan_timer ++;
}
- // wake up any blocked scanning threads so they can check $interrupted and
kill themselves
- scanq.nuke();
-
return 0;
}
@@ -2589,36 +2620,44 @@ void groom()
static void*
thread_main_groom (void* /*arg*/)
{
- unsigned groom_timer = 0;
sig_atomic_t forced_groom_count = 0;
- set_metric("thread_timer_max", "role", "groom", groom_s);
set_metric("thread_tid", "role", "groom", tid());
add_metric("thread_count", "role", "groom", 1);
- while (! interrupted)
+
+ time_t last_groom = 0;
+
+ while (1)
{
- set_metric("thread_timer", "role", "groom", groom_timer);
- // set_metric("thread_forced_total", "role", "groom",
forced_groom_count);
- if (groom_s && groom_timer > groom_s)
- groom_timer = 0;
+ sleep (1);
+ scanq.wait_idle(); // PR25394: block scanners during grooming!
+ if (interrupted) break;
+
+ time_t now = time(NULL);
+ bool groom_now = false;
+ if (last_groom == 0) // at least one initial groom is documented even
for -g0
+ groom_now = true;
+ if (groom_s > 0 && now > last_groom + groom_s)
+ groom_now = true;
if (sigusr2 != forced_groom_count)
{
forced_groom_count = sigusr2;
- groom_timer = 0;
+ groom_now = true;
}
- if (groom_timer == 0)
+ if (groom_now)
try
{
set_metric("thread_busy", "role", "groom", 1);
- inc_metric("thread_work_total", "role", "groom");
groom ();
+ last_groom = time(NULL); // NB: now was before grooming
+ inc_metric("thread_work_total", "role", "groom");
set_metric("thread_busy", "role", "groom", 0);
}
catch (const sqlite_exception& e)
{
obatched(cerr) << e.message << endl;
}
- sleep (1);
- groom_timer ++;
+
+ scanq.done_idle();
}
return 0;
@@ -2860,26 +2899,27 @@ main (int argc, char *argv[])
if (du && du[0] != '\0') // set to non-empty string?
obatched(clog) << "upstream debuginfod servers: " << du << endl;
- vector<pthread_t> scanner_threads;
- pthread_t groom_thread;
+ vector<pthread_t> all_threads;
- rc = pthread_create (& groom_thread, NULL, thread_main_groom, NULL);
+ pthread_t pt;
+ rc = pthread_create (& pt, NULL, thread_main_groom, NULL);
if (rc < 0)
error (0, 0, "warning: cannot spawn thread (%d) to groom database\n", rc);
+ else
+ all_threads.push_back(pt);
if (scan_files || scan_archives.size() > 0)
{
- pthread_t pt;
pthread_create (& pt, NULL, thread_main_fts_source_paths, NULL);
if (rc < 0)
error (0, 0, "warning: cannot spawn thread (%d) to traverse source
paths\n", rc);
- scanner_threads.push_back(pt);
+ all_threads.push_back(pt);
for (unsigned i=0; i<concurrency; i++)
{
pthread_create (& pt, NULL, thread_main_scanner, NULL);
if (rc < 0)
error (0, 0, "warning: cannot spawn thread (%d) to scan source
files / archives\n", rc);
- scanner_threads.push_back(pt);
+ all_threads.push_back(pt);
}
}
@@ -2887,15 +2927,15 @@ main (int argc, char *argv[])
set_metric("ready", 1);
while (! interrupted)
pause ();
+ scanq.nuke(); // wake up any remaining scanq-related threads, let them die
set_metric("ready", 0);
if (verbose)
obatched(clog) << "stopping" << endl;
/* Join all our threads. */
- for (auto&& it : scanner_threads)
+ for (auto&& it : all_threads)
pthread_join (it, NULL);
- pthread_join (groom_thread, NULL);
/* Stop all the web service threads. */
if (d4) MHD_stop_daemon (d4);
diff --git a/tests/ChangeLog b/tests/ChangeLog
index 8936e410b3d6..8b03204182af 100644
--- a/tests/ChangeLog
+++ b/tests/ChangeLog
@@ -1,3 +1,7 @@
+2020-01-19 Frank Ch. Eigler <f...@redhat.com>
+
+ * run-debuginfod-find.sh: Check for proper groom completion count.
+
2020-01-18 Frank Ch. Eigler <f...@redhat.com>
* run-debuginfod-find.sh: Test empty source_paths[].
diff --git a/tests/run-debuginfod-find.sh b/tests/run-debuginfod-find.sh
index 2a3d591b4e77..33bbd1a6dafe 100755
--- a/tests/run-debuginfod-find.sh
+++ b/tests/run-debuginfod-find.sh
@@ -257,6 +257,7 @@ RPM_BUILDID=d44d42cbd7d915bc938c81333a21e355a6022fb7 # in
rhel6/ subdir, for a l
rm -r R/debuginfod-rpms/rhel6/*
kill -USR2 $PID1 # groom cycle
# Expect 3 rpms to be deleted by the groom
+wait_ready $PORT1 'thread_work_total{role="groom"}' 1
wait_ready $PORT1 'groom{statistic="file d/e"}' 3
rm -rf $DEBUGINFOD_CACHE_PATH # clean it from previous tests
