Merge pull request #10 from FranciscoKloganB/develop

Ready for Metropolis Hastings insertion

Author: FranciscoKloganB

hive/app/basic_simulation.py

@@ -8,7 +8,7 @@ def usage():
     print(" -------------------------------------------------------------------------")
     print(" Francisco Barros (francisco.teixeira.de.barros@tecnico.ulisboa.pt\n")
     print(" Run a simulation for Markov Chain Based Swarm Guidance algorithm on a P2P Network that persists files\n")
-    print(" Typical usage: basic_simulation.py --simfile=sim01.json --filepath=yourfile.txt\n")
+    print(" Typical usage: basic_simulation.py --simfile=sim01.json\n")
     print(" Display all optional flags and other important notices: basic_simulation.py --help\n")
     print(" -------------------------------------------------------------------------\n")
     sys.exit(" ")
@@ -25,8 +25,7 @@ def help():
 def main():
 
     try:
-        options, args = getopt.getopt(sys.argv[1:], "uhs:f:", ["usage", "help", "simfile=", "filepath="])
-
+        options, args = getopt.getopt(sys.argv[1:], "uhs:", ["usage", "help", "simfile="])
         for options, args in options:
             if options in ("-u", "--usage"):
                 usage()
@@ -36,10 +35,6 @@ def main():
                 sim_file_path = str(args).strip()
                 if not sim_file_path:
                     sys.exit("Invalid simulation filepath. A simulation file is required for execution rules.")
-            if options in ("-f", "--filepath"):
-                file_path = str(args)
-                if not file_path:
-                    sys.exit("Invalid share file. Simulation requires file to simulate sharing.")
 
     except getopt.GetoptError:
         usage()

hive/app/domain/Enums.py

@@ -11,3 +11,4 @@ class HttpCodes(Enum):
     OK = 200
     NOT_FOUND = 404
     TIME_OUT = 408
+    SERVER_DOWN = 521

hive/app/domain/Hivemind.py

@@ -3,14 +3,17 @@
 
 
 class MarkovMatrix:
+    # region docstrings
     """
     Implements a matrix which adheres to markov chain theory and implements some basic markov chains' behaviour
     :ivar states: identifiers for the buckets existing on network passed in matching order to their transition arrays
     :type list<str>
     :ivar transition_matrix: concrete markov matrix data structure with named rows and columns according to passed states
     :type 2D pandas.DataFrame
     """
+    # endregion
 
+    # region class variables, instance variables and constructors
     def __init__(self, states, transition_arrays):
         """
         Initialize the Markov Chain instance.
@@ -26,7 +29,9 @@ def __init__(self, states, transition_arrays):
             columns=states,
             index=states
         )
+    # endregion
 
+    # region instance methods
     def next_state(self, current_state):
         """
         Choose list variable given probability of each variable
@@ -36,4 +41,5 @@ def next_state(self, current_state):
         :type str
         """
         # https://docs.scipy.org/doc/numpy-1.14.0/reference/generated/numpy.random.choice.html
-        return np.random.choice(self.states, p=self.transition_matrix[current_state])
+        return np.random.choice(a=self.states, p=self.transition_matrix[current_state])
+    # endregion

hive/app/domain/MarkovMatrix.py

@@ -5,6 +5,7 @@
 
 
 class SharedFilePart:
+    # region docstrings
     """
     Represents a simulation over the P2P Network that tries to persist a file using stochastic swarm guidance
     :ivar part_name: original name of the file this part belongs to
@@ -22,7 +23,9 @@ class SharedFilePart:
     :ivar markov_matrix: container object describing and implementing Markov Chain behaviour
     :type hive.domain.MarkovChain
     """
+    # endregion
 
+    # region class variables, instance variables and constructors
     def __init__(self, part_name, part_number, part_data, ddv=None, transition_matrix_definition=None):
         """
         :param part_name: original name of the file this part belongs to
