如何在使用多處理時將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被調用時父進程中的值不同。

但是，只是模塊級別常量的全局變量不會引起任何問題。