eon.communicator

Module Contents

Classes

Communicator
MPI
Local
Script

Functions

bundled_job_name	Job id for slot slot inside a bundle named after the first job.
tryint
alphanum_key	Turn a string into a list of string and number chunks. “z23a” -> [“z”, 23, “a”]
sort_nicely	Sort the given list in the way that humans expect.
reset_communicators	Clear non-MPI communicator cache (tests / multi-config isolation).
get_communicator	Return a communicator for config.
read_return_files_manifest	Return the filenames listed in return_files.dat, or None if absent.
harvest_job_files	Harvest one non-bundled job directory into {filename: StringIO}.
client_environment	Environment for an eonclient the server starts outside an MPI launcher.

Data

logger
RETURN_FILES_MANIFEST

API

eon.communicator.logger

‘getLogger(…)’

eon.communicator.bundled_job_name(bundle_dirname, slot)

Job id for slot slot inside a bundle named after the first job.

Bundles live in a directory named {state}_{first_wuid}. Jobs in the chunk are consecutive wuids, so slot n is first_wuid + n.

eon.communicator.tryint(s)

eon.communicator.alphanum_key(s)

Turn a string into a list of string and number chunks. “z23a” -> [“z”, 23, “a”]

eon.communicator.sort_nicely(l)

Sort the given list in the way that humans expect.

eon.communicator.reset_communicators()

Clear non-MPI communicator cache (tests / multi-config isolation).

eon.communicator.get_communicator(config: eon.config.ConfigClass)

Return a communicator for config.

config is required (no process-wide default). Non-MPI communicators are memoized by id(config). MPI uses a single process-wide instance.

exception eon.communicator.NotImplementedError

Bases: Exception

exception eon.communicator.CommunicatorError

Bases: Exception

exception eon.communicator.EONClientError

Bases: Exception

An eOn client finished without outputting results, it probably crashed.

Initialization

Initialize self. See help(type(self)) for accurate signature.

eon.communicator.RETURN_FILES_MANIFEST

‘return_files.dat’

eon.communicator.read_return_files_manifest(jobpath)

Return the filenames listed in return_files.dat, or None if absent.

eon.communicator.harvest_job_files(jobpath)

Harvest one non-bundled job directory into {filename: StringIO}.

When return_files.dat is present, only those names (plus results.dat) are kept, any extension. A listed file that was never written is logged and omitted. Without a manifest, keep .con/.dat/.json/.log.

class eon.communicator.Communicator(scratchpath, bundle_size=1, config: eon.config.ConfigClass = None)

Initialization

abstractmethod submit_jobs(data, invariants)

Throws CommunicatorError if fails.

abstractmethod get_results(results)

Returns a list of dictionaries containing the results.

abstractmethod get_queue_size()

Returns the number of items waiting to run in the queue.

queued_search_count()

Searches waiting, using the bundle size this communicator was built with.

get_queue_size counts bundles. Multiplying by the live config comm_job_bundle_size is wrong if that value changed after construction.

in_progress_search_count()

Searches running, using this communicator’s bundle size.

abstractmethod cancel_state(statenumber)

Returns the number of workunits that were canceled.

get_bundle_size(job_path)

unbundle(resultpath, keep_result)

This method unbundles multiple jobs into multiple single jobs so the akmc script can process them.

If the job did not return results (probably because it crashed or was canceled), this method will raise EONClientError.

make_bundles(data, invariants)

This method is a generator that bundles together multiple jobs into a single job. Example usage: for jobpath in self.make_bundles(data, invariants): do_stuff()

class eon.communicator.MPI(scratchpath, bundle_size, config: eon.config.ConfigClass = None)

Bases: eon.communicator.Communicator

Initialization

submit_jobs(data, invariants)

run_resume_jobs()

get_ready_ranks()

get_queue_size()

get_results(resultspath, keep_result)

Moves work from scratchpath to results path.

get_number_in_progress()

stop_clients()

Send STOPCAR to every ready client rank (ClientEON exits on that path).

cancel_state(state)

eon.communicator.client_environment(base=None)

Environment for an eonclient the server starts outside an MPI launcher.

An eonclient built with MPI (rgpot calculator groups) loads libmpi, and with it UCX, in every run. UCX patches the memory functions as the library loads, which on conda’s OpenMPI costs about 0.4 s per process against 0.02 s for the job itself. A client the Local communicator starts is one process with no MPI peers, so the memory events are of no use to it. An explicit UCX_MEM_EVENTS from the user wins.

class eon.communicator.Local(scratchpath, client, ncpus, bundle_size, config: eon.config.ConfigClass = None)

Bases: eon.communicator.Communicator

Initialization

cleanup()

Kills the running eonclients and everything they started.

get_results(resultspath, keep_result)

Moves work from scratchpath to results path.

check_job(job)

submit_jobs(data, invariants)

Run up to ncpu number of clients to process the work in jobpaths. The job directories are moved to the scratch path before the calculation is run. This method doesn’t return anything.

cancel_state(state)

get_queue_size()

get_number_in_progress()

class eon.communicator.Script(scratch_path, bundle_size, name_prefix, scripts_path, queued_jobs_cmd, cancel_job_cmd, submit_job_cmd, config: eon.config.ConfigClass = None)

Bases: eon.communicator.Communicator

Initialization

save_jobids()

get_results(resultspath, keep_result)

Moves work from scratchpath to results path.

check_command(status, output, cmdname)

submit_jobs(data, invariants)

cancel_state(state)

Cancel the jobs still in the queue and drop their directories.

Jobs that already left the queue keep their ids and directories, so the next get_results() harvests them; the explorer’s keep_result filter then decides whether they still count. A cancel that fails (the job finished in between) is logged, not fatal. Returns the number of jobs cancelled.

get_queued_jobs()

get_number_in_progress()

get_queue_size()