Newer
Older
Andreas Klöckner
committed
"a[i] = 5+i+j",
],
[
lp.GlobalArg("a", np.float32),
lp.ValueArg("n", np.int32, approximately=1000),
],
assumptions="n>=1")
Andreas Klöckner
committed
knl = lp.tag_inames(knl, dict(j="ilp"))
from loopy.check import WriteRaceConditionError
import pytest
with pytest.raises(WriteRaceConditionError):
list(lp.generate_loop_schedules(knl))
def test_ilp_write_race_avoidance_local(ctx_factory):
ctx = ctx_factory()
knl = lp.make_kernel(ctx.devices[0],
Andreas Klöckner
committed
"<> a[i] = 5+i+j",
Andreas Klöckner
committed
knl = lp.tag_inames(knl, dict(i="l.0", j="ilp"))
for k in lp.generate_loop_schedules(knl):
assert k.temporary_variables["a"].shape == (16,17)
def test_ilp_write_race_avoidance_private(ctx_factory):
ctx = ctx_factory()
knl = lp.make_kernel(ctx.devices[0],
"{[j]: 0<=j<16 }",
[
Andreas Klöckner
committed
"<> a = 5+j",
Andreas Klöckner
committed
knl = lp.tag_inames(knl, dict(j="ilp"))
for k in lp.generate_loop_schedules(knl):
assert k.temporary_variables["a"].shape == (16,)
# }}}
Andreas Klöckner
committed
def test_write_parameter(ctx_factory):
dtype = np.float32
ctx = ctx_factory()
knl = lp.make_kernel(ctx.devices[0], [
"{[i,j]: 0<=i,j<n }",
],
"""
a = sum((i,j), i*j)
b = sum(i, sum(j, i*j))
n = 15
""",
[
lp.GlobalArg("a", dtype, shape=()),
lp.GlobalArg("b", dtype, shape=()),
lp.ValueArg("n", np.int32, approximately=1000),
],
assumptions="n>=1")
Andreas Klöckner
committed
import pytest
with pytest.raises(RuntimeError):
Andreas Klöckner
committed
lp.CompiledKernel(ctx, knl).get_code()
Andreas Klöckner
committed
def test_arg_shape_guessing(ctx_factory):
ctx = ctx_factory()
knl = lp.make_kernel(ctx.devices[0], [
"{[i,j]: 0<=i,j<n }",
],
"""
a = 1.5 + sum((i,j), i*j)
b[i, j] = i*j
c[i+j, j] = b[j,i]
""",
[
lp.GlobalArg("a", shape=lp.auto),
lp.GlobalArg("b", shape=lp.auto),
lp.GlobalArg("c", shape=lp.auto),
Andreas Klöckner
committed
lp.ValueArg("n"),
],
assumptions="n>=1")
print knl
print lp.CompiledKernel(ctx, knl).get_highlighted_code()
def test_arg_guessing(ctx_factory):
ctx = ctx_factory()
knl = lp.make_kernel(ctx.devices[0], [
"{[i,j]: 0<=i,j<n }",
],
"""
a = 1.5 + sum((i,j), i*j)
b[i, j] = i*j
c[i+j, j] = b[j,i]
""",
assumptions="n>=1")
print knl
print lp.CompiledKernel(ctx, knl).get_highlighted_code()
def test_arg_guessing_with_reduction(ctx_factory):
#logging.basicConfig(level=logging.DEBUG)
ctx = ctx_factory()
knl = lp.make_kernel(ctx.devices[0], [
"{[i,j]: 0<=i,j<n }",
],
"""
a = 1.5 + sum((i,j), i*j)
d = 1.5 + sum((i,j), b[i,j])
b[i, j] = i*j
c[i+j, j] = b[j,i]
""",
assumptions="n>=1")
print knl
print lp.CompiledKernel(ctx, knl).get_highlighted_code()
# }}}
def test_nonlinear_index(ctx_factory):
ctx = ctx_factory()
knl = lp.make_kernel(ctx.devices[0], [
"{[i,j]: 0<=i,j<n }",
],
"""
a[i*i] = 17
""",
[
lp.GlobalArg("a", shape="n"),
lp.ValueArg("n"),
],
assumptions="n>=1")
print knl
print lp.CompiledKernel(ctx, knl).get_highlighted_code()
def test_triangle_domain(ctx_factory):
ctx = ctx_factory()
knl = lp.make_kernel(ctx.devices[0], [
"{[i,j]: 0<=i,j<n and i <= j}",
],
"a[i,j] = 17",
assumptions="n>=1")
print knl
print lp.CompiledKernel(ctx, knl).get_highlighted_code()
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
def test_array_with_offset(ctx_factory):
dtype = np.float32
ctx = ctx_factory()
queue = cl.CommandQueue(ctx)
n = 5
knl = lp.make_kernel(ctx.devices[0], [
"{[i,j]: 0<=i<n and 0<=j<m }",
],
"""
b[i,j] = 2*a[i,j]
""",
assumptions="n>=1 and m>=1")
cknl = lp.CompiledKernel(ctx, knl)
a_full = cl.clrandom.rand(queue, (n, n), np.float64)
a = a_full[3:10]
print cknl.get_highlighted_code({"a": a.dtype}, {"a": True, "b": False})
evt, (b,) = cknl(queue, a=a, n=a.shape[0], m=a.shape[1])
import numpy.linalg as la
assert la.norm(b.get() - 2*a.get()) < 1e-13
if __name__ == "__main__":
import sys
if len(sys.argv) > 1:
exec(sys.argv[1])
else:
from py.test.cmdline import main
main([__file__])
# vim: foldmethod=marker