如何允许某些线程在锁定互斥锁时具有优先权使用 PTHREADS

Question

假设以下代码由 10 个线程执行。

pthread_mutex_lock(&lock)
Some trivial code
pthread_mutex_unlock(&lock)

为了便于解释，假设线程是 T1、T2、T3.....T10。 我的要求是，只要 T1 或 T2 或 T3（即 T1、T2 或 T3 中的任何一个）正在等待获取锁，它 T4、T5、T6 .....T10 的其他线程应该无法获取锁，即 T1、T2 和 T3 应该优先于其他线程获取锁。

我想这可以通过增加线程T1、T2和T3的优先级来完成

这里是伪代码

if this thread is T1 or T2 or T3
increase its priority 
pthread_mutex_lock(&lock)
Some trivial code
pthread_mutex_unlock(&lock)
if this thread is T1 or T2 or T3 decrease it priority to normal

请注意，我想要一个适用于 Linux 平台并且应该使用 pthreads 的解决方案。我真的不关心任何其他平台。

另外请注意，我真的不想让这 3 个线程成为实时线程，我希望它们展示它们的默认行为（调度和优先级），除了在上面提到的一小段代码中我希望它们始终具有优先权在获取锁。

我已经阅读了一些关于 Linux 中调度策略和调度优先级的手册页，但无法真正理解：(

这行得通吗？您能帮我了解完成上述任务所需的确切 pthread API 吗？

问候 拉里

Answer 1

这是我的实现。低优先级线程使用prio_lock_low()和prio_unlock_low()进行锁定和解锁，高优先级线程使用prio_lock_high()和prio_unlock_high()。

设计很简单。高优先级线程保存在临界区互斥体->cs_mutex，低优先级线程保存在条件变量中。条件变量 mutex 仅在对共享变量的更新和条件变量的信号传递时保持。

#include <pthread.h>

typedef struct prio_lock {
    pthread_cond_t cond;
    pthread_mutex_t cv_mutex; /* Condition variable mutex */
    pthread_mutex_t cs_mutex; /* Critical section mutex */
    unsigned long high_waiters;
} prio_lock_t;

#define PRIO_LOCK_INITIALIZER { PTHREAD_COND_INITIALIZER, PTHREAD_MUTEX_INITIALIZER, PTHREAD_MUTEX_INITIALIZER }

void prio_lock_low(prio_lock_t *prio_lock)
{
    pthread_mutex_lock(&prio_lock->cv_mutex);
    while (prio_lock->high_waiters || pthread_mutex_trylock(&prio_lock->cs_mutex))
    {
        pthread_cond_wait(&prio_lock->cond, &prio_lock->cv_mutex);
    }
    pthread_mutex_unlock(&prio_lock->cv_mutex);
}

void prio_unlock_low(prio_lock_t *prio_lock)
{
    pthread_mutex_unlock(&prio_lock->cs_mutex);

    pthread_mutex_lock(&prio_lock->cv_mutex);
    if (!prio_lock->high_waiters)
        pthread_cond_signal(&prio_lock->cond);
    pthread_mutex_unlock(&prio_lock->cv_mutex);
}

void prio_lock_high(prio_lock_t *prio_lock)
{
    pthread_mutex_lock(&prio_lock->cv_mutex);
    prio_lock->high_waiters++;
    pthread_mutex_unlock(&prio_lock->cv_mutex);

    pthread_mutex_lock(&prio_lock->cs_mutex);
}

void prio_unlock_high(prio_lock_t *prio_lock)
{
    pthread_mutex_unlock(&prio_lock->cs_mutex);

    pthread_mutex_lock(&prio_lock->cv_mutex);
    prio_lock->high_waiters--;
    if (!prio_lock->high_waiters)
        pthread_cond_signal(&prio_lock->cond);
    pthread_mutex_unlock(&prio_lock->cv_mutex);
}

Answer 2

据我了解，您真正能保证这一点的唯一方法是自己编写一个这样工作的锁。但是 @xryl669's answer 建议使用线程优先级和优先级继承，如果它适用于您的用例，当然值得考虑。

要自己实现它，您需要条件变量和等待低/高优先级线程数的计数。

就您需要的概念和 API 而言，它与实现读/写锁相对相似（但显然，您需要的语义完全不同 - 但如果您了解 r/w 锁的工作原理，你就会明白如何实现你想要的）。

您可以在此处查看读写锁的实现：

http://ptgmedia.pearsoncmg.com/images/0201633922/sourcecode/rwlock.c

在较低优先级的线程中，您需要等待高优先级的线程完成，就像读取器等待写入器完成一样。

（上面的代码取自这本书，顺便说一句，这本书也是一本很棒的 posix 线程书，http://www.informit.com/store/product.aspx?isbn=0201633922）

Answer 3

本机方法是为您的互斥锁启用优先级继承（使用 pthread_mutex_attr），并使用 pthread 的线程优先级来执行您需要的操作。 它只需要很少的代码行，而且您无需重新发明轮子。 从好的方面来说，它也可以与 RT 或 FIFO 调度程序一起使用，而您的自制版本则不能。

然后，每当具有高优先级的线程等待由低优先级线程获取的互斥锁时，内核“提升”低优先级线程，以便可以调度它来代替高优先级线程，从而给它释放锁的时间片。一旦锁被释放，高优先级的线程就会被调度。这是自从它在内核中完成以来您可以获得的最低延迟。

Answer 4

或者，您可以只为更高优先级的线程引入另一个锁。考虑以下伪代码（我不熟悉 pthread 语义，但我相信将代码映射到所需的调用并不难）

编辑（感谢 JosephH）

引入 exec 信号量设置为 3（高优先级线程数） 请注意，pend(exec,3); 表示此挂起将休眠，直到所有 3 个插槽都可用并将它们全部消耗完

//init
exec = semaphore(3,3);

//========================

if this is NOT thread (t1,t2,t3)
    lock(low_prio);
    sem_pend(exec,3);
else
    sem_pend(exec,1);
lock(high_prio);
//...
unlock(high_prio);
if this is NOT thread (t1,t2,t3)
    sem_release(exec,3);
    sleep(0); //yield();  //ensures that sem_pend(exec,1) is executed
    unlock(low_prio);
else
    sem_release(exec,1);

Answer 5

（前两次尝试有错误，请跳转到EDIT2）

也许这行得通？

if NOT this thread is T1 or T2 or T3
    pthread_mutex_lock(&lock1) // see note below
    pthread_mutex_lock(&lock2)
    Some trivial code
    pthread_mutex_unlock(&lock2)
    pthread_mutex_unlock(&lock1)
else
    pthread_mutex_lock(&lock2)
    Some trivial code
    pthread_mutex_unlock(&lock2)        
end if

推理： 一些线程将竞争两个锁，因此具有较低的优先级，而一些线程将只竞争一个锁，因此具有较高的优先级。 差异可能仍然很小，然后解决方案是在获取第一个锁和为更高优先级线程尝试第二个锁之间引入一些延迟，在此期间，更高优先级的线程将有机会获得 lock2。
（免责声明：在这方面我是新手）

编辑： 另一种尝试/方法

if NOT (this thread is T1 or T2 or T3)  
    pthread_mutex_lock(&lock1)
    if pthread_mutex_trylock(&lock2) == 0  // low priority threads will not get queued
        Some trivial code
        pthread_mutex_unlock(&lock2)
    end if
    pthread_mutex_unlock(&lock1)
else 
    if (this thread is T1 or T2 or T3)
        pthread_mutex_lock(&lock2)
        Some trivial code
        pthread_mutex_unlock(&lock2)        
    end if
end if

EDIT2：另一个尝试（试图在这里学习一些东西）

if NOT (this thread is T1 or T2 or T3)  
    pthread_mutex_lock(&lock1)
    while !(pthread_mutex_trylock(&lock2) == 0)
        pthread_yield()
    Some trivial code
    pthread_mutex_unlock(&lock2)
    pthread_mutex_unlock(&lock1)
else 
    if (this thread is T1 or T2 or T3)
        pthread_mutex_lock(&lock2)
        Some trivial code
        pthread_mutex_unlock(&lock2)        
    end if
end if

Answer 6

要使用 pthread 实现这一点，您需要 N 个列表，每个线程优先级一个。列表将包含指向线程的 pthread_cond_t 变量的指针。

未经测试的示意图元代码：

/* the main lock */
pthread_mutex_t TheLock = PTHREAD_MUTEX_INITIALIZER;

/* service structures: prio lists and the lock for them */
pthread_mutex_t prio_list_guard = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t *prio_lists[MY_MAX_PRIO][MY_MAX_THREAD]; /* 0 == highest prio */

/* lock */
void
prio_lock(int myprio)
{
    pthread_cond_t x;

    pthread_mutex_lock( &prio_list_guard );

    if (0 == pthread_mutex_trylock( &TheLock )) {
        pthread_mutex_unlock( &prio_list_guard );
        return 0;
    }

    pthread_cond_init( &x, 0 );
    LIST_ADD( prio_lists[myprio], &x )

    while(1)    /* handle spurious wake-ups */
    {
        pthread_cond_wait( &prio_list_guard, &x );
        if (0 == pthread_mutex_trylock( &TheLock )) 
        {
            LIST_REMOVE( prio_lists[myprio], &x );
            pthread_mutex_unlock( &prio_list_guard );
            return 0;
        }
    }
}

/* unlock */
void
prio_unlock()
{
    int i;
    pthread_cond_t *p;

    pthread_mutex_lock( &prio_list_guard );

    for (i=0; i<MY_MAX_PRIO; i++)
    {
        if ((p = LIST_GETFIRST( prio_lists[i] )))
        {
            pthread_cond_signal( p );
            break;
        }
    }

    pthread_mutex_unlock( &TheLock );

    pthread_mutex_unlock( &prio_list_guard );
}

该代码还处理来自pthread_cond_wait() 的虚假唤醒，但坦率地说，我从未见过这种情况发生。

编辑1。请注意，上面的prio_lists 是priority queue 的原始形式。