def execnet_gateways(self):
        # TODO: Eventlet hangs on Mac OS X with popen.
        # execnet.set_execmodel("eventlet", "thread")
        gw = self.gateway

        if not gw:
            for _ in range(self.job_num):
                yield execnet.makegateway(
                    "" if "FOWLER_PYTHON" not in os.environ else "popen//python={}".format(os.environ["FOWLER_PYTHON"])
                )

        for gateway_spec in gw:
            if "*" in gateway_spec:
                num, spec = gateway_spec.split("*")
                num = int(num)

                group = execnet.Group()
                group.defaultspec = spec

                xspec = execnet.XSpec(spec)
                master_spec = "ssh={xspec.via}//" "id={xspec.via}//" "python={xspec.python}" "".format(xspec=xspec)

                logger.debug("Connecting to master %s to create %s gateways.", master_spec, num)
                group.makegateway(master_spec)

                for _ in range(num):
                    yield group.makegateway()
            else:
                yield execnet.makegateway()

def node_recon(nodelist, interactive=True):
    """grab system information from a list of hosts and create or update
    nodes' db entries.
    """
    import execnet
    from dirt.tasks import system_info
    from dirt.core.db import db

    nodes = db.get_nodes()
    for node in nodelist:
        log.write("Connecting to host %s" % node)
        try:
            gw = execnet.makegateway("ssh=%s" % node)
        except execnet.HostNotFound:
            log.write("Host not found: %s" % node)
            continue
        log.write("Connected to host %s" % node)

        ch = gw.remote_exec(system_info)
        sys_info = ch.receive()

        # update the db
        if sys_info["fqdn"] in nodes:
            d = nodes[sys_info["fqdn"]]
            d["sys_info"] = sys_info
            d["enabled"] = True
        else:
            d = {"type": "node", "fqdn": sys_info["fqdn"], "sys_info": sys_info}
            log.write("Adding new node %(fqdn)s to database" % d)
            d["enabled"] = settings.node_enable_default
        db.save(d)

def turn(responses):
    if turn_way == "left":
        z = 1 
    elif turn_way == "right":
        z = 1 
    else:
        responses.put("failure")
        return

    action_code = """
    import rospy, time
    from geometry_msgs.msg import Twist

    rospy.init_node('move')
    p = rospy.Publisher('/base_controller/command', Twist)
    turn = Twist()
    turn.linear.x = 0;
    turn.linear.y = 0; turn.linear.z = 0;       
    turn.angular.x = 0; turn.angular.y = 0;
    turn.angular.z = %i;
    times = 0
    while 1:
        times += 1
        p.publish(turn)
        rospy.sleep(%s)
        if times == %i:
            break
    twist = Twist()
    p.publish(twist)
    channel.send("success")
    """ % (z, "0.1", 40)
    gw = execnet.makegateway("popen//python=python2.7")
    channel = gw.remote_exec(action_code)
    if channel.receive() == "success":
        responses.put("success")

 def connect(self):
     if self._cache_key not in RPC_CACHE:
         ctrl = self.runner._ploy_ctrl
         instance = ctrl.instances[self.host]
         if hasattr(instance, '_status'):
             if instance._status() != 'running':
                 raise errors.AnsibleError("Instance '%s' unavailable." % instance.config_id)
         try:
             ssh_info = instance.init_ssh_key(user=self.user)
         except instance.paramiko.SSHException as e:
             raise errors.AnsibleError("Couldn't validate fingerprint for '%s': %s" % (instance.config_id, e))
         spec = execnet.XSpec('ssh')
         ssh_args = instance.ssh_args_from_info(ssh_info)
         if utils.VERBOSITY > 3:
             ssh_args += ["-vvv"]
         spec.ssh = SSHArgs(ssh_args)
         vars = self.runner.inventory.get_variables(self.host)
         spec.python = vars.get('ansible_python_interpreter', 'python')
         gw = execnet.makegateway(spec)
         try:
             channel = gw.remote_exec(remote)
         except IOError as e:
             raise errors.AnsibleError("Couldn't open execnet channel for '%s': %s" % (instance.config_id, e))
         RPC_CACHE[self._cache_key] = RPCWrapper(channel)
     self.rpc = RPC_CACHE[self._cache_key]
     return self

