batch_dispatcher.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "\n",
    "###background generator (from your generator)\n",
    "import threading\n",
    "import Queue\n",
    "import time \n",
    "\n",
    "class BatchProcessor(threading.Thread):\n",
    "    def __init__(self,process_fun,max_batch_size=10,max_delay=5,dt=0.1):\n",
    "        threading.Thread.__init__(self)\n",
    "        self.queue = Queue.Queue(max_batch_size)\n",
    "        self.process_fun = process_fun\n",
    "        self.max_batch_size=max_batch_size\n",
    "        self.max_delay = max_delay\n",
    "        self.dt = dt\n",
    "        \n",
    "        self.daemon = True\n",
    "        self.start()\n",
    "    \n",
    "    def assign_task(self,task):\n",
    "        self.queue.put(task)\n",
    "        \n",
    "    def run(self):\n",
    "        while True:\n",
    "            max_start_time = time.time() + self.max_delay\n",
    "            minibatch = []\n",
    "\n",
    "            while True:\n",
    "                if time.time() >= max_start_time:\n",
    "                    print 'started due to exceeding max time'\n",
    "                    break\n",
    "                if len(minibatch) >= self.max_batch_size:\n",
    "                    print 'started due to reaching max batch size'\n",
    "                    break\n",
    "\n",
    "                if not self.queue.empty():\n",
    "                    minibatch.append(self.queue.get())\n",
    "                else:\n",
    "                    time.sleep(self.dt)\n",
    "                    \n",
    "            if len(minibatch) == 0:\n",
    "                print \"skipped due to zero tasks\"\n",
    "                continue\n",
    "                \n",
    "            self.process_fun(minibatch)\n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def process_batch(batch):\n",
    "    \n",
    "    for key,value in batch:\n",
    "        print key,\"gets\",value\n",
    "    print \"%i samples processed with mah super neural network...\\n\\n\\n\"%len(batch)\n",
    "    return 0\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "worker = BatchProcessor(process_batch,)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "started due to exceeding max time\n",
      "bob gets cake\n",
      "jeff gets card\n",
      "mike gets punch\n",
      "3 samples processed with mah super neural network...\n",
      "\n",
      "\n",
      "\n",
      "started due to exceeding max time\n",
      "skipped due to zero tasks\n",
      "started due to exceeding max time\n",
      "skipped due to zero tasks\n"
     ]
    }
   ],
   "source": [
    "worker.assign_task(['bob','cake'])\n",
    "worker.assign_task(['jeff','card'])\n",
    "worker.assign_task(['mike','punched'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "started due to reaching max batch size\n",
      "john gets fired\n",
      "mike gets email\n",
      "kyle gets fired\n",
      "kyle gets email\n",
      "john gets fired\n",
      "mike gets fired\n",
      "john gets email\n",
      "butters gets reply\n",
      "mike gets fired\n",
      "john gets email\n",
      "10 samples processed with mah super neural network...\n",
      "\n",
      "\n",
      "\n",
      "started due to reaching max batch size\n",
      "mike gets fired\n",
      "kyle gets reply\n",
      "john gets punch\n",
      "cartman gets reply\n",
      "butters gets punch\n",
      "cartman gets reply\n",
      "john gets punch\n",
      "butters gets punch\n",
      "john gets email\n",
      "butters gets fired\n",
      "10 samples processed with mah super neural network...\n",
      "\n",
      "\n",
      "\n",
      "started due to exceeding max time\n",
      "kyle gets punch\n",
      "john gets email\n",
      "2 samples processed with mah super neural network...\n",
      "\n",
      "\n",
      "\n",
      "started due to exceeding max time\n",
      "skipped due to zero tasks\n"
     ]
    }
   ],
   "source": [
    "import random\n",
    "for i in range(22):\n",
    "    worker.assign_task((random.choice([\"mike\",'john','butters','kyle','cartman']),\n",
    "                       random.choice(['cake','fired','punch','reply','email'])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python [Root]",
   "language": "python",
   "name": "Python [Root]"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}