dragon.utils

Functions

`b64decode`(the_str)
`b64encode`(the_bytes)
`get_cpu_count`()
`get_hugepage_mount`()
`get_local_kv`(key[, timeout])
`get_local_rt_uid`()
`getlasterrstr`()
`hash`(byte_str)
`host_id`()
`host_id_from_k8s`(pod_uid)	This is used to get a hostid based on the k8s pod uid.
`set_core_affinity`(core)
`set_host_id`(new_id)
`set_local_kv`(key, value[, timeout])
`set_procname`(name)
`strtobool`(val)	Convert a string representation of truth to true (1) or false (0).

Classes

`B64`	Cython wrapper for Dragon's byte <> string conversion routines.
`ExceptionalThread`	Enhanced threading.Thread that can be killed from the outside by raising an exception inside the running instance.
`SerType`
`TimeKeeper`	Tracks elapsed times by id and optionally streams them to the Dragon telemetry infrastructure on a background thread.
`XByteBufferPickler`	A Pickler that has X-language support and is compatible with the C++ SerializableByteBuffer.
`XNumPy2DMatrixPickler`	A Pickler that has X-language support and is compatible with the C++ Serializable2DMatrix<Type>.
`XNumPyVectorPickler`	A Pickler that has X-language support and is compatible with the C++ SerializableVector<Type>.
`XPickler`	A Pickler that has X-language support and is compatible with the C++ Serializables.
`XScalarPickler`	A Pickler that has X-language support and is compatible with the C++ Scalars like int and double.
`XStringPickler`	A Pickler that has X-language support and is compatible with the C++ SerializableString.

class B64

Bases: object

Cython wrapper for Dragon’s byte <> string conversion routines.

__init__(): Convert a bytes array into a base64 encoded string. :param data: The list of bytes to convert. :return: A new B64String object containing the base64 encoded string.

classmethod bytes_to_str(the_bytes): Converts bytes into a string by base64 encoding it. Convenience function to convert bytes objects to base64 encoded strings. :param the_bytes: bytes to get encoded :return: string

decode()

classmethod from_str(serialized_str)

classmethod str_to_bytes(the_str): Converts a base64 encoded string to a bytes object. Convenience function to unpack strings. :param the_str: base64 encoded string. :return: original bytes representation.

class ExceptionalThread

Bases: Thread

Enhanced threading.Thread that can be killed from the outside by raising an exception inside the running instance. The thread’s running target function must reach a point where the exception can be noticed such as when the GIL swaps executing threads or some other wait state occurs. Note that external native library (C/C++/Fortran) code invoked within the target function will not notice the exception being raised and so control must be returned to Python for the exception to be noticed.

>>> import time
>>> t1 = ExceptionalThread(
...     target=lambda n: sum(1 if y % 2 == 0 else -1 for y in range(n)),
...     args=(10_000_000_000,)
... )  # Compute should take several minutes on modern processor core
>>> start_time = time.monotonic(); t1.start()
>>> t1.kill_by_exception()
1
>>> t1.join()
>>> (time.monotonic() - start_time) < 1  # Killed and joined in under 1s
True
>>> # time.sleep ignores exception until very end like native code would
>>> t2 = ExceptionalThread(target=time.sleep, args=(1,))
>>> start_time = time.monotonic(); t2.start()
>>> t2.kill_by_exception()  # Still indicates exception trigger success
1
>>> t2.join()
>>> (time.monotonic() - start_time) < 1  # No early termination of thread
False

kill_by_exception(exc_type=<class 'Exception'>): When called, will raise the specified exception inside the running thread instance and will return an int to indicate success (1), failure to find the thread (0), or unhandled error followed by an attempt to revert the thread state change (2 or greater).

class SerType

Bases: IntFlag

TY_NONE = 0

TY_STR = 1

TY_INT = 2

TY_DOUBLE = 3

TY_INTVECTOR = 4

TY_DOUBLEVECTOR = 5

TY_INTMATRIX = 6

TY_DOUBLEMATRIX = 7

TY_BYTEBUFFER = 8

class TimeKeeper

Bases: object

Tracks elapsed times by id and optionally streams them to the Dragon telemetry infrastructure on a background thread.

A Telemetry instance is created internally. When its level > 0 (i.e. the runtime was launched with --telemetry-level greater than zero), a background thread is started that wakes every collection_window seconds, acquires an threading.RLock , ships every accumulated timing to the telemetry service via add_data, resets all accumulators, then releases the lock.

timekeeper_name becomes the ts_metric_name in every add_data call (the OpenTSDB metric name).
Each timing id is sent as tagk="id", tagv=str(id).
The local hostname is sent as tagk="hostname", tagv=<hostname>.

When telemetry is inactive (telemetry.level == 0) no thread is started and no lock is created, preserving the original single-threaded behaviour.

class Recorder

Bases: object

__init__(timekeeper, id, start=None)

__init__(timekeeper_name: str = '', recording: bool = False, collection_window: float = 10.0, clear_all: bool = True)

Parameters:

timekeeper_name (str ) – Name used as the telemetry metric name.
recording (bool ) – Enable timing accumulation.
collection_window (float ) – Seconds between telemetry flushes.

add(id, start, end=None)

add_elapsed(id, elapsed)

empty()

get_timings()

now()

record(id, start=None)

reset(id)

reset_all()

