{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "ename": "WorkerException",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mValueError\u001b[0m                                Traceback (most recent call last)",
      "\u001b[0;31mValueError\u001b[0m: hi",
      "\nThe above exception was the direct cause of the following exception:\n",
      "\u001b[0;31mWorkerException\u001b[0m                           Traceback (most recent call last)",
      "\u001b[1;32m/data/projects/great_ai/docs/map.ipynb Cell 1'\u001b[0m in \u001b[0;36m<cell line: 8>\u001b[0;34m()\u001b[0m\n\u001b[1;32m      <a href='vscode-notebook-cell:/data/projects/great_ai/docs/map.ipynb#ch0000000?line=3'>4</a>\u001b[0m \u001b[39mdef\u001b[39;00m \u001b[39moh_no\u001b[39m(_):\n\u001b[1;32m      <a href='vscode-notebook-cell:/data/projects/great_ai/docs/map.ipynb#ch0000000?line=4'>5</a>\u001b[0m     \u001b[39mraise\u001b[39;00m \u001b[39mValueError\u001b[39;00m(\u001b[39m\"\u001b[39m\u001b[39mhi\u001b[39m\u001b[39m\"\u001b[39m)\n\u001b[0;32m----> <a href='vscode-notebook-cell:/data/projects/great_ai/docs/map.ipynb#ch0000000?line=7'>8</a>\u001b[0m \u001b[39mlist\u001b[39;49m(parallel_map(oh_no, \u001b[39mrange\u001b[39;49m(\u001b[39m10000\u001b[39;49m), concurrency\u001b[39m=\u001b[39;49m\u001b[39m2\u001b[39;49m))\n",
      "File \u001b[0;32m/data/projects/great_ai/src/great_ai/utilities/parallel_map/parallel_map.py:120\u001b[0m, in \u001b[0;36mparallel_map\u001b[0;34m(function, input_values, chunk_size, concurrency, unordered, ignore_exceptions)\u001b[0m\n\u001b[1;32m    117\u001b[0m     p\u001b[39m.\u001b[39mstart()\n\u001b[1;32m    119\u001b[0m \u001b[39mtry\u001b[39;00m:\n\u001b[0;32m--> 120\u001b[0m     \u001b[39myield from\u001b[39;00m manage_communication(\n\u001b[1;32m    121\u001b[0m         input_values\u001b[39m=\u001b[39minput_values,\n\u001b[1;32m    122\u001b[0m         chunk_size\u001b[39m=\u001b[39mconfig\u001b[39m.\u001b[39mchunk_size,\n\u001b[1;32m    123\u001b[0m         input_queue\u001b[39m=\u001b[39minput_queue,\n\u001b[1;32m    124\u001b[0m         output_queue\u001b[39m=\u001b[39moutput_queue,\n\u001b[1;32m    125\u001b[0m         unordered\u001b[39m=\u001b[39munordered,\n\u001b[1;32m    126\u001b[0m         ignore_exceptions\u001b[39m=\u001b[39mignore_exceptions,\n\u001b[1;32m    127\u001b[0m     )\n\u001b[1;32m    128\u001b[0m     should_stop\u001b[39m.\u001b[39mset()\n\u001b[1;32m    129\u001b[0m \u001b[39mexcept\u001b[39;00m WorkerException:\n",
      "File \u001b[0;32m/data/projects/great_ai/src/great_ai/utilities/parallel_map/manage_communication.py:53\u001b[0m, in \u001b[0;36mmanage_communication\u001b[0;34m(input_values, chunk_size, input_queue, output_queue, unordered, ignore_exceptions)\u001b[0m\n\u001b[1;32m     51\u001b[0m e, tb \u001b[39m=\u001b[39m exception\n\u001b[1;32m     52\u001b[0m \u001b[39mif\u001b[39;00m \u001b[39mnot\u001b[39;00m ignore_exceptions:\n\u001b[0;32m---> 53\u001b[0m     \u001b[39mraise\u001b[39;00m WorkerException \u001b[39mfrom\u001b[39;00m \u001b[39me\u001b[39;00m\n\u001b[1;32m     54\u001b[0m \u001b[39melse\u001b[39;00m:\n\u001b[1;32m     55\u001b[0m     logger\u001b[39m.\u001b[39merror(\n\u001b[1;32m     56\u001b[0m         \u001b[39mf\u001b[39m\u001b[39m\"\u001b[39m\u001b[39mException \u001b[39m\u001b[39m{\u001b[39;00me\u001b[39m}\u001b[39;00m\u001b[39m encountered in worker, traceback:\u001b[39m\u001b[39m\\n\u001b[39;00m\u001b[39m{\u001b[39;00mtb\u001b[39m}\u001b[39;00m\u001b[39m\"\u001b[39m\n\u001b[1;32m     57\u001b[0m     )\n",
      "\u001b[0;31mWorkerException\u001b[0m: "
     ]
    }
   ],
   "source": [
    "from great_ai.utilities import parallel_map\n",
    "\n",
    "\n",
    "def oh_no(_):\n",
    "    raise ValueError(\"hi\")\n",
    "\n",
    "\n",
    "list(parallel_map(oh_no, range(10000), concurrency=2))\n",
    "\n",
    "# list(parallel_map(oh_no, range(10), concurrency=1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from great_ai.utilities import parallel_map\n",
    "\n",
    "COUNT = int(1e5) + 3\n",
    "\n",
    "\n",
    "inputs = range(COUNT)\n",
    "expected = [v**2 for v in range(COUNT)]\n",
    "\n",
    "assert list(parallel_map(lambda v: v**2, inputs, disable_logging=False)) == expected"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# from time import sleep\n",
    "# from great_ai.utilities import chunk, threaded_parallel_map, parallel_map, unchunk\n",
    "\n",
    "\n",
    "# def generate_urls():\n",
    "#     for i in range(20000):\n",
    "#         # print(f\"yield {i}.com\")\n",
    "#         yield f\"{i}.com\"\n",
    "\n",
    "\n",
    "# def scrape(url):\n",
    "#     sleep(2)  # scrape\n",
    "#     # print(f\"scraped {url}\")\n",
    "#     return url + \" html\"\n",
    "\n",
    "\n",
    "# def scrape_bulk(urls):\n",
    "#     return list(\n",
    "#         threaded_parallel_map(scrape, urls, concurrency=100, disable_logging=True)\n",
    "#     )\n",
    "\n",
    "\n",
    "# def process(html):\n",
    "#     sleep(1)  # process\n",
    "#     # print(f\"processed {html}\")\n",
    "\n",
    "\n",
    "# scraped_pages_stream = parallel_map(\n",
    "#     scrape_bulk,\n",
    "#     chunk(generate_urls(), 100),\n",
    "#     chunk_size=1,\n",
    "#     concurrency=1,\n",
    "#     disable_logging=True,\n",
    "#     unordered=True,\n",
    "# )\n",
    "# list(\n",
    "#     threaded_parallel_map(\n",
    "#         process,\n",
    "#         unchunk(scraped_pages_stream),\n",
    "#         chunk_size=1,\n",
    "#         concurrency=50,\n",
    "#         unordered=True,\n",
    "#     )\n",
    "# )\n",
    "\n",
    "# None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# def a():\n",
    "#     yield [2]\n",
    "#     yield [6]\n",
    "#     raise Exception()\n",
    "#     yield 3\n",
    "\n",
    "\n",
    "# a = a()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "list(unchunk(a))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# list(map(lambda x: print(x), [1, 2, 4]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# from time import sleep\n",
    "\n",
    "\n",
    "# def my_generator():\n",
    "#     for i in range(10):\n",
    "#         yield i\n",
    "#         sleep(0.2)\n",
    "\n",
    "\n",
    "# list(parallel_map(lambda x: x**3, my_generator(), chunk_length=1))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.10.4 ('.env': 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"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "d88b0dd276e3f918f7798b7e97af2e3c2f843817b9e5b55a9df0a682ffd80f44"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}