def execnet_master_main(install_hook=True, callback=None, gw=None, slave_main=execnet_slave_main):
    callback = callback or handle_ch
    exit_event.clear()  # to allow multiple test cases to run despite our gross global state

    # This directory contains the 'baz' module.  Adding it to sys.path means that 'baz' will be importable on the
    # master, but not (normally) on any of the slaves.
    baz_path = os.path.join(os.path.abspath(os.path.dirname(__file__)), '../test/not-in-default-path')
    if baz_path not in sys.path:
        sys.path.append(baz_path)

    # Create another Python interpreter to be our slave.  (This could just as well be any other remote interpreter that
    # execnet supports.)
    gw = gw or execnet.makegateway()

    # This is all of the setup that we need to do!  This creates a channel to the gateway (and inserts itself into the
    # slave's import machinery) as well as a callback to service requests on this side.
    if install_hook:
        execnet_importhook.install_import_hook(gw)

    # Now we can go about our business.
    ch = gw.remote_exec(slave_main)
    ch.setcallback(partial(callback, ch), ENDMARKER)

    # When execnet_slave_main() finishes, our callback (handle_ch) will get ENDMARKER and set() the event, so that we
    # know to let the master end too.
    exit_event.wait()

def getinfo(sshname, ssh_config=None, loginfo=sys.stdout):
    if ssh_config:
        spec = "ssh=-F %s %s" % (ssh_config, sshname)
    else:
        spec += "ssh=%s" % sshname
    debug("connecting to", repr(spec))
    try:
        gw = execnet.makegateway(spec)
    except IOError:
        error("could not get sshgatway", sshname)
    else:
        ri = RemoteInfo(gw)
        #print "%s info:" % sshname
        prefix = sshname.upper() + " "
        print >>loginfo, prefix, "fqdn:", ri.getfqdn()
        for attr in (
            "sys.platform",
            "sys.version_info",
        ):
            loginfo.write("%s %s: " %(prefix, attr,))
            loginfo.flush()
            value = ri.getmodattr(attr)
            loginfo.write(str(value))
            loginfo.write("\n")
            loginfo.flush()
        memswap = ri.getmemswap()
        if memswap:
            mem,swap = memswap
            print >>loginfo, prefix, "Memory:", mem, "Swap:", swap
        cpuinfo = ri.getcpuinfo()
        if cpuinfo:
            numcpu, model = cpuinfo
            print >>loginfo, prefix, "number of cpus:",  numcpu
            print >>loginfo, prefix, "cpu model", model
        return ri

 def load_workers(self, workers):
     workers_dict = helpers.x_to_dict[type(workers)](workers)
     for worker in workers_dict["workers"]:
         ## if not configured to be in a pool, have a pool of your own.
         pool_ids = worker.setdefault("pools", [worker["id"]])
         ctrl_sock = self.ctx.socket(zmq.REQ)
         ctrl_lock = multiprocessing.Lock()
         ctrl_lock.acquire()
         self.workers[worker["id"]] = (ctrl_sock, ctrl_lock, worker)
         if "execnet_address" in worker:
             channel = execnet.makegateway(worker["execnet_address"]).remote_exec(remote_agent)
             channel.send(worker["control_address"])
         ctrl_sock.connect(worker["control_address"])
         ctrl_sock.send_json({"cmd":[{"instruction":"set_attr","attr":"name", "value": worker["id"]}]})
         print ctrl_sock.recv_json()
         ctrl_lock.release()
         for pool_id in pool_ids:
             pool = self.add_pool(pool_id)
             pool[2].add(worker["id"])
             ctrl_lock.acquire()
             ctrl_sock.send_json({"cmd":[{"instruction": "task_attach", "remote": pool[1]} ] } )
             print "task_attach:", ctrl_sock.recv_json()
             ctrl_sock.send_json({"cmd":[{"instruction": "results_attach", "remote": self.config["result_bind"]} ] } )
             print "results_attach:", ctrl_sock.recv_json()
             ctrl_lock.release()
         ctrl_lock.acquire()
         ctrl_lock.release()

    def load_workers(self, workers):
        workers_dict = helpers.x_to_dict[type(workers)](workers)
        for worker in workers_dict["workers"]:
            worker.setdefault("exec_id",0)
            ## if not configured to be in a pool, have a pool of your own.
            pool_id = worker.setdefault("pool", [worker["id"]])
            ctrl_sock = self.ctx().socket(zmq.REQ)
            ctrl_lock = threading.Lock()
            ctrl_lock.acquire()
            self.workers[worker["id"]] = (ctrl_sock, ctrl_lock, worker)
            if "execnet_address" in worker:
                channel = execnet.makegateway(worker["execnet_address"]).remote_exec(remote_agent)
                channel.send(worker["control_address"])
            #print "{id:s}: connecting to {control_address:s}...".format(**worker)
            ctrl_sock.connect(worker["control_address"])
            #print worker["id"], "connected"

            self._send_ctrl(worker, {"instruction":"set_attr","attr":"name", "value": worker["id"]})
            ctrl_lock.release()
            pool = self.add_pool(pool_id)
            pool[2].add(worker["id"])
            ctrl_lock.acquire()
            self._send_ctrl(worker, {"instruction": "task_attach", "remote": pool[1]})
            self._send_ctrl(worker, {"instruction": "results_attach", "remote": self.config["result_bind"]})
            ctrl_lock.release()
            ctrl_lock.acquire()
            ctrl_lock.release()

