subinterpreter_test_mandelbrot.py

import math
import time
from threading import Thread
import multiprocessing as mp
import extrainterpreters as ei
from multiprocessing.dummy import Pool as ThreadPool


def mandelbrot(c, max_iterations=100):
    z = 0
    for i in range(max_iterations):
        z = z ** 2 + c
        if z.real ** 2 + z.imag ** 2 > 4:
            return i + 1
        if i == max_iterations - 1:
            return 0


def mandelbrot_serial(xmin, xmax, ymin, ymax, N=100, iter=100):
    incx = math.fabs((xmax - xmin) / N)
    incy = math.fabs((ymax - ymin) / N)
    x, y, myList, n, arr = xmin, ymax, [], 1, []
    while y > ymin and n <= N:
        while x < xmax:
            i = mandelbrot(complex(x, y), iter)
            myList.append(i)
            x += incx
        arr.append(myList[:N])
        x = xmin
        y -= incy
        myList = []
        n += 1
    return arr


def call_mandelbrot_serial(xmin, xmax, ymin, ymax, N, iter, results, i):
    results.put({i: mandelbrot_serial(xmin, xmax, ymin, ymax, N, iter)})


def call_mandelbrot_serial2(xmin, xmax, ymin, ymax, N, iter, results, i):
    results[i] = mandelbrot_serial(xmin, xmax, ymin, ymax, N, iter)


