02_Gossip_Services.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2. Adding services to the simulation\n",
    "\n",
    "Service is a key component to the simulation that defines a running scenario for your experiment.\n",
    "Each service has access to peer class and Simpy environment. \n",
    "\n",
    "At the moment simulation has two abstract service classes: \n",
    "1. Runner  \n",
    "2. Handler\n",
    "\n",
    "\n",
    "### **Runner**\n",
    "\n",
    "Runner works as a producer of events, or when you require an action initiative from a peer. For example, a message producer is a runner.\n",
    "\n",
    "Function to implement: `run()`\n",
    "\n",
    "\n",
    "### **Handler**\n",
    "\n",
    "Handler works as a reactive service that handles incoming messages. \n",
    "\n",
    "Functions to implement: \n",
    " - `handle_message(msg)` - react upon recieving a message.\n",
    " - `@property messages` - list of messages that this service can handle.  \n",
    "\n",
    "\n",
    "For example, a connection manager is both a runner and a handler as it produces and reacts on messages. \n",
    "\n",
    "\n",
    "\n",
    "# Creating new services\n",
    "\n",
    "\n",
    "The simulation at this point is not very useful. \n",
    "Let's create our own service on top of what we have now, i.e. let's build a gossip network.\n",
    "\n",
    "\n",
    "A gossip protocol ([wiki](https://en.wikipedia.org/wiki/Gossip_protocol)) is an important building block used in almost any P2P system(for example: [hyperledger gossip](https://hyperledger-fabric.readthedocs.io/en/release-1.4/gossip.html), [course slides](http://www.cs.cornell.edu/courses/cs6410/2016fa/slides/19-p2p-gossip.pdf)). \n",
    "\n",
    "\n",
    "## Gossip Message Handler\n",
    "\n",
    "There are multiple ways to implement Gossip: [ttl, anti-entropy etc](http://www.cs.cornell.edu/Projects/Quicksilver/public_pdfs/2007PromiseAndLimitations.pdf).\n",
    "\n",
    "We will implement one of the simplest versions based on ttl (Time to live), where each message has a predefined $ttl$ parameter. $ttl$ shows a stoping condition for the peer, i.e. if $ttl$=0, the peer should stop, otherwise $ttl$ must be decremented and the message should be gossiped further.\n",
    "\n",
    "Let's first implement a handler: a service that reacts on incoming gossip messages: \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:16:57.059695Z",
     "start_time": "2020-03-27T09:16:56.964066Z"
    }
   },
   "outputs": [],
   "source": [
    "# Message gossip\n",
    "from p2psimpy import *\n",
    "from p2psimpy.services.base import BaseHandler\n",
    "from p2psimpy.messages import BaseMessage\n",
    "\n",
    "from re import split \n",
    "from copy import copy\n",
    "\n",
    "# Define a special message GossipMessage: Message with ttl\n",
    "class GossipMessage(BaseMessage):\n",
    "\n",
    "    __slots__ = ('sender', 'data', 'ttl')\n",
    "    size = 1024\n",
    "    \n",
    "    def __init__(self, sender, data, ttl):\n",
    "        super().__init__(sender, data)\n",
    "        self.ttl = ttl\n",
    "        \n",
    "class GossipService(BaseHandler):\n",
    "    \"\"\"\n",
    "    A simple gossip service to handle the gossip messages and pass them to the neighbors. \n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, peer, fanout=3, exclude_peers: set=None, exclude_types: set=None):\n",
    "        super().__init__(peer)\n",
    "        \n",
    "        self.fanout = fanout\n",
    "        if exclude_peers is None:\n",
    "            self.exclude_peers = set() \n",
    "        else:\n",
    "            self.exclude_peers = exclude_peers\n",
    "        self.exclude_types = exclude_types\n",
    "        \n",
    "        self.strg_name = 'msg_time'\n",
    "        self.times_name = 'times_seen'\n",
    "        \n",
    "\n",
    "    def handle_message(self, msg):\n",
    "        # Store message localy \n",
    "        msg_id = msg.data\n",
    "        # Store the message id received with the current timestamp\n",
    "        if not self.peer.get_value(self.strg_name, msg_id):\n",
    "            self.peer.store_value(self.strg_name, msg_id, self.peer.env.now)\n",
    "            self.peer.store_value(self.times_name, msg_id, 1)\n",
    "        else: \n",
    "            old_value = self.peer.get_value(self.times_name, msg_id)\n",
    "            new_value = old_value + 1\n",
    "            self.peer.store_value(self.times_name, msg_id, new_value)\n",
    "        if msg.ttl > 0:\n",
    "            # Rely message further, modify the message\n",
    "            exclude_peers = {msg.sender} | self.exclude_peers\n",
    "            # Use peer gossip - it will sample self.config.fanout and exclude sender\n",
    "            # If you need to exclude some peers: add it to the set\n",
    "            self.peer.gossip( GossipMessage(self.peer, msg.data, msg.ttl-1), \n",
    "                             self.fanout, except_peers=exclude_peers, except_type=self.exclude_types)\n",
    "\n",
    "    @property\n",
    "    def messages(self):\n",
    "        return GossipMessage,\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Peer object has a gossip function that will send a message to a random sample of `fanout` connected peers.\n",
    "Peers are included in `except_peers` and `except_type` are excluded from the sampling. \n",
    "\n",
    "\n",
    "The service implements a reaction on the messages specified in the `messages` property and reacts as follows: \n",
    "- Peer stores the message when it is received and keeps the timestamp when the message was received - `self.peer.env.now`. Note also that to store the message first we need to initilize it via `self.peer.add_storage(storage_name, storage_object)`. \n",
    "- If the message ttl is positive, the peer sends the message to the `fanout` number of random connected peers with a function `self.peer.gossip(Msg, fanout)`. You can also exclude some peers from the sampling like excluding the sender of the message (from which you get the message) is a reasonable optimization. \n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Message producer \n",
    "\n",
    "Finally, we need to create a service that will first create (produce) the message. \n",
    "We will inherit a MessageProducer from a `BaseRunner` class.\n",
    "\n",
    "MessageProducer will have `msg_rate`, `ttl`, `fanout` and a `timeout_delay`.\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:16:57.129301Z",
     "start_time": "2020-03-27T09:16:57.073884Z"
    }
   },
   "outputs": [],
   "source": [
    "from p2psimpy.services.base import BaseRunner\n",
    "\n",
    "class MessageProducer(BaseRunner):\n",
    "\n",
    "    def __init__(self, peer, init_timeout=1000, msg_rate=5, init_ttl=3, init_fanout=10):\n",
    "        '''\n",
    "        init_timeout: milliseconds to wait before starting the message production. \n",
    "        msg_rate: number of messages per second\n",
    "        init_ttl: ttl to set up for the message \n",
    "        init_fanout: to how many peer send the message to\n",
    "        '''\n",
    "        super().__init__(peer)\n",
    "\n",
    "        # calculate tx_interval\n",
    "        self.init_timeout = init_timeout\n",
    "        self.init_ttl = init_ttl\n",
    "        self.init_fanout = init_fanout\n",
    "        \n",
    "        self.tx_interval = 1000 / msg_rate\n",
    "        self.counter = 1 \n",
    "        \n",
    "        # Let's add a storage layer to store messages\n",
    "        self.strg_name = 'msg_time'\n",
    "        self.times_name = 'times_seen'\n",
    "\n",
    "\n",
    "    def produce_transaction(self):\n",
    "        # Create a gossip message: message counter, peer_id and gossip it   \n",
    "        self.peer.gossip(GossipMessage(self.peer,\n",
    "                                       '_'.join((str(self.counter), str(self.peer.peer_id))), \n",
    "                                       self.init_ttl), \n",
    "                         self.init_fanout)\n",
    "        # Locally store the message counter \n",
    "        self.peer.store_value(self.strg_name, str(self.counter), self.peer.env.now)\n",
    "        self.peer.store_value(self.times_name, str(self.counter), 1)\n",
    "        self.counter+=1\n",
    "        \n",
    "\n",
    "    def run(self):\n",
    "        # Wait the initial timeout\n",
    "        yield self.env.timeout(self.init_timeout)\n",
    "        while True:\n",
    "            self.produce_transaction()\n",
    "            yield self.env.timeout(self.tx_interval)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice that `yield time_to_wait` is used to simulate time and waiting in a simulation. \n",
    "\n",
    "-----"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Preparing an Experiment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:17:05.902935Z",
     "start_time": "2020-03-27T09:16:57.140734Z"
    }
   },
   "outputs": [],
   "source": [
    "# Define locations \n",
    "from p2psimpy.config import *\n",
    "from p2psimpy.consts import *\n",
    "from p2psimpy.services.connection_manager import BaseConnectionManager\n",
    "import networkx as nx\n",
    "from random import choice\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "# We take the locations from AWS \n",
    "class Locations(Config):\n",
    "    locations = ['Ohio', 'Ireland', 'Tokyo']\n",
    "    latencies = {\n",
    "        'Ohio': {'Ohio': Dist('invgamma', (5.54090, 0.333305, 0.987249)),\n",
    "                 'Ireland': Dist('norm', (73.6995, 1.19583092197097127)),\n",
    "                 'Tokyo': Dist('norm', (156.00904977375566, 0.09469886668079797))\n",
    "                },\n",
    "        'Ireland':{'Ireland': Dist('invgamma', (6.4360455224301525, 0.8312748033308526, 1.086191852963273)),\n",
    "                   'Tokyo': Dist('norm', (131.0275, 0.25834811785650774))\n",
    "                  },\n",
    "        'Tokyo': {'Tokyo':  Dist('invgamma', (11.104508341331055, 0.3371934865734555, 2.0258998705983737))}\n",
    "    }\n",
    "    \n",
    "# Define peer     \n",
    "class PeerConfig(Config):\n",
    "    location = Dist('sample', Locations.locations)\n",
    "    bandwidth_ul = Dist( 'norm', (50*MBit, 10*MBit))\n",
    "    bandwidth_dl = Dist( 'norm', (50*MBit, 10*MBit))\n",
    "\n",
    "# Configuration used for our GossipService\n",
    "class GossipConfig(Config):\n",
    "    exclude_types={'client',}\n",
    "\n",
    "# We have not two types of nodes: *peer* and *client*\n",
    "def prepare_peer_types():\n",
    "    return { 'peer': PeerType(PeerConfig,   {BaseConnectionManager: None,\n",
    "                                                GossipService: GossipConfig }),\n",
    "             'client': PeerType(PeerConfig, (BaseConnectionManager, MessageProducer))}\n",
    "\n",
    "def prepare_topology(num_peers=25, num_clients=1):    \n",
    "    # Create network topology\n",
    "    G = nx.erdos_renyi_graph(num_peers, 0.4)   \n",
    "    nx.relabel_nodes(G, {k: k+1 for k in G.nodes()} ,copy=False)\n",
    "    \n",
    "    # Connect the client node to a random peer\n",
    "    client_edges = [(i, choice(list(G.nodes()))) for i in range(num_peers+1, num_clients+num_peers+1)]\n",
    "    G.add_edges_from(client_edges)\n",
    "\n",
    "    types_map = {k: 'peer' if k < num_peers+1 else 'client' for k in G.nodes()}\n",
    "    # Assign a peer type to the peers \n",
    "    nx.set_node_attributes(G, types_map , 'type')\n",
    "    return G\n",
    "\n",
    "def visualize_peer_client_network(G):\n",
    "    plt.figure(figsize=(10,10))\n",
    "\n",
    "    # Draw client/ peer network \n",
    "\n",
    "    master_nodes = [n for (n,ty) in \\\n",
    "        nx.get_node_attributes(G,'type').items() if ty == 'peer']\n",
    "    client_nodes = [n for (n,ty) in \\\n",
    "        nx.get_node_attributes(G,'type').items() if ty == 'client']\n",
    "\n",
    "    pos = nx.kamada_kawai_layout(G)\n",
    "\n",
    "    nx.draw_networkx_nodes(G, pos, nodelist=master_nodes, \\\n",
    "        node_color='blue', node_shape='o', node_size=500)\n",
    "    nx.draw_networkx_nodes(G, pos, nodelist=client_nodes,  \\\n",
    "        node_color='green', node_shape='^', node_size=100, label=1)\n",
    "\n",
    "    nx.draw_networkx_labels(G, pos, labels={k:k for k in master_nodes}, font_color='w')\n",
    "\n",
    "    nx.draw_networkx_edges(G, pos, edgelist=G.subgraph(master_nodes).edges(), width=1.5)\n",
    "    nx.draw_networkx_edges(G, pos, edgelist=G.edges(nbunch=client_nodes),  style='dotted')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:17:06.835115Z",
     "start_time": "2020-03-27T09:17:05.906070Z"
    }
   },
   "outputs": [],
   "source": [
    "# Prepare and show target network topology \n",
    "G = prepare_topology()\n",
    "visualize_peer_client_network(G)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:17:07.031220Z",
     "start_time": "2020-03-27T09:17:06.837122Z"
    }
   },
   "outputs": [],
   "source": [
    "from time import time\n",
    "from p2psimpy.simulation import Simulation\n",
    "\n",
    "net_sim = Simulation(Locations, G, prepare_peer_types())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:17:08.103312Z",
     "start_time": "2020-03-27T09:17:07.034437Z"
    }
   },
   "outputs": [],
   "source": [
    "\n",
    "net_sim.run(5_200)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "One of the main properties that we need to achieve with gossip is **convergence**.\n",
    "**Convergence** is a state where every peer has received the required message. \n",
    "The gossip protocols are compared with each other using the **time to convergence** metric.\n",
    "\n",
    "\n",
    "We can find it by inspecting the time when peer has the received message by looking at `peer.storage`.\n",
    "For this experiment we used a simple dictionary - *first-seen* storage. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:17:08.224114Z",
     "start_time": "2020-03-27T09:17:08.144922Z"
    }
   },
   "outputs": [],
   "source": [
    "client_id = 26\n",
    "print(\"Time when message was produced by the client: \\n\", net_sim.peers[client_id].storage['msg_time'].keys())\n",
    "\n",
    "# Put any peer id you want to inspect\n",
    "peer_id = 1 \n",
    "\n",
    "print(\"Time when the message was received by the peer: \\n\", net_sim.peers[peer_id].storage['msg_time'].keys())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:17:08.314634Z",
     "start_time": "2020-03-27T09:17:08.257650Z"
    }
   },
   "outputs": [],
   "source": [
    "# Calculate the delay \n",
    "\n",
    "def total_delay(sim, peer_id, storage_name):\n",
    "    store = sim.peers[peer_id].storage[storage_name].items()\n",
    "    for k, peer_time in store:\n",
    "        msg_num, client_id = k.split('_')\n",
    "        client_time = sim.peers[int(client_id)].storage[storage_name][msg_num]\n",
    "        yield (int(msg_num), peer_time - float(client_time))\n",
    "        \n",
    "dict(total_delay(net_sim, peer_id, 'msg_time'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:17:08.401407Z",
     "start_time": "2020-03-27T09:17:08.355154Z"
    }
   },
   "outputs": [],
   "source": [
    "dict(total_delay(net_sim, 12, 'msg_time'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise\n",
    "\n",
    "Try to play around with different peer_id values. Try peer that is directly connected to the client, then try peer that is on the edge of the network.\n",
    "\n",
    "Try the same with an emergent topology."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "____________________________"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Analyze and visualize  gossip\n",
    "\n",
    "\n",
    "When implementing gossip it is important to analyze following: \n",
    " - What is the average time for the peer to receive a message? \n",
    " - What is the convergence speed for my gossip protocol? \n",
    " - Can we guarantee that all peers will see the message? At what time message is finalized?   \n",
    "\n",
    "\n",
    "Let's go one by one and answer these questions for this gossip protocol implementation. \n",
    "\n",
    "\n",
    "## Convergence"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:17:11.084223Z",
     "start_time": "2020-03-27T09:17:08.414765Z"
    }
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "def total_delay(sim, peer_id, storage_name):\n",
    "    store = sim.peers[peer_id].storage[storage_name].items()\n",
    "    for k, peer_time in store:\n",
    "        msg_num, client_id = k.split('_')\n",
    "        client_time = sim.peers[int(client_id)].storage[storage_name][msg_num]\n",
    "        yield (int(msg_num), peer_time - float(client_time))\n",
    "        \n",
    "def get_gossip_table(sim, storage_name):\n",
    "    return pd.DataFrame({k: dict(total_delay(sim, k, storage_name)) \n",
    "                         for k in sim.types_peers['peer']}).sort_index()\n",
    "df = get_gossip_table(net_sim, 'msg_time')\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Code above should produce a table with the message number on the rows and peer id on the columns. \n",
    "\n",
    "You can also probably see `NaN` values in the table. These indicate that the gossip protocol didn't fully converge, and some peers never saw some messages. To guarantee convergence we either need to implement a gossip based on [set reconciliation](https://www.cse.ust.hk/~yike/sigmod14.pdf)  or tweak the parameters: `fanout` and `ttl`. We need to fine-tune the parameter to achieve the required effect. \n",
    "\n",
    "Usually gossip is fast to converge, but depends on the network topology. \n",
    "\n",
    "We can also represent our table as a heatmap. White color corresponds to `Nan`. You can see what peers are slower, or what message is least/most seen:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:17:16.386660Z",
     "start_time": "2020-03-27T09:17:11.121737Z"
    }
   },
   "outputs": [],
   "source": [
    "import seaborn as sns\n",
    "sns.set()\n",
    "\n",
    "def show_heat_map(df):\n",
    "    plt.figure(figsize=(12, 7))\n",
    "    ax = plt.axes()\n",
    "\n",
    "    sns.heatmap(df, ax=ax)\n",
    "\n",
    "    plt.xlabel('Peer number', fontsize = 12) # x-axis label with fontsize 15\n",
    "    plt.ylabel('Message number', fontsize = 12) # y-axis label with fontsize 15\n",
    "\n",
    "    plt.show()\n",
    "\n",
    "show_heat_map(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Average time "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:17:19.949408Z",
     "start_time": "2020-03-27T09:17:16.416905Z"
    }
   },
   "outputs": [],
   "source": [
    "def show_avg_time(df):\n",
    "    plt.figure(figsize=(10, 5))\n",
    "    ax = sns.boxplot(data=df, )\n",
    "    ax.set_title('Average time to arrival', fontsize= 20 )\n",
    "    ax.set_ylabel('Time (ms)', fontsize=12)\n",
    "    ax.set_xlabel('Peer id', fontsize=12)\n",
    "    plt.show()\n",
    "show_avg_time(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Convergence speed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:17:20.748150Z",
     "start_time": "2020-03-27T09:17:19.952022Z"
    }
   },
   "outputs": [],
   "source": [
    "def conv(df):\n",
    "    num_peers = len(df.columns)\n",
    "    num_messages = len(df.index)\n",
    "    \n",
    "    \n",
    "    a = df.values\n",
    "    a.sort(axis=1)\n",
    "    df2 = pd.DataFrame(a)\n",
    " \n",
    "\n",
    "    df2.columns = ((i+1)/num_peers for i in df2.columns)\n",
    "    df2.index = pd.RangeIndex(start=1, stop=num_messages+1, step=1)\n",
    "\n",
    "    return df2.stack().reset_index().rename(columns={'level_0':'msg_num', 'level_1':'convergance', 0: 'time'})\n",
    "\n",
    "def show_convergence(df):\n",
    "    plt.figure(figsize=(10,6))\n",
    "    sns.lineplot(x='time', y='convergance', data=conv(df), ci='sd', estimator=\"median\", hue='msg_num')\n",
    "    \n",
    "show_convergence(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Overhead on peers \n",
    "\n",
    "Each gossip creates an overhead for peers. \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "net_sim.peers[peer_id].storage['msg_time']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def calc_overhead(sim, peer_id, storage_name):\n",
    "    store = sim.peers[peer_id].storage[storage_name].items()\n",
    "    for k, times in store:\n",
    "        msg_num, client_id = k.split('_')\n",
    "        yield (int(msg_num), int(times))\n",
    "    \n",
    "def get_overhead_table(sim, storage_name):\n",
    "    return pd.DataFrame({k: dict(calc_overhead(sim, k, storage_name)) \n",
    "                         for k in sim.types_peers['peer']}).sort_index()\n",
    "\n",
    "oh = get_overhead_table(net_sim, 'times_seen')\n",
    "show_heat_map(oh)            "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Show average overhead on message\n",
    "def show_overhead_hist(overhead_table):\n",
    "    plt.figure(figsize=(10, 6))\n",
    "    sns.histplot(overhead_table)\n",
    "    plt.xlabel('Overhead (# times the message is seen by the same peer)', fontsize=15)\n",
    "\n",
    "show_overhead_hist(oh)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "\n",
    "\n",
    "# Achieving full convergence \n",
    "\n",
    "\n",
    "This is an exploratory section where you have one task: **full convergence**.\n",
    "\n",
    "As it is clear from our gossip protocol, full convergence is not possible with this set of parameter configuration. What can we do to achieve it? \n",
    "\n",
    "Possible options: \n",
    "\n",
    "1. Tweak parameter of gossip protocol. Change `ttl` and `fanout`. But keep in mind that it will increase the load on the network. As our gossip protocol is blind, it might take a lot of resources to guarantee convergence. Also, `ttl` and `fanout` highly depend on the network topology. Try to change that and find out what the optimal `ttl` is as a function of the number of peers?     \n",
    "2. Change the network topology. What is an optimal topology for convergence?  \n",
    "3. Implement better gossip protocols. We were using a naive gossip, but there better variations based on [rumor-mongering and anti-entropy](http://www.cs.cornell.edu/courses/cs6410/2016fa/slides/19-p2p-gossip.pdf). A good start will be change gossip protocol to *pull-based* instead of *push-based*. \n",
    "\n",
    "\n",
    "\n",
    "\n",
    "Report for your gossip protocol: \n",
    "- If it achieves convergence. \n",
    "- Convergence speed.\n",
    "- Average time to convergence.\n",
    "- The overhead on peers. \n",
    "\n",
    "\n",
    "Report your findings: \n",
    "- How can you improve your gossip protocol further? \n",
    "- Try to change the topology? Can you still achieve convergence? \n",
    "- How big is the overhead on peers? How can you improve it? \n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, for reproducibility you can save the experiment data. Use `.save_experiment(dir)` to save configurations for your experiment to a `dir` directory: locations, topology and peer services.  Whole experiment will be saved in yaml files. As they are saved in human readable format, you can always open and read the experiment configurations.\n",
    "\n",
    "By default, it will not save the reference to implementations as it is not always [safe to do](https://github.com/yaml/pyyaml/wiki/PyYAML-yaml.load(input)-Deprecation). But if you are confident enough that other notebooks can find the service implementation feel free to use `include_module_classes=True`. \n",
    "\n",
    "For example, `GossipService` and `MessageProducer` cannot be saved as they are defined only within the scope of one notebook. If you want to include also implementations of services make sure to save them first somewhere where other notebooks can find it, e.g. to save into `py` file in `p2psimpy.services` module. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2020-03-27T09:20:52.914213Z",
     "start_time": "2020-03-27T09:20:52.858287Z"
    }
   },
   "outputs": [],
   "source": [
    "net_sim.save_experiment(expr_dir='gossip_expr')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "hide_input": false,
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.9"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "position": {
    "height": "484px",
    "left": "808px",
    "right": "20px",
    "top": "142px",
    "width": "353px"
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}