def set_shared(self, params, tasks, force=False):
        """Configures the mandatory shared resources.

        Updates `params`. Places tasks in `taskq`.

        Args:
            params (dict): Parameters.
            tasks (dict): Tasks.

        Keyword Args:
            force (bool): Forces task into `taskq`, regardless of whether it is already complete.

        """
        # Create a client that will the set the actual resource data.
        params["save"] = self.save
        client = SCManager(user=self.user, address=self.address,
                           authkey=self.authkey)
        client.connect()
        # Update params.
        client.params().update(params)
        # Update tasks.
        taskq = client.taskq()
        for task in tasks.itervalues():
            if force or not task.get("complete", False):
                taskq.put(task)
        util.info("[Server] Shared data successfully created.")
        del client

def _send_result(local, remote, retries):
    """Send a task's result file to remote location.

    Args:
        local (str): Path of local file.
        remote (str): Path of remote file.
        retries (int): Number of retries to perform.

    Return:
        (bool): Indicates whether send was successful.

    """
    send_str = "{} (local) > {} (remote)".format(local, remote)
    sent_pths = ()
    for r in xrange(retries):
        sent_pths = ssh.put(local, remote)
        if sent_pths.succeeded and remote in sent_pths:
            sent = True
            break
        else:
            msg = "[Client] {}/{} failed send retries ({}){{}}".format(
                r + 1, retries, send_str)
            if r + 1 < retries:
                util.info(msg.format(", retrying..."))
            else:
                util.info(msg.format(", exiting."))
    else:
        sent = False
        # sys.exit(0)  # raise NetworkError(msg.format("."))
    return sent

    def bootstrap_2_6_6(self):
        """Pulled from python 2.6.6. Needed to ensure we have the fix from
        http://bugs.python.org/issue5313 when running on python version 2.6.2
        or lower."""

        try:
            self._children = set()
            self._counter = itertools.count(1)
            try:
                sys.stdin.close()
                sys.stdin = open(os.devnull)
            except (OSError, ValueError):
                pass
            multiprocessing._current_process = self
            util._finalizer_registry.clear()
            util._run_after_forkers()
            util.info("child process calling self.run()")
            try:
                self.run()
                exitcode = 0
            finally:
                util._exit_function()
        except SystemExit, e:
            if not e.args:
                exitcode = 1
            elif type(e.args[0]) is int:
                exitcode = e.args[0]
            else:
                sys.stderr.write(e.args[0] + "\n")
                sys.stderr.flush()
                exitcode = 1

    def shutdown(self, c):
        '''
        Shutdown this process
        '''
        try:
            try:
                util.debug('manager received shutdown message')
                c.send(('#RETURN', None))

                if sys.stdout != sys.__stdout__:
                    util.debug('resetting stdout, stderr')
                    sys.stdout = sys.__stdout__
                    sys.stderr = sys.__stderr__

                util._run_finalizers(0)

                for p in active_children():
                    util.debug('terminating a child process of manager')
                    p.terminate()

                for p in active_children():
                    util.debug('terminating a child process of manager')
                    p.join()

                util._run_finalizers()
                util.info('manager exiting with exitcode 0')
            except:
                import traceback
                traceback.print_exc()
        finally:
            exit(0)

	def _bootstrap(self):
		from multiprocessing import util
		global _current_process

		try:
			self._children = set()
			self._counter = itertools.count(1)
			try:
				# sys.stdin.close()
				sys.stdin = open(os.devnull)
			except (OSError, ValueError):
				pass
			_current_process = self
			util._finalizer_registry.clear()
			util._run_after_forkers()
			util.info('child process calling self.run()')
			try:
				self.run()
				exitcode = 0
			finally:
				pass
				# util._exit_function()
		except SystemExit, e:
			if not e.args:
				exitcode = 1
			elif isinstance(e.args[0], int):
				exitcode = e.args[0]
			else:
				sys.stderr.write(str(e.args[0]) + '\n')
				sys.stderr.flush()
				exitcode = 1

    def _feed(buffer, notempty, send, writelock, close, ignore_epipe):
        debug('starting thread to feed data to pipe')
        from .util import is_exiting

        nacquire = notempty.acquire
        nrelease = notempty.release
        nwait = notempty.wait
        bpopleft = buffer.popleft
        sentinel = _sentinel
        if sys.platform != 'win32':
            wacquire = writelock.acquire
            wrelease = writelock.release
        else:
            wacquire = None

        try:
            while 1:
                nacquire()
                try:
                    if not buffer:
                        nwait()
                finally:
                    nrelease()
                try:
                    while 1:
                        obj = bpopleft()
                        if obj is sentinel:
                            debug('feeder thread got sentinel -- exiting')
                            close()
                            return

                        if wacquire is None:
                            send(obj)
                            # Delete references to object. See issue16284
                            del obj
                        else:
                            wacquire()
                            try:
                                send(obj)
                                # Delete references to object. See issue16284
                                del obj
                            finally:
                                wrelease()
                except IndexError:
                    pass
        except Exception as e:
            if ignore_epipe and getattr(e, 'errno', 0) == errno.EPIPE:
                return
            # Since this runs in a daemon thread the resources it uses
            # may be become unusable while the process is cleaning up.
            # We ignore errors which happen after the process has
            # started to cleanup.
            try:
                if is_exiting():
                    info('error in queue thread: %s', e)
                else:
                    import traceback
                    traceback.print_exc()
            except Exception:
                pass

    def _after_fork(self):
        self._manager = None
        try:
            self._incref()
        except Exception as e:
            util.info('incref failed: %s' % e)

        return

    def _feed(buffer, notempty, send_bytes, writelock, close, reducers,
              ignore_epipe, onerror, queue_sem):
        util.debug('starting thread to feed data to pipe')
        nacquire = notempty.acquire
        nrelease = notempty.release
        nwait = notempty.wait
        bpopleft = buffer.popleft
        sentinel = _sentinel
        if sys.platform != 'win32':
            wacquire = writelock.acquire
            wrelease = writelock.release
        else:
            wacquire = None

        while 1:
            try:
                nacquire()
                try:
                    if not buffer:
                        nwait()
                finally:
                    nrelease()
                try:
                    while 1:
                        obj = bpopleft()
                        if obj is sentinel:
                            util.debug('feeder thread got sentinel -- exiting')
                            close()
                            return

                        # serialize the data before acquiring the lock
                        obj_ = CustomizableLokyPickler.dumps(
                            obj, reducers=reducers)
                        if wacquire is None:
                            send_bytes(obj_)
                        else:
                            wacquire()
                            try:
                                send_bytes(obj_)
                            finally:
                                wrelease()
                        # Remove references early to avoid leaking memory
                        del obj, obj_
                except IndexError:
                    pass
            except BaseException as e:
                if ignore_epipe and getattr(e, 'errno', 0) == errno.EPIPE:
                    return
                # Since this runs in a daemon thread the resources it uses
                # may be become unusable while the process is cleaning up.
                # We ignore errors which happen after the process has
                # started to cleanup.
                if util.is_exiting():
                    util.info('error in queue thread: %s', e)
                    return
                else:
                    queue_sem.release()
                    onerror(e, obj)

    def _feed(buffer, notempty, send, writelock, close):
        debug('starting thread to feed data to pipe')
        from .util import is_exiting
        nacquire = notempty.acquire
        nrelease = notempty.release
        nwait = notempty.wait
        bpopleft = buffer.popleft
        sentinel = _sentinel
        if sys.platform != 'win32':
            wacquire = writelock.acquire
            wrelease = writelock.release
        else:
            wacquire = None
        try:
            while 1:
                nacquire()
                try:
                    if not buffer:
                        nwait()
                finally:
                    nrelease()

                try:
                    while 1:
                        obj = bpopleft()
                        if obj is sentinel:
                            debug('feeder thread got sentinel -- exiting')
                            close()
                            return
                        if wacquire is None:
                            send(obj)
                        else:
                            wacquire()
                            try:
                                send(obj)
                            finally:
                                wrelease()

                except IndexError:
                    pass

        except Exception as e:
            try:
                if is_exiting():
                    info('error in queue thread: %s', e)
                else:
                    import traceback
                    traceback.print_exc()
            except Exception:
                pass

        return

 def _run_server(cls, registry, address, authkey, serializer, writer,
                 conn_writer, initializer=None, initargs=()):
     """ Create a server, report its address and run it."""
     if initializer is not None:
         initializer(*initargs)
     # Create server.
     server = cls._Server(registry, fqaddr(address), authkey, serializer,
                          conn_writer)
     # Inform parent process of the server's address.
     writer.send(server.address)
     writer.close()
     # Run the manager.
     util.info("Server running at {}:{}.".format(*server.address))
     server.serve_forever()

 def _finalize_manager(process, address, authkey, state, conns, _Client):
     """ Shutdown the manager process; will be registered as a
     finalizer."""
     if process.is_alive():
         util.info("Sending shutdown message to manager.")
         try:
             conn = _Client(fqaddr(address), authkey=authkey)
             try:
                 dispatch(conn, None, "shutdown")
             finally:
                 conn.close()
         except Exception:
             pass
         for conn in conns:
             conn.close()
         process.join(timeout=0.2)
         if process.is_alive():
             util.info("Manager still alive.")
             if hasattr(process, "terminate"):
                 util.info("Trying to `terminate()` manager process.")
                 process.terminate()
                 process.join(timeout=0.1)
                 if process.is_alive():
                     util.info("Manager still alive after terminate.")
     state.value = State.SHUTDOWN
     try:
         del BaseProxy._address_to_local[fqaddr(address)]
     except KeyError:
         pass

    def _finalize_manager(process, address, authkey, state, _Client):
        '''
        Shutdown the manager process; will be registered as a finalizer
        '''
        if process.is_alive():
            util.info('sending shutdown message to manager')
            try:
                conn = _Client(address, authkey=authkey)
                try:
                    managers.dispatch(conn, None, 'shutdown')
                finally:
                    conn.close()
            except Exception:
                pass

            process.join(timeout=0.2)
            if process.is_alive():
                util.info('manager still alive')
                if hasattr(process, 'terminate'):
                    util.info('trying to `terminate()` manager process')

                    try:
                        process.terminate()
                        process.join(timeout=0.1)
            # XXX: catch the OS error ... something weird is going on here..
                    except OSError:
                        pass
                    if process.is_alive():
                        util.info('manager still alive after terminate')

        state.value = managers.State.SHUTDOWN
        try:
            del managers.BaseProxy._address_to_local[address]
        except KeyError:
            pass

    def _run_server(cls, registry, address, authkey, serializer, writer):
        '''
        Create a server, report its address and run it
        '''
        # create server
        server = cls._Server(registry, address, authkey, serializer)

        # inform parent process of the server's address
        writer.send(server.address)
        writer.close()

        # run the manager
        util.info('manager serving at %r', server.address)
        server.serve_forever()

 def _worker_process(self, finish, job, taskq, doneq, max_run_retries=3,
                     max_send_retries=10):
     """Run a set of jobs."""
     job.setup()
     run_retries = 0
     exitcode = 0
     # Loop over tasks.
     while not taskq.empty():
         # Pop task off queue.
         task = taskq.get()
         success = False
         T0 = Time.time()
         try:
             progress, tmp_fid = job.run(task)
         except EOFError as err:
             # Report what went wrong and retry.
             util.debug("[Client] {}".format(err.msg))
             msg = "[Client] {}/{} failed run retries, {{}}".format(
                 run_retries, max_run_retries)
             if run_retries < max_run_retries:
                 run_retries += 1
                 util.debug(msg.format("retrying..."))
             else:
                 util.debug(msg.format("exiting."))
                 raise err
         else:
             # Send results.
             sent = send_result(tmp_fid, task, retries=max_send_retries)
             if sent or not self.save:
                 # Mark simulation as complete.
                 task["complete"] = True
                 # Task is done.
                 doneq.put(task)
                 success = True
                 # Report progress.
                 progress.task = (taskq.qsize(), T0.delta())
                 progress.report()
         finally:
             if not success:
                 # Task did not complete successfully: put it back in taskq.
                 taskq.put(task)
             taskq.task_done()
             exitcode = 0
             if finish.is_set():
                 exitcode = 100
                 break
     job.teardown()
     util.info("[Client] Process complete: {}.".format(ProcLabel()))
     sys.exit(exitcode)

