0
\$\begingroup\$

This week I dived into the issue of speeding up my program by either threading or multiprocessing. I did not understand well the difference between the two, but that became quite clear when I got the results. Below program is processing 12 text data files that has a Pandas DataFrame as output.

It compares the execution times for a simple sequential method; a multiprocessing pool method; a multiprocessing process method; and threading. My computer has 8 cpu's.

What is obvious is that threading in this case has no benefit over simple sequential processing (it took both just over 1000 s [17 minutes]), but multiprocessing pool and multiprocessing process does give an almost equivalent speed up of 3.4 times running the process in about 303 s [5 minutes].

''' program to run extended QC parsing module
    using various options: sequential, multiprocessing, threading
'''
import multiprocessing as mp
import threading
import queue
from pathlib import Path
from vp_extended_qc import ExtendedQc
from Utils.plogger import Logger, timed
from pprint import pprint

# Logging setup
logformat = '%(asctime)s:%(levelname)s:%(message)s'
Logger.set_logger(Path('./logs/vp_extended_qc.log'), logformat, 'INFO')
logger = Logger.getlogger()

def log_message(message):
    print(message)
    logger.info(message)

extended_qc_files = [
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB01.txt'),
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB02.txt'),
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB03.txt'),
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB04.txt'),
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB05.txt'),
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB07.txt'),
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB08.txt'),
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB09.txt'),
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB10.txt'),
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB11.txt'),
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB12.txt'),
    Path('./data_files/211006 - check VAPS/20210920/210920_VIB13.txt'),
]


@timed(logger, print_log=True)
def extended_qc_pool(file_name):
    log_message(f'run extended qc for: {file_name}')
    ext_qc = ExtendedQc(file_name)
    ext_qc.read_extended_qc()
    return ext_qc.avg_peak_df


@timed(logger, print_log=True)
def extended_qc_thread_or_process(file_name, results_queue):
    log_message(f'run extended qc for: {file_name}')
    ext_qc = ExtendedQc(file_name)
    ext_qc.read_extended_qc()
    results_queue.put(ext_qc.avg_peak_df)


@timed(logger, print_log=True)
def run_sequential():
    results = []
    log_message('start sequential ...')
    results_queue = queue.Queue()
    for filename in extended_qc_files:
        results.append(extended_qc_pool(filename))

    pprint(results)


@timed(logger, print_log=True)
def run_pool():
    results = []
    log_message('start pool ...')
    cpus = mp.cpu_count()
    log_message(f'cpu\'s: {cpus}')
    with mp.Pool(cpus - 1) as pool:
        results.append(pool.map(extended_qc_pool, extended_qc_files))

    pprint(results)


@timed(logger, print_log=True)
def run_processes():
    log_message('start processes ...')
    processes = []
    results_queue = mp.Queue()
    results = []
    for filename in extended_qc_files:
        processes.append(
            mp.Process(
                target=extended_qc_thread_or_process,
                args=(filename, results_queue,),
            )
        )
        processes[-1].start()

    log_message('all processes have started ...')
    for _ in range(len(extended_qc_files)):
        results.append(results_queue.get())
        log_message(f'get results for vibrator: {results[-1].iloc[0]["vibrator"]}')

    log_message('all processes have completed ...')
    pprint(results)


@timed(logger, print_log=True)
def run_threading():
    results = []
    log_message('start threading ...')
    threads = []
    results_queue = queue.Queue()
    results = []
    for filename in extended_qc_files:
        threads.append(
            threading.Thread(
                target=extended_qc_thread_or_process,
                args=(filename, results_queue)
            )
        )
        threads[-1].start()

    log_message('all threats have started ...')
    _ = [t.join() for t in threads]

    log_message('all threats have completed ...')
    while not results_queue.empty():
        results.append(results_queue.get())

    pprint(results)


if __name__ == '__main__':
    log_message(f'=================== timed comparisons =================== ')
    run_sequential()
    run_pool()
    run_processes()
    run_threading()
    log_message(f'======================= completed ======================= ')

log results:

