Deap OneMax Example with JobLib

joblib_deap.py

import os
import time
from itertools import repeat
import random
from deap import creator, base, tools, algorithms

import numpy as np
from joblib import Parallel, delayed
from joblib import dump, load

folder = './joblib_memmap'
try:
    os.mkdir(folder)
except FileExistsError:
    pass

# just used to initialize memory mapped data
def init_memmap_data(original_data):
    data_filename_memmap = os.path.join(folder, 'data_memmap')
    dump(original_data, data_filename_memmap)
    shared_data = load(data_filename_memmap, mmap_mode='r')
    return shared_data

config = {
    'config_array': init_memmap_data(np.ones(500000))    
}

def joblib_map(f, *iters):
    return Parallel(n_jobs=-1, backend="multiprocessing")(delayed(f)(*args) for args in zip(*iters))


creator.create("FitnessMax", base.Fitness, weights=(1.0,))
creator.create("Individual", list, fitness=creator.FitnessMax)

toolbox = base.Toolbox()
toolbox.register("map", joblib_map)

toolbox.register("attr_bool", random.randint, 0, 1)
toolbox.register("individual", tools.initRepeat, creator.Individual, toolbox.attr_bool, n=100)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)

def evalOneMaxConf(individual, conf):
    # simulate long-running function
    time.sleep(0.1)
    return [sum(individual)]

toolbox.register("evaluate", evalOneMaxConf)
toolbox.register("mate", tools.cxTwoPoint)
toolbox.register("mutate", tools.mutFlipBit, indpb=0.05)
toolbox.register("select", tools.selTournament, tournsize=3)


population = toolbox.population(n=100)

NGEN=10
for gen in range(NGEN):
    offspring = algorithms.varAnd(population, toolbox, cxpb=0.5, mutpb=0.1)
    fits = toolbox.map(toolbox.evaluate, offspring, repeat(config))
    
    for fit, ind in zip(fits, offspring):
        ind.fitness.values = fit
    population = toolbox.select(offspring, k=len(population))
    
top10 = tools.selBest(population, k=10)