Skip to content
This repository was archived by the owner on Feb 2, 2024. It is now read-only.

WIP: interface for map-reduce style kernels #284

Open
Hardcode84 wants to merge 11 commits into
base: master
Choose a base branch
from
Open

WIP: interface for map-reduce style kernels #284

Changes from 1 commit
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
fd005ee
1
Oct 30, 2019
46af82a
2
Oct 30, 2019
0ccc2f9
some work
Nov 7, 2019
07b630f
fix
Nov 11, 2019
7aa49a5
parallel sort example
Nov 12, 2019
995955c
distribution depth
Nov 12, 2019
995ff31
remove commented code
Nov 12, 2019
34ca22d
reduce commented code
Nov 12, 2019
14ebb3d
style
Nov 12, 2019
a711ac3
fix
Nov 12, 2019
1c91eac
prange
Nov 12, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
some work
  • Loading branch information
Ivan Butygin committed Nov 11, 2019
commit 0ccc2f928b9877d5ddea8430c5c9b800b61a437a
45 changes: 26 additions & 19 deletions hpat/distributed.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,13 +120,13 @@ def __init__(self):
pass

def run_pass(self, state):
print('XXXXXXXXXXXXXXXXXXXXXXXXXXX')
state.func_ir.dump()
print('YYYYYYYYYYYYYYYYYYYYYYYYYYY')
# print('XXXXXXXXXXXXXXXXXXXXXXXXXXX')
# state.func_ir.dump()
# print('YYYYYYYYYYYYYYYYYYYYYYYYYYY')
res = DistributedPassImpl(state).run_pass()
print('XXXXXXXXXXXXXXXXXXXXXXXXXXX')
state.func_ir.dump()
print('YYYYYYYYYYYYYYYYYYYYYYYYYYY')
# print('XXXXXXXXXXXXXXXXXXXXXXXXXXX')
# state.func_ir.dump()
# print('YYYYYYYYYYYYYYYYYYYYYYYYYYY')
return res

class DistributedPassImpl(object):
Expand Down Expand Up @@ -219,7 +219,7 @@ def _run_dist_pass(self, blocks, depth):
self.state.targetctx, self)
elif isinstance(inst, Parfor):
print('********** process parfor **********')
out_nodes = self._run_parfor(inst, namevar_table, 0 == depth)
out_nodes = self._run_parfor(inst, namevar_table, depth)
# run dist pass recursively
p_blocks = wrap_parfor_blocks(inst)
# build_definitions(p_blocks, self.state.func_ir._definitions)
Expand Down Expand Up @@ -286,6 +286,9 @@ def _run_dist_pass(self, blocks, depth):

if not replaced:
blocks[label].body = new_body
# print('******************** new_body')
# print('\n'.join([str(a) for a in new_body]))
# print('********************')

return blocks

Expand Down Expand Up @@ -1700,12 +1703,16 @@ def f(A, start, step):

return out

def _run_parfor(self, parfor, namevar_table, distribute):
def _run_parfor(self, parfor, namevar_table, depth):
# stencil_accesses, neighborhood = get_stencil_accesses(
# parfor, self.state.typemap)

if not distribute:
parfor.no_sequential_lowering = True
dist_depth = 0

if depth > dist_depth:
# Do not distribute
if depth == (dist_depth + 1):
parfor.no_sequential_lowering = True
return [parfor]

# Thread and 1D parfors turn to gufunc in multithread mode
Expand All @@ -1718,7 +1725,7 @@ def _run_parfor(self, parfor, namevar_table, distribute):
return self._run_parfor_1D_Var(parfor, namevar_table)

if self._dist_analysis.parfor_dists[parfor.id] != Distribution.OneD:
if debug_prints(): # pragma: no cover
if True or debug_prints(): # pragma: no cover
print("parfor " + str(parfor.id) + " not parallelized.")
return [parfor]

Expand Down Expand Up @@ -1896,12 +1903,12 @@ def _gen_parfor_reductions(self, parfor, namevar_table):
_, reductions = get_parfor_reductions(
parfor, parfor.params, self.state.calltypes)

print('aaaaaaaaaaaaaaaaaaa')
parfor.dump()
print('aaaaaaaaaaaaaaaaaaa')
# print('aaaaaaaaaaaaaaaaaaa')
# parfor.dump()
# print('aaaaaaaaaaaaaaaaaaa')
for reduce_varname, (init_val, reduce_nodes) in reductions.items():
print(len(reduce_nodes))
print('\n'.join([str(a) for a in reduce_nodes]))
# print(len(reduce_nodes))
# print('\n'.join([str(a) for a in reduce_nodes]))
reduce_op = guard(self._get_reduce_op, reduce_nodes)
# TODO: initialize reduction vars (arrays)
reduce_var = namevar_table[reduce_varname]
Expand Down Expand Up @@ -2106,9 +2113,9 @@ def f(val, op): # pragma: no cover
replace_arg_nodes(block, [reduce_var, op_var])
dist_reduce_nodes = [op_assign] + block.body[:-3]
dist_reduce_nodes[-1].target = reduce_var
print('*****************************')
print('\n'.join([str(a) for a in dist_reduce_nodes]))
print('*****************************')
# print('*****************************')
# print('\n'.join([str(a) for a in dist_reduce_nodes]))
# print('*****************************')
return dist_reduce_nodes

def _get_reduce_op(self, reduce_nodes):
Expand Down