0

こんにちは、このワーカー ジェネレーター クラスを作成しています。終了時にハングします。それらを閉じるには a を使用する必要があることはわかっていますが、プロセス名を関数.joinに渡す方法がわかりません。terminate私の考えは、どういうわけかそれをグローバル変数に保存することでした。私は辞書を考えていました。次に、ワーカterminateを終了させるときに、メッセージ キューにポイズン ピルをドロップした後、その関数変数にアクセスし、該当するプロセスを終了します。

class worker_manager:
      i = test_imports()
      #someVarForP 
      #someVarForP2 

      def generate(control_queue, threadName, runNum):
          if threadName == 'one':
              print ("Starting import_1 number %d") % runNum
              p = multiprocessing.Process(target=i.import_1, args=(control_queue, runNum))
              #someVarForP = p
              p.start()        
          if threadName == 'two': 
              print ("Starting import_2 number %d") % runNum
              p = multiprocessing.Process(target=i.import_2, args=(control_queue, runNum))
              #someVarForP2 = p2
              p.start()
          if threadName == 'three':    
              p = multiprocessing.Process(target=i.import_1, args=(control_queue, runNum))
              print ("Starting import_1 number %d") % runNum
              p2 = multiprocessing.Process(target=i.import_2, args=(control_queue, runNum))
              print ("Starting import_2 number %d") % runNum
              #someVarForP = p
              #someVarForP2 = p2
              p.start()
              p2.start()

      def terminate(threadName):
           if threadName == 'one':
               #self.someVarForP.join()
           if threadName == 'two':
               #self.someFarForP2.join()
           if threadName == 'three':
               #self.someVarForP.join()
               #self.someVarForP2.join()

皆さんとギャルのアイデアをいただければ幸いです。私はこれを行うための特定の方法に固執していません。私はPythonに慣れていないので、どんな提案も大歓迎です!

編集：完全なコード

import multiprocessing 
import time 

class test_imports:#Test classes remove 
      def import_1(self, control_queue, thread_number):
          print ("Import_1 number %d started") % thread_number
          run = True
          count = 1
          while run:
                alive = control_queue.get()
                if alive == 't1kill':
                   print ("Killing thread type 1 number %d") % thread_number
                   run = False
                   break
                print ("Thread type 1 number %d run count %d") % (thread_number, count)
                count = count + 1

      def import_2(self, control_queue, thread_number):
          print ("Import_1 number %d started") % thread_number
          run = True
          count = 1
          while run:
                alive = control_queue.get()
                if alive == 't2kill':
                   print ("Killing thread type 2 number %d") % thread_number
                   run = False
                break
                print ("Thread type 2 number %d run count %d") % (thread_number, count)           
                count = count + 1

class worker_manager:
    # ...
    names = {'one': 'import_1', 'two': 'import_2'}
    def __init__(self):
        self.children = {}
    def generate(self, control_queue, threadName, runNum):
        name = self.names[threadName]
        target = i.getattr(name)
        print ("Starting %s number %d") % (name, runNum)
        p = multiprocessing.Process(target=target, args=(control_queue, runNum))
        self.children[threadName] = p
        p.start()
    def terminate(self, threadName):
        self.children[threadName].join()

if __name__ == '__main__':
    # Establish communication queues
    control = multiprocessing.Queue()
    manager = worker_manager()    
    runNum = int(raw_input("Enter a number: ")) 
    threadNum = int(raw_input("Enter number of threads: "))
    threadName = raw_input("Enter number: ")
    thread_Count = 0

    print ("Starting threads") 

    for i in range(threadNum):
        if threadName == 'three':
            manager.generate(control, 'one', i)
            manager.generate(control, 'two', i)
        manager.generate(control, threadName, i)
        thread_Count = thread_Count + 1              
        if threadName == 'three':
            thread_Count = thread_Count + 1 

    time.sleep(runNum)#let threads do their thing

    print ("Terminating threads")     

    for i in range(thread_Count):
        control.put("t1kill")
        control.put("t2kill")
    if threadName == 'three':
        manager.terminate('one')
        manager.terminate('two')
    else:
        manager.terminate(threadName)

編集のみworker_managerが、実際に使用される唯一のビットです。これの残りの部分は、それを開発するだけです。

4

1 に答える 1

1

joinまず、という関数から使用している場合、何をするのか理解できませんterminate。を呼び出すとjoin、子プロセスが終了するまで親プロセスを待機させるだけです。終了する必要があることを子プロセスに伝える必要があります (たとえば、キュ​​ーに何かを投稿するか、シグナルを送信することにより)。そうしないと、永遠に待機することになります。

次に、selfメソッドからパラメーターを除外し、古いスタイルのクラスを使用し、おそらくインスタンス属性が必要な場所にクラス属性を作成し、インスタンスに完全に適していると思われるもののグローバル変数について話し始めたという事実。属性は、クラスについても誤解がある可能性があることを意味します。

しかし、大きな問題は、文字列名をインスタンス属性 (または他の種類の変数) にマップする必要があるようです。これを行うことはできますが、ほとんどしたくありません。dict名前ごとに個別の属性を使用する代わりに、名前ごとに値を持つ単一の を使用します。例えば：

class worker_manager:
    # ...
    def __init__(self):
        self.children = {}
    def generate(self, control_queue, threadName, runNum):
        if threadName == 'one':
            print ("Starting import_1 number %d") % runNum
            p = multiprocessing.Process(target=i.import_1, args=(control_queue, runNum))
            self.children[threadName]
            p.start()
        # ...
    def terminate(self, threadName):
        self.children[threadName].join()

ifブロックの共通部分を除外することで、これをさらにクリーンアップできます。たとえば、ディスパッチ テーブルが必要な場合があります。そして、さらに標準的な名前、ランダムではなく 4 文字のインデント、新しいスタイルのクラスなどを使用しましょう。

class WorkerManager(object):
    # ...
    names = {'one': 'import_1', 'two': 'import_2', 'three': 'import_3'}
    def __init__(self):
        self.children = {}
    def generate(self, control_queue, threadName, runNum):
        name = WorkerManager.names[threadName]
        target = i.getattr(name)
        print ("Starting %s number %d") % (name, runNum)
        p = multiprocessing.Process(target=target, args=(control_queue, runNum))
        self.children[threadName] = p
        p.start()
    def terminate(self, threadName):
        self.children[threadName].join()

あなたのコードをもっと見ることができれば、おそらくリファクタリングできるものよりもはるかに多くのものがあるでしょうが、始めるにはこれで十分でしょう。

于 2013-07-31T18:30:40.890 に答える