递归互斥锁背后的思想

Question

我正在进行一个学校实验，我们被要求为计数程序创建一个递归互斥锁。我写了一些代码（不起作用），但我认为这主要是因为我不了解使用递归互斥锁的真正思路。请问有人能详细说明一下递归互斥锁应该如何工作/看起来像什么吗？

总体说明：我不是在寻求答案，只是想澄清递归互斥锁应该做什么。

此外，如果有人感兴趣，这里是必需的代码。我正在编辑/实现的代码是recmutex.c

recmutex.h

#include <pthread.h>

/*
 * The recursive_mutex structure. 
*/

struct recursive_mutex {

  pthread_cond_t    cond;
  pthread_mutex_t   mutex; //a non-recursive pthread mutex
  pthread_t         owner;
  unsigned int      count;
  unsigned int      wait_count;
};

typedef struct recursive_mutex  recursive_mutex_t;


/* Initialize the recursive mutex object. 
 *Return a non-zero integer if errors occur. 
 */

int recursive_mutex_init (recursive_mutex_t *mu);


/* Destroy the recursive mutex object. 
 *Return a non-zero integer if errors occur.
 */

int recursive_mutex_destroy (recursive_mutex_t *mu);


/* The recursive mutex object referenced by mu shall be 
   locked by calling pthread_mutex_lock(). When a thread 
   successfully acquires a mutex for the first time, 
   the lock count shall be set to one and successfully return. 
   Every time a thread relocks this mutex, the lock count 
   shall be incremented by one and return success immediately. 
   And any other calling thread can only wait on the conditional 
   variable until being waked up. Return a non-zero integer if errors occur. 
*/
int recursive_mutex_lock (recursive_mutex_t *mu);


/* The recursive_mutex_unlock() function shall release the 
  recursive mutex object referenced by mu. Each time the owner 
  thread unlocks the mutex, the lock count shall be decremented by one. 
  When the lock count reaches zero, the mutex shall become available 
  for other threads to acquire. If a thread attempts to unlock a 
  mutex that it has not locked or a mutex which is unlocked, 
  an error shall be returned. Return a non-zero integer if errors occur. 
*/

int recursive_mutex_unlock (recursive_mutex_t *mu);

recmutex.c: 包含递归互斥锁的函数。

#include <stdio.h>
#include <pthread.h>
#include <errno.h>
#include "recmutex.h"

int recursive_mutex_init (recursive_mutex_t *mu){
    int err;
    err = pthread_mutex_init(&mu->mutex, NULL);
    if(err != 0){
        perror("pthread_mutex_init");
        return -1;
    }else{
        return 0;   
    }
    return 0;
}

int recursive_mutex_destroy (recursive_mutex_t *mu){
    int err;
    err = pthread_mutex_destroy(&mu->mutex);
    if(err != 0){
        perror("pthread_mutex_destroy");
        return -1;
    }else{
        return 1;
    }
    return 0;
}

int recursive_mutex_lock (recursive_mutex_t *mu){

    if(mutex_lock_count == 0){
        pthread_mutex_lock(&mu->mutex);
        mu->count++;
        mu->owner = pthread_self();
        printf("%s", mu->owner);
        return 0;
    }else if(mutex_lock_count > 0){
        pthread_mutex_lock(&mu->mutex);
        mu->count++;
        mu->owner = pthread_self();
        return 0;
    }else{
        perror("Counter decremented incorrectly");
        return -1;
    }
}

int recursive_mutex_unlock (recursive_mutex_t *mu){

    if(mutex_lock_count <= 0){
        printf("Nothing to unlock");
        return -1;
    }else{
        mutex_lock_count--;
        pthread_mutex_unlock(&mu->mutex);
        return 0;
    }
}

count_recursive.cc：上面提到的计数程序。 使用recmutex函数。

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <ctype.h>
#include <unistd.h>
#include <assert.h>
#include <string.h>
#include "recmutex.h"

//argument structure for the thread
typedef struct _arg_{
    int n1;
    int n2;
    int ntimes;
}Arg;

int count; //global counter
recursive_mutex_t mutex; //the recursive mutex

void do_inc(int n){
    int ret;
    if(n == 0){ 
    return;
    }else{
    int c;
    ret = recursive_mutex_lock(&mutex);
    assert(ret == 0);
    c = count;
    c = c + 1;
    count = c;
    do_inc(n - 1);
    ret = recursive_mutex_unlock(&mutex);
    assert(ret == 0);
    }
}

/*  Counter increment function. It will increase the counter by n1 * n2 * ntimes. */
void inc(void *arg){
    Arg * a = (Arg *)arg;
    for(int i = 0; i < a->n1; i++){
    for(int j = 0; j < a->n2; j++){
        do_inc(a->ntimes);
    }
    }
}

int isPositiveInteger (const char * s)
{
    if (s == NULL || *s == '\0' || isspace(*s))
            return 0;
    char * p;
    int ret = strtol (s, &p, 10);
    if(*p == '\0' && ret > 0)
    return 1;
    else
    return 0;
}