2021-10-18 17:10:00,607:INFO:=================== timed comparisons =================== 
2021-10-18 17:10:00,608:INFO:start sequential ...
2021-10-18 17:10:00,608:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB01.txt
2021-10-18 17:12:17,395:INFO:==> extended_qc_pool ran in 136.787 s
2021-10-18 17:12:17,398:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB02.txt
2021-10-18 17:13:40,244:INFO:==> extended_qc_pool ran in 82.849 s
2021-10-18 17:13:40,245:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB03.txt
2021-10-18 17:13:52,424:INFO:==> extended_qc_pool ran in 12.179 s
2021-10-18 17:13:52,426:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB04.txt
2021-10-18 17:15:53,204:INFO:==> extended_qc_pool ran in 120.78 s
2021-10-18 17:15:53,206:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB05.txt
2021-10-18 17:17:29,051:INFO:==> extended_qc_pool ran in 95.844 s
2021-10-18 17:17:29,056:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB07.txt
2021-10-18 17:19:35,376:INFO:==> extended_qc_pool ran in 126.325 s
2021-10-18 17:19:35,380:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB08.txt
2021-10-18 17:21:18,993:INFO:==> extended_qc_pool ran in 103.615 s
2021-10-18 17:21:18,995:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB09.txt
2021-10-18 17:21:20,982:INFO:==> extended_qc_pool ran in 1.989 s
2021-10-18 17:21:20,988:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB10.txt
2021-10-18 17:23:04,654:INFO:==> extended_qc_pool ran in 103.669 s
2021-10-18 17:23:04,658:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB11.txt
2021-10-18 17:24:49,316:INFO:==> extended_qc_pool ran in 104.66 s
2021-10-18 17:24:49,317:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB12.txt
2021-10-18 17:26:14,879:INFO:==> extended_qc_pool ran in 85.563 s
2021-10-18 17:26:14,881:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB13.txt
2021-10-18 17:27:18,781:INFO:==> extended_qc_pool ran in 63.902 s
2021-10-18 17:27:19,127:INFO:==> run_sequential ran in 1038.52 s
2021-10-18 17:27:19,130:INFO:start pool ...
2021-10-18 17:27:19,130:INFO:cpu: 8
2021-10-18 17:27:21,550:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB01.txt
2021-10-18 17:27:21,581:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB02.txt
2021-10-18 17:27:21,680:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB03.txt
2021-10-18 17:27:21,758:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB04.txt
2021-10-18 17:27:21,810:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB05.txt
2021-10-18 17:27:21,959:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB07.txt
2021-10-18 17:27:22,086:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB08.txt
2021-10-18 17:27:41,817:INFO:==> extended_qc_pool ran in 20.137 s
2021-10-18 17:27:41,819:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB09.txt
2021-10-18 17:27:45,590:INFO:==> extended_qc_pool ran in 3.772 s
2021-10-18 17:27:45,608:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB10.txt
2021-10-18 17:29:54,069:INFO:==> extended_qc_pool ran in 152.49 s
2021-10-18 17:29:54,075:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB11.txt
2021-10-18 17:29:55,724:INFO:==> extended_qc_pool ran in 153.915 s
2021-10-18 17:29:55,737:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB12.txt
2021-10-18 17:30:27,502:INFO:==> extended_qc_pool ran in 185.744 s
2021-10-18 17:30:27,515:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB13.txt
2021-10-18 17:30:36,060:INFO:==> extended_qc_pool ran in 194.512 s
2021-10-18 17:30:43,957:INFO:==> extended_qc_pool ran in 201.891 s
2021-10-18 17:30:44,781:INFO:==> extended_qc_pool ran in 202.824 s
2021-10-18 17:31:00,955:INFO:==> extended_qc_pool ran in 195.362 s
2021-10-18 17:32:10,120:INFO:==> extended_qc_pool ran in 136.045 s
2021-10-18 17:32:23,161:INFO:==> extended_qc_pool ran in 115.646 s
2021-10-18 17:32:23,415:INFO:==> extended_qc_pool ran in 147.678 s
2021-10-18 17:32:24,020:INFO:==> run_pool ran in 304.891 s
2021-10-18 17:32:24,027:INFO:start processes ...
2021-10-18 17:32:24,094:INFO:all processes have started ...
2021-10-18 17:32:27,595:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB05.txt
2021-10-18 17:32:27,628:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB02.txt
2021-10-18 17:32:27,656:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB08.txt
2021-10-18 17:32:27,708:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB04.txt
2021-10-18 17:32:27,748:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB07.txt
2021-10-18 17:32:27,771:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB10.txt
2021-10-18 17:32:27,771:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB11.txt
2021-10-18 17:32:27,799:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB01.txt
2021-10-18 17:32:28,184:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB09.txt
2021-10-18 17:32:28,393:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB12.txt
2021-10-18 17:32:28,742:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB13.txt
2021-10-18 17:32:30,041:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB03.txt
2021-10-18 17:32:36,264:INFO:==> extended_qc_thread_or_process ran in 8.096 s
2021-10-18 17:32:36,389:INFO:get results for vibrator: 9
2021-10-18 17:33:09,257:INFO:==> extended_qc_thread_or_process ran in 39.309 s
2021-10-18 17:33:09,371:INFO:get results for vibrator: 3
2021-10-18 17:36:08,966:INFO:==> extended_qc_thread_or_process ran in 221.374 s
2021-10-18 17:36:09,068:INFO:get results for vibrator: 5
2021-10-18 17:36:19,484:INFO:==> extended_qc_thread_or_process ran in 231.862 s
2021-10-18 17:36:19,517:INFO:get results for vibrator: 2
2021-10-18 17:36:44,191:INFO:==> extended_qc_thread_or_process ran in 256.498 s
2021-10-18 17:36:44,223:INFO:get results for vibrator: 4
2021-10-18 17:37:05,644:INFO:==> extended_qc_thread_or_process ran in 276.977 s
2021-10-18 17:37:05,652:INFO:get results for vibrator: 13
2021-10-18 17:37:11,263:INFO:==> extended_qc_thread_or_process ran in 283.51 s
2021-10-18 17:37:11,337:INFO:get results for vibrator: 10
2021-10-18 17:37:13,987:INFO:==> extended_qc_thread_or_process ran in 286.245 s
2021-10-18 17:37:14,005:INFO:get results for vibrator: 7
2021-10-18 17:37:14,263:INFO:==> extended_qc_thread_or_process ran in 286.469 s
2021-10-18 17:37:14,293:INFO:get results for vibrator: 1
2021-10-18 17:37:17,111:INFO:==> extended_qc_thread_or_process ran in 289.455 s
2021-10-18 17:37:17,118:INFO:get results for vibrator: 8
2021-10-18 17:37:18,881:INFO:==> extended_qc_thread_or_process ran in 291.122 s
2021-10-18 17:37:18,884:INFO:get results for vibrator: 11
2021-10-18 17:37:26,190:INFO:==> extended_qc_thread_or_process ran in 297.854 s
2021-10-18 17:37:26,192:INFO:get results for vibrator: 12
2021-10-18 17:37:26,196:INFO:all processes have completed ...
2021-10-18 17:37:26,582:INFO:==> run_processes ran in 302.558 s
2021-10-18 17:37:26,583:INFO:start threading ...
2021-10-18 17:37:26,584:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB01.txt
2021-10-18 17:37:26,587:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB02.txt
2021-10-18 17:37:26,589:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB03.txt
2021-10-18 17:37:26,595:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB04.txt
2021-10-18 17:37:26,604:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB05.txt
2021-10-18 17:37:26,609:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB07.txt
2021-10-18 17:37:26,620:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB08.txt
2021-10-18 17:37:26,632:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB09.txt
2021-10-18 17:37:26,644:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB10.txt
2021-10-18 17:37:26,654:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB12.txt
2021-10-18 17:37:26,661:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB13.txt
2021-10-18 17:37:26,664:INFO:run extended qc for: data_files\211006 - check VAPS\20210920\210920_VIB11.txt
2021-10-18 17:37:26,671:INFO:all threats have started ...
2021-10-18 17:37:50,140:INFO:==> extended_qc_thread_or_process ran in 23.528 s
2021-10-18 17:39:20,498:INFO:==> extended_qc_thread_or_process ran in 113.911 s
2021-10-18 17:50:11,285:INFO:==> extended_qc_thread_or_process ran in 764.7 s
2021-10-18 17:51:30,443:INFO:==> extended_qc_thread_or_process ran in 843.846 s
2021-10-18 17:52:23,355:INFO:==> extended_qc_thread_or_process ran in 896.772 s
2021-10-18 17:53:12,968:INFO:==> extended_qc_thread_or_process ran in 946.374 s
2021-10-18 17:53:46,476:INFO:==> extended_qc_thread_or_process ran in 979.855 s
2021-10-18 17:53:50,388:INFO:==> extended_qc_thread_or_process ran in 983.78 s
2021-10-18 17:53:55,209:INFO:==> extended_qc_thread_or_process ran in 988.57 s
2021-10-18 17:54:01,452:INFO:==> extended_qc_thread_or_process ran in 994.826 s
2021-10-18 17:54:03,221:INFO:==> extended_qc_thread_or_process ran in 996.617 s
2021-10-18 17:54:09,694:INFO:==> extended_qc_thread_or_process ran in 1003.066 s
2021-10-18 17:54:09,697:INFO:all threats have completed ...
2021-10-18 17:54:10,039:INFO:==> run_threading ran in 1003.457 s
2021-10-18 17:54:10,041:INFO:======================= completed =======================

