Is it possible to add a timeout to a fitness evaluation function in DEAP?

Question

I'm trying to use DEAP to evolve regex strings. It worked for small inputs but when I try larger inputs, then it starts to evolve regex strings that exhibit catastrophic backtracking. This stalls the evolution process as the regex search never completes.

I've tried using multiprocessing, pebble and timeout_decorator to add a timeout to the regex matching, but I hit issues with the way threads are handled in Python.

Maybe I'm missing something obvious, but from what I understand, DEAP wants to define the toolbox in the main script, which includes the fitness evaluation function. Limited Python threading knowledge here but I think this causes issues because the threaded functions need to be guarded by a __main__ check.

So the following works fine:

@concurrent.process
def function(a, b):
    return a + b


if __name__ == '__main__':
    future = function(1, 1)

    print(future.result())

But this fails because the future can't be declared outside of a __main__ guard.

@concurrent.process
def function(a, b):
    return a + b

future = function(1, 1)

if __name__ == '__main__':
    print(future.result())

Giving this error:

RuntimeError:
    An attempt has been made to start a new process before the
    current process has finished its bootstrapping phase.

    This probably means that you are not using fork to start your
    child processes and you have forgotten to use the proper idiom
    in the main module:

        if __name__ == '__main__':
            freeze_support()
            ...

    The "freeze_support()" line can be omitted if the program
    is not going to be frozen to produce an executable.

Shortened version of my code showing why I have this issue (the toolbox needs to be globally accessible to the evalSymbReg that is registered with DEAP as the "evaluate" function. I can't see a way of having everything behind the __main__ check while also having the toolbox available to the evalSymbReg function that needs to use it.

# ...snip...
toolbox = base.Toolbox()
# ...snip...
@concurrent.process(timeout=1)
def get_match(regex, input_str):
    result = re.match(regex, input_str)
    return result[0] if result else ""


def evalSymbReg(individual):
    # Transform the tree expression in a callable function
    func = toolbox.compile(expr=individual)
    # Evaluate the mean squared error between the expression
    # and the real function
    try:
        result = get_match(func, args.goal)
        sqerrors += editdistance.eval(goal, result)**2
    except TimeoutError:
        sqerrors += editdistance.eval(goal, "")**2
    return sqerrors,


toolbox.register("evaluate", evalSymbReg)

# ...snip...

def main():
    # ...snip...
    pop, log = algorithms.eaSimple(pop, toolbox, 0.5, 0.1, G, stats=mstats,
                                halloffame=hof, verbose=True)
    # ...snip...

if __name__ == "__main__":
    main()