diff --git a/.gitignore b/.gitignore
index 737aedf2c9145e3de749eecc951e368cc18fb048..a520df17f99f8343ad33639bdee13125d2136d17 100644
--- a/.gitignore
+++ b/.gitignore
@@ -52,6 +52,7 @@ _build
__pycache__
*.o
.ipynb_checkpoints
+cscope.*
# needed by jenkins env
.env
diff --git a/pyopencl/__init__.py b/pyopencl/__init__.py
index f11dff9bab8b0b58fb1ee1c52181ff5f81a20890..41fa6a8927a3ccd0176b4a24b6587a60cdabc45b 100644
--- a/pyopencl/__init__.py
+++ b/pyopencl/__init__.py
@@ -330,7 +330,20 @@ def _add_functionality():
def platform_repr(self):
return "<pyopencl.Platform '%s' at 0x%x>" % (self.name, self.int_ptr)
+ def platform_get_cl_version(self):
+ import re
+ version_string = self.version
+ match = re.match(r"^OpenCL ([0-9]+)\.([0-9]+) .*$", version_string)
+ if match is None:
+ raise RuntimeError("platform %s returned non-conformant "
+ "platform version string '%s'" %
+ (self, version_string))
+
+ return int(match.group(1)), int(match.group(2))
+
+
Platform.__repr__ = platform_repr
+ Platform._get_cl_version = platform_get_cl_version
# }}}
@@ -357,16 +370,7 @@ def _add_functionality():
", ".join(repr(dev) for dev in self.devices))
def context_get_cl_version(self):
- import re
- platform = self.devices[0].platform
- plat_version_string = platform.version
- match = re.match(r"^OpenCL ([0-9]+)\.([0-9]+) .*$",
- plat_version_string)
- if match is None:
- raise RuntimeError("platform %s returned non-conformant "
- "platform version string '%s'" % (platform, plat_version_string))
-
- return int(match.group(1)), int(match.group(2))
+ return self.devices[0].platform._get_cl_version()
Context.__repr__ = context_repr
from pytools import memoize_method
diff --git a/test/test_wrapper.py b/test/test_wrapper.py
index 2e3636cb11019e4d152e42b9ba1fccae4ce77790..971ca27d6f959b9adb91843acc5038e35d305505 100644
--- a/test/test_wrapper.py
+++ b/test/test_wrapper.py
@@ -629,10 +629,8 @@ def test_wait_for_events(ctx_factory):
cl.wait_for_events([evt1, evt2])
-def test_unload_compiler(ctx_factory):
- ctx = ctx_factory()
- platform = ctx.devices[0].platform
- if (ctx._get_cl_version() < (1, 2) or
+def test_unload_compiler(platform):
+ if (platform._get_cl_version() < (1, 2) or
cl.get_cl_header_version() < (1, 2)):
from pytest import skip
skip("clUnloadPlatformCompiler is only available in OpenCL 1.2")
@@ -673,6 +671,94 @@ def test_enqueue_task(ctx_factory):
assert la.norm(a[::-1] - b) == 0
+def test_platform_get_devices(platform):
+ dev_types = [cl.device_type.ACCELERATOR, cl.device_type.ALL,
+ cl.device_type.CPU, cl.device_type.DEFAULT, cl.device_type.GPU]
+ if (platform._get_cl_version() >= (1, 2) and
+ cl.get_cl_header_version() >= (1, 2)):
+ dev_types.append(cl.device_type.CUSTOM)
+ for dev_type in dev_types:
+ devs = platform.get_devices(dev_type)
+ if dev_type in (cl.device_type.DEFAULT,
+ cl.device_type.ALL,
+ getattr(cl.device_type, 'CUSTOM', None)):
+ continue
+ for dev in devs:
+ assert dev.type == dev_type
+
+
+def test_user_event(ctx_factory):
+ ctx = ctx_factory()
+ if (ctx._get_cl_version() < (1, 1) and
+ cl.get_cl_header_version() < (1, 1)):
+ from pytest import skip
+ skip("UserEvent is only available in OpenCL 1.1")
+
+ status = {}
+
+ def event_waiter1(e, key):
+ e.wait()
+ status[key] = True
+
+ def event_waiter2(e, key):
+ cl.wait_for_events([e])
+ status[key] = True
+
+ from threading import Thread
+ from time import sleep
+ evt = cl.UserEvent(ctx)
+ Thread(target=event_waiter1, args=(evt, 1)).start()
+ sleep(.05)
+ if status.get(1, False):
+ raise RuntimeError('UserEvent triggered before set_status')
+ evt.set_status(cl.command_execution_status.COMPLETE)
+ sleep(.05)
+ if not status.get(1, False):
+ raise RuntimeError('UserEvent.wait timeout')
+ assert evt.command_execution_status == cl.command_execution_status.COMPLETE
+
+ evt = cl.UserEvent(ctx)
+ Thread(target=event_waiter2, args=(evt, 2)).start()
+ sleep(.05)
+ if status.get(2, False):
+ raise RuntimeError('UserEvent triggered before set_status')
+ evt.set_status(cl.command_execution_status.COMPLETE)
+ sleep(.05)
+ if not status.get(2, False):
+ raise RuntimeError('cl.wait_for_events timeout on UserEvent')
+ assert evt.command_execution_status == cl.command_execution_status.COMPLETE
+
+
+def test_buffer_get_host_array(ctx_factory):
+ ctx = ctx_factory()
+ mf = cl.mem_flags
+
+ host_buf = np.random.rand(25).astype(np.float32)
+ buf = cl.Buffer(ctx, mf.READ_WRITE | mf.USE_HOST_PTR, hostbuf=host_buf)
+ host_buf2 = buf.get_host_array(25, np.float32)
+ assert (host_buf == host_buf2).all()
+ assert (host_buf.__array_interface__['data'][0] ==
+ host_buf.__array_interface__['data'][0])
+ assert host_buf2.base is buf
+
+ buf = cl.Buffer(ctx, mf.READ_WRITE | mf.ALLOC_HOST_PTR, size=100)
+ try:
+ host_buf2 = buf.get_host_array(25, np.float32)
+ assert False, ("MemoryObject.get_host_array should not accept buffer "
+ "without USE_HOST_PTR")
+ except cl.LogicError:
+ pass
+
+ host_buf = np.random.rand(25).astype(np.float32)
+ buf = cl.Buffer(ctx, mf.READ_WRITE | mf.COPY_HOST_PTR, hostbuf=host_buf)
+ try:
+ host_buf2 = buf.get_host_array(25, np.float32)
+ assert False, ("MemoryObject.get_host_array should not accept buffer "
+ "without USE_HOST_PTR")
+ except cl.LogicError:
+ pass
+
+
if __name__ == "__main__":
# make sure that import failures get reported, instead of skipping the tests.
import pyopencl # noqa