def mandelbrot_static(xmin, xmax, ymin, ymax, N=100, iter=100):
    results = mp.Queue()
    processes = []
    count = mp.cpu_count() // 2
    arr_upper = []
    arr_lower = []
    incx = math.fabs((xmax - xmin) / count)

    x, i = xmin, 0
    while i < mp.cpu_count():
        i += 1
        if i == count + 1:
            x = xmin  # refresh x to xmin

        if i > count:
            p = mp.Process(target=call_mandelbrot_serial, args=(
                x, x + incx, ymin, 0, N // count, iter, results, i))
            p.start()
            processes.append(p)
            x += incx
        else:
            p = mp.Process(target=call_mandelbrot_serial, args=(
                x, x + incx, 0, ymax, N // count, iter, results, i))
            p.start()
            processes.append(p)
            x += incx

    result_set = {}
    for _ in processes:
        result_set.update(results.get())

    for p in processes:
        p.join()

    i = 1

    for _ in processes:
        if i <= count:
            if i == 1:
                arr_upper += result_set[i]
            else:
                for j in range(len(result_set[i])):
                    arr_upper[j] += result_set[i][j]
        else:
            if i == count + 1:
                arr_lower += result_set[i]
            else:
                for j in range(len(result_set[i])):
                    arr_lower[j] += result_set[i][j]
        i += 1

    return arr_upper + arr_lower


def mandelbrot_dynamic(xmin, xmax, ymin, ymax, N=100, iter=100):
    incx = math.fabs((xmax - xmin) / N)
    incy = math.fabs((ymax - ymin) / N)
    x, y, myList, lst, n = xmin, ymax, [], [], 1
    while y > ymin and n <= N:
        while x < xmax:
            lst.append(complex(x, y))
            x += incx
        x = xmin
        y -= incy
        myList.extend(lst[:N])
        n += 1
        lst = []

    with mp.Pool() as pool:
        result = pool.map(mandelbrot, (myList, iter))

    result = [result[i:i + N] for i in range(0, len(result), N)]

    return result


def mandelbrot_static_threads(xmin, xmax, ymin, ymax, N=100, iter=100):
    results = [[] for _ in range(mp.cpu_count())]
    threads = []
    count = mp.cpu_count() // 2
    arr_upper = []
    arr_lower = []
    incx = math.fabs((xmax - xmin) / count)

    x, i = xmin, 0
    while i < mp.cpu_count():
        i += 1
        if i == count + 1:
            x = xmin  # refresh x to xmin

        if i > count:
            th = Thread(target=call_mandelbrot_serial2, args=(
                x, x + incx, ymin, 0, N // count, iter, results, i - 1))
            th.start()
            threads.append(th)
            x += incx
        else:
            th = Thread(target=call_mandelbrot_serial2, args=(
                x, x + incx, 0, ymax, N // count, iter, results, i - 1))
            th.start()
            threads.append(th)
            x += incx

    for thread in threads:
        thread.join()

    i = 1

    for _ in threads:
        if i <= count:
            if i == 1:
                arr_upper += results[i - 1]
            else:
                for j in range(len(results[i - 1])):
                    arr_upper[j] += results[i - 1][j]
        else:
            if i == count + 1:
                arr_lower += results[i - 1]
            else:
                for j in range(len(results[i - 1])):
                    arr_lower[j] += results[i - 1][j]
        i += 1

    return arr_upper + arr_lower


def mandelbrot_dynamic_threads(xmin, xmax, ymin, ymax, N=100, iter=100):
    incx = math.fabs((xmax - xmin) / N)
    incy = math.fabs((ymax - ymin) / N)
    x, y, myList, lst, n = xmin, ymax, [], [], 1
    while y > ymin and n <= N:
        while x < xmax:
            lst.append(complex(x, y))
            x += incx
        x = xmin
        y -= incy
        myList.extend(lst[:N])
        n += 1
        lst = []

    with ThreadPool() as pool:
        result = pool.map(mandelbrot, (myList, iter))

    result = [result[i:i + N] for i in range(0, len(result), N)]

    return result


def mandelbrot_extrainterpreter(xmin, xmax, ymin, ymax, N=100, iter=100):
    interpreters = []
    count = mp.cpu_count() // 2
    arr_upper = []
    arr_lower = []
    incx = math.fabs((xmax - xmin) / count)

    x, i = xmin, 0
    while i < mp.cpu_count():
        i += 1
        if i == count + 1:
            x = xmin  # refresh x to xmin

        if i > count:
            interp = ei.Interpreter(target=mandelbrot_serial, args=(
                x, x + incx, ymin, 0, N // count, iter))
            interp.start()
            interpreters.append(interp)
            x += incx
        else:
            interp = ei.Interpreter(target=mandelbrot_serial, args=(
                x, x + incx, 0, ymax, N // count, iter))
            interp.start()
            interpreters.append(interp)
            x += incx

    for interp in interpreters:
        interp.join()

    i = 1
    result_set = {}
    for p in interpreters:
        while not p.done():
            pass
        result_set[i] = p.result()
        i += 1

    i = 1

    for _ in interpreters:
        if i <= count:
            if i == 1:
                arr_upper += result_set[i]
            else:
                for j in range(len(result_set[i])):
                    arr_upper[j] += result_set[i][j]
        else:
            if i == count + 1:
                arr_lower += result_set[i]
            else:
                for j in range(len(result_set[i])):
                    arr_lower[j] += result_set[i][j]
        i += 1

    return arr_upper + arr_lower


def run_test(func, step, *args):
    print(f'\nmandelbrot {step}, Resolution {args[-2]}x{args[-2]}, Iterations {args[-1]}:')
    print('Time:', end=' ')
    start = time.time()
    arr = func(*args)
    end = time.time()
    print(f'{end - start} secs')
    return arr


def show_mandelbrot(arr, xmin, xmax, ymin, ymax):
    import matplotlib.pyplot as plt
    plt.imshow(arr, extent=(xmin, xmax, ymin, ymax))
    plt.savefig('MandelBrot.png')


def main():
    xmin = -2.25
    xmax = 0.75
    ymin = -1.5
    ymax = 1.5

    test = {
        'serial': mandelbrot_serial,
        'parallel static': mandelbrot_static,
        'parallel dynamic': mandelbrot_dynamic,
        'static threads': mandelbrot_static_threads,
        'dynamic threads': mandelbrot_dynamic_threads,
        'subinterpreter': mandelbrot_extrainterpreter
    }

    for step, func in test.items():
        arr = run_test(func, step, xmin, xmax, ymin, ymax, 600, 100)

    show_mandelbrot(arr, xmin, xmax, ymin, ymax)


if __name__ == '__main__':
    main()