besmarts.core.compute module

besmarts.core.compute

Responsible for setting up and distributing large compute jobs

Architecture-wise, it interfaces the multiprocessing module

besmarts.core.compute.AutoProxy(token, serializer, manager=None, authkey=None, exposed=None, incref=True, manager_owned=False)

Return an auto-proxy for token

class besmarts.core.compute.BaseProxy(*args, **kwds)

Bases: BaseProxy

besmarts.core.compute.Client(address, family=None, authkey=None, timeout=240)

Returns a connection to the address of a Listener

class besmarts.core.compute.Connection(handle, readable=True, writable=True)

Bases: Connection

Connection class based on an arbitrary file descriptor (Unix only), or a socket handle (Windows).

class besmarts.core.compute.Listener(*args, **kwargs)

Bases: Listener

accept()

Accept a connection on the bound socket or named pipe of self.

Returns a Connection object.

besmarts.core.compute.MakeProxyType(name, exposed, _cache={})

Return a proxy type whose methods are given by exposed

besmarts.core.compute.Process(obj, *args, **kwds)

class besmarts.core.compute.Server(*args, **kwargs)

Bases: Server

accept_connection(c, name)

Spawn a new thread to serve this connection

accepter()

decref(c, ident)

handle_request(conn)

Handle a new connection

incref(c, ident)

recv_request(conn, out, err)

run_request(c, request, msg)

send_request(c, msg)

serve_client(conn)

Handle requests from the proxies in a particular process/thread

serve_forever()

Run the server forever

besmarts.core.compute.SocketClient(address, timeout=240)

Return a connection object connected to the socket given by address

besmarts.core.compute.close_pools()

besmarts.core.compute.close_workspaces()

besmarts.core.compute.compute_remote(addr, port, processes=1, queue_size=1)

besmarts.core.compute.dispatch(c, id, methodname, args=(), kwds={})

Send a message to manager using connection c and return response

besmarts.core.compute.dprint(*args, **kwds)

besmarts.core.compute.manager_connect(mgr: BaseManager, success: Event)

besmarts.core.compute.manager_remote_get_iqueue(mgr, timeout=240)

besmarts.core.compute.manager_remote_get_iqueue_thread(mgr, out)

besmarts.core.compute.manager_remote_get_oqueue(mgr, timeout=240)

besmarts.core.compute.manager_remote_get_oqueue_thread(mgr, out)

besmarts.core.compute.manager_remote_get_state(wq, timeout=240)

besmarts.core.compute.manager_remote_get_state_thread(ws, out)

besmarts.core.compute.manager_remote_get_status(wq, timeout=240)

besmarts.core.compute.manager_remote_get_status_thread(ws, out)

besmarts.core.compute.manager_remote_queue_put(oq, obj, timeout=240, n=1)

besmarts.core.compute.manager_remote_queue_put_thread(oq, obj, n, success)

besmarts.core.compute.manager_remote_queue_qsize(q, timeout=240)

besmarts.core.compute.manager_remote_queue_qsize_thread(q, out)

class besmarts.core.compute.myiqueue(maxsize=0)

Bases: Queue

get(block=True, timeout=240)

Remove and return an item from the queue.

If optional args ‘block’ is true and ‘timeout’ is None (the default), block if necessary until an item is available. If ‘timeout’ is a non-negative number, it blocks at most ‘timeout’ seconds and raises the Empty exception if no item was available within that time. Otherwise (‘block’ is false), return an item if one is immediately available, else raise the Empty exception (‘timeout’ is ignored in that case).

class besmarts.core.compute.myqueue(maxsize=0)

Bases: Queue

get(block=True, timeout=240, n=1)

Remove and return an item from the queue.

If optional args ‘block’ is true and ‘timeout’ is None (the default), block if necessary until an item is available. If ‘timeout’ is a non-negative number, it blocks at most ‘timeout’ seconds and raises the Empty exception if no item was available within that time. Otherwise (‘block’ is false), return an item if one is immediately available, else raise the Empty exception (‘timeout’ is ignored in that case).

put(item, block=True, timeout=240, n=1)

Put an item into the queue.

