xbps-rindex(8): remove obsoletes mode (-r) is now multithreaded.

NEWS

@@ -1,7 +1,8 @@
 xbps-0.19 (???):
 
- * xbps-rindex(8): in clean mode (-c) is now multithreaded and
+ * xbps-rindex(8): clean (-c) and remove obsoletes mode (-r) are
-   will spawn a thread per core to process a fraction of rindex.
+   now multithreaded and will spawn a thread per core to speed up
+   the process considerably.
 
  * xbps-pkgdb(8): -a, --all mode is now multithreaded and will
    spawn a thread per core to process a fraction of pkgdb.

bin/xbps-rindex/remove-obsoletes.c

@@ -32,20 +32,84 @@
 #include <dirent.h>
 #include <libgen.h>
 #include <assert.h>
+#include <pthread.h>
 
 #include <xbps_api.h>
 #include "defs.h"
 
+struct thread_data {
+	pthread_t thread;
+	prop_array_t array;
+	struct xbps_rindex *ri;
+	unsigned int start;
+	unsigned int end;
+	int thread_num;
+};
+
+static void *
+cleaner_thread(void *arg)
+{
+	prop_dictionary_t pkgd;
+	struct thread_data *thd = arg;
+	const char *binpkg, *pkgver, *arch;
+	unsigned int i;
+	int rv;
+
+	/* process pkgs from start until end */
+	for (i = thd->start; i < thd->end; i++) {
+		prop_array_get_cstring_nocopy(thd->array, i, &binpkg);
+		pkgd = xbps_get_pkg_plist_from_binpkg(binpkg, "./props.plist");
+		if (pkgd == NULL) {
+			rv = remove_pkg(thd->ri->uri, arch, binpkg);
+			if (rv != 0) {
+				fprintf(stderr, "xbps-rindex: failed to remove "
+				    "package `%s': %s\n", binpkg,
+				    strerror(rv));
+				prop_object_release(pkgd);
+				break;
+			}
+			printf("Removed broken package `%s'.\n", binpkg);
+		}
+		prop_dictionary_get_cstring_nocopy(pkgd, "pkgver", &pkgver);
+		prop_dictionary_get_cstring_nocopy(pkgd, "architecture", &arch);
+		/* ignore pkgs from other archs */
+		if (!xbps_pkg_arch_match(thd->ri->xhp, arch, NULL)) {
+			prop_object_release(pkgd);
+			continue;
+		}
+		xbps_dbg_printf(thd->ri->xhp, "thread[%d] checking %s (%s)\n",
+		    thd->thread_num, pkgver, binpkg);
+		/*
+		 * If binpkg is not registered in index, remove binpkg.
+		 */
+		if (!xbps_rindex_get_pkg(thd->ri, pkgver)) {
+			rv = remove_pkg(thd->ri->uri, arch, binpkg);
+			if (rv != 0) {
+				fprintf(stderr, "xbps-rindex: failed to remove "
+				    "package `%s': %s\n", binpkg,
+				    strerror(rv));
+				prop_object_release(pkgd);
+			}
+			printf("Removed obsolete package `%s'.\n", binpkg);
+		}
+		prop_object_release(pkgd);
+	}
+
+	return NULL;
+}
+
 int
 remove_obsoletes(struct xbps_handle *xhp, const char *repodir)
 {
-	prop_dictionary_t pkgd, idx;
+	prop_dictionary_t idx;
+	prop_array_t array = NULL;
 	struct xbps_rindex ri;
+	struct thread_data *thd;
 	DIR *dirp;
 	struct dirent *dp;
-	const char *pkgver, *arch;
 	char *plist, *ext;
-	int rv = 0;
+	int i, maxthreads, rv = 0;
+	size_t slicecount, pkgcount;
 
 	if ((plist = xbps_pkg_index_plist(xhp, repodir)) == NULL)
 		return -1;
@@ -86,39 +150,39 @@ remove_obsoletes(struct xbps_handle *xhp, const char *repodir)
 			continue;
 		if (strcmp(ext, ".xbps"))
 			continue;
+		if (array == NULL)
+			array = prop_array_create();
 
-		pkgd = xbps_get_pkg_plist_from_binpkg(dp->d_name,
+		prop_array_add_cstring(array, dp->d_name);
-		    "./props.plist");
-		if (pkgd == NULL) {
-			rv = remove_pkg(repodir, arch, dp->d_name);
-			if (rv != 0) {
-				fprintf(stderr, "xbps-rindex: failed to remove "
-				    "package `%s': %s\n", dp->d_name,
-				    strerror(rv));
-				prop_object_release(pkgd);
-				break;
-			}
-			printf("Removed broken package `%s'.\n", dp->d_name);
-		}
-		prop_dictionary_get_cstring_nocopy(pkgd, "pkgver", &pkgver);
-		prop_dictionary_get_cstring_nocopy(pkgd, "architecture", &arch);
-		/*
-		 * If binpkg is not registered in index, remove binpkg.
-		 */
-		if (!xbps_rindex_get_pkg(&ri, pkgver)) {
-			rv = remove_pkg(repodir, arch, dp->d_name);
-			if (rv != 0) {
-				fprintf(stderr, "xbps-rindex: failed to remove "
-				    "package `%s': %s\n", dp->d_name,
-				    strerror(rv));
-				prop_object_release(pkgd);
-				break;
-			}
-			printf("Removed obsolete package `%s'.\n", dp->d_name);
-		}
-		prop_object_release(pkgd);
 	}
 	(void)closedir(dirp);
+
+	maxthreads = (int)sysconf(_SC_NPROCESSORS_ONLN);
+	thd = calloc(maxthreads, sizeof(*thd));
+
+	slicecount = prop_array_count(array) / maxthreads;
+	pkgcount = 0;
+
+	for (i = 0; i < maxthreads; i++) {
+		thd[i].thread_num = i;
+		thd[i].array = array;
+		thd[i].ri = &ri;
+		thd[i].start = pkgcount;
+		if (i + 1 >= maxthreads)
+			thd[i].end = prop_array_count(array);
+		else
+			thd[i].end = pkgcount + slicecount;
+		pthread_create(&thd[i].thread, NULL, cleaner_thread, &thd[i]);
+		pkgcount += slicecount;
+	}
+
+	/* wait for all threads */
+	for (i = 0; i < maxthreads; i++)
+		pthread_join(thd[i].thread, NULL);
+
+	free(thd);
+	prop_object_release(array);
 	prop_object_release(idx);
+
 	return rv;
 }