"""Multiprocessing utilities for the pipeline."""
import os
import os.path
from datetime import datetime
import faulthandler
import logging
import re
import signal
from glob import glob
import sys
import platformdirs
from autowisp.database.interface import set_project_home
from autowisp.data_reduction.data_reduction_file import DataReductionFile
from autowisp.catalog import WISPGaia
from autowisp.error_context import ErrorContext, set_error_context
# Used by this module's ``__main__``; canonical home is
# ``autowisp.miscellaneous``.
from autowisp.miscellaneous import get_code_version_str
default_config = {
"task": "calculate",
"fname_datetime_format": "%Y%m%d%H%M%S",
"std_out_err_fname": "{task}_{now!s}_{pid:d}.outerr",
"logging_fname": "{task}_{now!s}_{pid:d}.log",
"logging_verbosity": "info",
"logging_message_format": (
"%(levelname)s %(asctime)s %(name)s: %(message)s | "
"%(pathname)s.%(funcName)s:%(lineno)d"
),
}
[docs]
def get_log_outerr_filenames(existing_pid=False, **config):
"""Return the filenames where `setup_process()` redirects log and output."""
config.update(
now=(
"*"
if existing_pid
else datetime.now().strftime(config["fname_datetime_format"])
),
pid=(existing_pid or os.getpid()),
)
if existing_pid == "*":
pid_rex = re.compile(r"\{pid[^}]*\}")
def prepare(format_str):
return "*".join(pid_rex.split(format_str))
else:
def prepare(format_str):
return format_str
if config["std_out_err_fname"] is None:
std_out_err_fname = None
else:
std_out_err_fname = prepare(config["std_out_err_fname"]).format_map(
config
)
result = (
prepare(config["logging_fname"]).format_map(config),
std_out_err_fname,
)
if config.get("parent_pid"):
result = tuple(
os.path.join(
os.path.dirname(fname),
str(config["parent_pid"]),
os.path.basename(fname),
)
for fname in result
)
if existing_pid:
return tuple(sorted(glob(glob_str)) for glob_str in result)
return result
[docs]
def _enable_faulthandler(stream):
"""Point ``faulthandler`` at ``stream`` so a fatal signal self-reports.
A segfault / abort / FPE produces no Python exception, so a worker
that dies of one is "silent" and -- because the pool executor collapses
every pending future to the same ``BrokenProcessPool`` and discards
which worker died -- otherwise unattributable. With ``faulthandler``
enabled against the worker's redirected stderr, the *faulting* worker
dumps a native traceback into its own log before dying; workers the
executor merely ``terminate()``s (``SIGTERM``) dump nothing, so the log
carrying a dump singles out the culprit. Also arms ``SIGUSR1`` (POSIX)
so a *hung* worker can be prodded to dump where it is stuck.
Best-effort: silently does nothing where it cannot attach (e.g. a
captured stderr with no real ``fileno``); never fails bootstrap.
Args:
stream: The (already redirected) stderr file object to dump to.
Returns:
None
"""
try:
faulthandler.enable(file=stream, all_threads=True)
except (AttributeError, ValueError, OSError, RuntimeError):
return # e.g. stream has no real fileno -- nothing we can do
if hasattr(signal, "SIGUSR1"):
try:
faulthandler.register(
signal.SIGUSR1, file=stream, all_threads=True, chain=True
)
except (AttributeError, ValueError, OSError, RuntimeError):
pass
[docs]
def setup_process_map(config):
"""
Logging and I/O setup for the current processes.
KWArgs:
std_out_err_fname(str): Format string for the standard output/error
file name with substitutions including any keyword arguments passed
to this function, ``now`` which gets replaced by current date/time,
``pid`` which gets replaced by the process ID, ``task`` which
gets the value ``'calculate'`` by default but can be overwritten
here.
logging_fname(str): Format string for the logging file name (see
``std_out_err_fname``).
fname_datetime_format(str): The format for the date and time string
to be inserted in the file names.
logging_message_format(str): The format for the logging messages (see
logging module documentation)
logging_verbosity(str): The verbosity of logging (see logging module
documentation)
All other keyword arguments are used to substitute into the format
strings for the filenames.
Returns:
None
"""
def ensure_directory(fname):
"""Make sure the directory containing the given name exists."""
dirname = os.path.dirname(fname)
if dirname and not os.path.exists(dirname):
try:
os.makedirs(dirname)
except FileExistsError:
if not os.path.isdir(dirname):
raise
for param, value in default_config.items():
if param not in config and (
param != "logging_verbosity" or "verbose" not in config
):
config[param] = value
app_data_dir = platformdirs.user_data_dir("autowisp")
logging_fname, std_out_err_fname = get_log_outerr_filenames(**config)
if os.path.exists(app_data_dir): # allow running on GitHub Actions
with open(
os.path.join(app_data_dir, "setup_process.outerr"),
"a",
encoding="utf-8",
) as info_file:
info_file.write(
f"Setting up process with project home {config['project_home']}"
"and configuration:\n\t"
+ "\n\t".join(
f"{key!r}: {value!r}" for key, value in config.items()
)
+ "\n"
)
info_file.write(
f"Logging to {logging_fname!r}, "
f"stdout/stderr to {std_out_err_fname!r}\n"
)
all_loggers = [logging.root] + [
lgr
for lgr in logging.Logger.manager.loggerDict.values()
if isinstance(lgr, logging.Logger)
]
for lgr in all_loggers:
for handler in lgr.handlers[:]:
lgr.removeHandler(handler)
handler.close()
if std_out_err_fname is not None:
sys.stdout.flush()
sys.stderr.flush()
sys.stdout.close()
sys.stderr.close()
ensure_directory(std_out_err_fname)
sys.stdout = open( # pylint: disable=consider-using-with
std_out_err_fname, "w", encoding="utf-8", buffering=1
)
sys.stderr = sys.stdout
# After the redirect, so a fatal-signal dump lands in the process's own
# (collectable) log rather than a lost stderr. Enabled unconditionally
# -- for a non-redirected process it dumps to the inherited stderr.
_enable_faulthandler(sys.stderr)
ensure_directory(logging_fname)
logging_config = {
"filename": logging_fname,
"level": getattr(
logging,
config.get("logging_verbosity", config.get("verbose")).upper(),
),
"format": config["logging_message_format"],
"force": True,
}
if config.get("logging_datetime_format") is not None:
logging_config["datefmt"] = config["logging_datetime_format"]
logging.basicConfig(**logging_config)
logging.info("Starting process with configuration: %s", repr(config))
set_project_home(config["project_home"])
if "data_reduction_fname" in config:
DataReductionFile.fname_template = config["data_reduction_fname"]
if config.get("gaia_user") and config.get("gaia_password"):
WISPGaia.set_credentials(
user=config["gaia_user"], password=config["gaia_password"]
)
# Establish the ambient error context for this process (main process
# or worker).
set_error_context(ErrorContext.from_config(config))
[docs]
def setup_process(**config):
"""Like `setup_process_map`, but accepts keyword arguments."""
setup_process_map(config)
if __name__ == "__main__":
print(f"Code version: {get_code_version_str()}")