import multiprocessing, random, time\n",
"import zmq\n",
"import collector, democfg, distributor, worker\n",
"\n",
"def main(tasks):\n",
" # Initialize random number generator\n",
" random.seed()\n",
" print("Starting task workers ...")\n",
" for worker_id in range(democfg.worker_pool_size):\n",
" worker_proc = multiprocessing.Process(\n",
" target=worker.work, args=[worker_id])\n",
" worker_proc.start()\n",
" print("Starting task collector ...")\n",
" collector_proc = multiprocessing.Process(target=collector.collect)\n",
" collector_proc.start()\n",
" time.sleep(democfg.pause_time)\n",
" print("Starting task distributor ...")\n",
" distributor.distribute(tasks)\n",
"\n",
"if __name__ == "__main__":\n",
" tasks = [1e8 / democfg.task_count] * democfg.task_count\n",
" main(tasks)\n",
"import time\n",
"import zmq\n",
"import democfg\n",
"\n",
"def distribute(tasks):\n",
" context = zmq.Context()\n",
" # Socket to send messages on\n",
" sender = context.socket(zmq.PUSH)\n",
" sender.bind(democfg.routing_table["receiver"])\n",
" # Socket with direct access to the sink used to syncronize start of batch\n",
" syncher = context.socket(zmq.PUSH)\n",
" syncher.connect(democfg.routing_table["sender"])\n",
" # Give 0MQ time to start up\n",
" time.sleep(democfg.pause_time)\n",
" syncher.send(democfg.start_flag)\n",
" for task in tasks:\n",
" sender.send_pyobj(task)\n",
" # Give 0MQ time to deliver message\n",
" time.sleep(democfg.pause_time)\n",
"import random\n",
"import zmq\n",
"import democfg, tasker\n",
"\n",
"def is_done(socks, controller):\n",
" if (socks.get(controller) == zmq.POLLIN and\n",
" controller.recv() == democfg.done_msg):\n",
" return True\n",
" return False\n",
"\n",
"def work(worker_id):\n",
" context = zmq.Context()\n",
" # Socket to receive messages on\n",
" receiver = context.socket(zmq.PULL)\n",
" receiver.connect(democfg.routing_table["receiver"])\n",
" # Socket to send messages to\n",
" sender = context.socket(zmq.PUSH)\n",
" sender.connect(democfg.routing_table["sender"])\n",
" # Socket for control input\n",
" controller = context.socket(zmq.SUB)\n",
" controller.connect(democfg.routing_table["controller"])\n",
" controller.setsockopt(zmq.SUBSCRIBE, b"")\n",
" # Process messages from receiver and controller\n",
" poller = zmq.Poller()\n",
" poller.register(receiver, zmq.POLLIN)\n",
" poller.register(controller, zmq.POLLIN)\n",
" # Process messages from both sockets\n",
" run_loop = True\n",
" while run_loop:\n",
" socks = dict(poller.poll())\n",
" if socks.get(receiver) == zmq.POLLIN:\n",
" task_data = receiver.recv_pyobj()\n",
" # Process task data\n",
" result = tasker.task(task_data)\n",
" sender.send_pyobj(result)\n",
" if is_done(socks, controller):\n",
" run_loop = False\n",
"import random\n",
"import zmq\n",
"import democfg\n",
"\n",
"def task(task_data):\n",
" in_circle = 0\n",
" for _ in range(int(task_data)):\n",
" (x, y) = (random.random(), random.random())\n",
" if (x ** 2 + y ** 2) <= 1:\n",
" in_circle += 1\n",
" return 4 * in_circle / task_data\n",
"import time\n",
"import zmq\n",
"import democfg, processor\n",
"\n",
"def collect():\n",
" context = zmq.Context()\n",
" # Socket to receive messages on (collect results from worker)\n",
" receiver = context.socket(zmq.PULL)\n",
" receiver.bind(democfg.routing_table["sender"])\n",
" # Socket for worker control\n",
" controller = context.socket(zmq.PUB)\n",
" controller.bind(democfg.routing_table["controller"])\n",
" # Wait for start signal\n",
" assert receiver.recv() == democfg.start_flag\n",
" processor.process(receiver)\n",
" # Let workers know that all results have been processed\n",
" controller.send(democfg.done_msg)\n",
" # Finished\n",
" time.sleep(5 * democfg.pause_time) # Give 0MQ time to deliver\n",
"import time\n",
"import zmq\n",
"import democfg\n",
"\n",
"def process(receiver):\n",
" return get_results(receiver)\n",
"\n",
"def get_results(receiver):\n",
" data = []\n",
" # Process results from the workers\n",
" for _ in range(democfg.task_count):\n",
" result = receiver.recv_pyobj()\n",
" print("Processing result: {}".format(result))\n",
" data.append(result)\n",
" # Calculate final result\n",
" print("Final result: {}".format(sum(data)/len(data)))\n",
" return data\n",
"worker_pool_size = 8\n",
"task_count = 8\n",
"delay_range = (1, 100) # milliseconds\n",
"pause_time = 1 # seconds\n",
"start_flag = b"0"\n",
"done_msg = b"DONE"\n",
"routing_table = {\n",
" "receiver": "tcp://127.0.0.1:7557",\n",
" "sender": "tcp://127.0.0.1:7558",\n",
" "controller": "tcp://127.0.0.1:7559"}\n",
"![](\"ipy-overview.png\")
\n",
"\n",
"Due to this architecture, IPython supports the following different styles of parallelism:\n",
"\n",
"* Single program, multiple data\n",
"* Multiple program, multiple data\n",
"* Message passing using MPI\n",
"* Task farming\n",
"* Data parallel\n",
"* Combinations of these approaches\n",
"* Custom user defined approaches\n",
"\n",
"Practically speaking, this allows IPython users to tackle the following use cases:\n",
"\n",
"* Quickly parallelize algorithms that are embarrassingly parallel using a number of simple approaches. Many simple things can be parallelized interactively in one or two lines of code.\n",
"* Steer traditional MPI applications on a supercomputer from an IPython session on your laptop.\n",
"* Analyze and visualize large datasets (that could be remote and/or distributed) interactively using IPython and tools like matplotlib/TVTK.\n",
"* Develop, test and debug new parallel algorithms (that may use MPI) interactively.\n",
"* Tie together multiple MPI jobs running on different systems into one giant distributed and parallel system.\n",
"* Start a parallel job on your cluster and then have a remote collaborator connect to it and pull back data into their local IPython session for plotting and analysis.\n",
"* Run a set of tasks on a set of CPUs using dynamic load balancing."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Getting Started"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"To use IPython for parallel computing, you need to start one instance of the controller and one or more instances of the engine. Initially, it is best to simply start a controller and engines on a single host using the ``ipcluster`` command. To start a controller and 4 engines on your localhost, switch to a terminal window with this notebook's virtual environment activated and execute the following:\n",
"\n",
"```\n",
"$ ipcluster start -n 4\n",
"```\n",
"\n",
"This will run the ipcluster app in the foreground, showing output such as:\n",
"\n",
"```\n",
"2015-04-28 09:50:29.584 [IPClusterStart] Using existing profile dir: '/Users/oubiwann/.ipython/profile_default'\n",
"2015-04-28 09:50:29.587 [IPClusterStart] Starting ipcluster with [daemon=False]\n",
"2015-04-28 09:50:29.589 [IPClusterStart] Creating pid file: /Users/oubiwann/.ipython/profile_default/pid/ipcluster.pid\n",
"2015-04-28 09:50:29.589 [IPClusterStart] Starting Controller with LocalControllerLauncher\n",
"2015-04-28 09:50:30.593 [IPClusterStart] Starting 4 Engines with LocalEngineSetLauncher\n",
"```"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Direct View"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"The “direct view” is called such because the DirectView object offers an interface to the cluster which doesn't go through the schedulers, but instead offers direct, manual execution control."
]
},
{
"cell_type": "code",
"execution_count": 20,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"[0, 1, 2, 3]"
]
},
"execution_count": 20,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"from IPython.parallel import Client\n",
"client = Client() \n",
"client.ids "
]
},
{
"cell_type": "code",
"execution_count": 21,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"StarCluster is an open source cluster-computing toolkit for Amazon’s Elastic Compute Cloud (EC2). StarCluster has been designed to simplify the process of building, configuring, and managing clusters of virtual machines on Amazon’s EC2 cloud.\n", "\n", "\n", "
![](\"sc-overview.png\")
\n",
"\n",
"StarCluster was developed at MIT for both use by researchers as well as instruction in the classroom. Early uses of it included protein visualization, genetic simulation, and quantum mechanical computations in molecular dynamics. \n",
"\n",
"StarCluster takes care of the heavy lifting when it comes to creating and configuring a cluster on EC2. StarCluster configures the following items out-of-the-box:\n",
"\n",
"* Security Groups\n",
"* User-friendly Hostnames\n",
"* User Accounts\n",
"* Password-less SSH\n",
"* NFS Shares\n",
"* EBS Volumes\n",
"* Queueing System\n",
"* Scratch Space\n",
"* Extensibility via Plugin System\n",
"* Cluster Creation and Management\n",
"* Dynamic Cluster Resizing\n",
"* Easy Data Copying to and from Clusters\n",
"\n",
"StarCluster is written in Python using the boto, crypto, and paramiko libraries, among others. It supports various Amazon OS images and multiple versions of those OSs. "
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Code and Images"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"We have made modifications to StarCluster and some of its dependencies in order to run on Python 3. This notebook's dependencies download the following Python 3 versions of libraries:\n",
"\n",
"* https://github.com/oubiwann/StarCluster\n",
"* https://github.com/oubiwann/optcomplete\n",
"* https://github.com/oubiwann/workerpool\n",
"\n",
"Note that these are still in an experimental state, but are suitable for our needs.\n",
"\n",
"We have created an updated StarCluster AMI with Ubuntu 14.04.2 LTS and IPython for Python 2 and 3, as well as the many scientific computing libraries you've come to know and love:\n",
" * [ami-a73c0a97](https://console.aws.amazon.com/ec2/v2/home?region=us-west-2#LaunchInstanceWizard:ami=ami-a73c0a97)\n",
" \n",
"Only one region was used for the example in this notebook; if you would like to run in an AWS EC2 region other than ``us-west-2``, search for \"ami-a73c0a97\" in the \"AMIs\" pane of your AWS web console, and then from the \"Actions\" drop-down, select \"Copy\". You will then be presented with a dialog box where you can select a region as well as give it a name and description.\n",
"\n",
"A best attempt was made at securing the AMI according to the instructions provided by Amazon [here](https://aws.amazon.com/articles/0155828273219400). However, no guarantees are made -- as usual, use at your own risk."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Configuring StarCluster"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"In the process of running ``make`` for this notebook, you will have downloaded and installed all the dependencies. In a terminal, activate the Python virtual environment from the directory you started this notebook:\n",
"\n",
"```\n",
". ../.venv-mmpl/bin/activate\n",
"```\n",
"\n",
"Now the ``startcluster`` command line utility will be available to you. Let's use it to create a default configuration file for you:\n",
"\n",
"```\n",
"$ starcluster help\n",
"```\n",
"```\n",
"StarCluster - (http://star.mit.edu/cluster) (v. 0.95.7+py3k-oubiwann)\n",
"Software Tools for Academics and Researchers (STAR)\n",
"Please submit bug reports to starcluster@mit.edu\n",
"!!! ERROR - config file /Users/oubiwann/.starcluster/config does not exist\n",
"\n",
"Options:\n",
"--------\n",
"[1] Show the StarCluster config template\n",
"[2] Write config template to /Users/oubiwann/.starcluster/config\n",
"[q] Quit\n",
"\n",
"Please enter your selection: 2\n",
"```\n",
"\n",
"At which point your configuration will be generated:\n",
"\n",
"```\n",
">>> Config template written to /Users/oubiwann/.starcluster/config\n",
">>> Please customize the config template\n",
"```\n",
"\n",
"You will need to make the following edits to your new ``~/.starcluster/config`` file:\n",
" * Add a value for the ``AWS_ACCESS_KEY_ID`` entry\n",
" * Add a value for the ``AWS_SECRET_ACCESS_KEY`` entry\n",
" * Add a value for the ``AWS_USER_ID`` entry\n",
" * If you are using a different region than the default of ``us-east-1``:\n",
" * Uncomment and set a value for ``AWS_REGION_NAME`` \n",
" * Uncomment and set a value for ``AWS_REGION_HOST`` for your chosen region\n",
" * Change the value for ``KEY_LOCATION`` to the key ``.pem`` file you downloaded from AWS (this is the full path and filename, including the ``.pem`` file extension)\n",
" * Change the value for ``KEYNAME`` to the filename of the ``.pem`` file, without the extension\n",
" * Change the value for ``NODE_IMAGE_ID`` to the AMI ID given above: ``ami-a73c0a97``\n",
" * Change the value for ``CLUSTER_SIZE`` to ``1`` -- we will add more shortly\n",
" \n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Starting a Cluster"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"With these changes in place, we're ready to start up our cluster:\n",
"\n",
"```\n",
"$ starcluster start my-cluster-name\n",
"```\n",
"\n",
"At which point you will start to see output as StarCluster makes requests to AWS for starting up (and waiting for) nodes. You will see ASCII progress bars and/or status messages for things such as:\n",
" * Waiting for the instances to come up\n",
" * Waiting for SSH to be ready on all nodes\n",
" * Cluster configuration\n",
" * Setting up NFS\n",
" * Setting up SSH keys\n",
" * Setting up Sun Grid Engine\n",
"\n",
"You can expect a four-node cluster to take about 3-4 minutes for a complete start up."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Running a Job"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"TBD"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## IPython Clusters Revisited"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"TBD"
]
}
],
"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.4.2"
}
},
"nbformat": 4,
"nbformat_minor": 0
}