def map(mod, args, specs=[('popen', 2)]):
	gateways = []
	channels = []
	
	for spec, count in specs:
		for i in range(count):
			gw = execnet.makegateway(spec)
			gateways.append(gw)
			channels.append(gw.remote_exec(mod))
	
	cyc = itertools.cycle(channels)
	
	for i, arg in enumerate(args):
		channel = next(cyc)
		channel.send((i, arg))
	
	mch = execnet.MultiChannel(channels)
	queue = mch.make_receive_queue()
	l = len(args)
	results = [None] * l
	
	for j in range(l):
		channel, (i, result) = queue.get()
		results[i] = result
	
	for gw in gateways:
		gw.exit()
	
	return results

    def _start_cluster_simulation(self, subdomains, cluster=None):
        """Starts a simulation on a cluster of nodes."""

        if cluster is None:
            try:
                cluster = imp.load_source('cluster', self.config.cluster_spec)
            except IOError:
                cluster = imp.load_source('cluster',
                        os.path.expanduser('~/.sailfish/{0}'.format(self.config.cluster_spec)))

        self._cluster_gateways = []
        self._node_subdomains = split_subdomains_between_nodes(cluster.nodes, subdomains)

        import execnet
        for _, node in zip(self._node_subdomains, cluster.nodes):
            self._cluster_gateways.append(execnet.makegateway(node.host))

        # Copy files to remote nodes if necessary.
        if self.config.cluster_sync:
            local, dest = self.config.cluster_sync.split(':')
            assert dest[0] != '/', 'Only relative paths are supported on remote nodes.'
            rsync = execnet.RSync(local)
            for gw in self._cluster_gateways:
                rsync.add_target(gw, dest)
            rsync.send()

        subdomain_id_to_addr = {}
        for node_id, subdomains in enumerate(self._node_subdomains):
            for subdomain in subdomains:
                subdomain_id_to_addr[subdomain.id] = cluster.nodes[node_id].addr

        self._cluster_channels = []
        import sys
        for i, (node, gw) in enumerate(zip(cluster.nodes, self._cluster_gateways)):
            # Assign specific GPUs from this node, as defined by the cluster
            # config file.
            node_config = copy.copy(self.config)
            node_config.gpus = cluster.nodes[i].gpus
            for k, v in node.settings.items():
                setattr(node_config, k, v)

            self._cluster_channels.append(
                    gw.remote_exec(_start_cluster_machine_master,
                    args=pickle.dumps((node_config, self._node_subdomains[i])),
                    main_script=sys.argv[0],
                    lb_class_name=self._lb_class.__name__,
                    subdomain_addr_map=subdomain_id_to_addr,
                    iface=node.iface))

        ports = {}
        for channel in self._cluster_channels:
            data = channel.receive()
            # If a string is received, print it to help with debugging.
            if type(data) is str:
                print(data)
            else:
                ports.update(data)

        for channel in self._cluster_channels:
            channel.send(ports)

 def create_connection(self, hostname, connection_id, niceness):
   """Create a new connection to a prepared host."""
   gateway_cmd = "ssh="+hostname+"//id="+connection_id+"//nice="+ niceness
   gw = execnet.makegateway(gateway_cmd)
   # keep track of how we created this gw so we can re-create it if necessary.
   self.worker_attrs_[gw] = (hostname, connection_id, niceness)
   return gw

