dragon.data.DDict

class DDict

Bases: object

The Distributed Dictionary provides a key/value store that is distributed across a series of managers and one nodes of a Dragon run-time. The goal is to evenly distribute data across all managers to provide a scalable implementation of a dictionary with a high degree of allowable parallelism. Clients attach to the Distributed Dictionary and store key/value pairs in it just like accessing a local dictionary in Python. However, the Distributed Dictionary goes beyond what the standard Python dictionary supports by including support for distributing data, checkpointing, and various other optimization opportunities for specific applications.

__init__(managers_per_node: int = 1, n_nodes: int = 1, total_mem: int = 3145728, *, working_set_size: int = 1, wait_for_keys: bool = False, wait_for_writers: bool = False, policy: ~dragon.infrastructure.policy.Policy = None, managers_per_policy: int = 1, orc_policy: ~dragon.infrastructure.policy.Policy = None, persist_freq: int = 0, name: str = '', timeout: float = None, trace: bool = False, restart: bool = False, read_only: bool = False, restore_from: int = None, persist_count: int = 0, persist_path: str = '', persister_class: ~dragon.data.ddict.ddict.CheckpointPersister = <class 'dragon.data.ddict.ddict.NULLCheckpointPersister'>, streams_per_manager=5) → None 

Construct a Distributed Dictionary to be shared amongst distributed processes running in the Dragon Runtime. The distributed dictionary creates the specified number of managers and shards the data across all managers. The total memory of the dictionary is split across all the managers, so you want to allocate more space than is required by perhaps 30 percent, but that should be determined via some experimentation and depends on the application being developed. See the Dragon documentation’s section on the Distributed Dictionary design for more details about creating and using a distributed dictionary.

Parameters:

managers_per_node – The number of managers on each node. The total_mem is divided up amongst the managers. If a list of policies is provided then this is the number of managers per policy. Each policy could be used to start more than one manager per node, in a potentially heterogeneous way. Defaults to 1.
streams_per_manager – A tuning parameter that when non-zero will reduce the time needed for a manager to receive requests from clients. If there are many clients connecting, then some clients will use their own stream channels when connecting.
n_nodes – The number of nodes that will have managers deployed on them. This must be set to None if a list of policies is provided. Defaults to 1.
total_mem – The total memory in bytes that will be sharded evenly across all managers. Defaults to DDICT_MIN_SIZE but this is really a minimum size for a single manager and should be specified by the user.
working_set_size – This sets the size of the checkpoint, in memory, working set. This determines how much state each manager will keep internally. This is the number of different, simultaneous checkpoints that may be active at any point in time. Defaults to 1.
wait_for_keys – Setting this to true means that each manager will keep track of a set of keys at each checkpoint and clients advancing to a new checkpoint will block until the set of keys at the oldest, retiring working set checkpoint are all written. By specifying this all clients will remain in sync with each other relative to the size of the working set. Defaults to False. It is also possible to store key/values that are not part of the checkpointing set of key/values. Those keys are called persistent keys and will not be affected by setting this argument to true. Specifying wait_for_keys also means that readers will block while waiting for a non-persistent key to be written until the key is found or a timeout occurs. Setting this to true requires a working set size of at least 2.
wait_for_writers – Setting this to true means that each manager will wait for the set of clients which have previously written keys to a checkpoint in a manager to have all advanced their checkpoint id beyond the oldest checkpointing id before retiring a checkpoint from the working set. Setting this to true will cause clients that are advancing rapidly to block while others catch up. Defaults to False. Setting this to true requires a working set size of at least 2.
policy – A policy can be supplied for starting the managers. Please read about policies in the Process Group documentation. Managers are started via a Process Group and placement of managers and other characteristics can be controlled via a policy or list of policies. If a list of policies is given then managers_per_node processes are started for each policy. Defaults to None which applies a Round-Robin policy.
managers_per_policy – The number of managers started with each policy when a list of policies is provided. The total_mem is divided up evenly amongst the managers. This is the Defaults to 1.
orc_policy – A policy can be supplied for starting the orchestrator. The placement of the orchestrator is most impactful when re-constituting a distributed dictionary that has previously been destroyed with allow_restart=True. In this case, placing the orchestrator on the same node as the previous orchestrator allows the dictionary to be re-created. The policy defaults to None which applies a Round-Robin policy.
persist_freq – This is the frequency that a checkpoint will be persisted to disk. This is independent of the working set size and can be any frequency desired. Defaults to 0 which means that no persisting will be done.
name – This is a base file name to be applied to persisted state for the dictionary. This base name along with a checkpoint number is used to restore a distributed dictionary from a persisted checkpoint. Defaults to “”.
timeout – This is a timeout that will be used for all timeouts on the creating client and all managers during communication between the distributed components of the dictionary. New clients wishing to set their own timeout can use the attach method to specify their own local timeout. Defaults to None (block).
trace – Defaults to False. If set to true, all interaction between clients and managers is logged. This results in large logs, but may help in debugging.

