testparallel.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import io, os, sys, types"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Vendor:  Continuum Analytics, Inc.\n",
      "Package: mkl\n",
      "Message: trial mode expires in 30 days\n"
     ]
    }
   ],
   "source": [
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([1, 1])"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "np.array([1,1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "$$\\alpha = 5$$"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import multiprocessing as mp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "mp.cpu_count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Serial time: 135.262062\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/john/anaconda2/lib/python2.7/site-packages/ipykernel/__main__.py:7: DeprecationWarning: using a non-integer number instead of an integer will result in an error in the future\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Parallel time: 75.328563\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Pool\n",
    "import time\n",
    "import numpy\n",
    " \n",
    "def takeuptime(ntrials):\n",
    "    for ii in xrange(ntrials):\n",
    "        junk = numpy.std(numpy.random.randn(1e5))\n",
    "    return junk\n",
    " \n",
    "if __name__ == \"__main__\":\n",
    "    start = time.time()\n",
    "    map(takeuptime, [100000,100000])\n",
    "    print \"Serial time: %f\" % (time.time() - start)\n",
    " \n",
    "    start = time.time()\n",
    "    pool = Pool(processes=8)\n",
    "    pool.map(takeuptime, [100000, 100000])\n",
    "    print \"Parallel time: %f\" % (time.time() - start)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import ipyparallel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0, 1, 2, 3, 4, 5, 6, 7]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "clients = ipyparallel.Client()\n",
    "clients.ids"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "importing numpy on engine(s)\n"
     ]
    }
   ],
   "source": [
    "dview = clients.direct_view()\n",
    "with dview.sync_imports():\n",
    "    import numpy  #using 'as np' does not work!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def mc_pi(n):\n",
    "    count = 0\n",
    "    for i in range(n):\n",
    "        x = numpy.random.rand()\n",
    "        y = numpy.random.rand()\n",
    "        if (x**2 + y**2) <= 1:\n",
    "            count += 1\n",
    "    return 1.0 * count/n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Serial time: 53.868618\n",
      "Pi = 3.141501\n"
     ]
    }
   ],
   "source": [
    "start = time.time()\n",
    "est_pi = 4.0 * mc_pi(10**8)\n",
    "print \"Serial time: %f\" % (time.time() - start)\n",
    "print \"Pi = %f\" % est_pi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 142,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.78564736, 0.78527352, 0.78545552, 0.78540392, 0.7853624, 0.78537504, 0.7855432, 0.78551376]\n"
     ]
    }
   ],
   "source": [
    "results = dview.map_sync(mc_pi, [125*10**5]*8)\n",
    "print(results)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3.14191492\n"
     ]
    }
   ],
   "source": [
    "n_proc = 8;\n",
    "n = 10**8\n",
    "n_pass = n/n_proc\n",
    "est_pis = dview.map_sync(mc_pi, [n_pass]*n_proc)\n",
    "print(4.0 * sum(est_pis)/n_proc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We should be able to import variables across engines if we import the function with dview.sync_imports()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def funct(nstart,nend):\n",
    "    \n",
    "    def iter_loadtxt(filename, delimiter=',', skiprows=1, dtype=float):\n",
    "        def iter_func():\n",
    "            with open(filename, 'r') as infile:\n",
    "                for _ in range(skiprows):\n",
    "                    next(infile)\n",
    "                for line in infile:\n",
    "                    line = line.rstrip().split(delimiter)\n",
    "                    for item in line:\n",
    "                        yield dtype(item)\n",
    "            iter_loadtxt.rowlength = len(line)\n",
    "\n",
    "        data = numpy.fromiter(iter_func(), dtype=dtype)\n",
    "        data = data.reshape((-1, iter_loadtxt.rowlength))\n",
    "        return data\n",
    "    \n",
    "    file1 = iter_loadtxt('/media/john/My Passport/slabs/serge_tests/data0.csv',delimiter=',',skiprows=1,dtype=float)\n",
    "    polar0 = [row[0] for row in file1]\n",
    "    polar1 = [row[1] for row in file1]\n",
    "    polar2 = [row[2] for row in file1]\n",
    "    nodeID = [row[3] for row in file1]\n",
    "    r0 = [row[5] for row in file1]\n",
    "    r1 = [row[6] for row in file1]\n",
    "    r2 = [row[7] for row in file1]\n",
    "    \n",
    "    polar0ave = numpy.mean(polar0);\n",
    "    polar1ave = numpy.mean(polar1);\n",
    "    polar2ave = numpy.mean(polar2);\n",
    "  \n",
    "    Cr = numpy.zeros(len(polar0));\n",
    "    R = numpy.zeros(len(polar0));\n",
    "    for i in range(1,len(polar0)):\n",
    "        for j in range(nstart,nend):\n",
    "            if i != j:\n",
    "                Cr[i] += 1/(0.8*0.8)*((polar0[i] - polar0ave)* (polar0[j] - polar0ave) + (polar1[i] - polar1ave) * (polar1[j] - polar1ave) + (polar2[i] - polar2ave) * (polar2[j] - polar2ave));\n",
    "        R[i] = (r0[i]*r0[i] +  r1[i]*r1[i] + r2[i]*r2[i])**(0.5);\n",
    "    return [Cr, R];\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "ename": "AttributeError",
     "evalue": "'tuple' object has no attribute 'shape'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mAttributeError\u001b[0m                            Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-83-a6ce210bce06>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[0mfunct\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;36m10\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mshape\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;31mAttributeError\u001b[0m: 'tuple' object has no attribute 'shape'"
     ]
    }
   ],
   "source": [
    "funct(10).shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Parallel time: 235.973917\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "\n",
    "start = time.time()\n",
    "n_proc = len(clients.ids);\n",
    "n = 4000; #should be multiple of n_proc\n",
    "dof_pass = n/n_proc;\n",
    "\n",
    "Crfunc = dview.map_sync(funct,range(0,n,dof_pass),range(0 + dof_pass,n + 1,dof_pass))\n",
    "\n",
    "print \"Parallel time: %f\" % (time.time() - start)\n",
    "#the above code uses two input arguments (two ranges) for the distribution of the outer loop\n",
    "\n",
    "#how do call/save the solution? first need to catenate them -- Cr = Crfunc[pid][0], R = Crfunc[pid][1];"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 111,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#4 processor version:\n",
    "\n",
    "CR = numpy.concatenate(\n",
    "   (\n",
    "    Crfunc[0][0][numpy.nonzero(Crfunc[0][0])], \n",
    "    Crfunc[1][0][numpy.nonzero(Crfunc[1][0])],\n",
    "    Crfunc[2][0][numpy.nonzero(Crfunc[2][0])],\n",
    "    Crfunc[3][0][numpy.nonzero(Crfunc[3][0])]\n",
    ")\n",
    "                               );\n",
    "\n",
    "R = numpy.concatenate(\n",
    "   (\n",
    "    Crfunc[0][1][numpy.nonzero(Crfunc[0][1])], \n",
    "    Crfunc[1][1][numpy.nonzero(Crfunc[1][1])],\n",
    "    Crfunc[2][1][numpy.nonzero(Crfunc[2][1])],\n",
    "    Crfunc[3][1][numpy.nonzero(Crfunc[3][1])]\n",
    ")\n",
    "                               );"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "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.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}