Merge branch 'release-0.3'

Author: bbengfort

.gitignore

@@ -61,3 +61,4 @@ target/
 conf/gvas.yaml
 docs/slate
 .ipynb_checkpoints
+fixtures/*

conf/gvas-example.yaml

@@ -29,14 +29,37 @@ vizualization:
 random_seed:  42
 max_sim_time: 1000
 
+# generalized default values used across simulations
 defaults:
   network:
-    capacity: 500
+    capacity: 1000
     base_latency: 10
   cluster:
-    size: 4
+    size: 2
   rack:
-    size: 96
+    size: 4
+    egress_latency: 20
   node:
     cpus: 4
     memory: 16
+  program:
+    cpus: 1
+    memory: 4
+
+# simulation specific parameters
+simulations:
+
+  # a basic simulation to exercise the cluster objects
+  simple:
+
+    # the number of nodes to add to the cluster
+    node_count: 8
+
+    # the number of nodes that should start in a work phase
+    start_team_size: 4
+
+    # ranges for message size and values
+    min_msg_size: 10
+    max_msg_size: 50
+    min_msg_value: 10
+    max_msg_value: 50

gvas/__init__.py

@@ -17,32 +17,7 @@
 ## Imports
 ##########################################################################
 
-
-##########################################################################
-## Module Info
-##########################################################################
-
-__version_info__ = {
-    'major': 0,
-    'minor': 2,
-    'micro': 0,
-    'releaselevel': 'final',
-    'serial': 0,
-}
-
-
-def get_version(short=False):
-    """
-    Prints the version.
-    """
-    assert __version_info__['releaselevel'] in ('alpha', 'beta', 'final')
-    vers = ["%(major)i.%(minor)i" % __version_info__, ]
-    if __version_info__['micro']:
-        vers.append(".%(micro)i" % __version_info__)
-    if __version_info__['releaselevel'] != 'final' and not short:
-        vers.append('%s%i' % (__version_info__['releaselevel'][0],
-                              __version_info__['serial']))
-    return ''.join(vers)
+from .version import get_version
 
 ##########################################################################
 ## Package Version

gvas/base.py

@@ -22,6 +22,7 @@
 
 from gvas.config import settings
 from gvas.dynamo import Sequence
+from gvas.results import Results
 
 ##########################################################################
 ## Base Process Objects
@@ -74,16 +75,25 @@ def __init__(self, **kwargs):
 
         self.max_sim_time = kwargs.get('max_sim_time', settings.max_sim_time)
         self.env = simpy.Environment()
+        self.diary = Results()
 
     def script(self):
         """
         Use the environment to generate a script.
         """
         raise NotImplementedError("Every simulation requires a script.")
 
+    def complete(self):
+        """
+        Override for a final report or cleanup at the end of the run.
+        """
+        pass
+
     def run(self):
         """
         The entry point for all simulations.
         """
         self.script()
-        self.env.run(until=self.max_sim_time)
+        with self.diary.timer:
+            self.env.run(until=self.max_sim_time)
+        self.complete()

gvas/cluster/cluster.py

@@ -17,6 +17,8 @@
 # Imports
 ##########################################################################
 
+import random
+
 from gvas.config import settings
 from gvas.exceptions import ClusterLacksCapacity
 from .base import Machine
@@ -69,14 +71,24 @@ def filter(self, evaluator):
         Uses the evaluator function to test against the Node instances and
         return a list of matches.
         """
-        pass
+        return filter(evaluator, self.nodes)
 
     def first(self, evaluator):
         """
         Uses the evaluator function to test against the Node instances and
         return the first match.
         """
-        pass
+        for n in self.nodes:
+            if evaluator(n):
+                return n
+        return None
+
+    def random(self, evaluator=lambda n: True):
+        """
+        Uses the evaluator function to test against the Node instances and
+        return a random match.
+        """
+        return random.choice(self.filter(evaluator))
 
     def send(self, *args, **kwargs):
         """
@@ -117,7 +129,6 @@ def run(self):
         """
         Method to kickoff process simulation.
         """
-        # TODO: placeholder code
         yield self.env.timeout(1)
 
     @property
@@ -131,25 +142,38 @@ def id(self):
         return self._id
 
     @property
-    def first_available_rack(self):
+    def nodes(self):
+        """
+        Returns a generator to iterate through all of the nodes in the racks.
         """
+        for r in self.racks.itervalues():
+            for n in r.nodes.itervalues():
+                yield n
 
+    @property
+    def first_available_rack(self):
+        """
+        Returns the first rack with room for a node or raises
+        ClusterLacksCapacity.
         """
         ids = sorted(self.racks.keys())
 
         for id in ids:
             if not self.racks[id].full:
                 return self.racks[id]
 
-        else:
-            raise ClusterLacksCapacity()
-
-
-
-
-
+        raise ClusterLacksCapacity()
 
+    def __str__(self):
+        nodes = sum([len(r.nodes) for r in self.racks.itervalues()])
+        return "Cluster: id: {}, racks={},  nodes={}".format(
+            self.id,
+            self.size,
+            nodes
+        )
 
+    def __repr__(self):
+        return "<{}>".format(self.__str__())
 
 ##########################################################################
 # Execution

gvas/cluster/network.py

@@ -31,6 +31,7 @@ class Network(object):
 
     def __init__(self, env, parent=None, *args, **kwargs):
         self.parent = parent
+        self.message_count = 0
         self.capacity = kwargs.get(
             'capacity',
             settings.defaults.network.capacity
@@ -59,10 +60,10 @@ def send(self, size):
         Removes available bandwidth thereby simulating additional traffic on
         the network medium.
         """
-        try:
-            self.medium.get(size)
-        except ValueError:
+        if self.medium.level < size:
             raise BandwidthExceeded()
+        self.medium.get(size)
+        self.message_count += 1
 
     def recv(self, size):
         """
@@ -71,6 +72,7 @@ def recv(self, size):
         """
         try:
             self.medium.put(size)
+            self.message_count -= 1
         except ValueError:
             raise
 
@@ -93,9 +95,11 @@ def latency(self):
         return self.base_latency + delay
 
     def __str__(self):
-        return "Network: capacity={},  base_latency={}".format(
+        return "Network: capacity={},  bandwidth={}, base_latency={}, latency={}".format(
             self.capacity,
-            self.base_latency
+            self.bandwidth,
+            self.base_latency,
+            self.latency,
         )
 
     def __repr__(self):

gvas/cluster/node.py

@@ -27,6 +27,7 @@
 # Classes
 ##########################################################################
 
+
 class Node(Machine):
 
     def __init__(self, env, *args, **kwargs):
@@ -45,44 +46,55 @@ def create(cls, env, *args, **kwargs):
         while True:
             yield cls(env, *args, **kwargs)
 
-    def send(self, address, size, value=None):
+    def send(self, address, port, size, value=None):
         """
-        Puts a message onto the containing Rack.
+        Puts a message onto the parent Rack.
         """
-        pass
+        self.rack.send(address=address, port=port, size=size, value=value)
 
-    def recv(self):
+    def recv(self, port, size, value=None):
         """
-        Obtains  a message from the containing Rack.
+        Obtains a message from the parent Rack.
         """
-        pass
+        program = None
+        for p in self.programs.itervalues():
+            if port in p.ports:
+                program = p
+
+        if program:
+            program.recv(value)
 
     def assign(self, program):
         """
         Ingests a new Program for processing.  If there aren't enough resources
         available then raises `NodeLacksCapacity`.
         """
-        # TODO: ensure we have capacity
+        if self.idle_cpus < program.cpus:
+            raise NodeLacksCapacity('{} cpus requested but only {} are free.'
+                                    .format(program.cpus, self.idle_cpus))
+
+        if self.idle_memory < program.memory:
+            raise NodeLacksCapacity('{}GB requested but only {}GB are free.'
+                                    .format(program.memory, self.idle_memory))
 
         self.programs[program.id] = program
         program.node = self
-        program.run()
 
     def run(self):
         """
         Method to kickoff process simulation.
         """
-        # TODO: replace with actual code or returned Program process
-        while True:
-            yield self.env.timeout(1)
-            # print "Node {} checking in at {}".format(self.id, self.env.now)
+        yield self.env.timeout(1)
 
     @property
     def address(self):
         """
         Addressable identifier for this node containing the Rack and Node ID.
         """
-        pass
+        return "{}:{}".format(
+            self.rack.id,
+            self.id
+        )
 
     @property
     def id(self):
@@ -121,8 +133,6 @@ def __repr__(self):
         return "<{}>".format(self.__str__())
 
 
-
-
 ##########################################################################
 # Execution
 ##########################################################################

gvas/cluster/program.py

@@ -27,10 +27,10 @@
 class Program(NamedProcess):
 
     def __init__(self, env, *args, **kwargs):
-        self.cpus = kwargs.get('cpus', settings.defaults.program.cpus)
-        self.memory = kwargs.get('memory', settings.defaults.program.memory)
-        self.ports = kwargs.get('ports', [])
-        self.node = kwargs.get('node', None)
+        self.cpus = kwargs.pop('cpus', settings.defaults.program.cpus)
+        self.memory = kwargs.pop('memory', settings.defaults.program.memory)
+        self.ports = kwargs.pop('ports', [])
+        self.node = kwargs.pop('node', None)
         super(Program, self).__init__(env, *args, **kwargs)
 
     @classmethod