As the documents say threading seems only useful if the process is I/O bound and multiprocessing works if you have more than 1 cpu.

\$\endgroup\$
0

1 Answer 1

Reset to default
2
\$\begingroup\$

Overall this looks like throwaway code meant to give you a quick answer. It's not worth trying to improve since you will never re-use or maintain it.

If you did want to improve it, the issue I'd tackle is that the code is too long and repetitive.

Reduce Repetition

  • The definition of extended_qc_files is long. Say "run once for each file in this folder" instead of this is all the files in the given folder.
  • extended_qc_pool and extended_qc_thread_or_process are almost identical, factor out the part in common. Since run_sequential uses a queue, you could pass in the queue for both.
  • run_* have a lot of repetition across methods, mostly around setting up results and logging. Remove it. You could
    • Pull this out into a wrapping loop (loop over each run function and have this outside the function)
    • Make a custom decorator, and have the function take results as a parameter
    • Make a context handler, and have the function take results or the context handler as a parameter
  • Stop copy-pasting, which leads to errors and repetition. run_threading defines results twice and copy-pasted the spelling error threats. If you want a short program, typing everything by hand is one viable method. Factor things out instead of copy-pasting them.

Logging

Replace log_message by having the logger itself print to stderr/stdout in a different format.

Right now your logging is scattered through all the functions. I think that's fine since this is a quick test. For bigger projects I'd recommend instead returning timing information, and having main print the results. This also would let you group together summary statistics at the end.

