Basic configuration for queues.

Author: wtpayne

src/xact/cfg/__init__.py

@@ -17,6 +17,7 @@
 import xact.cfg.exception
 import xact.cfg.load
 import xact.cfg.override
+import xact.cfg.queue
 import xact.cfg.validate
 import xact.util
 import xact.util.serialization
@@ -98,8 +99,9 @@ def denormalize(cfg):
     such information explicit.
 
     """
-    return xact.cfg.edge.denormalize(
-                xact.cfg.data.denormalize(cfg))
+    return  xact.cfg.queue.denormalize(
+                xact.cfg.edge.denormalize(
+                    xact.cfg.data.denormalize(cfg)))
 
 
 # -----------------------------------------------------------------------------

src/xact/cfg/data/_meta/conftest.py

@@ -85,5 +85,8 @@ def valid_partly_denormalized_config():
                 }
             ],
             'some_type_alias': 'py_dict'
+        },
+        'queue': {
+
         }
     }

src/xact/cfg/queue.py

@@ -0,0 +1,50 @@
+
+# -*- coding: utf-8 -*-
+"""
+Module of functions that support the configuration of queues.
+
+"""
+
+# -----------------------------------------------------------------------------
+def denormalize(cfg):
+    """
+    Add default queue config.
+
+    """
+    if 'queue' not in cfg:
+        cfg['queue'] = dict()
+        cfg['queue']['inter_process']     = 'xact.queue.multiprocessing'
+        cfg['queue']['inter_host_server'] = 'xact.queue.zmq_server'
+        cfg['queue']['inter_host_client'] = 'xact.queue.zmq_client'
+
+    return cfg
+
+    # default_rules: (
+    #     {'edge_type': 'inter_host_server',
+    #      'from_node': ('python',),
+    #      'to_node':   ('python',),
+    #      'tags':      ('*',),
+    #      'logic':     'xact.queue.zmq_server'},
+    #     {'edge_type': 'inter_host_client',
+    #      'from_node': ('python',),
+    #      'to_node':   ('python',),
+    #      'tags':      ('*',),
+    #      'logic':     'xact.queue.zmq_client'},
+    #     {'edge_type': 'inter_process_server',
+    #      'from_node': ('python',),
+    #      'to_node':   ('python',),
+    #      'tags':      ('*',),
+    #      'logic':     'xact.queue.multiprocessing'},
+    #     {'edge_type': 'inter_process_client',
+    #      'from_node': ('python',),
+    #      'to_node':   ('python',),
+    #      'tags':      ('*',),
+    #      'logic':     'xact.queue.multiprocessing'},
+    #     {'edge_type': 'intra_process',
+    #     'from_node': ('python',),
+    #     'to_node':   ('python',),
+    #     'tags':      ('*',),
+    #     'logic':      None})
+    # cfg['queue']['default_rules'] = default_rules
+
+

src/xact/cfg/validate.py

@@ -196,6 +196,9 @@ def _normalized_cfg_schema():
                     'additionalProperties':  False
                 }
             },
+            'queue': {
+                'type': 'object'
+            },
             'req_host_cfg': {
                 'type': 'object'
             },
@@ -269,6 +272,7 @@ def _denormalized_cfg_schema():
     _denormalize_host_section(schema)
     _denormalize_node_section(schema)
     _denormalize_edge_section(schema)
+    schema['required'].append('queue')
     return schema
 
 

src/xact/host/__init__.py

@@ -21,7 +21,6 @@
 import xact.host.util
 import xact.log
 import xact.proc
-import xact.queue
 
 
 FIFO_HOST_CONTROL = "xact_host_control"
@@ -153,23 +152,35 @@ def _group_edges_by_class(cfg, id_host_local):
 
     """
     map_id_by_class = {
-        'ipc':    set(),
-        'server': set(),
-        'client': set()
+        'intra_process':     set(),
+        'inter_process':     set(),
+        'inter_host_server': set(),
+        'inter_host_client': set()
     }
 
     for cfg_edge in cfg['edge']:
         id_edge = cfg_edge['id_edge']
 
-        if id_host_local not in cfg_edge['list_id_host']:
+        # Ignore edges that don't impact the current host.
+        #
+        is_on_host_local = id_host_local in cfg_edge['list_id_host']
+        if not is_on_host_local:
             continue
-        if cfg_edge['ipc_type'] == 'inter_process':
-            map_id_by_class['ipc'].add(id_edge)
+
+        # Inter-host (i.e. remote) queues have a server end and a client end.
+        #
         if cfg_edge['ipc_type'] == 'inter_host':
             if id_host_local == cfg_edge['id_host_owner']:
-                map_id_by_class['server'].add(id_edge)
+                queue_type = 'inter_host_server'
             else:
-                map_id_by_class['client'].add(id_edge)
+                queue_type = 'inter_host_client'
+
+        # Inter-process and intra-process queues are the same class both ends.
+        #
+        else:
+            queue_type = cfg_edge['ipc_type']  # inter_process or intra_process
+
+        map_id_by_class[cfg_edge['ipc_type']].add(id_edge)
 
     return map_id_by_class
 
@@ -180,16 +191,22 @@ def _construct_queues(cfg, map_cfg_edge, map_id_by_class, id_host_local):
     Return a map from edge id to queue instance.
 
     """
-    map_queues = dict()
-    for id_edge in map_id_by_class['server']:
-        map_queues[id_edge] = xact.queue.RemoteQueueServer(
-                                    cfg, map_cfg_edge[id_edge], id_host_local)
-
-    for id_edge in map_id_by_class['ipc']:
-        map_queues[id_edge] = xact.queue.LocalQueue()
+    # Ensure queue implementations are loaded.
+    #
+    map_queue_impl = dict()
+    for (id_edge_class, spec_module) in cfg['queue'].items():
+        module = xact.proc.ensure_imported(spec_module)
+        if 'Queue' not in module.__dict__:
+            raise xact.signal.NonRecoverableError(
+                                cause = 'No Queue class in specified module.')
+        map_queue_impl[id_edge_class] = module.Queue
 
-    for id_edge in map_id_by_class['client']:
-        map_queues[id_edge] = xact.queue.RemoteQueueClient(
+    # Select the correct queue implementation for each edge.
+    #
+    map_queues = dict()
+    for (id_edge_class, queue_impl) in map_queue_impl.items():
+        for id_edge in map_id_by_class[id_edge_class]:
+            map_queues[id_edge] = queue_impl(
                                     cfg, map_cfg_edge[id_edge], id_host_local)
 
     return map_queues

src/xact/node/__init__.py

@@ -12,6 +12,7 @@
 import xact.log
 import xact.signal
 import xact.util
+import xact.proc
 
 
 # =============================================================================
@@ -115,7 +116,7 @@ def _load_from_module(spec_module):
     Log any syntax errors.
 
     """
-    module  = _ensure_imported(spec_module)
+    module = xact.proc.ensure_imported(spec_module)
 
     if _is_step(map_func = module.__dict__):
         fcn_reset = module.reset
@@ -143,20 +144,6 @@ def _load_serialized(spec, unpacker):
     return (fcn_reset, fcn_step)
 
 
-# -----------------------------------------------------------------------------
-def _ensure_imported(spec_module):
-    """
-    Import the specified module or throw a NonRecoverableError
-
-    """
-    module = None
-    with xact.log.logger.catch():
-        module = importlib.import_module(spec_module)
-    if module is None:
-        raise xact.signal.NonRecoverableError(cause = 'Module not found.')
-    return module
-
-
 # -----------------------------------------------------------------------------
 def _is_step(map_func):
     """

src/xact/proc/__init__.py

@@ -6,6 +6,7 @@
 
 
 import collections
+import importlib
 import itertools
 import multiprocessing
 
@@ -57,6 +58,20 @@ def start(cfg, id_process, id_process_host, map_queues):
     raise RuntimeError('Termination condition not recognized.')
 
 
+# -----------------------------------------------------------------------------
+def ensure_imported(spec_module):
+    """
+    Import the specified module or throw a NonRecoverableError
+
+    """
+    module = None
+    with xact.log.logger.catch():
+        module = importlib.import_module(spec_module)
+    if module is None:
+        raise xact.signal.NonRecoverableError(cause = 'Module not found.')
+    return module
+
+
 # -----------------------------------------------------------------------------
 def _configure(id_process,
                map_cfg_node,
@@ -73,11 +88,13 @@ def _configure(id_process,
                                    map_cfg_node,
                                    map_alloc,
                                    runtime)
-    _config_edges(id_process,
-                  iter_cfg_edge,
-                  map_node,
-                  map_queues,
-                  map_alloc)
+
+    _configure_edges(id_process,
+                     iter_cfg_edge,
+                     map_node,
+                     map_queues,
+                     map_alloc)
+
     list_node = _get_list_node_in_runorder(id_process,
                                            map_cfg_node,
                                            iter_cfg_edge,
@@ -121,7 +138,11 @@ def _instantiate_nodes(id_process, map_cfg_node, map_alloc, runtime):
 
 
 # -----------------------------------------------------------------------------
-def _config_edges(id_process, iter_cfg_edge, map_node, map_queues, map_alloc):
+def _configure_edges(id_process,
+                     iter_cfg_edge,
+                     map_node,
+                     map_queues,
+                     map_alloc):
     """
     Configure the edges in the data flow graph.