If optional args ‘block’ is true and ‘timeout’ is None (the default), block if necessary until a free slot is available. If ‘timeout’ is a non-negative number, it blocks at most ‘timeout’ seconds and raises the Full exception if no free slot was available within that time. Otherwise (‘block’ is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (‘timeout’ is ignored in that case).

besmarts.core.compute.queue_get_nowait(q, block=False, timeout=240, n=1)

besmarts.core.compute.queue_get_nowait_thread(q, out, n, block, timeout)

besmarts.core.compute.register_pool(p)

besmarts.core.compute.register_workspace(p)

besmarts.core.compute.remote_connect(mgr, timeout=240)

besmarts.core.compute.remote_init_thread(remote_init, shm_proxy, out)

besmarts.core.compute.shm_init(proxy)

class besmarts.core.compute.shm_local(procs_per_task=1, data=None)

Bases: object

get()

remote_init()

besmarts.core.compute.signal_kill_processes(sig, frame)

besmarts.core.compute.thread_name_set(name)

besmarts.core.compute.unregister_pool(p)

besmarts.core.compute.unregister_workspace(p)

class besmarts.core.compute.workqueue(addr, port)

Bases: object

get_state()

get_status()

get_workspaces()

besmarts.core.compute.workqueue_get_workspace(wq: workqueue_remote, addr, port) → workspace

besmarts.core.compute.workqueue_is_active(wq)

besmarts.core.compute.workqueue_list_workspaces(wq: workqueue_remote) → Dict

class besmarts.core.compute.workqueue_local(addr, port)

Bases: workqueue

close()

get_state()

get_threads()

get_workspaces()

put_workspaces(wss)

remove_workspace(ws)

class besmarts.core.compute.workqueue_manager(address=None, authkey=None, serializer='pickle', ctx=None, *, shutdown_timeout=1.0)

Bases: SyncManager

get_iqueue() → Queue

get_oqueue() → Queue

get_rqueue() → Queue

get_state() → Dict

besmarts.core.compute.workqueue_new_workspace(wq: workqueue_local, address=None, shm=None, nproc=-1)

besmarts.core.compute.workqueue_push_workspace(wq: workqueue_local, ws: workspace)

class besmarts.core.compute.workqueue_remote(addr, port)

Bases: workqueue

connect()

get_state()

get_status()

get_workspaces()

put_workspaces(wss)

besmarts.core.compute.workqueue_remote_get_workspaces(wq, timeout=240)

besmarts.core.compute.workqueue_remote_get_workspaces_thread(wss, out, success)

besmarts.core.compute.workqueue_remote_is_active(wq)

besmarts.core.compute.workqueue_remote_put_workspaces(wq, wss, timeout=240)

besmarts.core.compute.workqueue_remote_put_workspaces_thread(mgr, inp, success)

besmarts.core.compute.workqueue_remove_workspace(wq: workqueue_local, ws: workspace_local)

class besmarts.core.compute.workspace(addr, port)

Bases: object

get_iqueue()

get_oqueue()

get_rqueue()

get_state()

get_status()

besmarts.core.compute.workspace_flush(ws: workspace_local, indices, timeout: float = 240, maxwait=None, verbose=True)

besmarts.core.compute.workspace_is_active(ws: workspace_remote)

class besmarts.core.compute.workspace_local(addr, port, shm: shm_local = None, nproc=-1)

Bases: workspace

Assumes that we are process-local to all needed resources and do not need to use the proxy interface i.e. no connection needed

clear_iqueue()

clear_oqueue()

close()

get_iqueue()

get_oqueue()

get_state()

get_status()

manager_start()

pool_close()

pool_start()

reset()

set_status(status)

start()

besmarts.core.compute.workspace_local_remote_gather_thread(ws: workspace_local)

pull results from the remote workers and put them in the local queue

besmarts.core.compute.workspace_local_remote_loadbalance_thread(ws: workspace_local)

pull results from the remote workers and put them in the local queue

besmarts.core.compute.workspace_local_run(ws: workspace_local)

Take jobs from the input queue and distribute to the processing queues, which can be a (low latency) local pool or a (high latency) manager

besmarts.core.compute.workspace_local_run_thread(ws: workspace_local)

besmarts.core.compute.workspace_local_submit(ws, work)

class besmarts.core.compute.workspace_manager(address=None, authkey=None, serializer='pickle', ctx=None, *, shutdown_timeout=1.0)

Bases: SyncManager

create(*args, **kwargs)

get_state() → Dict

get_status() → workspace_status

get_workspaces() → Dict

class besmarts.core.compute.workspace_pool(*args, **kwds)

Bases: Pool

Process(*args, **kwds)

class besmarts.core.compute.workspace_remote(addr, port, nproc=1)

Bases: workspace

Assumes that we need to go through the proxy interface to access resources that are not local i.e. need to go through a connection

close()

connect()

get_iqueue()

get_oqueue()

get_state()

get_status()

iqueue_get(n=1)

iqueue_put(obj, n=1)

oqueue_put(obj, n=1)

start(input_queue_size=2)

besmarts.core.compute.workspace_remote_compute(wq: workqueue_remote, ws: workspace_remote)

besmarts.core.compute.workspace_remote_local_gather_thread(ws: workspace_remote)

pull results from the remote workers and put them in the local queue

besmarts.core.compute.workspace_remote_local_pusher_thread(ws: workspace_remote)

pull results from the remote workers and put them in the local queue

besmarts.core.compute.workspace_remote_shm_init(ws, timeout=240)

besmarts.core.compute.workspace_remote_shm_init_thread(mgr, out)

besmarts.core.compute.workspace_run(distfun: Tuple[Callable, Sequence, Mapping], workspace_address=None)

besmarts.core.compute.workspace_run_init(shm, t0=None)

class besmarts.core.compute.workspace_status

Bases: object

DONE = 5

EMPTY = 0

INACTIVE = 1

INVALID = -1

RUNNING = 4

SUBMITTING = 2

WAITING = 3

besmarts.core.compute.workspace_submit_and_flush(ws, fn, iterable: Dict, chunksize=1, timeout=0.0, batchsize=0, verbose=False, clear=True) → Dict