Returns:

None and a new instance of a distributed dictionary is initialized.

Raises:

AttributeError – If incorrect parameters are supplied.
RuntimeError – If there was an unexpected error during initialization.

Methods

`__init__`([managers_per_node, n_nodes, ...])	Construct a Distributed Dictionary to be shared amongst distributed processes running in the Dragon Runtime.
`advance`()	Advance to next available persisted checkpoint.
`attach`(serialized_dict, *[, timeout, trace])	Within Python you typically do not need to call this method explicitly.
`bget`(key)	Read the key written through bput.
`bput`(key, value)	Store a non-persistent key/value pair by brodcasting to all managers across the Distributed Dictionary.
`checkpoint`()	Calling checkpoint advances the checkpoint for the distributed dictionary.
`clear`()	Empty the distributed dictionary of all keys and values.
`clone`(clone_list)	Clone dictionary to the list of dictionaries.
`copy`([name])	Returns a copy of the Distributed Dictionary.
`destroy`([allow_restart])	Destroy a Distributed Dictionary instance, freeing all the resources that were allocated when it was created.
`detach`()	Detach from the Distributed Dictionary and free all local resources of this client.
`end_batch_put`()
`filter`(mgr_code, mgr_code_args, comparator)	Calling this instantiates a tree of process groups where mgr_code is expected to be a function that is invoked as mgr_code(args) where args are (dd, out_queue)+mgr_code_args.
`freeze`()
`get_name`()
`items`()	Returns a list of all key/value pairs in the Distributed Dictionary.
`keys`()	Returns a keys view of the distributed dictionary.
`local_items`()	Returns a DDictItemsView of the keys that are local to the process invoking this method.
`local_keys`()	Returns a DDictKeysView of the keys that are local to the process invoking this method.
`local_len`()
`local_values`()	Returns a DDictValuesView of the keys that are local to the process invoking this method.
`manager`(id)	Return a version of the current ddict that will always choose the given manager for storing and retrieving data.
`persist`()	Persist current checkpoint to disk.
`persisted_ids`()	Get a list of persisted checkpoint IDs.
`pickler`([key_pickler, value_pickler])
`pop`(key[, default])	Pop the given key from the distributed dictionary and return the associated value.
`popitem`()	Returns a random key/value pair from the Distributed Dictionary.
`pput`(key, value)	Persistently store a key/value pair within the Distributed Dictionary.
`restore`(chkpt)	Restore a persisted checkpoint.
`rollback`()	Calling rollback decrements the checkpoint id to its previous value.
`serialize`()	Returns a serialized, base64 encoded descriptor (i.e. string) that may be shared with other processes for attaching.
`setup_logging`()
`start_batch_put`([persist])	Calling other APIs except for put before the batch put ends could leads to a hang or exception.
`sync_to_newest_checkpoint`()	Advance the checkpoint identifier of this client to the newest checkpoint across all managers.
`synchronize_ddicts`(serialized_ddicts)	Synchronize managers across all parallel dictionaries.
`unfreeze`()
`update`(dict2)	Adds all key/value pairs from dict2 into this Distributed Dictionary.
`values`()	When called this returns a list of all values in the Distributed Dictionary.
`which_manager`(key)	Return the manager id of the manager to which this key would be sent on a put/store operation.

Attributes

`checkpoint_id`	This returns the clients current checkpoint id.
`dstats`	Returns a dict of manager stats, one for each manager of the distributed dictionary.
`empty_managers`	Return a list of manager IDs that restarted on new nodes.
`is_frozen`
`local_manager`	Returns a local manager id if one exists.
`local_managers`	Returns all local manager ids of all managers that are local to this node.
`main_manager`	Returns the main manager id.
`manager_nodes`	For each manager, the serialized, base64 encoded FLI of the manager is returned.
`stats`	Returns a list of manager stats, one for each manager of the distributed dictionary.

