xbps-pkgdb: checking all pkgs with -a/--all is now multithreaded.

NEWS

@@ -1,5 +1,8 @@
 xbps-0.19 (???):
 
+ * xbps-pkgdb(8): -a, --all mode is now multithreaded and will
+   spawn a thread per core to process a fraction of pkgdb.
+
  * xbps-create(8): hardlinks can be added and archived correctly.
 
  * xbps-rindex(8): in clean mode (-c), entries with unmatched hashes

bin/xbps-pkgdb/Makefile

@@ -6,6 +6,7 @@ OBJS =	main.o check.o check_pkg_files.o
 OBJS +=	check_pkg_rundeps.o
 OBJS +=	check_pkg_symlinks.o check_pkg_unneeded.o
 OBJS +=	convert.o
+CFLAGS += -pthread
 
 MAN = $(BIN).8
 

bin/xbps-pkgdb/check.c

@@ -23,66 +23,94 @@
  * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  */
 
+#include <sys/param.h>
 #include <stdio.h>
 #include <stdbool.h>
 #include <stdlib.h>
 #include <string.h>
 #include <errno.h>
 #include <unistd.h>
-#include <sys/param.h>
+#include <pthread.h>
+#include <assert.h>
 
 #include <xbps_api.h>
 #include "defs.h"
 
-struct checkpkg {
+struct thread_data {
-	size_t totalpkgs;
+	pthread_t thread;
-	size_t npkgs;
+	struct xbps_handle *xhp;
-	size_t nbrokenpkgs;
+	unsigned int start;
+	unsigned int end;
+	int thread_num;
 };
 
-static int
+static void *
-cb_pkg_integrity(struct xbps_handle *xhp,
+pkgdb_thread_worker(void *arg)
-		 prop_object_t obj,
-		 void *arg,
-		 bool *done)
 {
-	struct checkpkg *cpkg = arg;
+	prop_dictionary_t pkgd;
-	const char *pkgname, *version;
+	struct thread_data *thd = arg;
+	const char *pkgname, *pkgver;
+	unsigned int i;
+	int rv;
 
-	(void)done;
+	/* process pkgs from start until end */
+	for (i = thd->start; i < thd->end; i++) {
+		pkgd = prop_array_get(thd->xhp->pkgdb, i);
+		prop_dictionary_get_cstring_nocopy(pkgd, "pkgname", &pkgname);
+		prop_dictionary_get_cstring_nocopy(pkgd, "pkgver", &pkgver);
+		if (thd->xhp->flags & XBPS_FLAG_VERBOSE)
+			printf("Checking %s ...\n", pkgver);
 
-	prop_dictionary_get_cstring_nocopy(obj, "pkgname", &pkgname);
+		rv = check_pkg_integrity(thd->xhp, pkgd, pkgname);
-	prop_dictionary_get_cstring_nocopy(obj, "version", &version);
+		if (rv != 0)
+			fprintf(stderr, "pkgdb[%d] failed for %s: %s\n",
+			    thd->thread_num, pkgver, strerror(rv));
+	}
 
-	printf("[%zu/%zu] checking %s-%s ...\n",
+	return NULL;
-	    cpkg->npkgs, cpkg->totalpkgs, pkgname, version);
-
-	if (check_pkg_integrity(xhp, obj, pkgname) != 0)
-		cpkg->nbrokenpkgs++;
-
-	cpkg->npkgs++;
-	return 0;
 }
 
 int
 check_pkg_integrity_all(struct xbps_handle *xhp)
 {
-	struct checkpkg cpkg;
+	struct thread_data *thd;
-	int rv;
+	unsigned int slicecount, pkgcount;
+	int rv, maxthreads, i;
 
-	memset(&cpkg, 0, sizeof(cpkg));
 	/* force an update to get total pkg count */
 	(void)xbps_pkgdb_update(xhp, false);
-	cpkg.totalpkgs = prop_array_count(xhp->pkgdb);
 
-	(void)xbps_pkgdb_foreach_cb(xhp, cb_pkg_integrity, &cpkg);
+	maxthreads = (int)sysconf(_SC_NPROCESSORS_ONLN);
+	thd = calloc(maxthreads, sizeof(*thd));
+	assert(thd);
+
+	slicecount = prop_array_count(xhp->pkgdb) / maxthreads;
+	pkgcount = 0;
+
+	for (i = 0; i < maxthreads; i++) {
+		thd[i].thread_num = i;
+		thd[i].xhp = xhp;
+		thd[i].start = pkgcount;
+		if (i + 1 >= maxthreads)
+			thd[i].end = prop_array_count(xhp->pkgdb);
+		else
+			thd[i].end = pkgcount + slicecount;
+		pthread_create(&thd[i].thread, NULL,
+		    pkgdb_thread_worker, &thd[i]);
+		pkgcount += slicecount;
+	}
+
+	/* wait for all threads */
+	for (i = 0; i < maxthreads; i++)
+		pthread_join(thd[i].thread, NULL);
+
+	free(thd);
+
 	if ((rv = xbps_pkgdb_update(xhp, true)) != 0) {
 		xbps_error_printf("failed to write pkgdb: %s\n",
 		    strerror(rv));
 		return rv;
 	}
-	printf("%zu package%s processed: %zu broken.\n", cpkg.npkgs,
-	    cpkg.npkgs == 1 ? "" : "s", cpkg.nbrokenpkgs);
 	return 0;
 }