Merge pull request #38 from FranciscoKloganB/finding-where-files-are-lost

Ready for release

Author: FranciscoKloganB

hive/app/domain/Hive.py

@@ -23,14 +23,18 @@ class Hive:
     :ivar List[float, float] corruption_chances: used to simulate file corruption on behalf of the workers, to avoid keeping independant distributions for each part and each replica
     :ivar str id: unique identifier in str format
     :ivar Hivemind hivemind: reference to the master server, which in this case is just a simulator program
-    :ivar FileData Union[None, FileData]: instance of class FileData which contains information regarding the file persisted by this hive
     :ivar Dict[str, Worker] members: Workers that belong to this P2P Hive, key is worker.id, value is the respective Worker instance
+    :ivar FileData file: instance of class FileData which contains information regarding the file persisted by this hive
+    :ivar DataFrame desired_distribution: distribution hive members are seeking to achieve for each the files they persist together.
     :ivar int critical_size: minimum number of replicas required for data recovery plus the number of peer faults the system must support during replication.
     :ivar int sufficient_size: depends on churn-rate and equals critical_size plus the number of peers expected to fail between two successive recovery phases
-    :ivar int redudant_size: application-specific system parameter, but basically represents that the hive is to big
-    :ivar DataFrame desired_distribution: distribution hive members are seeking to achieve for each the files they persist together.
-    :ivar Dict[str, SharedFilePart] recoverable_parts: just an hammer
+    :ivar int original_size: stores the initial hive size
+    :ivar int redundant_size: application-specific system parameter, but basically represents that the hive is to big
+    :ivar int set_recovery_epoch_sum: stores the sum of the values returned by all SharedFilePart.set_recovery_epoch calls - used for simulation output purposes
+    :ivar int set_recovery_epoch_calls: stores how many times SharedFilePart.set_recovery_epoch calls was called  during the current epoch
+    :ivar bool running: indicates if the hive has terminated - used for simulation purposes
     """
+
     # region Class Variables, Instance Variables and Constructors
     def __init__(self, hivemind: hm.Hivemind, file_name: str, members: Dict[str, Worker]) -> None:
         """