__init__(managers_per_node: int = 1, n_nodes: int = 1, total_mem: int = 3145728, *, working_set_size: int = 1, wait_for_keys: bool = False, wait_for_writers: bool = False, policy: ~dragon.infrastructure.policy.Policy = None, managers_per_policy: int = 1, orc_policy: ~dragon.infrastructure.policy.Policy = None, persist_freq: int = 0, name: str = '', timeout: float = None, trace: bool = False, restart: bool = False, read_only: bool = False, restore_from: int = None, persist_count: int = 0, persist_path: str = '', persister_class: ~dragon.data.ddict.ddict.CheckpointPersister = <class 'dragon.data.ddict.ddict.NULLCheckpointPersister'>, streams_per_manager=5) → None 

Construct a Distributed Dictionary to be shared amongst distributed processes running in the Dragon Runtime. The distributed dictionary creates the specified number of managers and shards the data across all managers. The total memory of the dictionary is split across all the managers, so you want to allocate more space than is required by perhaps 30 percent, but that should be determined via some experimentation and depends on the application being developed. See the Dragon documentation’s section on the Distributed Dictionary design for more details about creating and using a distributed dictionary.

Parameters:

managers_per_node – The number of managers on each node. The total_mem is divided up amongst the managers. If a list of policies is provided then this is the number of managers per policy. Each policy could be used to start more than one manager per node, in a potentially heterogeneous way. Defaults to 1.
streams_per_manager – A tuning parameter that when non-zero will reduce the time needed for a manager to receive requests from clients. If there are many clients connecting, then some clients will use their own stream channels when connecting.
n_nodes – The number of nodes that will have managers deployed on them. This must be set to None if a list of policies is provided. Defaults to 1.
total_mem – The total memory in bytes that will be sharded evenly across all managers. Defaults to DDICT_MIN_SIZE but this is really a minimum size for a single manager and should be specified by the user.
working_set_size – This sets the size of the checkpoint, in memory, working set. This determines how much state each manager will keep internally. This is the number of different, simultaneous checkpoints that may be active at any point in time. Defaults to 1.
wait_for_keys – Setting this to true means that each manager will keep track of a set of keys at each checkpoint and clients advancing to a new checkpoint will block until the set of keys at the oldest, retiring working set checkpoint are all written. By specifying this all clients will remain in sync with each other relative to the size of the working set. Defaults to False. It is also possible to store key/values that are not part of the checkpointing set of key/values. Those keys are called persistent keys and will not be affected by setting this argument to true. Specifying wait_for_keys also means that readers will block while waiting for a non-persistent key to be written until the key is found or a timeout occurs. Setting this to true requires a working set size of at least 2.
wait_for_writers – Setting this to true means that each manager will wait for the set of clients which have previously written keys to a checkpoint in a manager to have all advanced their checkpoint id beyond the oldest checkpointing id before retiring a checkpoint from the working set. Setting this to true will cause clients that are advancing rapidly to block while others catch up. Defaults to False. Setting this to true requires a working set size of at least 2.
policy – A policy can be supplied for starting the managers. Please read about policies in the Process Group documentation. Managers are started via a Process Group and placement of managers and other characteristics can be controlled via a policy or list of policies. If a list of policies is given then managers_per_node processes are started for each policy. Defaults to None which applies a Round-Robin policy.
managers_per_policy – The number of managers started with each policy when a list of policies is provided. The total_mem is divided up evenly amongst the managers. This is the Defaults to 1.
orc_policy – A policy can be supplied for starting the orchestrator. The placement of the orchestrator is most impactful when re-constituting a distributed dictionary that has previously been destroyed with allow_restart=True. In this case, placing the orchestrator on the same node as the previous orchestrator allows the dictionary to be re-created. The policy defaults to None which applies a Round-Robin policy.
persist_freq – This is the frequency that a checkpoint will be persisted to disk. This is independent of the working set size and can be any frequency desired. Defaults to 0 which means that no persisting will be done.
name – This is a base file name to be applied to persisted state for the dictionary. This base name along with a checkpoint number is used to restore a distributed dictionary from a persisted checkpoint. Defaults to “”.
timeout – This is a timeout that will be used for all timeouts on the creating client and all managers during communication between the distributed components of the dictionary. New clients wishing to set their own timeout can use the attach method to specify their own local timeout. Defaults to None (block).
trace – Defaults to False. If set to true, all interaction between clients and managers is logged. This results in large logs, but may help in debugging.

