notebook.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import decimal\n",
    "import json\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import seaborn as sns\n",
    "import matplotlib.pyplot as plt\n",
    "from pprint import pprint\n",
    "\n",
    "%matplotlib inline\n",
    "\n",
    "import warnings\n",
    "warnings.filterwarnings(\"ignore\", category=FutureWarning)\n",
    "\n",
    "def merge_nonbuilder_latencies(latency_files, node_type):\n",
    "    \"\"\"\n",
    "    Merge all validator latency files into one dataframe\n",
    "    \"\"\"\n",
    "    \n",
    "    latency_dfs = []\n",
    "    for latency_file in latency_files:\n",
    "        latency_df = pd.read_csv(latency_file)\n",
    "        latency_df['filename'] = latency_file.replace(\"_latency_stats_\" +node_type+ \".csv\", \"\")\n",
    "        latency_df[\"node_type\"] = node_type\n",
    "        latency_df['block_id'] = range(len(latency_df))\n",
    "        latency_df = latency_df[ [\"node_type\", \"filename\", \"block_id\", \"Total Sampling Latency (us)\"] ]\n",
    "        \n",
    "        latency_dfs.append(latency_df)\n",
    "\n",
    "    return pd.concat(latency_dfs)\n",
    "\n",
    "def get_nonbuilder_df(latency_files, node_type):\n",
    "    latency_df = merge_nonbuilder_latencies(latency_files, node_type)\n",
    "    latency_df['Latency (s)'] = latency_df['Total Sampling Latency (us)'] / 1_000_000\n",
    "    latency_df = latency_df[ [\"node_type\", \"filename\", \"block_id\", \"Latency (s)\"] ]\n",
    "\n",
    "    return latency_df\n",
    "\n",
    "def get_latency_files(node_type):\n",
    "    if node_type == \"builder\":\n",
    "        return [file for file in os.listdir() if file.endswith('builder.csv') and \"latency_stats\" in file.lower()][0]\n",
    "    else:\n",
    "        return [file for file in os.listdir() if file.endswith('_' +node_type+ '.csv') and \"latency_stats\" in file.lower()]\n",
    "\n",
    "def get_total_putgets(node_id, node_type):\n",
    "    total_stats_files = [file for file in os.listdir() if node_id in file and 'total_stats' in file.lower()]\n",
    "    \n",
    "    total_stats_file = total_stats_files[0]\n",
    "\n",
    "    total_stats_df = pd.read_csv(total_stats_file)\n",
    "\n",
    "    if node_type == \"builder\":\n",
    "        total_putgets = total_stats_df['Total PUT messages'][0]\n",
    "        failed_putgets = total_stats_df['Total failed PUTs'][0]\n",
    "    else:\n",
    "        total_putgets = total_stats_df['Total GET messages'][0]\n",
    "        failed_putgets = total_stats_df['Total failed GETs'][0]\n",
    "\n",
    "    return total_putgets, failed_putgets\n",
    "\n",
    "def get_total_stats(latency_df):\n",
    "    for index, row in latency_df.iterrows():\n",
    "        total_putgets, failed_putgets = get_total_putgets(row['filename'], row['node_type'])\n",
    "        success_putgets = total_putgets - failed_putgets\n",
    "        latency_df.loc[index, 'Total PUT/GETs'] = total_putgets\n",
    "        latency_df.loc[index, 'Successful PUT/GETs'] = success_putgets\n",
    "\n",
    "    return latency_df\n",
    "\n",
    "def check_premature_cases(operations_df, print_premature_cases=False):\n",
    "    grouped_op_df = operations_df.groupby('Block ID')\n",
    "    for block_id, block_df in grouped_op_df:\n",
    "        block_case_count = 0\n",
    "        total_count = 0\n",
    "        block_df['Parcel Key Hash Int'] = block_df['Parcel Key Hash Int'].apply(decimal.Decimal)\n",
    "        for hash_int, hash_int_df in block_df.groupby(\"Parcel Key Hash Int\"):\n",
    "            node_count = len(hash_int_df)\n",
    "\n",
    "            if node_count == 1:\n",
    "                continue\n",
    "\n",
    "            earliest_timestamp = hash_int_df['timestamps'].min()\n",
    "            earliest_node_type = hash_int_df[hash_int_df['timestamps'] == earliest_timestamp]['node_type'].values[0]\n",
    "\n",
    "            # Is the earliest timestamp from a non builder node?\n",
    "\n",
    "            if node_count > 1 and earliest_node_type != \"builder\":\n",
    "                # get the rows with the hash_int value and set the parcel status as premature\n",
    "                operations_df.loc[\n",
    "                    (operations_df['Parcel Key Hash Int'] == hash_int) & \n",
    "                    (operations_df['node_type'] != \"builder\") & \n",
    "                    (operations_df['Parcel Status'] != \"timeout\"), \n",
    "                    'Parcel Status'\n",
    "                ] = 'premature'\n",
    "                if print_premature_cases:\n",
    "                    print(\"Block \" + str(block_id) + \" Parcel Key Hash Int \" + str(hash_int) + \" has \" + str(node_count) + \" nodes with earliest timestamp from \" + earliest_node_type)\n",
    "                block_case_count += 1\n",
    "\n",
    "            total_count += 1\n",
    "\n",
    "        if block_case_count > 0:\n",
    "            print(\"Block \" + str(block_id) + \" has \" + str(block_case_count) + \" premature cases out of \" + str(total_count) + \" \")\n",
    "\n",
    "def get_operations(check_premature=False):\n",
    "    # Get the operation files for each node\n",
    "    operation_files = [file for file in os.listdir() if file.endswith(\".csv\") and \"operations\" in file.lower() and \"putget_operations.csv\" not in file.lower()]\n",
    "    \n",
    "    operations_df = pd.DataFrame()\n",
    "\n",
    "    for operation_file in operation_files:\n",
    "        operation_df = pd.read_csv(operation_file)\n",
    "        \n",
    "        # Look for cases where Block ID is nan\n",
    "        nan_block_cases = operation_df[operation_df['Block ID'].isna()]\n",
    "        if not nan_block_cases.empty:\n",
    "            # Delete the rows with nan block id\n",
    "            operation_df = operation_df.dropna(subset=['Block ID'])\n",
    "\n",
    "        operation_df['PUT timestamps'] = operation_df['PUT timestamps'].astype(str).str.split('+').str[0].apply(lambda x: x.strip() if isinstance(x, str) else x)\n",
    "        operation_df['GET timestamps'] = operation_df['GET timestamps'].astype(str).str.split('+').str[0].apply(lambda x: x.strip() if isinstance(x, str) else x)\n",
    "\n",
    "        # If the value is 0001-01-01 00:00:00 then add the .000000 to the end\n",
    "        operation_df['PUT timestamps'] = operation_df['PUT timestamps'].apply(lambda x: \"0001-01-01 00:00:00.0\" if x == \"0001-01-01 00:00:00\" else x)\n",
    "        operation_df['GET timestamps'] = operation_df['GET timestamps'].apply(lambda x: \"0001-01-01 00:00:00.0\" if x == \"0001-01-01 00:00:00\" else x)\n",
    "\n",
    "        operation_df['PUT timestamps'] = pd.to_datetime(operation_df['PUT timestamps'], format='%Y-%m-%d %H:%M:%S.%f', errors='coerce')\n",
    "        operation_df['GET timestamps'] = pd.to_datetime(operation_df['GET timestamps'], format='%Y-%m-%d %H:%M:%S.%f', errors='coerce')\n",
    "\n",
    "        operation_df['node_id'] = operation_file.split(\"_\")[0]\n",
    "        operation_df['node_type'] = operation_file.split(\"_\")[2].replace(\".csv\", \"\")\n",
    "        operation_df['timestamps'] = pd.to_datetime(operation_df['PUT timestamps'].fillna(operation_df['GET timestamps']), format='%Y-%m-%d %H:%M:%S.%f')\n",
    "        operation_df['latencies_us'] = operation_df['PUT latencies'].fillna(operation_df['GET latencies'])\n",
    "        operation_df['Block ID'] = operation_df['Block ID'].astype(int)\n",
    "        operation_df['Block ID'] = operation_df['Block ID'] + 1\n",
    "        \n",
    "        operations_df = pd.concat([operations_df, operation_df])\n",
    "\n",
    "    # Convert the parcel key hash hexadecimal string into an integer\n",
    "    len_before = len(operations_df)\n",
    "    operations_df = operations_df.dropna(subset=['Parcel Key Hashes'])\n",
    "    len_after = len(operations_df)\n",
    "    print(\"Dropped \" + str(len_before - len_after) + \" rows with nan parcel key hashes\")\n",
    "    operations_df['Parcel Key Hash Int'] = operations_df['Parcel Key Hashes'].astype(str).apply(lambda x: int(x, 16))\n",
    "    \n",
    "    # Delete the PUT/GET timestamps and latencies columns\n",
    "    operations_df = operations_df.drop(columns=['PUT timestamps', 'PUT latencies', 'GET timestamps', 'GET latencies', 'GET hops'])\n",
    "\n",
    "    # Remove data for the first and last block id\n",
    "    operations_df = operations_df[(operations_df['Block ID'] != 1) & (operations_df['Block ID'] != operations_df['Block ID'].max())]\n",
    "\n",
    "    if check_premature:\n",
    "        check_premature_cases(operations_df)\n",
    "\n",
    "    operations_df.to_csv(\"putget_operations.csv\", index=False)\n",
    "\n",
    "    return operations_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Seeding/Sampling Latencies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "builder_latency_file = get_latency_files(\"builder\")\n",
    "validator_latency_files = get_latency_files(\"validator\")\n",
    "non_validator_latency_files = get_latency_files(\"nonvalidator\")\n",
    "\n",
    "validator_latency_df = get_nonbuilder_df(validator_latency_files, \"validator\")\n",
    "non_validator_latency_df = get_nonbuilder_df(non_validator_latency_files, \"nonvalidator\")\n",
    "\n",
    "validator_latency_df = get_total_stats(validator_latency_df)\n",
    "non_validator_latency_df = get_total_stats(non_validator_latency_df)\n",
    "\n",
    "# Get builder latency stats\n",
    "builder_latency_df = pd.read_csv(builder_latency_file)\n",
    "builder_latency_df['Latency (s)'] = builder_latency_df['Seeding Latency (us)'] / 1_000_000\n",
    "builder_latency_df['filename'] = builder_latency_file.replace(\"_latency_stats_builder.csv\", \"\")\n",
    "builder_latency_df['block_id'] = range(len(builder_latency_df))\n",
    "builder_latency_df[\"node_type\"] = \"builder\"\n",
    "builder_latency_df = builder_latency_df[ [\"node_type\", \"filename\", \"block_id\", \"Latency (s)\"] ]\n",
    "builder_latency_df = get_total_stats(builder_latency_df)\n",
    "\n",
    "# Merge all latency stats\n",
    "latency_df = pd.concat([builder_latency_df, validator_latency_df, non_validator_latency_df])\n",
    "latency_df['node_label'] = latency_df['node_type'] + \" \" + latency_df['filename'].apply(lambda x: x[:5])\n",
    "latency_df['Successful PUT/GETs'] = latency_df['Successful PUT/GETs'].astype(int)\n",
    "latency_df['Total PUT/GETs'] = latency_df['Total PUT/GETs'].astype(int)\n",
    "latency_df['Successful/Total Label'] = latency_df['Successful PUT/GETs'].astype(str) + '/' + latency_df['Total PUT/GETs'].astype(str)\n",
    "latency_df['Percentage'] = (latency_df['Successful PUT/GETs'] / latency_df['Total PUT/GETs'] * 100).round().astype(int)\n",
    "latency_df['node_label'] = latency_df['node_label'] + \" \" + latency_df['Successful/Total Label'].astype(str) + \" \" + latency_df['Percentage'].astype(str) + \"%\"\n",
    "\n",
    "# Remove first and last block\n",
    "latency_df = latency_df[ (latency_df['block_id'] != 0) & (latency_df['block_id'] != len(builder_latency_df)) ]\n",
    "\n",
    "fig, axes = plt.subplots(1, 2, figsize=(15, 5))\n",
    "\n",
    "ax = axes[0]\n",
    "plt.title(\"Seeding/Sampling Latency for Each Block\")\n",
    "sns.barplot(x=\"block_id\", y=\"Latency (s)\", hue=\"node_label\", data=latency_df, ax=ax)\n",
    "ax.set_xlabel(\"Block Number\")\n",
    "ax.set_ylabel(\"Latency (s)\")\n",
    "ax.legend()\n",
    "\n",
    "ax = axes[1]\n",
    "plt.title(\"Seeding/Sampling Latency for Each Block with 12 Second Threshold\")\n",
    "sns.barplot(x=\"block_id\", y=\"Latency (s)\", hue=\"node_label\", data=latency_df, ax=ax)\n",
    "ax.set_xlabel(\"Block Number\")\n",
    "ax.set_ylabel(\"Latency (s)\")\n",
    "ax.axhline(y=12, color='red', linestyle='--')\n",
    "ax.legend()\n",
    "\n",
    "plt.tight_layout()\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Parcel Key Hash Distributions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "grouped_op_df = get_operations().groupby(\"Block ID\")\n",
    "\n",
    "fig, axes = plt.subplots(len(grouped_op_df), 1, figsize=(15, 5 * len(grouped_op_df)))\n",
    "\n",
    "for block_id, block_df in grouped_op_df:\n",
    "    parcel_hash_ints = block_df['Parcel Key Hash Int']\n",
    "    parcel_hash_ints = parcel_hash_ints[parcel_hash_ints != -1]\n",
    "    parcel_hash_ints = parcel_hash_ints.apply(decimal.Decimal)\n",
    "\n",
    "    ax = axes[block_id - 2]\n",
    "    ax.set_title(\"Parcel Key Hash Ints for Block \" + str(block_id))\n",
    "    ax.hist(parcel_hash_ints, bins=100)\n",
    "    ax.set_xlabel(\"Parcel Key Hash Int\")\n",
    "    ax.set_ylabel(\"Count\")\n",
    "\n",
    "plt.tight_layout()\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PUT/GET Operation Scatter Plot"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "op_df = get_operations()\n",
    "grouped_op_df = op_df.groupby(\"Block ID\")\n",
    "\n",
    "fig, axes = plt.subplots(len(grouped_op_df) + 1, 1, figsize=(15, 4 * len(grouped_op_df)))\n",
    "\n",
    "colours = {\n",
    "    \"builder\": \"red\",\n",
    "    \"validator\": \"blue\",\n",
    "    \"nonvalidator\": \"green\"\n",
    "}\n",
    "\n",
    "for block_id, block_df in grouped_op_df:\n",
    "    block_df['Parcel Key Hash Int'] = block_df['Parcel Key Hash Int'].apply(decimal.Decimal)\n",
    "    block_df['timestamps'] = pd.to_datetime(block_df['timestamps'])\n",
    "\n",
    "    starting_signals = block_df[block_df['Parcel Key Hashes'] == '-1']\n",
    "\n",
    "    # Remove the -1 values which are the starting signals\n",
    "    block_df = block_df[block_df['Parcel Key Hash Int'] != -1]\n",
    "\n",
    "    block_df_grouped_by_nodetype = block_df.groupby(\"node_type\")\n",
    "\n",
    "    ax = axes[block_id - 2]\n",
    "    \n",
    "    for starting_signal_index, starting_signal_row in starting_signals.iterrows():\n",
    "        if starting_signal_row['node_type'] == \"builder\":\n",
    "            ax.axvline(x=starting_signal_row['timestamps'], color='red', linestyle='--')\n",
    "        elif starting_signal_row['node_type'] == \"validator\":\n",
    "            ax.axvline(x=starting_signal_row['timestamps'], color='blue', linestyle='--')\n",
    "        else:\n",
    "            ax.axvline(x=starting_signal_row['timestamps'], color='green', linestyle='--')\n",
    "\n",
    "    sns.scatterplot(\n",
    "        x=\"timestamps\", \n",
    "        y=\"Parcel Key Hash Int\", \n",
    "        hue=\"node_type\",\n",
    "        style=\"Parcel Status\",\n",
    "        palette=colours,\n",
    "        data=block_df, \n",
    "        ax=ax,\n",
    "        alpha=0.5\n",
    "    )\n",
    "\n",
    "sns.scatterplot(\n",
    "    x=\"timestamps\", \n",
    "    y=\"Parcel Key Hash Int\", \n",
    "    hue=\"node_type\",\n",
    "    style=\"Parcel Status\",\n",
    "    palette=colours,\n",
    "    data=op_df, \n",
    "    ax=axes[-1],\n",
    "    alpha=0.5\n",
    ")\n",
    "\n",
    "plt.tight_layout()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.11.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}