Parallelize calc_rank_pdfs

gstlal-inspiral/python/dags/layers/inspiral.py

@@ -578,7 +578,8 @@ def marginalize_layer(config, dag, prior_cache, dist_stat_cache):
 
 def calc_pdf_layer(config, dag, dist_stat_cache):
 	# FIXME: expose this in configuration
-	num_cores = 4
+	num_cores = config.rank.calc_rank_cores if config.rank.calc_rank_cores else 4
+	num_jobs = config.rank.calc_rank_jobs if config.rank.calc_rank_jobs else 1
 
 	layer = Layer(
 		"gstlal_inspiral_calc_rank_pdfs",
@@ -591,27 +592,36 @@ def calc_pdf_layer(config, dag, dist_stat_cache):
 		transfer_files=config.condor.transfer_files,
 	)
 
+	if num_jobs == 1:
+		svd_bins=config.svd.bins
+	else:
+		svd_bins = []
+		for svd_bin in config.svd.bins:
+			for i in range(num_jobs):
+				svd_bins.append(svd_bin + "_" + str(i))
+
 	pdf_cache = DataCache.generate(
 		DataType.DIST_STAT_PDFS,
 		config.all_ifos,
 		config.span,
-		svd_bins=config.svd.bins,
+		svd_bins=svd_bins,
 		root="rank",
 	)
 
 	dist_stats = dist_stat_cache.groupby("bin")
 	for svd_bin, pdfs in pdf_cache.groupby("bin").items():
-		layer += Node(
-			arguments = [
-				Option("ranking-stat-samples", config.rank.ranking_stat_samples),
-				Option("num-cores", num_cores),
-			],
-			inputs = [
-				Argument("mass-model", config.prior.mass_model, track=False, suppress=True),
-				Argument("dist-stats", dist_stats[svd_bin].files),
-			],
-			outputs = Option("output", pdfs.files)
-		)
+		for pdf in pdfs.files:
+			layer += Node(
+				arguments = [
+					Option("ranking-stat-samples", config.rank.ranking_stat_samples),
+					Option("num-cores", num_cores),
+				],
+				inputs = [
+					Argument("mass-model", config.prior.mass_model, track=False, suppress=True),
+					Argument("dist-stats", dist_stats[svd_bin].files),
+				],
+				outputs = Option("output", pdf)
+			)
 
 	dag.attach(layer)
 	return pdf_cache