@@ -43,15 +47,16 @@ def __init__(self, hivemind: hm.Hivemind, file_name: str, members: Dict[str, Wor
         self.corruption_chances: List[float] = [0, 0]
         self.id: str = str(uuid.uuid4())
         self.hivemind = hivemind
-        self.file: FileData = FileData(file_name)
         self.members: Dict[str, Worker] = members
+        self.file: FileData = FileData(file_name)
+        self.desired_distribution: pd.DataFrame = pd.DataFrame()
         self.critical_size: int = REPLICATION_LEVEL
         self.sufficient_size: int = self.critical_size + math.ceil(len(self.members) * 0.34)
         self.original_size: int = len(members)
-        self.redudant_size: int = self.sufficient_size + len(self.members)
-        self.desired_distribution = None
-        self.file.simulation_data.set_membership_maintenace_at_index(status="stable", size_before=len(members), size_after=len(members), i=0)
-        self.running = True
+        self.redundant_size: int = self.sufficient_size + len(self.members)
+        self.running: bool = True
+        self.set_recovery_epoch_sum: int = 0
+        self.set_recovery_epoch_calls: int = 0
         self.broadcast_transition_matrix(self.new_transition_matrix())  # implicitly inits self.desired_distribution within new_transition_matrix()
     # endregion
 
@@ -203,6 +208,7 @@ def execute_epoch(self, epoch: int) -> None:
                 self.running = False
         except Exception as e:
             self.set_fail("Unexpected exception: ".join(traceback.format_exception(etype=type(e), value=e, tb=e.__traceback__)))
+        self.file.simulation_data.set_delay_at_index(self.set_recovery_epoch_sum, self.set_recovery_epoch_calls, self.current_epoch)
 
     def is_running(self) -> bool:
         return self.running
@@ -224,7 +230,7 @@ def evaluate_hive_convergence(self):
             else:
                 self.file.current_distribution.at[worker.id, DEFAULT_COL] = 0
 
-        self.file.simulation_data.parts_in_hive[self.current_epoch] = parts_in_hive
+        self.file.simulation_data.set_parts_at_index(parts_in_hive, self.current_epoch)
 
         if not parts_in_hive:
             self.set_fail("hive has no remaining parts")
@@ -239,8 +245,10 @@ def evaluate_hive_convergence(self):
     # region Helpers
     def setup_epoch(self, epoch: int) -> None:
         self.current_epoch = epoch
-        self.corruption_chances[0] = 0.0  # np.log10(epoch).item() / 100.0
+        self.corruption_chances[0] = np.log10(epoch).item() / 300.0
         self.corruption_chances[1] = 1.0 - self.corruption_chances[0]
+        self.set_recovery_epoch_sum = 0
+        self.set_recovery_epoch_calls = 0
 
     def workers_execute_epoch(self, lost_parts_count: int = 0) -> List[Worker]:
         """
@@ -257,11 +265,11 @@ def workers_execute_epoch(self, lost_parts_count: int = 0) -> List[Worker]:
                 lost_parts_count += len(lost_parts)
                 offline_workers.append(worker)
                 for part in lost_parts.values():
-                    part.set_epochs_to_recover(self.current_epoch)
+                    self.set_recovery_epoch(part)
                     if part.decrease_and_get_references() == 0:
-                        self.set_fail("lost all replicas of at least one file part")
+                        self.set_fail("lost all replicas of file part with id: {}".format(part.id))
         if len(offline_workers) >= len(self.members):
-            self.set_fail("lost all replicas of at least one file part")
+            self.set_fail("all hive members disconnected simultaneously")
         self.file.simulation_data.set_disconnected_and_losses(len(offline_workers), lost_parts_count, self.current_epoch)
         return offline_workers
 
@@ -278,20 +286,22 @@ def membership_maintenance(self, offline_workers: List[Worker]) -> None:
         if damaged_hive_size >= self.sufficient_size:
             self.remove_cloud_reference()
 
-        if damaged_hive_size >= self.redudant_size:
+        if damaged_hive_size >= self.redundant_size:
             status_before_recovery = "redundant"  # TODO: future-iterations evict worse members
-        elif self.original_size <= damaged_hive_size < self.redudant_size:
+        elif self.original_size <= damaged_hive_size < self.redundant_size:
             status_before_recovery = "stable"
         elif self.sufficient_size <= damaged_hive_size < self.original_size:
             status_before_recovery = "sufficient"
             self.members.update(self.__get_new_members())
         elif self.critical_size < damaged_hive_size < self.sufficient_size:
             status_before_recovery = "unstable"
             self.members.update(self.__get_new_members())
-        else:
+        elif 0 < damaged_hive_size <= self.critical_size:
             status_before_recovery = "critical"
             self.members.update(self.__get_new_members())
             self.add_cloud_reference()
+        else:
+            status_before_recovery = "dead"
 
         healed_hive_size = len(self.members)
         if damaged_hive_size != healed_hive_size:
@@ -302,12 +312,15 @@ def membership_maintenance(self, offline_workers: List[Worker]) -> None:
     def __get_new_members(self) -> Dict[str, Worker]:
         return self.hivemind.find_replacement_worker(self.members, self.original_size - len(self.members))
 
-    def set_fail(self, msg: str) -> bool:
+    def set_fail(self, msg: str) -> None:
         self.running = False
-        return self.file.simulation_data.set_fail(self.current_epoch, msg)
+        self.file.simulation_data.set_fail(self.current_epoch, msg)
 
     def tear_down(self, epoch: int) -> None:
         # self.hivemind.append_epoch_results(self.id, self.file.simulation_data.__repr__()) TODO: future-iterations where Hivemind has multiple hives
         self.file.jwrite(self.file.simulation_data, epoch)
 
+    def set_recovery_epoch(self, part: SharedFilePart) -> None:
+        self.set_recovery_epoch_sum += part.set_recovery_epoch(self.current_epoch)
+        self.set_recovery_epoch_calls += 1
     # endregion

hive/app/domain/Worker.py

@@ -30,7 +30,7 @@ class Worker:
     # region Class Variables, Instance Variables and Constructors
     def __init__(self, worker_id: str, worker_uptime: float):
         self.id: str = worker_id
-        self.uptime: float = math.ceil(worker_uptime * MAX_EPOCHS)
+        self.uptime: float = float('inf') if worker_uptime == 1.0 else math.ceil(worker_uptime * MAX_EPOCHS)
         self.hives: Dict[str, h.Hive] = {}
         self.files: Dict[str, Dict[int, SharedFilePart]] = {}
         self.routing_table: Dict[str, pd.DataFrame] = {}
@@ -167,9 +167,9 @@ def discard_part(self, name: str, number: int, corrupt: bool = False, hive: h.Hi
         part: SharedFilePart = self.files.get(name, {}).pop(number, None)
         if part and corrupt:
             if part.decrease_and_get_references() == 0:
-                hive.set_fail("lost all replicas of at least one file part")
+                hive.set_fail("lost all replicas of file part with id: {}, and last loss was due to corruption".format(part.id))
             else:
-                part.set_epochs_to_recover(hive.current_epoch)
+                hive.set_recovery_epoch(part)
 
     def get_file_parts(self, file_name: str) -> Dict[int, SharedFilePart]:
         """

hive/app/domain/helpers/SharedFilePart.py

@@ -37,7 +37,7 @@ def __init__(self, hive_id: str, name: str, number: int, data: bytes):
     # endregion
 
     # region Simulation Interface
-    def set_epochs_to_recover(self, epoch: int) -> int:
+    def set_recovery_epoch(self, epoch: int) -> int:
         """
         Assigns a value to the instance's recovery_epoch attribute that indicates when a Worker who posses a reference to it, can replicate the part.
         :param int epoch: current simulation's epoch

hive/app/domain/helpers/file_data.py

@@ -37,6 +37,9 @@ def equal_distributions(self, parts_in_hive: int) -> bool:
         """
         Delegates distribution comparison to ConvergenceData.equal_distributions static method
         """
+        if parts_in_hive == 0:
+            return False
+
         normalized_cdv = self.current_distribution.divide(parts_in_hive)
         if DEBUG:
             self.fwrite("Desired Distribution:\n{}\nCurrent Distribution:\n{}\n".format(
@@ -54,14 +57,17 @@ def fwrite(self, string: str) -> None:
         self.out_file.write(string + "\n")
 
     def jwrite(self, data: SimulationData, epoch: int):
-        print(self.simulation_data.msg)
-        stop: int = epoch+1
-        data.disconnected_workers = data.disconnected_workers[0:stop]
-        data.lost_parts = data.lost_parts[0:stop]
-        data.hive_status_before_maintenance = data.hive_status_before_maintenance[0:stop]
-        data.hive_size_before_maintenance = data.hive_size_before_maintenance[0:stop]
-        data.hive_size_after_maintenance = data.hive_size_after_maintenance[0:stop]
-        data.delay = data.delay[0:stop]
+        if not data.msg:
+            data.msg.append("completed simulation successfully")
+        if DEBUG:
+            [print("* {};".format(reason)) for reason in data.msg]
+        data.disconnected_workers = data.disconnected_workers[:epoch]
+        data.delay = data.delay[:epoch]
+        data.lost_parts = data.lost_parts[:epoch]
+        data.hive_status_before_maintenance = data.hive_status_before_maintenance[:epoch]
+        data.hive_size_before_maintenance = data.hive_size_before_maintenance[:epoch]
+        data.hive_size_after_maintenance = data.hive_size_after_maintenance[:epoch]
+        data.delay = data.delay[:epoch]
         json_string = json.dumps(data.__dict__, indent=4, sort_keys=True, ensure_ascii=False)
         self.fwrite(json_string)
 

hive/app/domain/helpers/simulation_data.py

@@ -31,21 +31,21 @@ def __init__(self):
         # Updated on Hive.execute_epoch
         self.terminated: int = MAX_EPOCHS  # gathered
         self.successfull: bool = True  # gathered
-        self.msg = "completed simulation successfully"  # gathered
-        self.disconnected_workers: List[int] = [0] * MAX_EPOCHS_PLUS  # gathered
+        self.msg = []  # gathered
+        self.disconnected_workers: List[int] = [0] * MAX_EPOCHS  # gathered
         self.lost_parts: List[int] = [0] * MAX_EPOCHS_PLUS  # gathered
-        self.hive_status_before_maintenance: List[str] = [""] * MAX_EPOCHS_PLUS  # gathered
-        self.hive_size_before_maintenance: List[int] = [0] * MAX_EPOCHS_PLUS  # gathered
-        self.hive_size_after_maintenance: List[int] = [0] * MAX_EPOCHS_PLUS  # gathered
+        self.hive_status_before_maintenance: List[str] = [""] * MAX_EPOCHS  # gathered
+        self.hive_size_before_maintenance: List[int] = [0] * MAX_EPOCHS  # gathered
+        self.hive_size_after_maintenance: List[int] = [0] * MAX_EPOCHS  # gathered
         self.delay: List[float] = [0.0] * MAX_EPOCHS_PLUS
         ###############################
         ###############################
         # Updated on Hive.route_part
-        self.moved_parts: List[int] = [0] * MAX_EPOCHS_PLUS  # gathered
-        self.corrupted_parts: List[int] = [0] * MAX_EPOCHS_PLUS  # gathered
-        self.lost_messages: List[int] = [0] * MAX_EPOCHS_PLUS  # gathered
+        self.moved_parts: List[int] = [0] * MAX_EPOCHS  # gathered
+        self.corrupted_parts: List[int] = [0] * MAX_EPOCHS  # gathered
+        self.lost_messages: List[int] = [0] * MAX_EPOCHS  # gathered
         ###############################
-        self.parts_in_hive: List[int] = [0] * MAX_EPOCHS_PLUS  # gathered
+        self.parts_in_hive: List[int] = [0] * MAX_EPOCHS  # gathered
 
     # endregion
 
@@ -119,12 +119,13 @@ def __repr__(self):
     # endregion
 
     # region Helpers
-    def set_delay_at_index(self, n: float, i: int) -> None:
+    def set_delay_at_index(self, delay: int, calls: int, i: int) -> None:
         """
-        :param float n: the delay at epoch i
+        :param int delay: the delay sum
+        :param int calls: number of times a delay was generated
         :param int i: index of epoch i in SimulationData.delay list
         """
-        self.delay[i] = n
+        self.delay[i-1] = 0 if calls == 0 else delay / calls
 
     def set_disconnected_and_losses(self, disconnected=0, lost=0, i=0):
         """
@@ -138,50 +139,53 @@ def set_moved_parts_at_index(self, n: int, i: int) -> None:
         :param int n: the quantity of parts moved at epoch i
         :param int i: index of epoch i in SimulationData.moved_parts list
         """
-        self.moved_parts[i] += n
+        self.moved_parts[i-1] += n
+
+    def set_parts_at_index(self, n: int, i: int) -> None:
+        """
+        :param int n: the quantity of parts moved at epoch i
+        :param int i: index of epoch i in SimulationData.parts_in_hive list
+        """
+        self.parts_in_hive[i-1] += n
 
     def set_disconnected_workers_at_index(self, n: int, i: int) -> None:
         """
         :param int n: the quantity of disconnected workers at epoch i
         :param int i: index of epoch i in SimulationData.disconnected_workers list
         """
-        self.disconnected_workers[i] += n
+        self.disconnected_workers[i-1] += n
 
     def set_lost_parts_at_index(self, n: int, i: int) -> None:
         """
         :param int n: the quantity of lost parts at epoch i
         :param int i: index of epoch i in SimulationData.lost_parts list
         """
-        self.lost_parts[i] += n
+        self.lost_parts[i-1] += n
 
     def set_lost_messages_at_index(self, n: int, i: int) -> None:
         """
         :param int n: the quantity of loss messages at epoch i
         :param int i: index of epoch i in SimulationData.lost_messages list
         """
-        self.lost_messages[i] += n
+        self.lost_messages[i-1] += n
 
     def set_corrupt_files_at_index(self, n: int, i: int) -> None:
         """
         :param int n: the quantity of corrupted parts at epoch i
         :param int i: index of epoch i in SimulationData.corrupted_parts list
         """
-        self.corrupted_parts[i] += n
+        self.corrupted_parts[i-1] += n
 
-    def set_fail(self, i: int, msg: str = "") -> bool:
+    def set_fail(self, i: int, msg: str = "") -> None:
         """
         Records the epoch at which the Hive terminated, should only be called if it finished early.
         Default, Hive.terminated = MAX_EPOCHS and Hive.successfull = True.
         :param int i: epoch at which Hive terminated
         :param str msg: a message
-        :returns bool: usually returns False, only returns True when param i, representing epoch is qual to MAX_EPOCHS
         """
-        if i == MAX_EPOCHS:
-            return True
         self.terminated = i
         self.successfull = False
-        self.msg = msg
-        return False
+        self.msg.append(msg)
 
     def set_membership_maintenace_at_index(self, status: str, size_before: int, size_after: int, i: int) -> None:
         """
@@ -190,13 +194,8 @@ def set_membership_maintenace_at_index(self, status: str, size_before: int, size
         :param int size_after: size of the hive after maintenace
         :param int i: index of epoch i in SimulationData.delay list
         """
-        if status is None or size_before is None or size_after is None:
-            self.hive_status_before_maintenance[i] = self.hive_status_before_maintenance[i-1]
-            self.hive_size_before_maintenance[i] = self.hive_size_before_maintenance[i-1]
-            self.hive_size_after_maintenance[i] = self.hive_size_after_maintenance[i-1]
-        else:
-            self.hive_status_before_maintenance[i] = status
-            self.hive_size_before_maintenance[i] = size_before
-            self.hive_size_after_maintenance[i] = size_after
+        self.hive_status_before_maintenance[i-1] = status
+        self.hive_size_before_maintenance[i-1] = size_before
+        self.hive_size_after_maintenance[i-1] = size_after
     # endregion
 

hive/app/globals/globals.py

@@ -3,12 +3,12 @@
 DEBUG: bool = True
 
 # region Simulation Settings
-READ_SIZE: int = 20971520  # 512KB = 524288 bytes || 32KB = 32768 bytes. Defines the raw size of each SharedFilePart.
-MAX_EPOCHS = 50  # One day has 24h, meaning that one epoch per minute wwould be 1440, 720 defines one epoch every two minutes
+READ_SIZE: int = 131072  # 32KB = 32768b || 128KB = 131072b || 512KB = 524288b || 20MB = 20971520b. Defines the raw size of each SharedFilePart.
+MAX_EPOCHS = 720  # One day has 24h, meaning that one epoch per minute wwould be 1440, 720 defines one epoch every two minutes
 MAX_EPOCHS_PLUS = MAX_EPOCHS + 1
 MIN_DETECTION_DELAY: int = 1  # 2 minutes
-MAX_DETECTION_DELAY: int = 2  # 2 minutes
-REPLICATION_LEVEL: int = 3  # Each file part has 3 copies, for simulation purposes, this copies are soft copies.
+MAX_DETECTION_DELAY: int = 3  # 6 minutes
+REPLICATION_LEVEL: int = 5  # Each file part has 3 copies, for simulation purposes, this copies are soft copies.
 MIN_CONVERGENCE_THRESHOLD: int = 3
 LOSS_CHANCE: float = 0.04  # Each sent file as a 4% chance of timing out due to message being lost in travel
 DELIVER_CHANCE: float = 1.0 - LOSS_CHANCE  # Each sent file as a 4% chance of timing out due to message being lost in travel