patch 8.2.4250: channel out callback test is flaky on Mac
Problem: Channel out callback test is flaky on Mac.
Solution: Assign high priority to the test process. (Ozaki Kiichi,
closes #9653)
diff --git a/src/testdir/thread_util.py b/src/testdir/thread_util.py
new file mode 100644
index 0000000..e82f23c
--- /dev/null
+++ b/src/testdir/thread_util.py
@@ -0,0 +1,90 @@
+import platform
+
+if platform.system() == 'Darwin':
+ from ctypes import (
+ CDLL,
+ POINTER,
+ Structure,
+ byref,
+ c_int,
+ c_uint,
+ c_uint32,
+ c_void_p,
+ sizeof
+ )
+ from ctypes.util import find_library
+
+ class ThreadTimeConstraintPolicy(Structure):
+ _fields_ = [
+ ("period", c_uint32),
+ ("computation", c_uint32),
+ ("constraint", c_uint32),
+ ("preemptible", c_uint)
+ ]
+
+ _libc = CDLL(find_library('c'))
+
+ THREAD_TIME_CONSTRAINT_POLICY = c_uint(2)
+
+ THREAD_TIME_CONSTRAINT_POLICY_COUNT = c_uint(
+ int(sizeof(ThreadTimeConstraintPolicy) / sizeof(c_int)))
+
+ _libc.pthread_self.restype = c_void_p
+
+ _libc.pthread_mach_thread_np.restype = c_uint
+ _libc.pthread_mach_thread_np.argtypes = [c_void_p]
+
+ _libc.thread_policy_get.restype = c_int
+ _libc.thread_policy_get.argtypes = [
+ c_uint,
+ c_uint,
+ c_void_p,
+ POINTER(c_uint),
+ POINTER(c_uint)
+ ]
+
+ _libc.thread_policy_set.restype = c_int
+ _libc.thread_policy_set.argtypes = [
+ c_uint,
+ c_uint,
+ c_void_p,
+ c_uint
+ ]
+
+ def _mach_thread_self():
+ return _libc.pthread_mach_thread_np(_libc.pthread_self())
+
+ def _get_time_constraint_policy(default=False):
+ thread = _mach_thread_self()
+ policy_info = ThreadTimeConstraintPolicy()
+ policy_infoCnt = THREAD_TIME_CONSTRAINT_POLICY_COUNT
+ get_default = c_uint(default)
+
+ kret = _libc.thread_policy_get(
+ thread,
+ THREAD_TIME_CONSTRAINT_POLICY,
+ byref(policy_info),
+ byref(policy_infoCnt),
+ byref(get_default))
+ if kret != 0:
+ return None
+ return policy_info
+
+ def _set_time_constraint_policy(policy_info):
+ thread = _mach_thread_self()
+ policy_infoCnt = THREAD_TIME_CONSTRAINT_POLICY_COUNT
+
+ kret = _libc.thread_policy_set(
+ thread,
+ THREAD_TIME_CONSTRAINT_POLICY,
+ byref(policy_info),
+ policy_infoCnt)
+ if kret != 0:
+ raise OSError(kret)
+
+ def set_high_priority():
+ policy_info = _get_time_constraint_policy(default=True)
+ if not policy_info:
+ return
+ policy_info.preemptible = c_uint(False)
+ _set_time_constraint_policy(policy_info)