21331


pythonのプログラムを並列化しようと考え、pythonのモジュールであるmultiprocessingを実装した所以下のエラーが発生致しました。  
使用しているモジュールは主に遺伝的アルゴリズム用のDEAPというモジュールです。  
[DEAPのリファレンス](http://deap.readthedocs.org/en/master/index.html)  
コメントであるように`pool =
multiprocessing.Pool(3)`をmainの下に置いた所プロセスは5つ生まれマルチプロセスにはなれたのですが、タスクマネージャーで見たところそのプロセスのうち1つのみ稼働している見た目だけマルチプロセスになってしまいました。  
<http://gazo.shitao.info/r/i/20160125185509_000.png>  
1つだけ25%でそれ以外0％であり、尚且つプロセスを3つ生み出す命令しているのにプロセスが4つ生み出されているため、プロセスを生み出す処理と動作する処理を別に行っていると考えられるのですがその理由がわかりません。どのようにするべきか教えてくれれば助かります。

[code]

    import random
    import numpy
    import matplotlib.pyplot
    import time
    import multiprocessing
    from multiprocessing import Lock
    #from scoop import futures
    
    from deap import algorithms
    from deap import base
    from deap import creator
    from deap import tools
    
    toolbox = base.Toolbox()
    
    def main():
    #     random.seed(64)
        NGEN = 5
        MU = 75
        LAMBDA = 75
        CXPB = 0.6
        MUTPB = 0.3
    
        pop = toolbox.population(n=MU)
        hof = tools.ParetoFront()
        stats = tools.Statistics(lambda ind: ind.fitness.values)
        stats.register("avg", numpy.mean, axis=0)
        stats.register("std", numpy.std, axis=0)
        stats.register("min", numpy.min, axis=0)
        stats.register("max", numpy.max, axis=0)
    
        algorithms.eaMuPlusLambda(pop, toolbox, MU, LAMBDA, CXPB, MUTPB, NGEN, stats,
                                  halloffame=hof)
        return pop, stats, hof
    if __name__ == '__main__':
        pool = multiprocessing.Pool(processes=3)
        toolbox.register('map', pool.map_async)
        for var in range(0,5):
    　　　　　・・・・・
    
[/code]

**追記**  
並列化に関してですが、このページを参考にしております。[DEAPの並列化](http://deap.readthedocs.org/en/master/tutorials/basic/part4.html?highlight=multiprocessing)  
ここにあるようにMultiprocessing
Moduleを用いて並列化しており、中身としては以下のアルゴリズム中にあるmap関数で`multiprocessing.Pool.map_async`を呼び出すことで並列化計算していると考えています。  
どこを並列化したいのかという点ですが、コメントにあるようにアルゴリズム中でmap関数が使われている点が個体の評価という点ですのでそこを並列化したいと考えています。  
並列化していないバージョンですが、`if __name__ == '__main__':`以下の

[code]

    pool = multiprocessing.Pool(processes=3)
    toolbox.register('map', pool.map_async)
    
[/code]

をコメントアウトするなどで動作させない場合プロセスが生まれず通常通りに動きました。  
また、[参考にしたコード](http://pastebin.com/0WazVPKy)を実装した場合しっかりマルチプロセスで動いておりPCの問題ではないと思いますが、自分のコードではどこが原因となって見た目マルチプロセスになっているのかわかりません。  
よろしくお願いします。  
**追記2**  
自分のコードですがこのような形になります。  
[全コード](http://pastebin.com/4wFneYgX)  
[参考にしたコード]同様にDEAPモジュールのalgorithmsを以下のように変更しております。  
**変更点**  
def eaMuPlusLambdaの

[code]

    fitnesses = toolbox.map(toolbox.evaluate, invalid_ind)
    
[/code]

を

[code]

    jobs = toolbox.map(toolbox.evaluate, invalid_ind)
    fitnesses=jobs.get()
    
[/code]

としています。全文を以下に載せます。

**algorithms**

[code]

    def eaMuPlusLambda(population, toolbox, mu, lambda_, cxpb, mutpb, ngen,
                       stats=None, halloffame=None, verbose=__debug__):
        logbook = tools.Logbook()
        logbook.header = ['gen', 'nevals'] + (stats.fields if stats else [])
    
        # Evaluate the individuals with an invalid fitness
        invalid_ind = [ind for ind in population if not ind.fitness.valid]
        jobs= toolbox.map(toolbox.evaluate, invalid_ind)
        fitnesses=jobs.get()
        for ind, fit in zip(invalid_ind, fitnesses):
            ind.fitness.values = fit
    
        if halloffame is not None:
            halloffame.update(population)
    
        record = stats.compile(population) if stats is not None else {}
        logbook.record(gen=0, nevals=len(invalid_ind), **record)
        if verbose:
            print logbook.stream
    
        # Begin the generational process
        for gen in range(1, ngen+1):
            # Vary the population
            offspring = varOr(population, toolbox, lambda_, cxpb, mutpb)
    
            # Evaluate the individuals with an invalid fitness
            invalid_ind = [ind for ind in offspring if not ind.fitness.valid]
            jobs = toolbox.map(toolbox.evaluate, invalid_ind)
            fitnesses=jobs.get()
            for ind, fit in zip(invalid_ind, fitnesses):
                ind.fitness.values = fit
    
            # Update the hall of fame with the generated individuals
            if halloffame is not None:
                halloffame.update(offspring)
    
            # Select the next generation population
            population[:] = toolbox.select(population + offspring, mu)
    
            # Update the statistics with the new population
            record = stats.compile(population) if stats is not None else {}
            logbook.record(gen=gen, nevals=len(invalid_ind), **record)
            if verbose:
                print logbook.stream
    
        return population, logbook
    
[/code]

**追記3**  
簡単なコード

[code]

    # coding: UTF-8
    '''
    
    '''
    import multiprocessing
    
    def nibai(x): 
        print multiprocessing.current_process()
        return x*x
    
    def aaa():
        p = Pool(3) 
        print p.map(nibai, range(100)) 
    
    
    if __name__ == "__main__":
        aaa()
    
[/code]


CC BY-SA 3.0
2016-01-18T11:21:32.897
21034
2020-05-27T00:33:51.660
2016-01-26T10:49:31.883
13593
13593
question
python
pythonの並列化について


multiprocess版のeaMuPlusLambdaをコピペした他はimport周りを数行修正しただけですので説明は不要かと思います。ちなみに僕の環境ではmultiprocess化しないほうが速かったです。プロセス生成とプロセス間通信は一般にオーバーヘッドが大きいです。それで皆スレッドを使うわけですね。もっともPythonのスレッドはGIL(グローバル・インタープリタ・ロック）のせいでスピードが出ませんが。

[code]

    diff -u original.py  my.py
    --- original.py 2016-01-26 01:00:54.000000000 -0800
    +++ my.py   2016-01-26 01:02:21.000000000 -0800
    @@ -5,7 +5,7 @@
     '''
     import random
     import numpy
    -import matplotlib.pyplot
    +#import matplotlib.pyplot
     import time
     import multiprocessing
     # from multiprocessing import Lock
    @@ -42,6 +42,7 @@
     toolbox.register("population", tools.initRepeat, list, toolbox.individual)
    
     def evalKnapsack(individual):
    +    print multiprocessing.current_process()
         energy = 0.0
         power = 0.0
         for item in individual:
    @@ -49,7 +50,7 @@
             power += items[item][0]
         if power < MIN_POWER or energy < MIN_ENERGY:
             return 100000000000,1000000000000
    -    print multiprocessing.current_process()
    +
         return energy, power
    
     def cxSet(ind1, ind2):
    @@ -77,6 +78,54 @@
     toolbox.register("mutate", mutSet)
     toolbox.register("select", tools.selSPEA2)
    
    +def eaMuPlusLambda(population, toolbox, mu, lambda_, cxpb, mutpb, ngen,
    +                   stats=None, halloffame=None, verbose=__debug__):
    +    logbook = tools.Logbook()
    +    logbook.header = ['gen', 'nevals'] + (stats.fields if stats else [])
    +
    +    # Evaluate the individuals with an invalid fitness
    +    invalid_ind = [ind for ind in population if not ind.fitness.valid]
    +    jobs= toolbox.map(toolbox.evaluate, invalid_ind)
    +    print jobs
    +    fitnesses=jobs.get()
    +    for ind, fit in zip(invalid_ind, fitnesses):
    +        ind.fitness.values = fit
    +
    +    if halloffame is not None:
    +        halloffame.update(population)
    +
    +    record = stats.compile(population) if stats is not None else {}
    +    logbook.record(gen=0, nevals=len(invalid_ind), **record)
    +    if verbose:
    +        print logbook.stream
    +
    +    # Begin the generational process
    +    for gen in range(1, ngen+1):
    +        # Vary the population
    +        offspring = algorithms.varOr(population, toolbox, lambda_, cxpb, mutpb)
    +
    +        # Evaluate the individuals with an invalid fitness
    +        invalid_ind = [ind for ind in offspring if not ind.fitness.valid]
    +        jobs = toolbox.map(toolbox.evaluate, invalid_ind)
    +        fitnesses=jobs.get()
    +        for ind, fit in zip(invalid_ind, fitnesses):
    +            ind.fitness.values = fit
    +
    +        # Update the hall of fame with the generated individuals
    +        if halloffame is not None:
    +            halloffame.update(offspring)
    +
    +        # Select the next generation population
    +        population[:] = toolbox.select(population + offspring, mu)
    +
    +        # Update the statistics with the new population
    +        record = stats.compile(population) if stats is not None else {}
    +        logbook.record(gen=gen, nevals=len(invalid_ind), **record)
    +        if verbose:
    +            print logbook.stream
    +
    +    return population, logbook
    +
     def main():
         NGEN =10
         MU = 30
    @@ -95,7 +144,7 @@
         stats.register("min", numpy.min, axis=0)
         stats.register("max", numpy.max, axis=0)
    
    -    algorithms.eaMuPlusLambda(pop, toolbox, MU, LAMBDA, CXPB, MUTPB, NGEN, stats,
    +    eaMuPlusLambda(pop, toolbox, MU, LAMBDA, CXPB, MUTPB, NGEN, stats,
                                   halloffame=hof)
    
         return pop, stats, hof
    
[/code]

**出力(抜粋)**

10世代目で27個体の評価が行われた部分の抜粋です。27個`<Process(PoolWorker-X, started
daemon)>`とあるのでワーカーが並列に実行していることがわかります。IDの値が後の方になるにつれて入り乱れているのも並列化されている証拠。次の行にある`10
27...`とはGA/EAの10世代目で27個の個体が新たに評価されたことを示しています。先の27と一致していますね。

[code]

    <Process(PoolWorker-1, started daemon)>
    <Process(PoolWorker-1, started daemon)>
    <Process(PoolWorker-1, started daemon)>
    <Process(PoolWorker-3, started daemon)>
    <Process(PoolWorker-3, started daemon)>
    <Process(PoolWorker-3, started daemon)>
    <Process(PoolWorker-2, started daemon)>
    <Process(PoolWorker-2, started daemon)>
    <Process(PoolWorker-2, started daemon)>
    <Process(PoolWorker-1, started daemon)>
    <Process(PoolWorker-1, started daemon)>
    <Process(PoolWorker-1, started daemon)>
    <Process(PoolWorker-3, started daemon)>
    <Process(PoolWorker-3, started daemon)>
    <Process(PoolWorker-3, started daemon)>
    <Process(PoolWorker-2, started daemon)>
    <Process(PoolWorker-2, started daemon)>
    <Process(PoolWorker-1, started daemon)>
    <Process(PoolWorker-2, started daemon)>
    <Process(PoolWorker-1, started daemon)>
    <Process(PoolWorker-1, started daemon)>
    <Process(PoolWorker-3, started daemon)>
    <Process(PoolWorker-3, started daemon)>
    <Process(PoolWorker-2, started daemon)>
    <Process(PoolWorker-2, started daemon)>
    <Process(PoolWorker-3, started daemon)>
    <Process(PoolWorker-2, started daemon)>
    10  27      [  3418.33333333  34263.33333333]       [  559.66110777  3609.56907246]     [  2540.  30400.]       [  4040.  41500.]
    Individual([513, 8439, 9095, 3082, 8203, 2830, 5391, 9876, 1430, 281, 9370, 7962, 7072, 5154, 9635, 8357, 5883, 2473, 4778, 2987, 4785, 5619, 4405, 1718, 9655, 8896, 6593, 9666, 5899, 4555, 8398, 2147, 468, 8014, 9815, 6104, 3988, 347, 9692, 9313, 355, 2020, 5483, 2369, 1523, 1140, 4905, 148, 6523, 4348, 1662, 4223]) (2540.0, 30400.0)
    []
    [28]
    [24]
    []
    [deap.creator.Fitness((2540.0, 30400.0))]
    elapsed_time:5.64607787132[sec]
    Timer unit: 1e-06 s
    
    Total time: 5.64599 s
    File: my.py
    Function: x at line 152
    
[/code]

**なぜCPUの利用率が上がらないのか**

並列化されている関数は次の極単純な関数です。計算コストは小さそうです。この関数を実行する時間よりも並列化のオーバーヘッド（計算の配信、結果の取得に伴うプロセス間通信など）が高ければCPUの利用率は上がりません。

[code]

    def evalKnapsack(individual):
        print multiprocessing.current_process()
        energy = 0.0
        power = 0.0
        for item in individual:
            energy += items[item][1]
            power += items[item][0]
        if power < MIN_POWER or energy < MIN_ENERGY:
            return 100000000000,1000000000000
    
        return energy, power
    
[/code]

**CPU利用率を見る実験**

上の関数に時間のかかる処理を入れて、CPUの利用率をみます。オーバーヘッドよりも計算コストが高ければ利用率が上がるはず。ここでは`time.sleep(5)`などとやらず、forループで無駄にCPUを消費するのがポイント。

[code]

    def evalKnapsack(individual):
        print multiprocessing.current_process()
        print "busy loop"
        n = 0
        for i in range(100000):
            for j in range(100000):
                n += 1.1
        energy = 0.0
        power = 0.0
        for item in individual:
            energy += items[item][1]
            power += items[item][0]
        if power < MIN_POWER or energy < MIN_ENERGY:
            return 100000000000,1000000000000
    
        return energy, power
    
[/code]

**実験結果**

Windows環境が今手元にないので、Macでtopコマンドを実行したところ、Pythonのプロセスが３つあり100%近く利用中になっていることがわかります。

[![並列化の様子](https://i.stack.imgur.com/un1f0.png)](https://i.stack.imgur.com/un1f0.png)

**アドバイス**

もっと基礎的なプログラミングの練習をしてから、こういう大きな問題に取り組むことをお勧めしますよ。それに使い方の分からないモジュールはまず小さな問題で使って理解しましょう。例えばmultiprocessなら`def
nibai(x): return
x+x`みたいな簡単な関数で挙動を確かめるとか。いきなり組み合わせて結果を出そうとすると、個別に理解してから組み合わせるよりずっと時間がかかります。理解していないものを組み合わせても、どちらに動かない原因があるのかさえ分からないからね。


CC BY-SA 4.0
2016-01-26T09:06:16.893
21331
2020-05-27T00:33:51.660
2020-05-27T00:33:51.660
40341
7837
21034
answer