Returns:

None and a new instance of a distributed dictionary is initialized.

Raises:

AttributeError – If incorrect parameters are supplied.
RuntimeError – If there was an unexpected error during initialization.

setup_logging()

destroy(allow_restart=False) → None : Destroy a Distributed Dictionary instance, freeing all the resources that were allocated when it was created. Any clients that are still attached to the dictionary and try to do an operation on it will experience an exception when attempting subsequent operations.

serialize() → str 

Returns a serialized, base64 encoded descriptor (i.e. string) that may be shared with other processes for attaching. This is especially useful when sharing with C/C++ or Fortran code though not all clients are available yet. Within Python you can pass the Distributed Dictionary to another process and it will be automatically serialized and attached so using this method is not needed when passing to another Python process.

Returns:: A serialized, base64 encoded string that may be used for attaching to the dictionary.

classmethod attach(serialized_dict: str , *, timeout: float = None, trace: bool = False) → DDict

Within Python you typically do not need to call this method explicitly. It will be done automatically when you pass a Distributed Dictionary from one process to another. However, you can do this explicitly if desired/needed.

Parameters:

serialized_dict – A serialized distributed dictionary.
timeout – None or a float or int value. A value of None means to wait forever. Otherwise it is the number of seconds to wait while an operation is performed. This timeout is applied to all subsequent client operations that are performed by the process that is attaching this DDict.
trace – If True, specifies that all operations on the distributed dictionary should be logged in detail within the client log.

Returns:

An attached serialized dictionary.

Return type:

DDict

Raises:

TimeoutError – If the timeout expires.
Exception – Other exceptions are possible if for instance the serialized dictionary no longer exists.

detach() → None : Detach from the Distributed Dictionary and free all local resources of this client. But leave in place the DDict for other clients and processes.

classmethod synchronize_ddicts(serialized_ddicts: list [str ]) → None 

Synchronize managers across all parallel dictionaries.

Parameters:: serialized_ddicts – a list of serialized dictionaries.

clone(clone_list: list [str ]) → None 

Clone dictionary to the list of dictionaries.

Parameters:: clone_list – a list of serialized dictionaries.

manager(id: int ) → DDict

Return a version of the current ddict that will always choose the given manager for storing and retrieving data.

Parameters:: id – The manager id of the chosen manager.
Raises:: Exception – If the manager id is not a valid id.

pickler(key_pickler=None, value_pickler=None) → DDict

which_manager(key: object ) → int 

Return the manager id of the manager to which this key would be sent on a put/store operation. This can be useful when wanting to minimize the movement of data.

Parameters:: key – A key that might be stored at some future time. It must be serializable.
Returns:: The manager id of the manager where this key would be stored.

__setitem__(key: object , value: object ) → None 

Store the key/value pair in the current checkpoint within the Distributed Dictionary. Due to the nature of a parallel, distributed dictionary, insertion order into the distributed dictionary is not maintained.

Parameters:

key – The key of the pair. It must be serializable.
value – the value of the pair. It also must be serializable.

Raises:

Exception – Various exceptions can be raised including TimeoutError.

__getitem__(key: object ) → object 

Get the value that is associated with the given key.

Parameters:: key – The key of a stored key/value pair.
Returns:: The value associated with the key.
Raises:: Exception – Various exceptions can be raised including TimeoutError and KeyError.

__contains__(key: object ) → bool 

Returns True if key is in the Distributed Dictionary and False otherwise.

Parameters:: key – A possible key stored in the DDict.
Returns bool:: True or False depending on if the key is there or not.
Raises:: Various exceptions can be raised including TimeoutError.

__len__() → int 

Returns the number of keys stored in the entire Distributed Dictionary.

Returns int:: The number of stored keys in the current checkpoint plus any persistent keys.
Raises:: Various exceptions can be raised including TimeoutError.

__delitem__(key: object ) → None 

