Skip to content

Support for platforms where multiprocessing does not work #173

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Support for platforms where multiprocessing does not work #173

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
aacb677
Minor fix: unmatched parenthesis in comment
Grimler91 Sep 17, 2020
f6ca8c6
Simplify code a bit in do_multiprocessing
Grimler91 Sep 17, 2020
7ff120b
Support platforms where multiprocessing does not work
Grimler91 Sep 17, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
162 changes: 88 additions & 74 deletions pythontex/pythontex3.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,13 +57,19 @@
from collections import defaultdict, OrderedDict, namedtuple
from re import match, sub, search
import subprocess
import multiprocessing
from pygments.styles import get_all_styles
from pythontex_engines import *
import textwrap
import platform
import itertools

try:
import multiprocessing.synchronize # noqa: F401
from multiprocessing import Pool
except ImportError:
multiprocessing = False


if sys.version_info[0] == 2:
try:
import cPickle as pickle
Expand Down Expand Up @@ -192,8 +198,9 @@ def process_argv(data, temp_data):
temp_data['hashdependencies'] = False
if args.jobs is None:
try:
jobs = multiprocessing.cpu_count()
jobs = os.cpu_count()
except NotImplementedError:
# Is this exception needed with os.cpu_count?
jobs = 1
temp_data['jobs'] = jobs
else:
Expand Down Expand Up @@ -1145,6 +1152,13 @@ def negative_one():



def eval_task(fun_name, arg_list):
'''
Some platforms, like android, does not support multiprocessing. For such
platforms we fall back to this function when running tasks.
'''
arg_tuple = tuple(arg_list)
return eval(fun_name)(*arg_tuple)

def do_multiprocessing(data, temp_data, old_data, engine_dict):
jobname = data['jobname']
Expand Down Expand Up @@ -1269,7 +1283,14 @@ def do_multiprocessing(data, temp_data, old_data, engine_dict):
# concurrent processes to a user-specified value for jobs. If the user
# has not specified a value, then it will be None, and
# multiprocessing.Pool() will use cpu_count().
pool = multiprocessing.Pool(jobs)
if multiprocessing:
pool = Pool(jobs)
# Add task to list of tasks to run asynchronously, from function
# name and list of args.
# globals()[fun] converts string with function name into function handle
async_or_eval = lambda fun, args: pool.apply_async(globals()[fun], args)
else:
async_or_eval = eval_task
tasks = []

