view src/testdir/thread_util.py @ 33920:28e1e956f42c v9.0.2159

patch 9.0.2159: screenpos() may crash with neg. column Commit: https://github.com/vim/vim/commit/ec54af4e26952d954a4cc009f62c80ea01445d30 Author: zeertzjq <zeertzjq@outlook.com> Date: Tue Dec 12 16:43:44 2023 +0100 patch 9.0.2159: screenpos() may crash with neg. column Problem: screenpos() may crash with neg. column Solution: validate and correct column closes: #13669 Signed-off-by: zeertzjq <zeertzjq@outlook.com> Signed-off-by: Christian Brabandt <cb@256bit.org>
author Christian Brabandt <cb@256bit.org>
date Tue, 12 Dec 2023 17:00:08 +0100
parents 06e3c6bac36d
children
line wrap: on
line source

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)