Deletes a key/value pair from the Distributed Dictionary if it exists.

Raises:: Various exceptions can be raised including TimeoutError and KeyError.

pput(key: object , value: object ) → None 

Persistently store a key/value pair within the Distributed Dictionary. This is useful when checkpointing is employed in the dictionary. A persistent put of a key/value pair means that the key/value pair persists across checkpoints. Persistent key/value pairs are useful when putting constant values or other values that don’t change across checkpoints.

Parameters:

key – A serializable object that will be stored as the key in the DDict.
value – A serializable object that will be stored as the value.

bput(key: object , value: object ) → None 

Store a non-persistent key/value pair by brodcasting to all managers across the Distributed Dictionary. This is useful when multiple clients requesting the same key. This should be used carefully as each manager holds a duplicate of the key/value pair.

Parameters:

key – A serializable object that will be stored as the key in the DDict.
value – A serializable object that will be stored as the value.

bget(key: object ) → object 

Read the key written through bput. Each manager has a copy of the key, the client should be able to request the key from its main manager. Client request the key from chosen manager if there is one. Otherwise the client request key from its main manager.

Parameters:: key – The key of a stored key/value pair.
Returns:: The value associated with the key.
Raises:: Exception – Various exceptions can be raised including TimeoutError and KeyError.

local_len() → int 

local_keys() → list [object ]

Returns a DDictKeysView of the keys that are local to the process invoking this method.

Returns:: A DDictKeysView of the current DDict which has only the

co-located node local keys of the DDict in it.

keys() → list [object ]

Returns a keys view of the distributed dictionary. From this view you can iterate over the keys or get the number of keys (i.e. length operation). See dict view objects for the methods available on a ddict keys view.

Returns:: A DDictKeysView object which is a live view of the DDict.

pop(key: object , default: object = None) → object 

Pop the given key from the distributed dictionary and return the associated value. If the given key is not found in the dictionary, then KeyError is raised unless a default value is provided, in which case the default value is returned if the key is not found in the dictionary.

Parameters:

key – A key to be popped from the distributed dictionary.
default – A default value to be returned if the key is not in the distributed dictionary.

Returns:

The associated value if key is popped and the default value otherwise.

clear() → None : Empty the distributed dictionary of all keys and values.

get_name()

start_batch_put(persist=False) → None : Calling other APIs except for put before the batch put ends could leads to a hang or exception.

end_batch_put() → None 

values() → list [object ]

When called this returns a list of all values in the Distributed Dictionary.

Returns list[object]:: A list of all values in the DDict.

local_values() → list [object ]

Returns a DDictValuesView of the keys that are local to the process invoking this method.

Returns:: A DDictValuesView of the current DDict which has only the

co-located node local values of the DDict in it.

items() → list [tuple [object , object ]]

Returns a list of all key/value pairs in the Distributed Dictionary.

Returns list[tuple[object,object]]:: A list of all key/value pairs.

local_items() → list [object ]

Returns a DDictItemsView of the keys that are local to the process invoking this method.

Returns:: A DDictItemsView of the current DDict which has only the

co-located node local items of the DDict in it.

update(dict2: DDict) → None 

Adds all key/value pairs from dict2 into this Distributed Dictionary.

Parameters:: dict2 – Another distributed dictionary.
Raises:: NotImplementedError – Not implemented.

popitem() → tuple [object , object ]

Returns a random key/value pair from the Distributed Dictionary.

Returns tuple[object,object]:: A random key/value pair.
Raises:: NotImplementedError – Not implemented.

copy(name: str = '') → DDict

Returns a copy of the Distributed Dictionary.

Returns DDict:: A second DDict that is a copy of the first assuming that no other processes were concurrently using this DDict.

property stats: list [DDictManagerStats]: Returns a list of manager stats, one for each manager of the distributed dictionary. See the DDictManagerStats structure for a description of its contents.

property dstats: dict [int , DDictManagerStats]: Returns a dict of manager stats, one for each manager of the distributed dictionary. See the DDictManagerStats structure for a description of its contents.

checkpoint() → None : Calling checkpoint advances the checkpoint for the distributed dictionary. In subsequent calls to the distributed dictionary, like gets or puts, if the chosen manager does not have the current checkpoint in its working set, the get/put operations will block until the checkpoint becomes available. But, calling this operation itself does not block.

