NPO-Accounting/conservancy_beancount Changeset - d9360f1ceafe

Changeset - d9360f1ceafe

Parent rev.

Child rev.

[Not reviewed]

0 1 0

Brett Smith - 4 years ago 2020-09-17 14:24:13
brettcsmith@brettcsmith.org

audit_report: Use concurrent.futures for parallelization.

This is basically a pure maintainability change: concurrent.futures is the
nicest API that's available in both Python 3.6 and 3.7, and our other tools
are using it.

1 file changed with 5 insertions and 5 deletions:

conservancy_beancount/tools/audit_report.py

0 comments (0 inline, 0 general)

conservancy_beancount/tools/audit_report.py

➞

Show inline comments

@@ ... / @@ -12,15 +12,15 @@ @@
 # GNU Affero General Public License for more details.
+#
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 import argparse
 import concurrent.futures as futmod
 import datetime
 import logging
 import multiprocessing
 import os
 import runpy
 import sys
 import tempfile
 from pathlib import Path
@@ ... / @@ -195,32 +195,32 @@ def main(arglist: Optional[Sequence[str]]=None, @@
     books = config.books_loader()
     if books is None:
         logger.critical("no books available to load")
         return os.EX_NOINPUT
-    with multiprocessing.Pool(args.jobs, maxtasksperchild=1) as pool:
+    with futmod.ProcessPoolExecutor(args.jobs) as pool:
         logger.debug("%s: process pool ready with %s workers", now_s(), args.jobs)
         fy_paths = books._iter_fy_books(fy.range(args.audit_year - 1, args.end_date))
-        check_results = pool.imap_unordered(bean_check, fy_paths)
+        check_results = pool.map(bean_check, fy_paths)
         if all(exitcode == 0 for exitcode in check_results):
             logger.debug("%s: bean-check passed", now_s())
         else:
             logger.log(
                 logging.WARNING if args.force else logging.ERROR,
                 "%s: bean-check failed",
                 now_s(),
+            )
             if not args.force:
                 return os.EX_DATAERR
         report_results = [
-            pool.apply_async(report_func, (arglist,), {'config': config})
+            pool.submit(report_func, arglist, config=config)
             for report_func, arglist in reports
+        ]
-        report_errors = [res.get() for res in report_results if res.get() != 0]
+        report_errors = [res.result() for res in report_results if res.result() != 0]
         if not report_errors:
             logger.debug("%s: all reports generated", now_s())
         else:
             logger.error("%s: %s reports generated errors", now_s(), len(report_errors))
             if not args.force:
                 return max(report_errors)

0 comments (0 inline, 0 general)