如何在使用多处理时将env.host转换为结构？

Question

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

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

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

[] 
None 
None 
***Warning*** Host None via ssh is down 

代码：

@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。那么execute d任务将采用您在运行时设置的env值。但不幸的是，由于fabric不是完全线程安全的，因此env是一个全局单例，因此只能在每个网关一次启动类似的任务。

织物的简单但积分方面是所谓的 “环境”：Python字典子类，其被用作 组合设置注册表和共享任务间的数据名称空间。

环境字典当前作为全局单例实现， fabric.state.env，为方便起见，它包含在fabric.api中。 env中的键 有时被称为“env变量”。

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有几个怪癖。该信息可能对你的情况尤为重要：

全局变量

记住，如果一个子进程运行的代码试图访问一个全局变量，则该值它看到（如果有的话）与Process.start被调用时父进程中的值不同。

但是，只是模块级别常量的全局变量不会引起任何问题。