@@ -41,9 +44,9 @@ def __init__(self, part_name, part_number, part_data, ddv=None, transition_matri
         self.__part_id = part_name + "_#_" + str(part_number)
         self.__part_data = ConvertUtils.bytes_to_base64_string(part_data)
         self.__sha256 = CryptoUtils.sha256(part_data)
-        self.__desired_distribution = np.array(ddv).transpose() if ddv is not None else ddv
-        self.__markov_matrix = MarkovMatrix(transition_matrix_definition[0], transition_matrix_definition[1])
+    # endregion
 
+    # region properties
     @property
     def part_name(self):
         return self.__part_name
@@ -63,14 +66,4 @@ def part_data(self):
     @property
     def sha256(self):
         return self.__sha256
-
-    @property
-    def desired_distribution(self):
-        return self.__desired_distribution
-
-    @property
-    def markov_matrix(self):
-        return self.__markov_matrix
-
-    def get_next_state(self, worker_id):
-        return self.__markov_matrix.next_state(worker_id)
+    # endregion

hive/app/domain/SharedFilePart.py

@@ -1,71 +1,125 @@
+import numpy as np
+
 from utils import CryptoUtils
 from utils.ResourceTracker import ResourceTracker as rT
 from domain.Enums import HttpCodes
 
 
 class Worker:
+    # region docstrings
     """
     Defines a node on the P2P network. Workers are subject to constraints imposed by Hivemind, constraints they inflict
     on themselves based on available computing power (CPU, RAM, etc...) and can have [0, N] shared file parts. Workers
     have the ability to reconstruct lost file parts when needed.
-    :ivar hivemind: coordinator of the unstructured Hybrid P2P network that enlisted this worker for a Hive
-    :type str
+    :ivar file_parts: key part_name maps to a dict of part_id keys whose values are SharedFilePart
+    :type dict<str, dict<str, SharedFilePart>
     :ivar name: id of this worker node that uniquely identifies him in the network
     :type str
-    :ivar file_parts: part_id is a key to a SharedFilePart
-    :type dict<string, SharedFilePart>
+    :ivar hivemind: coordinator of the unstructured Hybrid P2P network that enlisted this worker for a Hive
+    :type str
+    :ivar routing_table: maps file name with state transition probabilities, from this worker to other workers
+    :type dict<str, pandas.DataFrame>
     """
+    # endregion
 
+    # region class variables, instance variables and constructors
     def __init__(self, hivemind, name):
-        self.hivemind = hivemind
-        self.name = name
         self.file_parts = {}
+        self.__routing_table = {}
+        self.name = name
+        self.hivemind = hivemind
+    # endregion
 
+    # region overriden class methods
     def __hash__(self):
         # allows a worker object to be used as a dictionary key
         return hash(str(self.name))
 
     def __eq__(self, other):
+        if isinstance(other, str):
+            return self.name == other
         return (self.hivemind, self.name) == (other.hivemind, other.name)
 
     def __ne__(self, other):
         return not(self == other)
+    # endregion
 
+    # region file recovery methods
     def __init_recovery_protocol(self, part):
         """
-        # TODO
         When a corrupt file is received initiate recovery protocol, if this is the node with the most file parts
         The recovery protocol consists of reconstructing the damaged file part from other parts on the system, it may be
         necessary to obtain other files from other nodes to initiate reconstruction
+        # Note to self - This is not important right now! This is only important after MCMC with metropolis hastings works
+        # For now assume that when a node dies, if it had less than N-K parts, his parts are given to someone else
         """
+        # TODO:
+        #  corrupted or missing file recovery algorithm
         pass
+    # endregion
+
+    # region instance methods
+    def set_file_routing(self, file_name, labeled_transition_vector):
+        """
+        :param file_name: a file name that is being shared on the hive
+        :type str
+        :param labeled_transition_vector: probability vector indicating transitions to other states for the given file
+        :type 1-D numpy.Array in column format
+        """
+        self.__routing_table[file_name] = labeled_transition_vector
 
-    def receive_part(self, part):
-        if CryptoUtils.sha256(part.part_data) == part.sha256:
-            self.file_parts[part.part_id] = part
+    def receive_part(self, part, no_check=False):
+        if no_check or CryptoUtils.sha256(part.part_data) == part.sha256:
+            if part.name in self.file_parts:
+                self.file_parts[part.name][part.part_id] = part
+            else:
+                self.file_parts[part.name] = {}
+                self.file_parts[part.name][part.part_id] = part
         else:
             print("part_name: {}, part_number: {} - corrupted".format(part.part_name, str(part.part_number)))
             self.__init_recovery_protocol(part)
 
     def send_part(self):
-        tmp = {}
-        for part_id, part in self.file_parts.items():
-                dest_worker = part.get_next_state(self.name)
+        for part_name, part_id_sfp_dict in self.file_parts.items():
+            tmp = {}
+            for part_id, sfp_obj in part_id_sfp_dict.items():
+                dest_worker = self.get_next_state(file_name=part_name)
                 if dest_worker == self.name:
-                    tmp[part_id] = part
+                    tmp[part_id] = sfp_obj
                 else:
-                    response_code = self.hivemind.simulate_transmission(dest_worker, part)
+                    response_code = self.hivemind.simulate_transmission(dest_worker, sfp_obj)
                     if response_code != HttpCodes.OK:
-                        tmp[part_id] = part
-        self.file_parts = tmp
+                        # TODO:
+                        #  make use of the HttpCode responses with more than a binary behaviour
+                        tmp[part_id] = sfp_obj
+            self.file_parts[part_name] = tmp
 
     def leave_hive(self, orderly=True):
+        """
+        Resets the field of the Worker instance
+        :param orderly: When True asks the hivemind (master node) to redistribute files belonging to the Worker instance
+        :type bool
+        """
         if orderly:
             self.hivemind.simulate_redistribution(self.file_parts)
         self.hivemind = None
         self.name = None
         self.file_parts = None
 
+    def get_next_state(self, file_name):
+        """
+        :param file_name: the name of the file the part to be routed belongs to
+        :type: str
+        :return: the name of the worker to whom the file should be routed too
+        :type: str
+        """
+        file_routing_table = self.__routing_table[file_name]
+        row_labels = [*file_routing_table.index.values]
+        label_probabilities = [*file_routing_table[self.name]]
+        return np.random.choice(a=row_labels, p=label_probabilities)
+    # endregion
+
+    # region static methods
     @staticmethod
     def get_resource_utilization(*args):
         """
@@ -82,6 +136,4 @@ def get_resource_utilization(*args):
         for arg in args:
             results[arg] = rT.get_value(arg)
         return results
-
-
-
+    # endregion

hive/app/domain/Worker.py

@@ -0,0 +1,70 @@
+class ConvergenceData:
+    # region docstrings
+    """
+    Holds data that helps an domain.Hivemind keep track of converge in a simulation
+    :cvar __DEVIATION_TOLERANCE: percentage in which a value on a distribution vector can deviate from another in eq cmp
+    :type float
+    :cvar MIN_CONVERGENCE_THRESHOLD: how many consecutive convergent stages must a file have to be considered converged
+    :type int
+    :ivar cswc: indicates how many consecutive steps a file has in convergence
+    :type int
+    :ivar largest_convergence_set: indicates the biggest set of consecutive steps throughout the simulaton for this file
+    :type int
+    :ivar convergence_set: list registering stages in which a file has seen convergence. Registers only when above min conv. threshold
+    :type list<int>
+    :ivar convergence_sets: list with all convergence sets found for this file during a simulation
+    :type list<list<int>>
+    """
+    # endregion
+
+    # region class variables, instance variables and constructors
+    __DEVIATION_TOLERANCE = 0.01
+    MIN_CONVERGENCE_THRESHOLD = 3
+
+    def __init__(self):
+        self.cswc = 0
+        self.convergence_set = []
+        self.convergence_sets = []
+        self.largest_convergence_set = 0
+    # endregion
+
+    # region instance methods
+    def cswc_increment_and_get(self, increment):
+        self.cswc += increment
+        return self.cswc
+
+    def try_set_largest(self):
+        if len(self.convergence_set) > self.largest_convergence_set:
+            self.largest_convergence_set = self.cswc
+
+    def try_update_convergence_set(self, stage):
+        if self.cswc >= ConvergenceData.MIN_CONVERGENCE_THRESHOLD:
+            self.convergence_set.append(stage)
+            return True
+        else:
+            return False
+
+    def save_sets_and_reset_data(self):
+        self.cswc = 0
+        if self.convergence_set:
+            self.try_set_largest()
+            self.convergence_sets.append(self.largest_convergence_set)
+            self.convergence_set = []
+    # endregion
+
+    # region static methods
+    @staticmethod
+    def equal_distributions(one, another):
+        row_count = len(one)
+        if row_count != len(another):
+            return False
+        for i in range(0, row_count):
+            deviation = another[i] * ConvergenceData.__DEVIATION_TOLERANCE
+            lower_bound = another[i] - deviation
+            upper_bound = another[i] + deviation
+            if lower_bound < one[i] < upper_bound:
+                continue
+            else:
+                return False
+        return True
+    # endregion