python-performance/10-dask/sec3-sche/arr.py at master · tiagoantao/python-performance · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
from time import time

import numpy as np
import dask.array as da
from dask.distributed import Client
from PIL import Image

client = Client('127.0.0.1:8786')
#client = Client()

start = -1.5, -1.3
end = 0.5, 1.3


def prepare_pos_array(size, start, end, pos_array):
    size = pos_array.shape[0]
    startx, starty = start
    endx, endy = end
    for xp in range(size):
        x = (endx - startx)*(xp/size) + startx
        for yp in range(size):
            y = (endy - starty)*(yp/size) + starty
            pos_array[yp, xp] = complex(x, y)


def block_prepare_pos_array(size, pos_array):
    nrows, ncols = pos_array.shape
    ret = np.empty(shape=(nrows,ncols), dtype=np.complex128)
    startx, starty = start
    endx, endy = end

    for row in range(nrows):
        x = (endx - startx) * ((pos_array[row, 0] // size ) / size) + startx
        for col in range(ncols):
            y = (endy - starty) * ((pos_array[row, col] % size) / size) + starty
            ret[row, col] = complex(x, y)
    return ret


def compute_point(c):
    i = -1
    z = complex(0, 0)
    max_iter = 200
    while abs(z) < 2:
        i += 1
        if i == max_iter:
            break
        z = z**2 + c
    return 255 - (255 * i) // max_iter


size = 3
pos_array = da.empty((size, size), dtype=np.complex128)
prepare_pos_array(3, start, end, pos_array)
pos_array.visualize("10-size3.png", rankdir="LR")
# pos_array.persist()

size = 1000
# pos_array = da.empty((size, size), dtype=np.int64)
range_array = da.arange(0, size*size).reshape(size, size)
range_array
range_array = range_array.rechunk(size // 2, size // 2)
range_array.visualize("10-rechunk.png", rankdir="TB")
range_array = range_array.persist()
range_array
#pos_array.chunks
#pos_array.npartitions
#pos_array.chunks
#pos_array[0].shape
#pos_array.blockwise()
# block_prepare_pos_array(3, np.array([0,1,2,3,4,5,6,7,8]).reshape(3,3))
pos_array = da.blockwise(
    lambda x: block_prepare_pos_array(size, x),
    'ij', range_array, 'ij', dtype=np.complex128)
pos_array.visualize("10-blockwise.png", rankdir="TB")
pos_array = pos_array.persist()  # scheduler="single-threaded")
pos_array.size
pos_array.nbytes

u_compute_point = da.frompyfunc(compute_point, 1, 1)

image_arr = u_compute_point(pos_array)
image_arr.visualize("10-image_arr.png", rankdir="TB")
image_np = image_arr.compute().astype(np.uint8)
image = Image.fromarray(image_np, mode="P")
image.save("mandelbrot.png")


def time_scenario(size, persist_range, persist_pos, chunk_div=10):
    start_time = time()
    size = size
    range_array = da.arange(0, size*size).reshape(size, size).persist()
    range_array = range_array.rechunk(size // chunk_div, size // chunk_div)
    range_array = range_array.persist() if persist_range else range_array
    pos_array = da.blockwise(
        lambda x: block_prepare_pos_array(size, x),
        'ij', range_array, 'ij', dtype=np.complex128)

    pos_array = pos_array.persist() if persist_pos else pos_array
    image_arr = u_compute_point(pos_array)
    # image_arr.visualize("xx.png", rankdir="TB")
    image_arr.compute()
    return time() - start_time


for size in [500, 1000, 5000, 10000]:  # for size in [10000]:
    print()
    for persist_range in [False, True]:
        for persist_pos in [False, True]:
            client.restart()
            print(size, persist_range, persist_pos,
                  time_scenario(size, persist_range, persist_pos))

5000/7000

size = 500
print(size, False, False, 2, time_scenario(size, False, False, 2))

size = 5000
client = Client('127.0.0.1:8786')
client = client.restart()
print(size, False, False, 10, time_scenario(size, True, True, 10))

size = 500
print(size, False, False, 2, time_scenario(size, False, False, 2))
print(size, False, False, time_scenario(size, False, False))
print(size, True, True, time_scenario(size, True, True))

size = 1000
for chunk_div in [1, 2, 4, 10, 100]:
    client.restart()
    print(size, False, False, chunk_div, time_scenario(size, False, False, chunk_div))
    client.restart()
    print(size, True, True, chunk_div, time_scenario(size, True, True, chunk_div))


size = 10000
print(size, False, False, time_scenario(size, False, False))
client.restart()
print(size, True, True, time_scenario(size, True, True))