int test1(char **argv){

    printf("==========================Test 1===========================\n");
    int ret;
    //Get the arguments from the command line.
    int num_threads = atoi(argv[1]); //The number of threads to be created.
    int n1 = atoi(argv[2]);          //The outer loop count of the inc function.
    int n2 = atoi(argv[3]);          //The inner loop count of the inc function.
    int ntimes = atoi(argv[4]);      //The number of increments to be performed in the do_inc function.

    pthread_t *th_pool = new pthread_t[num_threads];
    pthread_attr_t attr;
    pthread_attr_init( &attr );
    pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);

    ret = recursive_mutex_init(&mutex);
    assert(ret == 0);

    printf("Start Test. Final count should be %d\n", num_threads * n1 * n2 * ntimes );

    // Create threads
    for(int i = 0; i < num_threads; i++){
    Arg *arg = (Arg *)malloc(sizeof(Arg));
    arg->n1 = n1;
    arg->n2 = n2;
    arg->ntimes = ntimes;
    ret = pthread_create(&(th_pool[i]), &attr, (void * (*)(void *)) inc, (void *)arg);
    assert(ret == 0);
    }

    // Wait until threads are done
    for(int i = 0; i < num_threads; i++){
    ret = pthread_join(th_pool[i], NULL);
    assert(ret == 0);
    }

    if ( count != num_threads * n1 * n2 * ntimes) {
    printf("\n****** Error. Final count is %d\n", count );
    printf("****** It should be %d\n", num_threads * n1 * n2 * ntimes );
    }
    else {
    printf("\n>>>>>> O.K. Final count is %d\n", count );
    }

    ret = recursive_mutex_destroy(&mutex);
    assert(ret == 0);

    delete [] th_pool;
    return 0;
}


int foo(){
    int ret;
    printf("Function foo\n");
    ret = recursive_mutex_unlock(&mutex);
    assert(ret != 0);
    return ret;
}

//test a thread call unlock without actually holding it.
int test2(){
    int ret;
    printf("\n==========================Test 2==========================\n");
    pthread_t th;
    pthread_attr_t attr;
    pthread_attr_init( &attr );
    pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);

    ret = recursive_mutex_init(&mutex);
    ret = pthread_create(&th, &attr, (void * (*)(void *))foo, NULL);

    printf("Waiting for thread to finish\n");
    ret = pthread_join(th, NULL);
    assert(ret == 0);
    return 0;
}


int main( int argc, char ** argv )
{
    int ret;
    count = 0;

    if( argc != 5 ) {
    printf("You must enter 4 arguments. \nUsage: ./count_recursive num_threads n1 n2 ntimes\n");
    return -1;
    }

    if(isPositiveInteger(argv[1]) != 1 || isPositiveInteger(argv[2]) != 1 || isPositiveInteger(argv[3]) != 1 || isPositiveInteger(argv[4]) != 1 ){
    printf("All the 4 arguments must be positive integers\n");
    return -1;
    }

    test1(argv);

    test2();

    return 0;
}

- Davis

Answer 1

递归互斥锁的概念是，当前持有锁的线程可以成功地重新获取该锁。例如：

如果我有像这样的一些互斥锁（这是伪代码）：

mutex l;
recursive_mutex r;

如果我在单线程中这样做：

l.lock();
l.lock(); // this would hang the thread.

但是

r.lock();
r.lock();
r.lock(); // this would all pass though with no issue.

在实现递归互斥锁时，需要检查哪个线程ID已经锁定了它，如果已经锁定，并且与当前线程ID匹配，则返回成功。

- AllenKll

Answer 2

递归互斥锁的作用是让你写出这样的代码：

recursive_mutext_t rmutex;

void foo(...) {
    recursive_lock_lock(&rmutex);
    ...
    recursive_lock_unlock(&rmutex);
}

void bar(...) {
    recursive_lock_lock(&rmutex);
    ...
    foo(...);
    ...
    recursive_lock_unlock(&rmutex);
}

void baz(...) {
    ...
    foo(...);
    ...
}

函数foo()需要锁定互斥量，但您希望能够从已经锁定相同互斥量的bar()中调用它，也可以从未锁定互斥量的baz()中调用它。如果使用普通的mutex()，当从bar()中调用foo()时，线程会自我死锁，因为普通的mutex lock()函数将不会返回，直到mutex被解锁，并且没有其他线程将其解锁。

您的recursive_mutex_lock()需要区分这些情况：(1)互斥量未锁定，(2)互斥量已被锁定，但调用线程是所有者，以及(3)互斥量已被其他线程锁定。

情况(3)需要阻塞调用线程，直到所有者完全解锁互斥量。在这一点上，它转换为情况(1)。这里有一个提示：使用条件变量处理情况(3)。也就是说，当调用线程不是所有者时，调用线程应该进行pthread_condition_wait(...)调用。

- Solomon Slow