fail_stop_agreement.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!pip install ipynb"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import threading\n",
    "import random\n",
    "from protocol_tests import test_all\n",
    "\n",
    "import sys\n",
    "try:\n",
    "    del sys.modules['ipynb.fs.full.weak_global_coin']\n",
    "except KeyError:\n",
    "    pass\n",
    "\n",
    "import ipynb.fs.full.weak_global_coin as QCF\n",
    "\n",
    "try:\n",
    "    del sys.modules['globals']\n",
    "except KeyError:\n",
    "    pass\n",
    "from globals import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "broadcasted_messages = []\n",
    "first_to_decide = None\n",
    "\n",
    "QUESTION_MARK = \"?\"\n",
    "WAITING_MESSAGE = \"waiting\"\n",
    "HALF_PLUS_ONE = int(np.floor(n/2)) + 1\n",
    "\n",
    "broadcasting_lock = threading.Lock()\n",
    "decision_lock = threading.Lock()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Process:\n",
    "    def __init__(self, id, input_val) -> None:\n",
    "        self.id = id\n",
    "        self.input_val = input_val\n",
    "        self.round_messages = {}\n",
    "        self.output = None\n",
    "        self.decision_epoch = None\n",
    "        self.non_faulty = True\n",
    "    def __str__(self):\n",
    "        return f\"id: {self.id} | round_messages: {self.round_messages}\"\n",
    "    \n",
    "class BroadcastMessage(Message):\n",
    "    def __init__(self, process_id, receivers, epoch, round, message) -> None:\n",
    "        super().__init__(process_id, receivers)\n",
    "        self.epoch = epoch\n",
    "        self.round = round\n",
    "        self.message = message\n",
    "        self.read = [False for number in range(n)]\n",
    "    def __str__(self):\n",
    "        return f\"sender: {self.sender} | epoch: {self.epoch} | round: {self.round} | message: {self.message}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def broadcast(process_id, epoch, round, message):\n",
    "    new_msg = BroadcastMessage(process_id, list(range(n)), epoch, round, message)\n",
    "    broadcasting_lock.acquire()\n",
    "    broadcasted_messages.append(new_msg)\n",
    "    broadcasting_lock.release()\n",
    "\n",
    "def waiting_condition(num_received_messages, round):\n",
    "    if round == 1 or round == 2:\n",
    "        actual_alive_processes = [1 for pr in processes if pr.non_faulty].count(1)\n",
    "        return num_received_messages < actual_alive_processes\n",
    "    elif round == 3:\n",
    "        return num_received_messages < MAX_ALIVE_PROCESSES\n",
    "\n",
    "def receive(process, epoch, round, required_val=None):\n",
    "    num_received_messages = 0\n",
    "    while waiting_condition(num_received_messages, round):\n",
    "        broadcasting_lock.acquire()\n",
    "        for msg in broadcasted_messages:\n",
    "            if msg.epoch == epoch and msg.round == round and process.id in msg.receivers and not msg.read[process.id]:\n",
    "                if round == 3:\n",
    "                    assert(msg.message == required_val)\n",
    "                existing_count = process.round_messages.get(msg.message, 0)\n",
    "                existing_count += 1\n",
    "                process.round_messages.update({msg.message:existing_count})            \n",
    "\n",
    "                num_received_messages += 1\n",
    "                msg.read[process.id] = True\n",
    "        broadcasting_lock.release()\n",
    "\n",
    "def get_majority_value(process):\n",
    "    for value, count in process.round_messages.items():\n",
    "        if count >= HALF_PLUS_ONE:\n",
    "            return value\n",
    "    return QUESTION_MARK\n",
    "\n",
    "def get_most_frequent_val(process):\n",
    "    most_frequent_val = max(process.round_messages, key=process.round_messages.get)\n",
    "    \n",
    "    if most_frequent_val == QUESTION_MARK:\n",
    "        process.round_messages.pop(most_frequent_val)\n",
    "        most_frequent_val = None\n",
    "        if process.round_messages:          # if the dictionary is not empty after deleting the (first found) most frequent value\n",
    "            most_frequent_val = max(process.round_messages, key=process.round_messages.get)\n",
    "    \n",
    "    answer = most_frequent_val\n",
    "    number = process.round_messages.get(most_frequent_val, 0)\n",
    "    return answer, number"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def agreement(process):\n",
    "    current = process.input_val\n",
    "    next = False\n",
    "    epoch = 0\n",
    "    while True:\n",
    "        epoch += 1\n",
    "        \n",
    "        broadcast(process.id, epoch, 1, current)\n",
    "        if not next:\n",
    "            receive(process, epoch, 1)        \n",
    "            current = get_majority_value(process)\n",
    "        process.round_messages.clear()                  # needed so that round_messages can be reused for the counts of the next round\n",
    "\n",
    "        broadcast(process.id, epoch, 2, current)\n",
    "        if not next:\n",
    "            receive(process, epoch, 2)\n",
    "            answer, number = get_most_frequent_val(process)\n",
    "        process.round_messages.clear()\n",
    "\n",
    "        broadcast(process.id, epoch, 3, WAITING_MESSAGE)\n",
    "        if not next:\n",
    "            receive(process, epoch, 3, WAITING_MESSAGE)\n",
    "        process.round_messages.clear()\n",
    "\n",
    "        coin = QCF.quantum_coin_flip(processes, process, epoch)\n",
    "\n",
    "        if next: \n",
    "            break\n",
    "        \n",
    "        if number >= HALF_PLUS_ONE:\n",
    "            current = answer\n",
    "            next = True\n",
    "            process.decision_epoch = epoch\n",
    "            decision_lock.acquire()\n",
    "            global first_to_decide\n",
    "            if first_to_decide == None:\n",
    "                first_to_decide = process.id\n",
    "            decision_lock.release()\n",
    "        elif number >= 1:\n",
    "            current = answer\n",
    "        else:\n",
    "            current = coin\n",
    "\n",
    "        if not process.non_faulty:\n",
    "            process.decision_epoch = epoch\n",
    "            break\n",
    "    process.output = current\n",
    "    return current"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "threads = []\n",
    "processes = []\n",
    "\n",
    "for i in range(0, n):\n",
    "    pr = Process(i, str(random.choice([0,1])))\n",
    "    processes.append(pr)\n",
    "    thr = threading.Thread(target=agreement, args=((pr,)))\n",
    "    threads.append(thr)\n",
    "\n",
    "for thr in threads:    \n",
    "    thr.start()\n",
    "\n",
    "for thr in threads:\n",
    "    thr.join()\n",
    "\n",
    "print(\"*******   SOLUTION:   *******\")\n",
    "for pr in processes:\n",
    "    print(\"process(\", pr.id, \") = \", pr.output, \" @epoch: \", pr.decision_epoch, \" | input: \", pr.input_val, \" | faulty: \", not pr.non_faulty)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_all(processes, first_to_decide, broadcasted_messages)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "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.10.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}