def map(mod, args, options=[('popen', 2)]): #2 = no of processes to spawn
        gateways = []
        channels = []

        for option, count in options:
                for i in range(count):
                        gw = execnet.makegateway(option)
                        gateways.append(gw)
                        channels.append(gw.remote_exec(mod))
        cyc = itertools.cycle(channels)

        for i, arg in enumerate(args):
                channel = cyc.next()
                channel.send((i, arg))

        mch = execnet.MultiChannel(channels)
        queue = mch.make_receive_queue()
        l = len(args)
        results = [None] * l

        for j in range(l):
                if j+1 <= sys.maxint:
                        print "\nProcessing document no. %d..." %(j+1)
                channel, (i, result) = queue.get()
                results[i] = result

        for gw in gateways:
                gw.exit()
        return results

def add_prole(ident):
    ''' for a given host, try to spawn the crack module on that host
    if it succeeds, add the channel to chans, the gateway to gateways
    and put this host in the task_tracker''' 

    host = ident.split("_")[0]
    try:           
        # establish connection to 'host'
        gw = execnet.makegateway("ssh="+host+"//id="+ident+"//python=python3//chdir=./execnet/crackr/")

        # run crack module on 'host'
        ch = gw.remote_exec(crack)
        ch.send((passwd, max_task_size))

        # put this chan, gw into a list of them
        chans[ident] = ch
        gateways[ident] = gw

        # task_tracker keeps track of what set of strings this machine/core
        # is working on so that if it fails these strings won't be skipped
        task_tracker[ident] = 0

        if VERBOSE: print("added " + ident)
        return 0

    except: 
        if VERBOSE: print("fail to add " + ident)
        return -1

def printver(python_executable):
    gw = execnet.makegateway(
        'popen//python={}'.format(python_executable))
    channel = gw.remote_exec("""
    import platform
    channel.send(platform.python_version())
    """)
    return channel.receive()

 def get_load(hostname):
     try:
         load_module = __import__('dirt.tasks.load', fromlist=['dirt.tasks'])
         gw = execnet.makegateway('ssh=%s' % hostname)
         ch = gw.remote_exec(load_module)
         return ch.receive()['load']
     except Exception:
         return None

def gateway(request):
    try:
        gw = execnet.makegateway('popen//python=%s' % request.param)
    except FileNotFoundError:
        pytest.skip('%s is not installed')

    gw.reconfigure(py2str_as_py3str=False, py3str_as_py2str=False)
    return gw

 def setup(self, ):
     self.testdir.chdir()
     #import os ; os.environ['EXECNET_DEBUG'] = "2"
     self.gateway = execnet.makegateway()
     self.config = config = self.testdir.parseconfigure()
     putevent = self.use_callback and self.events.put or None
     self.slp = SlaveController(None, self.gateway, config, putevent)
     self.request.addfinalizer(self.slp.ensure_teardown)
     self.slp.setup()

 def on_ssh_select(self, host_string):
     import execnet
     venv_paths = sublime.load_settings(SETTINGS_FILE).get("python_virtualenv_paths", [])
     try:
         gw = execnet.makegateway("ssh=" + host_string)
         ch = gw.remote_exec(VENV_SCAN_CODE)
     except Exception, e:
         sublime.error_message(repr(e))
         return

 def test_default_group(self):
     oldlist = list(execnet.default_group)
     gw = execnet.makegateway("popen")
     try:
         newlist = list(execnet.default_group)
         assert len(newlist) == len(oldlist) + 1
         assert gw in newlist
         assert gw not in oldlist
     finally:
         gw.exit()

    def execute_steps(self, loop, steps):
        if self._gw is None:
            self._gw = execnet.makegateway('ssh=%s' % self.host_string)

        channel = self._gw.remote_exec(reverse_some_text)
        channel.send('sent from the master')

        result = yield from loop.run_in_executor(None, channel.receive)

        logger.info('Recieved the result: %s' % result)