rollback() → None : Calling rollback decrements the checkpoint id to its previous value. Again this call does not block. If rollback causes the checkpoint id to roll back to a checkpoint that a chosen manager no longer has in its working set, then subsequent operations may fail with a exception indicating the Checkpoint is no longer available, raising a DDictCheckpointSync exception.

sync_to_newest_checkpoint() → None : Advance the checkpoint identifier of this client to the newest checkpoint across all managers. This does not guarantee that all managers have advanced to the same checkpoint. That is up to the application which may guarantee all managers are at the same checkpoint by setting and getting values from managers in checkpoints and checkpoints advance. See the ddict_checkpoint_pi.py demo in examples/dragon_data/ddict for an example of an application that uses this method.

advance() → None : Advance to next available persisted checkpoint. This operation is for read only mode.

persist() → None : Persist current checkpoint to disk.

restore(chkpt: int ) → None : Restore a persisted checkpoint.

persisted_ids() → list [int ]: Get a list of persisted checkpoint IDs.

filter(mgr_code: LambdaType, mgr_code_args: tuple , comparator: LambdaType, branching_factor: int = 5)

Calling this instantiates a tree of process groups where mgr_code is expected to be a function that is invoked as mgr_code(args) where args are (dd, out_queue)+mgr_code_args. For instance, if mgr_code_args are (x,) then mgr_code(dd, outqueue, x) is how mgr_code is invoked.

The dd of the mgr_code arguments is this distributed dictionary directed toward one manager in the collection of dd managers. In other words, dd is as if the manager method had been invoked on this distributed dictionary so mgr_code only interacts with the manager it was provided. In addition, mgr_code is executed on the same node where the manager it is directed toward is running. This means that mgr code will get the best possible performance while filtering data that is associated with its manager. The mgr_code can do whatever computation is desired, but its chosen output is put into the outqueue.

All data written to outqueue is aggregated with data coming from each manager in a tree-like fashion so as to be scalable to tens of thousands of nodes. All data put in the outqueue by mgr_code is assumed to be ordered from best to worst. When data is aggregated for sending up the tree, it is aggregated according to some kind of ordering which is determined by the comparator function. The comparator will be called as comparator(x,y) and should return True if x is better than y and False otherwise. If there is no ordering, or the ordering is not relevant to the filtering, then comparator(x,y) may return a constant value of False or True and there will be no ordering of the data.

The branching_factor of the filtering tree has a default value, but may be provided by the user to create a tree of whatever width is desired. Note that branching_factor is the max branching factor. Depending on the number of managers, some nodes in the tree may/will have smaller numbers of children.

The filter function returns a Context Manager that supplies an Iterator over which you can iterate on the filtered values. So you can write with dd.filter(...) as candidates: and then iterate over candidates inside the context to read the filtered values.

Assuming your distributed dictionary is called dd, this will get num_needed elements from the result of filtering the distributed dictionary by calling the function get_largest on each distributed dictionary manager.

Parameters:: mgr_code – A function taking arguments as described above that will

run on the same node as a distributed dictionary manager and will be directed toward that manager.

Parameters:: comparator – A function taking two arguments that should return

True if the first argument of the values being filtered is “better” than the second and False otherwise. Note that returning a constant value of True or False will result in the filtering imposing no order.

Parameters:: branching_factor – The maximum branching factor of any interior

node in the filtering tree (i.e. any aggregator).

Returns:: A Context Manager that supplies an iterator over which you can

iterate over the filtered values.

property is_frozen: bool 

freeze() → None 

unfreeze() → None 

property checkpoint_id: int 

This returns the clients current checkpoint id.

Returns:: int: The current checkpoint id of the client.

property local_managers: list [int ]

Returns all local manager ids of all managers that are local to this node.

Raises:: NotImplementedError – Not implemented yet.

property local_manager: int : Returns a local manager id if one exists. This is manager designated as the main manager for the client. If no local manager exists, the None is returned.

property main_manager: int : Returns the main manager id. This will always exist and will be the same as the local manager id if a local manager exists. Otherwise, it will be the id of a random manager from another node.

property manager_nodes: list [str ]: For each manager, the serialized, base64 encoded FLI of the manager is returned.

property empty_managers: Return a list of manager IDs that restarted on new nodes.