Paper draft

.gitignore

@@ -101,3 +101,4 @@ ENV/
 *.pdf
 *.log
 *.gz
+*.zip

dds_simulation/experiments/consistent_partitions.py

@@ -8,6 +8,9 @@
 
 
 class ConsistentExperiment(DDS):
+    """
+
+    """
     def _form_random_partitions(self):
         partitions_number = int(default.parameter('experiment', 'partitions'))
         consistent_partitions = []
@@ -55,10 +58,6 @@ def single_iteration(self, partition, nodes_original):
         i = 0
         consistent_partitions = self._form_partitions(partition)
         taking_count = 2
-        found_inconsistent_counts = [0 for i in
-                                     range(0, len(consistent_partitions))]
-
-        #for partition in consistent_partitions:
         nodes = deepcopy(nodes_original)
         random.shuffle(nodes)
 
@@ -75,6 +74,7 @@ def multiple_partitions(self, partitions_number):
         nodes = [node.identity for node in self.nodes]
 
         partitions_invariants = [i + 1 for i in range(len(nodes))]
+        print("part inv>>> ", partitions_invariants)
         partitions = list(
             filter(lambda partition: sum(partition) == len(nodes),
                    combinations_with_replacement(
@@ -83,7 +83,7 @@ def multiple_partitions(self, partitions_number):
         y = []
         x = []
         inconsistency_array = []
-
+        print("patritions: ", partitions)
         for part in partitions:
             print("=========================================")
             print("PARTITION: ", part)
@@ -96,6 +96,7 @@ def multiple_partitions(self, partitions_number):
 
                 probability = sum(inconsistency_array) / len(
                     inconsistency_array)
+
                 inconsistency_array.clear()
                 y.append(probability)
                 x.append(i)
@@ -105,16 +106,17 @@ def multiple_partitions(self, partitions_number):
 
             # probability of inconsistency:
             average = sum(y) / len(y)
+            maximum = max(y)
             print(average)
+            print(maximum)
             extrapolation.draw_probability_extrapolation(
-                x, y, part, len(nodes), average)
+                x, y, part, len(nodes), average, maximum)
             x.clear()
             y.clear()
 
         return x, y
 
     def start_experiment(self):
-
         partitions = int(default.parameter('experiment', 'partitions'))
 
         inconsistency_states = self.multiple_partitions(partitions)

dds_simulation/experiments/link.py

@@ -4,9 +4,10 @@ class Link(object):
 
     bandwidth = 1
 
-    def __init__(self, node1, node2):
+    def __init__(self, node1, node2, weight=None):
         self.node1 = node1
         self.node2 = node2
+        self.weight = weight
 
     def transmit(self, message):
         return self.bandwidth

dds_simulation/experiments/simulation.py

@@ -1,35 +1,79 @@
 import abc
 import random
 
+import networkx
+
 from dds_simulation.experiments.node import Node
 from dds_simulation.experiments.link import Link
 from dds_simulation.experiments.dataunit import Dataunit
 
+from dds_simulation.visualisation import graph_control, extrapolation
+from dds_simulation.conf import default
 
-class DDS(object):
 
+class DDS(object):
+    """
+    Base class for all experiments
+    """
     __metaclass__ = abc.ABCMeta
 
-    def __init__(self, graph, controller, distribution, dataunits_number):
+    def __init__(self, graph):
         self.nodes = list()
         self.links = set()
         self.graph = graph
-        self.controller = controller
-
+        self.nodes = []
         for node_ident in graph.nodes():
             neighbors = graph.neighbors(node_ident)
             self.nodes.append(Node(node_ident, neighbors))
 
         for node1, node2 in graph.edges():
             self.links.add(Link(node1, node2))
-
-        self.dataunits = DDS.generate_data(dataunits_number)
-        self.distribute_data(distribution)
+            int(default.parameter('topology', 'nodes'))
 
     @abc.abstractmethod
     def start_experiment(self):
         """Starting one of experiments"""
 
+
+class DDSMessaging(DDS):
+    """
+    Class simulating messaging broadcasting across DDS
+    """
+
+    def __init__(self, graph, distribution, dataunits_number):
+        super(DDSMessaging, self).__init__(graph)
+        self.controller = graph_control.GraphController(graph)
+        self.graph = graph
+        self.time_slots_taken = 0
+
+        self.nodes = []
+
+        weighted = bool(default.parameter('topology', 'weighted'))
+        eccentricity = {}
+        for node1, node2 in graph.edges():
+            self.links.add(Link(node1, node2))
+            if weighted:
+                graph[node1][node2]['weight'] = random.randint(1, 9)
+
+        for node_ident in graph.nodes():
+            neighbors = graph.neighbors(node_ident)
+            self.nodes.append(Node(node_ident, neighbors))
+            self._eccentricity(node_ident, eccentricity)
+
+        self.diameter = networkx.diameter(graph, e=eccentricity)
+
+        print("diameter > ", self.diameter)
+
+        self.dataunits = DDSMessaging.generate_data(dataunits_number)
+        self.distribute_data(distribution)
+
+    def _eccentricity(self, node, eccentricity):
+        path = networkx.single_source_dijkstra_path_length(
+            self.graph, node, weight='weight')
+        path = {k: v for k, v in path.items()
+                if not eccentricity.get(k) or v > eccentricity[k]}
+        eccentricity.update(path)
+
     @staticmethod
     def generate_data(dataunits):
         return [Dataunit(i) for i in range(dataunits)]
@@ -44,41 +88,106 @@ def distribute_data(self, distribution):
             for node in self.nodes:
                 node.add_dataunits(dataunits_per_node)
 
+    def start_experiment(self):
+        self.simulate_message()
+
+    def _simulate_message_single_iteration(self, node_identity,
+                                           nodes_processed):
+        neighbors = self.graph.neighbors(node_identity)
+        print("Node: ", node_identity)
+        print("NEIGHBORS: ", neighbors)
+
+        # assume now that go to each node takes 1 ts, broadcasting parallel
+        #print("ts taken>>> ", self.time_slots_taken)
+
+        nodes_processed.update(set(neighbors))
+        nodes_processed.update(set([node_identity, ]))
+        return neighbors
+
+    def _simulate_parallel_iterations(self):
+        pass
+
+    def _shortest_path_for_neighbors(self, i, neighbors):
+        if i in neighbors:
+            neighbors.remove(i)
+        path = networkx.single_source_dijkstra_path(
+            self.graph, i, weight='weight')
+
+        path.pop(i)
+        return [path[n][1] for n in neighbors if len(path[n]) > 1]
+
     def simulate_message(self):
         """Simulate communication process.
 
         Simulate communication process starting from one of
         nodes.
         """
-        dataunit_id = random.randint(0, len(self.dataunits))
-        node_identity = random.randint(0, len(self.nodes))
-        # 1 time slot when initial node updates own dataunit
-        time_slots_taken = 0
-        if dataunit_id in self.nodes[node_identity].dataunits_ids:
-            time_slots_taken += 1
-
-        self.controller.message_broadcast([node_identity], [])
+        # only first node is taken at random
+        node_identity = random.randint(0, len(self.nodes) - 1)
+        self.time_slots_taken = 0
 
+        new_neighbors = set()
         nodes_processed = set([node_identity])
-        while len(nodes_processed) < len(self.nodes):
-            time_slots_taken += 1
-            neighbors = self.nodes[node_identity].neighbors
-            # assume now that go to each node takes 1 ts, broadcasting parallel
-            time_slots_taken += 1
-
-            # assume that each node is updating itself also for 1 ts
-            time_slots_taken += 1
-            nodes_processed.update(set(neighbors))
-            self.controller.message_broadcast(list(nodes_processed), [])
-            node_identity = random.randint(0, len(neighbors))
+        print("node ident>>> ", node_identity)
+        neighbors = self.graph.neighbors(node_identity)
+        path_ahead = self._shortest_path_for_neighbors(node_identity, neighbors)
+        print("neighbors>>> ", neighbors)
+        nodes_processed.update(set(neighbors))
+        link_cost = min([self.graph[node_identity][n]['weight'] for n in path_ahead])
+        print("time slots taken first time>>> ", link_cost)
+        self.time_slots_taken += link_cost
 
-        return time_slots_taken
+        while len(nodes_processed) < len(self.nodes):
 
-    def track_delivery_time(self):
-        pass
+            # ноды должны тоже выбираться параллельно,
+            # а не только одна, должны передавать сообщение
+            # ВСЕ соседи ПАРАЛЛЕЛЬНО
+            link_costs = []
+            print("====================================")
+
+            for i in neighbors:
+                print("identity now>>>> ", i)
+                neighbors_candidates = set(self.graph.neighbors(i)) - nodes_processed
+                new_neighbors.update(neighbors_candidates)
+                print("neighbors now>>>> ", new_neighbors)
+                path_ahead = self._shortest_path_for_neighbors(
+                    i, new_neighbors)
+                print("path ahed>>> ", path_ahead)
+                max_weight = min([self.graph[i][n]['weight'] for n in path_ahead])
+                print("max weight now>>> ", max_weight)
+                link_costs.append(
+                    max_weight)
+            print("=============================")
+            print("link costs>> ", link_costs)
+
+            link_cost = min(link_costs)
+            print("link costs before addig >> ", link_cost)
+            nodes_processed.update(set(new_neighbors))
+            self.time_slots_taken += link_cost
+
+            print("current time slots taken>>> ", self.time_slots_taken)
+
+            neighbors.clear()
+            link_costs.clear()
+
+            neighbors = list(new_neighbors)[:]
+            print("neighbors in the end>>>> ", neighbors)
+            new_neighbors.clear()
+
+
+        if self.time_slots_taken > self.diameter:
+            labels = {k.identity: k.identity for k in self.nodes}
+            self.controller.draw_graph(
+                self.graph, labels, f'{self.time_slots_taken}-{self.diameter}')
+
+    def get_delivery_time(self):
+        """Useful for interactive mode now"""
+        return self.time_slots_taken
 
     def track_replicas_on_nodes(self):
+        """Useful for interactive mode now"""
         pass
 
     def represent_current_state(self):
+        """Useful for interactive mode now"""
         pass

dds_simulation/start_experiment.py

@@ -1,21 +1,54 @@
+import random
+
+import networkx
 
 from dds_simulation.conf import default
 from dds_simulation.visualisation import graph_building
-from dds_simulation.visualisation import graph_control, extrapolation
+
 from dds_simulation.experiments import simulation
 from dds_simulation.experiments import consistent_partitions
+from dds_simulation.visualisation.extrapolation import draw_function
+
+
+def _initiate_dds(distribution, dataunits_number):
+    degree = random.randint(2, 5) // 2 * 2
+    nodes = degree * 7
+    graph = graph_building.form_graph(6, 3)
+
+    try:
+        return simulation.DDSMessaging(graph, distribution,
+                                              dataunits_number)
+    except networkx.exception.NetworkXError:
+        return _initiate_dds(distribution, dataunits_number)
 
 
 if __name__ == '__main__':
 
     nodes = int(default.parameter('topology', 'nodes'))
+    edges = int(default.parameter('topology', 'links'))
     degree = int(default.parameter('topology', 'degree'))
-    graph = graph_building.form_graph(nodes, degree)
+    experiments = int(default.parameter('experiment', 'experiments'))
+
     dataunits_number = int(default.parameter('dataunit', 'dataunits'))
     distribution = default.parameter('dataunit', 'distribution')
+    x_vector = []
+    y_vector = []
+
+
+    for i in range(experiments):
+        print("====================")
+        print(i)
 
-    dds_service = consistent_partitions.ConsistentExperiment(
-        graph, None, distribution, dataunits_number)
-    dds_service.start_experiment()
+        dds_service = _initiate_dds(distribution, dataunits_number)
+        dds_service.start_experiment()
 
+        x_vector.append(dds_service.time_slots_taken)
+        y_vector.append(dds_service.diameter)
+    print("X>> ", len(x_vector),x_vector)
+    print("Y>>> ", len(y_vector), y_vector)
 
+    draw_function(x_vector, y_vector, 'T_c', 'D(G)',
+                  f'{experiments}-consistency-convergence-weighted')
+    # graph = graph_building.form_graph(nodes, degree)
+    # exp = consistent_partitions.ConsistentExperiment(graph)
+    # exp.start_experiment()