workgroup.hpp

Go to the documentation of this file.

#pragma once

 
#include <cstdint>
 
#if defined(__APPLE__)
#include <os/workgroup.h>
#include <pthread.h>
#include <mach/mach.h>
#include <mach/mach_time.h>
#include <mach/thread_policy.h>
#elif defined(__linux__)
#include <pthread.h>
#include <sched.h>
#endif
 
namespace pulp::audio {

class AudioWorkgroup {
public:
    AudioWorkgroup() = default;
    ~AudioWorkgroup() { leave(); }
 
    AudioWorkgroup(const AudioWorkgroup&) = delete;
    AudioWorkgroup& operator=(const AudioWorkgroup&) = delete;
 
#if defined(__APPLE__)
    void set_workgroup(os_workgroup_t workgroup) {
        workgroup_ = workgroup;
    }
#endif

    bool join_from_audio_thread() {
        if (joined_) return true;
 
#if defined(__APPLE__)
        if (workgroup_) {
            int result = os_workgroup_join(workgroup_, &token_);
            if (result == 0) {
                joined_ = true;
                return true;
            }
            return false;
        }
 
        // Fallback: set real-time thread priority via Mach
        joined_ = capture_and_set_realtime_priority();
        return joined_;
#else
        // Other platforms: best-effort high priority
        joined_ = set_high_priority();
        return joined_;
#endif
    }

    bool leave() {
        if (!joined_)
            return true;
 
#if defined(__APPLE__)
        if (workgroup_) {
            os_workgroup_leave(workgroup_, &token_);
        } else if (priority_thread_ != MACH_PORT_NULL) {
            const auto result =
                previous_constraint_is_default_
                    ? thread_policy_set(priority_thread_, THREAD_EXTENDED_POLICY,
                                        reinterpret_cast<thread_policy_t>(&previous_extended_),
                                        THREAD_EXTENDED_POLICY_COUNT)
                    : thread_policy_set(priority_thread_, THREAD_TIME_CONSTRAINT_POLICY,
                                        reinterpret_cast<thread_policy_t>(&previous_constraint_),
                                        THREAD_TIME_CONSTRAINT_POLICY_COUNT);
            if (result != KERN_SUCCESS)
                return false;
            mach_port_deallocate(mach_task_self(), priority_thread_);
            priority_thread_ = MACH_PORT_NULL;
        }
#endif
        joined_ = false;
        return true;
    }

    bool is_joined() const { return joined_; }

    static bool set_realtime_priority() {
#if defined(__APPLE__)
        mach_timebase_info_data_t timebase;
        mach_timebase_info(&timebase);
 
        // Request real-time scheduling: ~5ms period, ~3ms computation
        thread_time_constraint_policy_data_t policy;
        policy.period = static_cast<uint32_t>(5000000ULL * timebase.denom / timebase.numer);
        policy.computation = static_cast<uint32_t>(3000000ULL * timebase.denom / timebase.numer);
        policy.constraint = static_cast<uint32_t>(5000000ULL * timebase.denom / timebase.numer);
        policy.preemptible = TRUE;
 
        const auto thread = mach_thread_self();
        kern_return_t result = thread_policy_set(thread, THREAD_TIME_CONSTRAINT_POLICY,
                                                 reinterpret_cast<thread_policy_t>(&policy),
                                                 THREAD_TIME_CONSTRAINT_POLICY_COUNT);
 
        mach_port_deallocate(mach_task_self(), thread);
        return result == KERN_SUCCESS;
#else
        return set_high_priority();
#endif
    }

    static bool set_high_priority() {
#if defined(__APPLE__)
        auto result = pthread_set_qos_class_self_np(QOS_CLASS_USER_INTERACTIVE, 0);
        return result == 0;
#elif defined(__linux__)
        struct sched_param param;
        param.sched_priority = sched_get_priority_max(SCHED_FIFO);
        // Don't use SCHED_FIFO as it requires root — use elevated normal priority
        int policy = SCHED_OTHER;
        param.sched_priority = 0; // max for SCHED_OTHER
        pthread_setschedparam(pthread_self(), policy, &param);
        return true;
#elif defined(_WIN32)
        // SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_TIME_CRITICAL);
        return true; // Windows priority boost is not implemented here.
#else
        return false;
#endif
    }
 
private:
    bool joined_ = false;
 
#if defined(__APPLE__)
    bool capture_and_set_realtime_priority() {
        priority_thread_ = mach_thread_self();
        auto count = THREAD_TIME_CONSTRAINT_POLICY_COUNT;
        previous_constraint_is_default_ = FALSE;
        auto result = thread_policy_get(priority_thread_, THREAD_TIME_CONSTRAINT_POLICY,
                                        reinterpret_cast<thread_policy_t>(&previous_constraint_),
                                        &count, &previous_constraint_is_default_);
        if (result == KERN_SUCCESS && previous_constraint_is_default_) {
            count = THREAD_EXTENDED_POLICY_COUNT;
            boolean_t get_default = FALSE;
            result = thread_policy_get(priority_thread_, THREAD_EXTENDED_POLICY,
                                       reinterpret_cast<thread_policy_t>(&previous_extended_),
                                       &count, &get_default);
        }
        // Never change a policy we cannot subsequently restore.
        if (result != KERN_SUCCESS || !set_realtime_priority()) {
            mach_port_deallocate(mach_task_self(), priority_thread_);
            priority_thread_ = MACH_PORT_NULL;
            return false;
        }
        return true;
    }
 
    thread_t priority_thread_ = MACH_PORT_NULL;
    thread_time_constraint_policy_data_t previous_constraint_{};
    thread_extended_policy_data_t previous_extended_{};
    boolean_t previous_constraint_is_default_ = FALSE;
    os_workgroup_t workgroup_ = nullptr;
    os_workgroup_join_token_s token_{};
#endif
};
 
} // namespace pulp::audio