如何转移环境主机在使用多处理时要构造？

Question

我有几百个盒子只能用不同的网关访问，如下图所示：

gateway1:ip1,ip2
gateway2:ip3,ip4 ...

所有的工作都需要在一分钟内完成，所以我使用下面的多进程命令fab -f ytj_sto.py doitnow，错误如下。在

^{pr2}$

代码：

@parallel(pool_size=20)
def coll():
    print env.hosts
    print env.host
    print env.gateway
    if _is_ssh_ok(env.host):
        d = patt()

def doitnow():
    p=Pool(20)
    with open('ytj_sto.hn','r') as f:
        for line in f.readlines():
            line = line.strip()
            if not len(line) or line.startswith('#'):
                continue
            env.gateway = line.split(':')[0]
            env.hosts = line.split(':')[1].split(',')
            result = p.apply_async(coll, args=())
            result.get()
    p.close()
    p.join()

编辑时间： 我已经用fab-H-g解决了这个问题，谢谢大家

  def fabfun(Hosts,Gate,des,func1):
        with settings(hide('running'), warn_only=True):
           local(("fab -H %s -g %s -f %s %s ") % (Hosts,Gate,des,func1))

p=Pool(20)
starttime = time.asctime( time.localtime(time.time()) )
print('Waiting for all job done...%s' % starttime)
with open('ytj_sto.hn','r') as f:
    for line in f.readlines():
        line = line.strip()
        if not len(line) or line.startswith('#'):
            continue
        Hosts = line.split(':')[1]
        Gate = line.split(':')[0]
        p.apply_async(fabfun, args=(Hosts,Gate,des,func1))
    p.close()
    p.join()

Answer 1

如果您想像现在这样动态地设置env变量，应该使用execute。这样，executed任务接受您在运行时设置的env值。但是不幸的是，由于fabric不是完全线程安全的，您只能在您的情况下在每个网关并行启动一次这样的任务，因为{}是一个全局单例。在

A simple but integral aspect of Fabric is what is known as the “environment”: a Python dictionary subclass, which is used as a combination settings registry and shared inter-task data namespace.

The environment dict is currently implemented as a global singleton, fabric.state.env, and is included in fabric.api for convenience. Keys in env are sometimes referred to as “env variables”.

from fabric.context_managers import env

@parallel(pool_size=20)
def check_and_patt():
    if _is_ssh_ok(env.host):
        d = patt()

def doitnow():
    p=Pool(20)
    with open('ytj_sto.hn','r') as f:
        for line in f.readlines():
            line = line.strip()
            if not len(line) or line.startswith('#'):
                continue
            env.gateway = line.split(':')[0]
            env.hosts = line.split(':')[1].split(',')
            result = execute(check_and_patt)

Answer 2

我认为您可以将env作为参数提供给coll函数，如下所示：

@parallel(pool_size=20)
def coll(env): # <  updated
    print env.hosts
    print env.host
    print env.gateway
    if _is_ssh_ok(env.host):
        d = patt()

def doitnow():
    p=Pool(20)
    with open('ytj_sto.hn','r') as f:
        for line in f.readlines():
            line = line.strip()
            if not len(line) or line.startswith('#'):
                continue
            env.gateway = line.split(':')[0]
            env.hosts = line.split(':')[1].split(',')
            result = p.apply_async(coll, args=(env,)) # <  updated
            result.get()
    p.close()
    p.join()

使用multiprocessing library有一些怪癖。此信息可能与您的场景特别相关：

Global variables

Bear in mind that if code run in a child process tries to access a global variable, then the value it sees (if any) may not be the same as the value in the parent process at the time that Process.start was called.

However, global variables which are just module level constants cause no problems.