Run parallel bash jobs in python by multiprocessing and subprocess modules

demoParallel.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
- multiprossing 101
- whether multiprocessing could contain subprocess work
@contact: Yun Yan ([email protected])
"""
import subprocess
from multiprocessing import Pool
import os
import time
import sys
import logging
import random
from helper import *

def get_logger():
    logger = logging.getLogger("parallel_ex")
    logger.setLevel(logging.DEBUG)

    fh = logging.FileHandler("demoParallel.log")
    fmt = '%(asctime)s - %(levelname)s - %(message)s'
    formatter = logging.Formatter(fmt)
    fh.setFormatter(formatter)

    logger.addHandler(fh)
    return logger

cmd = "sh longwork.sh"

def runcmd(cmd):
    print_logger(paste0("Worker", os.getpid(), "starts"))
    try:
        p = subprocess.Popen(cmd, shell=True,
                             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    except ValueError as e:
        print(e)
    (pout, perr) = p.communicate()
    print_logger(paste0("Worker", os.getpid(), "ends"))

#print_logger("Test Single")
#runcmd(cmd)

print_logger("Test Parallel")
num_tasks = 10
size_pool = 2

def f(x):
    print('Run task %s (%s)...' % (x, os.getpid()))
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print('Task %s runs %0.2f seconds.' % (x, (end - start)))
    return(x+100)

ps = Pool(size_pool)
print_logger(paste("Main", os.getpid()))
for i in range(num_tasks):
#    ps.apply_async(f, args=(i, ))
    ps.apply_async(runcmd, args=(cmd, ))

ps.close()
ps.join()
print_logger("All Done")


cmd = "cat file* > allfile.txt"
p = subprocess.Popen(cmd, shell=True)

helper.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
@contact: Yun Yan ([email protected])
"""
import os
import time
import shutil

def join_path(*args):
    x = map(str, args)
    return(os.path.join(*x))


def paste(*args, sept=" "):
    x = map(str, args)
    return(sept.join(x))


def paste0(*args):
    return(paste(*args, sept=""))


def print_logger(msg):
    localtime = time.asctime(time.localtime(time.time()))
    print(paste(">>" + "[" + localtime + "]" + msg))
    return None

longwork.sh

#!/usr/bin/env bash
echo "Worker" $$  "begins at $(date)"

sleep  20
echo "Worker " $$ > "file"$$".txt"
echo "Worker" $$ "ends at $(date)"