start_telemetry()

stop() → None 

Signal the background telemetry worker to stop and wait for it.

Safe to call when telemetry is not active (no-op in that case).

class XByteBufferPickler

Bases: object

A Pickler that has X-language support and is compatible with the C++ SerializableByteBuffer. Pickling a byte buffer with this pickler can be read in C++ by deserializing using the SerializableByteBuffer class. Likewise, a C++ byte buffer serialized with the SerializableByteBuffer can be unpickled into a bytes object as well. This is meant to be used by code that wishes to access the bytes and perhaps provide code that interprets them in an application specific way.

__init__()

byte_size(val: bytes ) → int 

dump(val: bytes , file) → None 

dumps(val: bytes ) → bytes 

load(file)

loads(val: bytes ) → bytes 

property ty_val: This is the type value to identify the particular type of value to be unpickled. It is one of the SerType values.

class XNumPy2DMatrixPickler

Bases: object

A Pickler that has X-language support and is compatible with the C++ Serializable2DMatrix<Type>. Pickling a numpy array with this pickler can be read in C++ by deserializing using the Serializable2DMatrix<Type>. Likewise, a C++ matrix serialized with the Serializable2DMatrix<Type> can be unpickled into a numpy matrix was well. This is meant to be an efficient implementation for passing 2D matrices to and from other languages (at present that means C++) through a pickling/unpickling interface like native Queue or DDict.

__init__(data_type: dtype, type_val=<SerType.TY_NONE: 0>)

byte_size(nparr)

dump(nparr, file) → None 

dumps(nparr) → bytes 

load(file)

loads(val)

property ty_val

class XNumPyVectorPickler

Bases: object

A Pickler that has X-language support and is compatible with the C++ SerializableVector<Type>. Pickling a numpy array with this pickler can be read in C++ by deserializing using the SerializableVector<Type>. Likewise, a C++ matrix serialized with the SerializableVector<Type> can be unpickled into a numpy array as well. This is meant to be an efficient implementation for passing arrays/vectors to and from other languages (at present that means C++) through a pickling/unpickling interface like native Queue or DDict.

__init__(data_type: dtype, type_val=<SerType.TY_NONE: 0>)

byte_size(nparr)

dump(nparr, file) → None 

dumps(nparr) → bytes 

load(file)

loads(val)

property ty_val: This is the type value to identify the particular type of value to be unpickled. It is one of the SerType values.

class XPickler

Bases: object

A Pickler that has X-language support and is compatible with the C++ Serializables. Objects to be shared between Python and C++ must be supported by XPickler on the Python side and the Serializable class on the C++ side. Pickling a supported object with this pickler can be read in C++ by deserializing using the Seralizable class. Likewise, a C++ Serializable object can be unpickled into its corresponding Python object, enabling cross communication between Python and other languages (at present that means C++) through a pickling/unpickling interface like native Queue or DDict.

__init__()

choose_pickler(val)

dump(val, file) → None 

dumps(val) → bytes 

load(file) → object 

loads(val: bytes ) → object 

classmethod ty_bytes(pickler)

class XScalarPickler

Bases: object

A Pickler that has X-language support and is compatible with the C++ Scalars like int and double. Pickling a scalar with this pickler can be read in C++ by deserializing using the SeralizableInt or SerializableScalar class. Likewise, a C++ int/double serialized with SeralizableScalar and SerializableScalar can be unpickled into a int/float was well, enabling cross language communication between Python and other languages (at present that means C++) through a pickling/unpickling interface like native Queue or DDict.

__init__(data_type: dtype, type_val=<SerType.TY_NONE: 0>)

byte_size(val)

dump(val, file) → None 

dumps(val) → bytes 

load(file) → object 

loads(val: bytes ) → object 

property ty_val

class XStringPickler

Bases: object

A Pickler that has X-language support and is compatible with the C++ SerializableString. Pickling a string with this pickler can be read in C++ by deserializing using the SeralizableString class. Likewise, a C++ string serialized with SeralizableString can be unpickled into a str was well, enabling cross communication between Python and other languages (at present that means C++) through a pickling/unpickling interface like native Queue or DDict.

__init__()

byte_size(val)

dump(val, file) → None 

dumps(val) → bytes 

load(file) → str 

loads(val: bytes ) → str 

property ty_val

b64decode(the_str)

b64encode(the_bytes)

get_cpu_count()

get_hugepage_mount()

get_local_kv(key, timeout=None)

get_local_rt_uid()

getlasterrstr()

hash(byte_str: bytes )

host_id()

host_id_from_k8s(pod_uid): This is used to get a hostid based on the k8s pod uid. It is called when a process in one node needs to know the host_id of another node. It can be called from a node/pod as many times as needed in order to translate a pod uid to a dragon hostid. For example, a backend pod can query the hostids of the other backend pods. For setting/assigning the hostid of the pod itself, each pod needs to call set_host_id().

set_core_affinity(core)

set_host_id(new_id)

set_local_kv(key, value, timeout=None)

set_procname(name)

strtobool(val)

Convert a string representation of truth to true (1) or false (0).

True values are ‘y’, ‘yes’, ‘t’, ‘true’, ‘on’, and ‘1’; false values are ‘n’, ‘no’, ‘f’, ‘false’, ‘off’, and ‘0’. Raises ValueError if ‘val’ is anything else.