Run binary package generation and file classification in parallel threads

build/Makefile.am

@@ -2,6 +2,7 @@
 
 include $(top_srcdir)/rpm.am
 AM_CFLAGS = @RPMCFLAGS@
+AM_CFLAGS += @OPENMP_CFLAGS@
 
 AM_CPPFLAGS =  -I$(top_builddir) -I$(top_srcdir) -I$(top_builddir)/include/
 AM_CPPFLAGS += @WITH_BEECRYPT_INCLUDE@

build/pack.c

@@ -622,111 +622,147 @@ static rpmRC checkPackages(char *pkgcheck)
     return RPMRC_OK;
 }
 
-static rpmRC packageBinary(rpmSpec spec, Package pkg, const char *cookie, int cheating, char** filename)
+static rpmRC checkPackageSet(Package pkgs)
 {
-	const char *errorString;
-	rpmRC rc = RPMRC_OK;
+    char *pkglist = NULL;
+    char *pkgcheck_set = NULL;
 
-	if (pkg->fileList == NULL)
-	    return rc;
+    for (Package pkg = pkgs; pkg != NULL; pkg = pkg->next) {
+	if (pkg->filename)
+	    rstrscat(&pkglist, pkg->filename, " ", NULL);
+    }
 
-	if ((rc = processScriptFiles(spec, pkg)))
-	    return rc;
-
-	if (cookie) {
-	    headerPutString(pkg->header, RPMTAG_COOKIE, cookie);
-	}
+    pkgcheck_set = rpmExpand("%{?_build_pkgcheck_set} ", pkglist,  NULL);
 
-	/* Copy changelog from src rpm */
-	headerCopyTags(spec->packages->header, pkg->header, copyTags);
-
-	headerPutString(pkg->header, RPMTAG_RPMVERSION, VERSION);
-	headerPutString(pkg->header, RPMTAG_BUILDHOST, spec->buildHost);
-	headerPutUint32(pkg->header, RPMTAG_BUILDTIME, &(spec->buildTime), 1);
+    /* run only if _build_pkgcheck_set is defined */
+    if (pkgcheck_set[0] != ' ')
+	checkPackages(pkgcheck_set);
 
-	if (spec->sourcePkgId != NULL) {
-	    headerPutBin(pkg->header, RPMTAG_SOURCEPKGID, spec->sourcePkgId,16);
-	}
+    free(pkgcheck_set);
+    free(pkglist);
+    return RPMRC_OK;
+}
 