Other

  • Fuzzy conclusions and understanding of the data. What did you learn? You (implicitly) conclude your process is CPU-bound and not IO bound. Why? You shouldn't conclude it because of this test--that's indirect and possibly circular reasoning. Learn tools like top and iotop on Linux/Mac, or similar built-in programs on Widows, to test this theory directly. Is the CPU or disk maxed out during a run? Mention the size of the files and where they're stored--this can let you theoretically calculate how long reading the files will take--if it's 30 seconds, and the real process runs 17 minutes, it's probably not IO-bound. Replace ExtendedQc by something that reads the files and exits. How long does that take with each method? If it's still 17 minutes, it was IO bound after all.
  • Missing information. You did not include all code. Utils.plogger, Utils.timed, ExtendedQc, etc are all relevant to this review. What does ext_qc.read_extended_qc() do? You call this and ignore the return value.
  • Re-usability. Your functions are all some version of calling extended_qc_pool. Pass this in instead of hardcoding it in each of the four methods. This would actually make your code re-usable when you want to test how to speed something up later.
\$\endgroup\$
1
  • \$\begingroup\$ Thank you for the extensive reply. This was indeed some throwaway code to experiment with threading and multiprocessing. I agree with the refactoring of the function extended_qc, which I have done by adding a key word argument for when a queue is necessary. The functions run_pool, run_processes, run_threading are a handy reference on how to boilerplate this. \$\endgroup\$
    – Bruno Vermeulen
    Commented Oct 19, 2021 at 3:15

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.