{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Coffea-Casa Benchmark Example 7"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "%matplotlib inline\n",
    "import hist\n",
    "import coffea.processor as processor\n",
    "import awkward as ak\n",
    "from coffea.nanoevents import schemas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This program will graph the sum of Jet pT's which are greater than 30 GeV and farther than a Euclidean distance of 0.4 from any lepton with pT > 10 GeV.\n",
    "class Processor(processor.ProcessorABC):\n",
    "    def __init__(self):\n",
    "        dataset_axis = hist.axis.StrCategory(name=\"dataset\", label=\"\", categories=[], growth=True)\n",
    "        Jet_axis = hist.axis.Regular(name=\"Jet_pt\", label=\"Jet_pt [GeV]\", bins=50, start=15, stop=200)\n",
    "        \n",
    "        self.output = processor.dict_accumulator({\n",
    "            'Jet_pt': hist.Hist(dataset_axis, Jet_axis),\n",
    "            'cutflow': processor.defaultdict_accumulator(int)\n",
    "        })\n",
    "    \n",
    "    def process(self, events):\n",
    "        dataset = events.metadata[\"dataset\"]\n",
    "\n",
    "        muons = events.Muon\n",
    "        electrons = events.Electron\n",
    "        jets = events.Jet\n",
    "        \n",
    "        self.output['cutflow']['all events'] += ak.size(jets, axis=0)\n",
    "        self.output['cutflow']['all jets'] += ak.sum(ak.num(jets, axis=1))\n",
    "        \n",
    "        # Get jets with higher GeV than 30.\n",
    "        min_jetpt = (jets.pt > 30)\n",
    "        self.output['cutflow']['jets with pt > 30'] += ak.sum(ak.sum(min_jetpt, axis=1))\n",
    "        \n",
    "        # Get all leptons with higher GeV than 10.\n",
    "        min_muonpt = (muons.pt > 10)\n",
    "        self.output['cutflow']['muons with pt > 10'] += ak.sum(ak.sum(min_muonpt, axis=1))\n",
    "        min_electronpt = (electrons.pt > 10)\n",
    "        self.output['cutflow']['electrons with pt > 10'] += ak.sum(ak.sum(min_electronpt, axis=1))\n",
    "        \n",
    "        # Mask jets and leptons with their minimum requirements/\n",
    "        goodjets = jets[min_jetpt]\n",
    "        goodmuons = muons[min_muonpt]\n",
    "        goodelectrons = electrons[min_electronpt]\n",
    "    \n",
    "        jet_muon_pairs = ak.cartesian({'jets': goodjets, 'muons': goodmuons}, nested=True)\n",
    "        jet_electron_pairs = ak.cartesian({'jets': goodjets, 'electrons': goodelectrons}, nested=True)\n",
    "    \n",
    "        # This long conditional checks that the jet is at least 0.4 euclidean distance from each lepton. It then checks if each unique jet contains a False, i.e., that a jet is 0.4 euclidean distance from EVERY specific lepton in the event.\n",
    "        good_jm_pairs = goodjets.nearest(goodmuons).delta_r(goodjets) > 0.4\n",
    "        good_je_pairs = goodjets.nearest(goodelectrons).delta_r(goodjets) > 0.4\n",
    "        good_jl_pairs = good_jm_pairs & good_je_pairs\n",
    "        \n",
    "        self.output['cutflow']['jet-muon pairs'] += ak.sum(ak.sum(good_jm_pairs, axis=1))\n",
    "        self.output['cutflow']['jet-electron pairs'] += ak.sum(ak.sum(good_je_pairs, axis=1))\n",
    "        self.output['cutflow']['jet-lepton pairs'] += ak.sum(ak.sum(good_jl_pairs, axis=1))\n",
    "        \n",
    "        # We then mask our jets with all three of the above good pairs to get only jets that are 0.4 distance from every type of lepton, and sum them.\n",
    "        sumjets = ak.sum(goodjets[good_jl_pairs].pt, axis=1)\n",
    "        self.output['cutflow']['final jets'] += ak.sum(ak.num(goodjets[good_jl_pairs], axis=1))\n",
    "        self.output['Jet_pt'].fill(dataset=dataset, Jet_pt=sumjets)\n",
    "        \n",
    "        return self.output\n",
    "\n",
    "    def postprocess(self, accumulator):\n",
    "        pass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client\n",
    "\n",
    "client = Client(\"tls://localhost:8786\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "ename": "RuntimeError",
     "evalue": "can't start new thread",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mRuntimeError\u001b[0m                              Traceback (most recent call last)",
      "Input \u001b[0;32mIn [4]\u001b[0m, in \u001b[0;36m<cell line: 10>\u001b[0;34m()\u001b[0m\n\u001b[1;32m      3\u001b[0m executor \u001b[38;5;241m=\u001b[39m processor\u001b[38;5;241m.\u001b[39mDaskExecutor(client\u001b[38;5;241m=\u001b[39mclient)\n\u001b[1;32m      5\u001b[0m run \u001b[38;5;241m=\u001b[39m processor\u001b[38;5;241m.\u001b[39mRunner(executor\u001b[38;5;241m=\u001b[39mexecutor,\n\u001b[1;32m      6\u001b[0m                         schema\u001b[38;5;241m=\u001b[39mschemas\u001b[38;5;241m.\u001b[39mNanoAODSchema,\n\u001b[1;32m      7\u001b[0m                         savemetrics\u001b[38;5;241m=\u001b[39m\u001b[38;5;28;01mTrue\u001b[39;00m\n\u001b[1;32m      8\u001b[0m                       )\n\u001b[0;32m---> 10\u001b[0m output, metrics \u001b[38;5;241m=\u001b[39m \u001b[43mrun\u001b[49m\u001b[43m(\u001b[49m\u001b[43mfileset\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43mEvents\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mprocessor_instance\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mProcessor\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m     12\u001b[0m metrics\n",
      "File \u001b[0;32m/opt/conda/lib/python3.8/site-packages/coffea/processor/executor.py:1588\u001b[0m, in \u001b[0;36mRunner.__call__\u001b[0;34m(self, fileset, treename, processor_instance)\u001b[0m\n\u001b[1;32m   1567\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21m__call__\u001b[39m(\n\u001b[1;32m   1568\u001b[0m     \u001b[38;5;28mself\u001b[39m,\n\u001b[1;32m   1569\u001b[0m     fileset: Dict,\n\u001b[1;32m   1570\u001b[0m     treename: \u001b[38;5;28mstr\u001b[39m,\n\u001b[1;32m   1571\u001b[0m     processor_instance: ProcessorABC,\n\u001b[1;32m   1572\u001b[0m ) \u001b[38;5;241m-\u001b[39m\u001b[38;5;241m>\u001b[39m Accumulatable:\n\u001b[1;32m   1573\u001b[0m     \u001b[38;5;124;03m\"\"\"Run the processor_instance on a given fileset\u001b[39;00m\n\u001b[1;32m   1574\u001b[0m \n\u001b[1;32m   1575\u001b[0m \u001b[38;5;124;03m    Parameters\u001b[39;00m\n\u001b[0;32m   (...)\u001b[0m\n\u001b[1;32m   1585\u001b[0m \u001b[38;5;124;03m            An instance of a class deriving from ProcessorABC\u001b[39;00m\n\u001b[1;32m   1586\u001b[0m \u001b[38;5;124;03m    \"\"\"\u001b[39;00m\n\u001b[0;32m-> 1588\u001b[0m     wrapped_out \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mrun\u001b[49m\u001b[43m(\u001b[49m\u001b[43mfileset\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mprocessor_instance\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mtreename\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m   1589\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39muse_dataframes:\n\u001b[1;32m   1590\u001b[0m         \u001b[38;5;28;01mreturn\u001b[39;00m wrapped_out  \u001b[38;5;66;03m# not wrapped anymore\u001b[39;00m\n",
      "File \u001b[0;32m/opt/conda/lib/python3.8/site-packages/coffea/processor/executor.py:1670\u001b[0m, in \u001b[0;36mRunner.run\u001b[0;34m(self, fileset, processor_instance, treename)\u001b[0m\n\u001b[1;32m   1668\u001b[0m     chunks \u001b[38;5;241m=\u001b[39m fileset\n\u001b[1;32m   1669\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m-> 1670\u001b[0m     chunks \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mpreprocess\u001b[49m\u001b[43m(\u001b[49m\u001b[43mfileset\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mtreename\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m   1672\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mprocessor_compression \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[1;32m   1673\u001b[0m     pi_to_send \u001b[38;5;241m=\u001b[39m processor_instance\n",
      "File \u001b[0;32m/opt/conda/lib/python3.8/site-packages/coffea/processor/executor.py:1622\u001b[0m, in \u001b[0;36mRunner.preprocess\u001b[0;34m(self, fileset, treename)\u001b[0m\n\u001b[1;32m   1619\u001b[0m \u001b[38;5;28;01mfor\u001b[39;00m filemeta \u001b[38;5;129;01min\u001b[39;00m fileset:\n\u001b[1;32m   1620\u001b[0m     filemeta\u001b[38;5;241m.\u001b[39mmaybe_populate(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mmetadata_cache)\n\u001b[0;32m-> 1622\u001b[0m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_preprocess_fileset\u001b[49m\u001b[43m(\u001b[49m\u001b[43mfileset\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m   1623\u001b[0m fileset \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_filter_badfiles(fileset)\n\u001b[1;32m   1625\u001b[0m \u001b[38;5;66;03m# reverse fileset list to match the order of files as presented in version\u001b[39;00m\n\u001b[1;32m   1626\u001b[0m \u001b[38;5;66;03m# v0.7.4. This fixes tests using maxchunks.\u001b[39;00m\n",
      "File \u001b[0;32m/opt/conda/lib/python3.8/site-packages/coffea/processor/executor.py:1383\u001b[0m, in \u001b[0;36mRunner._preprocess_fileset\u001b[0;34m(self, fileset)\u001b[0m\n\u001b[1;32m   1376\u001b[0m pre_executor \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mpre_executor\u001b[38;5;241m.\u001b[39mcopy(\u001b[38;5;241m*\u001b[39m\u001b[38;5;241m*\u001b[39mpre_arg_override)\n\u001b[1;32m   1377\u001b[0m closure \u001b[38;5;241m=\u001b[39m partial(\n\u001b[1;32m   1378\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mautomatic_retries,\n\u001b[1;32m   1379\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mretries,\n\u001b[1;32m   1380\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mskipbadfiles,\n\u001b[1;32m   1381\u001b[0m     partial(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mmetadata_fetcher, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mxrootdtimeout, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39malign_clusters),\n\u001b[1;32m   1382\u001b[0m )\n\u001b[0;32m-> 1383\u001b[0m out, _ \u001b[38;5;241m=\u001b[39m \u001b[43mpre_executor\u001b[49m\u001b[43m(\u001b[49m\u001b[43mto_get\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mclosure\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mout\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m   1384\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m out:\n\u001b[1;32m   1385\u001b[0m     item \u001b[38;5;241m=\u001b[39m out\u001b[38;5;241m.\u001b[39mpop()\n",
      "File \u001b[0;32m/opt/conda/lib/python3.8/site-packages/coffea/processor/executor.py:964\u001b[0m, in \u001b[0;36mDaskExecutor.__call__\u001b[0;34m(self, items, function, accumulator)\u001b[0m\n\u001b[1;32m    961\u001b[0m         \u001b[38;5;28;01mfrom\u001b[39;00m \u001b[38;5;21;01mdistributed\u001b[39;00m \u001b[38;5;28;01mimport\u001b[39;00m progress\n\u001b[1;32m    963\u001b[0m         \u001b[38;5;66;03m# FIXME: fancy widget doesn't appear, have to live with boring pbar\u001b[39;00m\n\u001b[0;32m--> 964\u001b[0m         \u001b[43mprogress\u001b[49m\u001b[43m(\u001b[49m\u001b[43mwork\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mmulti\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[38;5;28;43;01mTrue\u001b[39;49;00m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mnotebook\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[38;5;28;43;01mFalse\u001b[39;49;00m\u001b[43m)\u001b[49m\n\u001b[1;32m    965\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m (\n\u001b[1;32m    966\u001b[0m         accumulate(\n\u001b[1;32m    967\u001b[0m             [\n\u001b[0;32m   (...)\u001b[0m\n\u001b[1;32m    974\u001b[0m         \u001b[38;5;241m0\u001b[39m,\n\u001b[1;32m    975\u001b[0m     )\n\u001b[1;32m    976\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m KilledWorker \u001b[38;5;28;01mas\u001b[39;00m ex:\n",
      "File \u001b[0;32m/opt/conda/lib/python3.8/site-packages/distributed/diagnostics/progressbar.py:448\u001b[0m, in \u001b[0;36mprogress\u001b[0;34m(notebook, multi, complete, *futures, **kwargs)\u001b[0m\n\u001b[1;32m    446\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m bar\n\u001b[1;32m    447\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m--> 448\u001b[0m     \u001b[43mTextProgressBar\u001b[49m\u001b[43m(\u001b[49m\u001b[43mfutures\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mcomplete\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mcomplete\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;241;43m*\u001b[39;49m\u001b[38;5;241;43m*\u001b[39;49m\u001b[43mkwargs\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m/opt/conda/lib/python3.8/site-packages/distributed/diagnostics/progressbar.py:129\u001b[0m, in \u001b[0;36mTextProgressBar.__init__\u001b[0;34m(self, keys, scheduler, interval, width, loop, complete, start, **kwargs)\u001b[0m\n\u001b[1;32m    127\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m start:\n\u001b[1;32m    128\u001b[0m     loop_runner \u001b[38;5;241m=\u001b[39m LoopRunner(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mloop)\n\u001b[0;32m--> 129\u001b[0m     \u001b[43mloop_runner\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mrun_sync\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mlisten\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m/opt/conda/lib/python3.8/site-packages/distributed/utils.py:535\u001b[0m, in \u001b[0;36mLoopRunner.run_sync\u001b[0;34m(self, func, *args, **kwargs)\u001b[0m\n\u001b[1;32m    533\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m sync(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mloop, func, \u001b[38;5;241m*\u001b[39margs, \u001b[38;5;241m*\u001b[39m\u001b[38;5;241m*\u001b[39mkwargs)\n\u001b[1;32m    534\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m--> 535\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mstart\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m    536\u001b[0m     \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[1;32m    537\u001b[0m         \u001b[38;5;28;01mreturn\u001b[39;00m sync(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mloop, func, \u001b[38;5;241m*\u001b[39margs, \u001b[38;5;241m*\u001b[39m\u001b[38;5;241m*\u001b[39mkwargs)\n",
      "File \u001b[0;32m/opt/conda/lib/python3.8/site-packages/distributed/utils.py:428\u001b[0m, in \u001b[0;36mLoopRunner.start\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    421\u001b[0m \u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[1;32m    422\u001b[0m \u001b[38;5;124;03mStart the IO loop if required.  The loop is run in a dedicated\u001b[39;00m\n\u001b[1;32m    423\u001b[0m \u001b[38;5;124;03mthread.\u001b[39;00m\n\u001b[1;32m    424\u001b[0m \n\u001b[1;32m    425\u001b[0m \u001b[38;5;124;03mIf the loop is already running, this method does nothing.\u001b[39;00m\n\u001b[1;32m    426\u001b[0m \u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[1;32m    427\u001b[0m \u001b[38;5;28;01mwith\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_lock:\n\u001b[0;32m--> 428\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_start_unlocked\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m/opt/conda/lib/python3.8/site-packages/distributed/utils.py:464\u001b[0m, in \u001b[0;36mLoopRunner._start_unlocked\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    462\u001b[0m thread \u001b[38;5;241m=\u001b[39m threading\u001b[38;5;241m.\u001b[39mThread(target\u001b[38;5;241m=\u001b[39mrun_loop, name\u001b[38;5;241m=\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mIO loop\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[1;32m    463\u001b[0m thread\u001b[38;5;241m.\u001b[39mdaemon \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mTrue\u001b[39;00m\n\u001b[0;32m--> 464\u001b[0m \u001b[43mthread\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mstart\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m    466\u001b[0m loop_evt\u001b[38;5;241m.\u001b[39mwait(timeout\u001b[38;5;241m=\u001b[39m\u001b[38;5;241m10\u001b[39m)\n\u001b[1;32m    467\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_started \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mTrue\u001b[39;00m\n",
      "File \u001b[0;32m/opt/conda/lib/python3.8/threading.py:852\u001b[0m, in \u001b[0;36mThread.start\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    850\u001b[0m     _limbo[\u001b[38;5;28mself\u001b[39m] \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\n\u001b[1;32m    851\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[0;32m--> 852\u001b[0m     \u001b[43m_start_new_thread\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_bootstrap\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m    853\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mException\u001b[39;00m:\n\u001b[1;32m    854\u001b[0m     \u001b[38;5;28;01mwith\u001b[39;00m _active_limbo_lock:\n",
      "\u001b[0;31mRuntimeError\u001b[0m: can't start new thread"
     ]
    }
   ],
   "source": [
    "fileset = {'SingleMu' : [\"root://eospublic.cern.ch//eos/root-eos/benchmark/Run2012B_SingleMu.root\"]}\n",
    "\n",
    "executor = processor.DaskExecutor(client=client)\n",
    "\n",
    "run = processor.Runner(executor=executor,\n",
    "                        schema=schemas.NanoAODSchema,\n",
    "                        savemetrics=True\n",
    "                      )\n",
    "\n",
    "output, metrics = run(fileset, \"Events\", processor_instance=Processor())\n",
    "\n",
    "metrics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "output['Jet_pt'].plot1d()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for key, value in output['cutflow'].items():\n",
    "    print(key, value)"
   ]
  }
 ],
 "metadata": {
  "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.8.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}