-	if (cheating) {
-	    (void) rpmlibNeedsFeature(pkg, "ShortCircuited", "4.9.0-1");
-	}
-
-	{   char *binFormat = rpmGetPath("%{_rpmfilename}", NULL);
-	    char *binRpm, *binDir;
-	    binRpm = headerFormat(pkg->header, binFormat, &errorString);
-	    free(binFormat);
-	    if (binRpm == NULL) {
-		rpmlog(RPMLOG_ERR, _("Could not generate output "
-		     "filename for package %s: %s\n"), 
-		     headerGetString(pkg->header, RPMTAG_NAME), errorString);
-		return RPMRC_FAIL;
-	    }
-	    *filename = rpmGetPath("%{_rpmdir}/", binRpm, NULL);
-	    if ((binDir = strchr(binRpm, '/')) != NULL) {
-		struct stat st;
-		char *dn;
-		*binDir = '\0';
-		dn = rpmGetPath("%{_rpmdir}/", binRpm, NULL);
-		if (stat(dn, &st) < 0) {
-		    switch (errno) {
-		    case  ENOENT:
-			if (mkdir(dn, 0755) == 0)
-			    break;
-		    default:
-			rpmlog(RPMLOG_ERR,_("cannot create %s: %s\n"),
-			    dn, strerror(errno));
-			break;
-		    }
-		}
-		free(dn);
+/* watchout, argument is modified */
+static rpmRC ensureDir(char *binRpm)
+{
+    rpmRC rc = RPMRC_OK;
+    char *binDir = strchr(binRpm, '/');
+    char *dn = NULL;
+    if (binDir) {
+	struct stat st;
+	*binDir = '\0';
+	dn = rpmGetPath("%{_rpmdir}/", binRpm, NULL);
+	if (stat(dn, &st) < 0) {
+	    switch (errno) {
+	    case ENOENT:
+		if (mkdir(dn, 0755) == 0 || errno == EEXIST)
+		    break;
+	    default:
+		rpmlog(RPMLOG_ERR,_("cannot create %s: %s\n"),
+			dn, strerror(errno));
+		rc = RPMRC_FAIL;
+		break;
 	    }
-	    free(binRpm);
 	}
+    }
+    free(dn);
+    return rc;
+}
 
-	rc = writeRPM(pkg, NULL, *filename, NULL, spec->buildTime, spec->buildHost);
-	if (rc == RPMRC_OK) {
-	    /* Do check each written package if enabled */
-	    char *pkgcheck = rpmExpand("%{?_build_pkgcheck} ", *filename, NULL);
-	    if (pkgcheck[0] != ' ') {
-		rc = checkPackages(pkgcheck);
-	    }
-	    free(pkgcheck);
-	}
+static rpmRC getPkgFilename(Header h, char **filename)
+{
+    const char *errorString;
+    char *binFormat = rpmGetPath("%{_rpmfilename}", NULL);
+    char *binRpm = headerFormat(h, binFormat, &errorString);
+    rpmRC rc = RPMRC_FAIL;
+
+    if (binRpm == NULL) {
+	rpmlog(RPMLOG_ERR, _("Could not generate output "
+	     "filename for package %s: %s\n"),
+	     headerGetString(h, RPMTAG_NAME), errorString);
+    } else {
+	*filename = rpmGetPath("%{_rpmdir}/", binRpm, NULL);
+	rc = ensureDir(binRpm);
+    }
+    free(binFormat);
+    free(binRpm);
+    return rc;
+}
+
+static rpmRC packageBinary(rpmSpec spec, Package pkg, const char *cookie, int cheating, char** filename)
+{
+    rpmRC rc = RPMRC_OK;
+
+    if (pkg->fileList == NULL)
+	return rc;
+
+    if ((rc = processScriptFiles(spec, pkg)))
 	return rc;
+
+    if (cookie) {
+	headerPutString(pkg->header, RPMTAG_COOKIE, cookie);
+    }
+
+    /* Copy changelog from src rpm */
+    headerCopyTags(spec->sourcePackage->header, pkg->header, copyTags);
+
+    headerPutString(pkg->header, RPMTAG_RPMVERSION, VERSION);
+    headerPutString(pkg->header, RPMTAG_BUILDHOST, spec->buildHost);
+    headerPutUint32(pkg->header, RPMTAG_BUILDTIME, &(spec->buildTime), 1);
+
+    if (spec->sourcePkgId != NULL) {
+	headerPutBin(pkg->header, RPMTAG_SOURCEPKGID, spec->sourcePkgId,16);
+    }
+
+    if (cheating) {
+	(void) rpmlibNeedsFeature(pkg, "ShortCircuited", "4.9.0-1");
+    }
+
+    if ((rc = getPkgFilename(pkg->header, filename)))
+	return rc;
+
+    rc = writeRPM(pkg, NULL, *filename, NULL, spec->buildTime, spec->buildHost);
+    if (rc == RPMRC_OK) {
+	/* Do check each written package if enabled */
+	char *pkgcheck = rpmExpand("%{?_build_pkgcheck} ", *filename, NULL);
+	if (pkgcheck[0] != ' ') {
+	    rc = checkPackages(pkgcheck);
+	}
+	free(pkgcheck);
+    }
+    return rc;
 }
 
 rpmRC packageBinaries(rpmSpec spec, const char *cookie, int cheating)
 {
-    rpmRC rc;
+    rpmRC rc = RPMRC_OK;
     Package pkg;
-    char *pkglist = NULL;
 
+    /* Run binary creation in parallel */
+    #pragma omp parallel
+    #pragma omp single
     for (pkg = spec->packages; pkg != NULL; pkg = pkg->next) {
-	char *fn = NULL;
-	rc = packageBinary(spec, pkg, cookie, cheating, &fn);
-	if (rc == RPMRC_OK) {
-	    rstrcat(&pkglist, fn);
-	    rstrcat(&pkglist, " ");
-	}
-	free(fn);
-	if (rc != RPMRC_OK) {
-	    pkglist = _free(pkglist);
-	    return rc;
+	#pragma omp task
+	{
+	pkg->rc = packageBinary(spec, pkg, cookie, cheating, &pkg->filename);
+	rpmlog(RPMLOG_DEBUG,
+		_("Finished binary package job, result %d, filename %s\n"),
+		pkg->rc, pkg->filename);
+	if (pkg->rc) {
+	    #pragma omp critical
+	    rc = pkg->rc;
 	}
+	} /* omp task */
+	if (rc)
+	    break;
     }
 
     /* Now check the package set if enabled */
-    if (pkglist != NULL) {
-	char *pkgcheck_set = rpmExpand("%{?_build_pkgcheck_set} ", pkglist,  NULL);
-	if (pkgcheck_set[0] != ' ') {	/* run only if _build_pkgcheck_set is defined */
-	    checkPackages(pkgcheck_set);
-	}
-	free(pkgcheck_set);
-	pkglist = _free(pkglist);
-    }
-
-    return RPMRC_OK;
+    if (rc == RPMRC_OK)
+	checkPackageSet(spec->packages);
+
+    return rc;
 }
 
 rpmRC packageSources(rpmSpec spec, char **cookie)
@@ -742,19 +778,18 @@ rpmRC packageSources(rpmSpec spec, char **cookie)
     headerPutUint32(sourcePkg->header, RPMTAG_SOURCEPACKAGE, &one, 1);
 
     /* XXX this should be %_srpmdir */
-    {	char *fn = rpmGetPath("%{_srcrpmdir}/", spec->sourceRpmName,NULL);
-	char *pkgcheck = rpmExpand("%{?_build_pkgcheck_srpm} ", fn, NULL);
+    {	sourcePkg->filename = rpmGetPath("%{_srcrpmdir}/", spec->sourceRpmName,NULL);
+	char *pkgcheck = rpmExpand("%{?_build_pkgcheck_srpm} ", sourcePkg->filename, NULL);
 
 	spec->sourcePkgId = NULL;
-	rc = writeRPM(sourcePkg, &spec->sourcePkgId, fn, cookie, spec->buildTime, spec->buildHost);
+	rc = writeRPM(sourcePkg, &spec->sourcePkgId, sourcePkg->filename, cookie, spec->buildTime, spec->buildHost);
 
 	/* Do check SRPM package if enabled */
 	if (rc == RPMRC_OK && pkgcheck[0] != ' ') {
 	    rc = checkPackages(pkgcheck);
 	}
 
 	free(pkgcheck);
-	free(fn);
     }
     return rc;
 }

build/parseSpec.c

@@ -6,6 +6,7 @@
 #include "system.h"
 
 #include <errno.h>
+#include <omp.h>
 #ifdef HAVE_ICONV
 #include <iconv.h>
 #endif
@@ -1003,6 +1004,12 @@ static rpmSpec parseSpec(const char *specFile, rpmSpecFlags flags,
 	}
     }
 
+    /* Set number of OMP threads centrally */
+    int ncpus = rpmExpandNumeric("%{?_smp_build_ncpus}");
+    if (ncpus <= 0)
+	ncpus = 1;
+    omp_set_num_threads(ncpus);
+
     if (spec->clean == NULL) {
 	char *body = rpmExpand("%{?buildroot: %{__rm} -rf %{buildroot}}", NULL);
 	spec->clean = newStringBuf();

build/rpmbuild_internal.h

@@ -143,6 +143,9 @@ struct Package_s {
     fileRenameHash fileRenameMap;
     ARGV_t policyList;
 
+    char *filename;
+    rpmRC rc;
+
     Package next;
 };
 

build/rpmfc.c

@@ -677,6 +677,7 @@ static void rpmfcAttributes(rpmfc fc, int ix, const char *ftype, const char *ful
 	/* Add attributes on libmagic type & path pattern matches */
 	if (matches(&(*attr)->incl, ftype, path, is_executable)) {
 	    argvAddTokens(&fc->fattrs[ix], (*attr)->name);
+	    #pragma omp critical(fahash)
 	    fattrHashAddEntry(fc->fahash, attr-fc->atypes, ix);
 	}
     }
@@ -1063,7 +1064,7 @@ static int initAttrs(rpmfc fc)
 rpmRC rpmfcClassify(rpmfc fc, ARGV_t argv, rpm_mode_t * fmode)
 {
     int msflags = MAGIC_CHECK | MAGIC_COMPRESS | MAGIC_NO_CHECK_TOKENS;
-    magic_t ms = NULL;
+    int nerrors = 0;
     rpmRC rc = RPMRC_FAIL;
 
     if (fc == NULL) {
@@ -1094,18 +1095,23 @@ rpmRC rpmfcClassify(rpmfc fc, ARGV_t argv, rpm_mode_t * fmode)
     /* Build (sorted) file class dictionary. */
     fc->cdict = rpmstrPoolCreate();
 
-    ms = magic_open(msflags);
+    #pragma omp parallel
+    {
+    /* libmagic is not thread-safe, each thread needs to a private handle */
+    magic_t ms = magic_open(msflags);
+
     if (ms == NULL) {
 	rpmlog(RPMLOG_ERR, _("magic_open(0x%x) failed: %s\n"),
 		msflags, strerror(errno));
-	goto exit;
+	#pragma omp cancel parallel
     }
 
     if (magic_load(ms, NULL) == -1) {
 	rpmlog(RPMLOG_ERR, _("magic_load failed: %s\n"), magic_error(ms));
-	goto exit;
+	#pragma omp cancel parallel
     }
 
+    #pragma omp for reduction(+:nerrors)
     for (int ix = 0; ix < fc->nfiles; ix++) {
 	rpmsid ftypeId;
 	const char * ftype;
@@ -1148,7 +1154,8 @@ rpmRC rpmfcClassify(rpmfc fc, ARGV_t argv, rpm_mode_t * fmode)
 		       s, mode, magic_error(ms));
 		/* only executable files are critical to dep extraction */
 		if (is_executable) {
-		    goto exit;
+		    nerrors++;
+		    #pragma omp cancel for
 		}
 		/* unrecognized non-executables get treated as "data" */
 		ftype = "data";
@@ -1171,21 +1178,28 @@ rpmRC rpmfcClassify(rpmfc fc, ARGV_t argv, rpm_mode_t * fmode)
 	/* Add to file class dictionary and index array */
 	if (fcolor != RPMFC_WHITE && (fcolor & RPMFC_INCLUDE)) {
 	    ftypeId = rpmstrPoolId(fc->cdict, ftype, 1);
+	    #pragma omp atomic
 	    fc->fknown++;
 	} else {
 	    ftypeId = rpmstrPoolId(fc->cdict, "", 1);
+	    #pragma omp atomic
 	    fc->fwhite++;
 	}
 	/* Pool id's start from 1, for headers we want it from 0 */
 	fc->fcdictx[ix] = ftypeId - 1;
     }
-    rc = RPMRC_OK;
+
+    if (ms != NULL)
+	magic_close(ms);
+
+    } /* omp parallel */
+
+    if (nerrors == 0)
+	rc = RPMRC_OK;
 
 exit:
     /* No more additions after this, freeze pool to minimize memory use */
     rpmstrPoolFreeze(fc->cdict, 0);
-    if (ms != NULL)
-	magic_close(ms);
 
     return rc;
 }

build/spec.c

@@ -146,6 +146,7 @@ Package freePackage(Package pkg)
 {
     if (pkg == NULL) return NULL;
 
+    pkg->filename = _free(pkg->filename);
     pkg->preInFile = _free(pkg->preInFile);
     pkg->postInFile = _free(pkg->postInFile);
     pkg->preUnFile = _free(pkg->preUnFile);

configure.ac

@@ -12,6 +12,7 @@ m4_ifdef([AM_SILENT_RULES], [AM_SILENT_RULES])
 AC_CONFIG_TESTDIR(tests)
 
 AC_USE_SYSTEM_EXTENSIONS
+AC_OPENMP
 
 AC_DISABLE_STATIC