# If verbose, print a list of processes
Expand All @@ -1284,75 +1305,63 @@ def do_multiprocessing(data, temp_data, old_data, engine_dict):
family = key.split('#')[0]
# Uncomment the following for debugging, and comment out what follows
'''run_code(encoding, outputdir,
workingdir,
cc_dict_begin[family],
code_dict[key],
cc_dict_end[family],
engine_dict[family].language,
engine_dict[family].commands,
engine_dict[family].created,
engine_dict[family].extension,
makestderr, stderrfilename,
code_index_dict[key],
engine_dict[family].errors,
engine_dict[family].warnings,
engine_dict[family].linenumbers,
engine_dict[family].lookbehind,
keeptemps, hashdependencies,
pygments_settings]))'''
tasks.append(pool.apply_async(run_code, [encoding, outputdir,
workingdir,
cc_dict_begin[family],
code_dict[key],
cc_dict_end[family],
engine_dict[family].language,
engine_dict[family].commands,
engine_dict[family].created,
engine_dict[family].extension,
makestderr, stderrfilename,
code_index_dict[key],
engine_dict[family].errors,
engine_dict[family].warnings,
engine_dict[family].linenumbers,
engine_dict[family].lookbehind,
keeptemps, hashdependencies,
pygments_settings]))
workingdir,
cc_dict_begin[family],
code_dict[key],
cc_dict_end[family],
engine_dict[family].language,
engine_dict[family].commands,
engine_dict[family].created,
engine_dict[family].extension,
makestderr, stderrfilename,
code_index_dict[key],
engine_dict[family].errors,
engine_dict[family].warnings,
engine_dict[family].linenumbers,
engine_dict[family].lookbehind,
keeptemps, hashdependencies,
pygments_settings)'''
tasks.append(async_or_eval('run_code', [encoding, outputdir,
workingdir,
cc_dict_begin[family],
code_dict[key],
cc_dict_end[family],
engine_dict[family].language,
engine_dict[family].commands,
engine_dict[family].created,
engine_dict[family].extension,
makestderr, stderrfilename,
code_index_dict[key],
engine_dict[family].errors,
engine_dict[family].warnings,
engine_dict[family].linenumbers,
engine_dict[family].lookbehind,
keeptemps, hashdependencies,
pygments_settings]))
if verbose:
print(' - Code process ' + key.replace('#', ':'))

# Add console processes
for key in cons_dict:
family = key.split('#')[0]
if engine_dict[family].language.startswith('python'):
if family in pygments_settings:
# Uncomment the following for debugging
'''python_console(jobname, encoding, outputdir, workingdir,
fvextfile, pygments_settings[family],
cc_dict_begin[family], cons_dict[key],
cc_dict_end[family], engine_dict[family].startup,
engine_dict[family].banner,
engine_dict[family].filename)'''
tasks.append(pool.apply_async(python_console, [jobname, encoding,
outputdir, workingdir,
fvextfile,
pygments_settings[family],
cc_dict_begin[family],
cons_dict[key],
cc_dict_end[family],
engine_dict[family].startup,
engine_dict[family].banner,
engine_dict[family].filename]))
else:
tasks.append(pool.apply_async(python_console, [jobname, encoding,
outputdir, workingdir,
fvextfile,
None,
cc_dict_begin[family],
cons_dict[key],
cc_dict_end[family],
engine_dict[family].startup,
engine_dict[family].banner,
engine_dict[family].filename]))
# Uncomment the following for debugging
'''python_console(jobname, encoding, outputdir, workingdir,
fvextfile, pygments_settings[family],
cc_dict_begin[family], cons_dict[key],
cc_dict_end[family], engine_dict[family].startup,
engine_dict[family].banner,
engine_dict[family].filename)'''
tasks.append(async_or_eval('python_console', [jobname, encoding,
outputdir, workingdir,
fvextfile,
pygments_settings[family] if family in pygments_settings else None,
cc_dict_begin[family],
cons_dict[key],
cc_dict_end[family],
engine_dict[family].startup,
engine_dict[family].banner,
engine_dict[family].filename]))
else:
print('* PythonTeX error')
print(' Currently, non-Python consoles are not supported')
Expand All @@ -1365,18 +1374,19 @@ def do_multiprocessing(data, temp_data, old_data, engine_dict):
# Uncomment the following for debugging
# do_pygments(encoding, outputdir, fvextfile, pygments_list,
# pygments_settings, typeset_cache, hashdependencies)
tasks.append(pool.apply_async(do_pygments, [encoding, outputdir,
fvextfile,
pygments_list,
pygments_settings,
typeset_cache,
hashdependencies]))
tasks.append(async_or_eval('do_pygments', [encoding, outputdir,
fvextfile,
pygments_list,
pygments_settings,
typeset_cache,
hashdependencies]))
if verbose:
print(' - Pygments process')

# Execute the processes
pool.close()
pool.join()
if multiprocessing:
pool.close()
pool.join()

# Get the outputs of processes
# Get the files and macros created. Get the number of errors and warnings
Expand All @@ -1387,7 +1397,11 @@ def do_multiprocessing(data, temp_data, old_data, engine_dict):
new_files = False
messages = []
for task in tasks:
result = task.get()
if multiprocessing:
result = task.get()
else:
result = task

if result['process'] == 'code':
key = result['key']
files[key].extend(result['files'])
Expand Down