def _global_after_fork():
    # Previously every app would call:
    #    `register_after_fork(app, app._after_fork)`
    # but this created a leak as `register_after_fork` stores concrete object
    # references and once registered an object cannot be removed without
    # touching and iterating over the private afterfork registry list.
    #
    # See Issue #1949
    from celery import _state
    from multiprocessing.util import info
    for app in _state.apps:
        try:
            app._after_fork()
        except Exception as exc:
            info('after forker raised exception: %r' % (exc, ), exc_info=1)

    def _feed(buffer, notempty, send, writelock, close):
        debug('starting thread to feed data to pipe')
        from .util import is_exiting

        nacquire = notempty.acquire
        nrelease = notempty.release
        nwait = notempty.wait
        bpopleft = buffer.popleft
        sentinel = _sentinel
        wacquire = writelock.acquire
        wrelease = writelock.release

        try:
            while 1:
                try:
                    while 1:
                        obj = bpopleft()
                        if obj is sentinel:
                            debug('feeder thread got sentinel -- exiting')
                            close()
                            return

                        if wacquire is None:
                            send(obj)
                        else:
                            wacquire()
                            try:
                                send(obj)
                            finally:
                                wrelease()
                except IndexError:
                    pass
                yield "Done for now"
        except Exception as e:
            # Since this runs in a daemon thread the resources it uses
            # may be become unusable while the process is cleaning up.
            # We ignore errors which happen after the process has
            # started to cleanup.
            try:
                if is_exiting():
                    info('error in queue thread: %s', e)
                else:
                    import traceback
                    traceback.print_exc()
            except Exception:
                pass

    def get_shared(self):
        """Get shared data from server.

        Return:
            (tuple: dict, multiprocessing.JoinableQueue,
            multiprocessing.Queue, multiprocessing.JoinableQueue):
            Parameters, shared queue for tasks, shared queue for done
            tasks, shared queue for active connections.

        """
        params = self.params()
        self.Job = params["Job"]
        taskq = self.taskq()
        doneq = self.doneq()
        activeq = self.activeq()
        util.info("[Client] Received shared data.")
        # Notify the server that this client is active.
        activeq.put(self.host)
        return params, taskq, doneq, activeq

    def start(self, params, tasks, force=False):
        """Start server.

        Args:
            params (dict): Parameters.
            tasks (dict): Tasks.

        Keyword Args:
            force (bool): Forces task into `taskq`, regardless of whether
                          it is already complete.

        """
        super(ServerManager, self).start()
        self.set_shared(params, tasks, force=force)
        # Monitor the tasks and update the tasks file as completed tasks
        # arrive from the clients.
        self.manage_tasks(tasks)
        self.shutdown()
        util.info("[Server] Jobs complete. Shutting down.")
        sys.exit(0)

    def _malloc(self, size):
        # returns a large enough block -- it might be much larger
        i = bisect.bisect_left(self._lengths, size)
        if i == len(self._lengths):
            length = self._roundup(max(self._size, size), mmap.PAGESIZE)
            self._size *= 2
            info('allocating a new mmap of length %d', length)
            arena = Arena(length)
            self._arenas.append(arena)
            return (arena, 0, length)
        else:
            length = self._lengths[i]
            seq = self._len_to_seq[length]
            block = seq.pop()
            if not seq:
                del self._len_to_seq[length], self._lengths[i]

        (arena, start, stop) = block
        del self._start_to_block[(arena, start)]
        del self._stop_to_block[(arena, stop)]
        return block

    def _spawn(self, finish, params, taskq, doneq, args, kwargs):
        """Spawn a job process.

        Args:
            finish (multiprocessing.Event): Signals that process is finished.
            params (dict): Parameters.
            taskq (multiprocessing.JoinableQueue): Shared queue for tasks.
            doneq (multiprocessing.Queue): Shared queue for done tasks.
            args (tuple): Arguments for ``_worker_process``.
            kwargs (dict): Keyword arguments for ``_worker_process``.

        Return:
            (multiprocessing.Process): Worker process.

        """
        # Initialize the job.
        job = self.Job(params, *args, **kwargs)
        proc = Process(target=self._worker_process,
                       args=(finish, job, taskq, doneq))
        util.info("[Client] Spawning process: '{}'".format(proc.name))
        return proc