《UNIX网络编程 卷2》读书笔记（四）

/**/ /* include globals */
#include     " unpipc.h "

#define     MAXNITEMS         1000000
#define     MAXNTHREADS            100

         /**/ /* globals shared by threads */
int         nitems;                 /**/ /* read-only by producer and consumer */
int         buff[MAXNITEMS];
struct   {
  pthread_mutex_t    mutex;
  int                nput;    /**//* next index to store */
  int                nval;    /**//* next value to store */
}  put  =   { PTHREAD_MUTEX_INITIALIZER } ;

struct   {
  pthread_mutex_t    mutex;
  pthread_cond_t    cond;
  int                nready;    /**//* number ready for consumer */
}  nready  =   { PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER } ;
/**/ /* end globals */

void      * produce( void   * ),  * consume( void   * );

/**/ /* include main */
int
main( int  argc,  char   ** argv)
{
    int            i, nthreads, count[MAXNTHREADS];
    pthread_t    tid_produce[MAXNTHREADS], tid_consume;

    if (argc != 3)
        err_quit("usage: prodcons6 <#items> <#threads>");
    nitems = min(atoi(argv[1]), MAXNITEMS);
    nthreads = min(atoi(argv[2]), MAXNTHREADS);

    Set_concurrency(nthreads + 1);
        /**//* 4create all producers and one consumer */
    for (i = 0; i < nthreads; i++) {
        count[i] = 0;
        Pthread_create(&tid_produce[i], NULL, produce, &count[i]);
    }
    Pthread_create(&tid_consume, NULL, consume, NULL);

        /**//* wait for all producers and the consumer */
    for (i = 0; i < nthreads; i++) {
        Pthread_join(tid_produce[i], NULL);
        printf("count[%d] = %d\n", i, count[i]);    
    }
    Pthread_join(tid_consume, NULL);

    exit(0);
}
/**/ /* end main */

/**/ /* include prodcons */
void   *
produce( void   * arg)
{
    for ( ; ; ) {
        Pthread_mutex_lock(&put.mutex);
        if (put.nput >= nitems) {
            Pthread_mutex_unlock(&put.mutex);
            return(NULL);        /**//* array is full, we're done */
        }
        buff[put.nput] = put.nval;
        put.nput++;
        put.nval++;
        Pthread_mutex_unlock(&put.mutex);

        Pthread_mutex_lock(&nready.mutex);
        if (nready.nready == 0)
            Pthread_cond_signal(&nready.cond);//发出信号
        nready.nready++;//置为1 
        Pthread_mutex_unlock(&nready.mutex);

        *((int *) arg) += 1;
    }
}

void   *
consume( void   * arg)
{
    int        i;

    for (i = 0; i < nitems; i++) {
        Pthread_mutex_lock(&nready.mutex);
        while (nready.nready == 0)
            Pthread_cond_wait(&nready.cond, &nready.mutex);//wait条件变量
        nready.nready--;//置为0
        Pthread_mutex_unlock(&nready.mutex);

        if (buff[i] != i)
            printf("buff[%d] = %d\n", i, buff[i]);
    }
    return(NULL);
}
/**/ /* end prodcons */

这里在生产者的代码中，当它获取到互斥锁时，若发出信号唤醒消费者，则此时可能系统立即调度唤醒消费者，但互斥锁任然在生产者之手，则消费者获取互斥锁必然失败，为了避免此种低效的情况出现，我们可以直到生产者释放互斥锁后才给与之关联的条件变量发送信号，这在Posix里是可以这么做的，但Posix又接着说：若要可预见的调度行为，则调用pthead_cond_signal的线程必须锁住该互斥锁。

当在进程间共享互斥锁时，持有该互斥锁的进程可能在持有期间终止，但无法让系统在终止时自动释放掉所持有的锁。一个线程也可以在持有互斥锁期间终止，可能是自己调用pthread_exit或被另一个线程取消，若是前者，则它应该知道自己还有一个互斥锁，若是后者，则该线程可以先安装一个将在被取消时调用的清理处理程序。但最致命的情况是由于此线程的终止导致整个进程的终止。即使一个进程终止时系统自动释放其持有的锁，但也会导致临界区内数据处于不一致状态，

读写锁的规则：

1，只要没有线程持有某个给定的读写锁用于写，则任意数目的线程可以持有

该读写锁用于读

2，仅当没有线程持有某个给定的读写锁用于读或用于写时，才能分配该读写锁用于写

这种锁在那些读数据比写数据频繁的应用中使用比较有用，允许多个读者提供了更高的并发度，同时在写者修改数据时保护数据，避免任何其他读者或写者的干扰。

这种对某个给定资源的共享访问也叫共享—独占上锁，获取一个读写锁用于读称为共享锁，获取一个读写锁用于写称为独占锁。在操作系统中就介绍过这种，经典的问题就是读者—写者问题，有多种类型：多读者，单写者或多读者，多写者。，此外还有要考虑的就是读者和写者谁优先，也就产生了1类和2类读写问题。

      读写锁类型为pthread_rwlock_t。pthread_rwlock_rdlock获取一个读出锁，若对应的读写锁已经被某个写者持有，则阻塞调用线程，pthread_rwlock_wrlock获取一个写出锁，若对应的读写锁已经被另一个写者持有或被一个或多个读者持有，则阻塞调用线程，pthread_rwlock_unlock释放一个读出锁或写入锁。

      使用互斥锁和条件变量实现读写锁（写者优先）

typedef  struct   {
  pthread_mutex_t    rw_mutex;        /**//* basic lock on this struct *///访问此读写锁使用的互斥锁
  pthread_cond_t    rw_condreaders;    /**//* for reader threads waiting  读者线程使用*/
  pthread_cond_t    rw_condwriters;    /**//* for writer threads waiting  写者线程使用*/
  int                rw_magic;        /**//* for error checking 初始化成功后， 被设置为RW_MAGIC,所有函数测试此成员，检查调用者是否作为参数传递了指向某个已经初始化的读写锁的指针，读写锁摧毁时，被设置为0*/
  int                rw_nwaitreaders;/**//* the number waiting  读者计数器*/
  int                rw_nwaitwriters;/**//* the number waiting   写者计数器*/
  int                rw_refcount;//本读写锁的当前状态，-1表示是写入锁（任意时刻只有一个），0表示可用，大于0表示当前容纳着的读出锁数目
    /**//* 4-1 if writer has the lock, else # readers holding the lock */
}  pthread_rwlock_t;

int  pthread_rwlock_init(pthread_rwlock_t  * rw, pthread_rwlockattr_t  * attr)
{
    int        result;

    if (attr != NULL)
        return(EINVAL);        /**//* not supported */

    if ( (result = pthread_mutex_init(&rw->rw_mutex, NULL)) != 0)
        goto err1;
    if ( (result = pthread_cond_init(&rw->rw_condreaders, NULL)) != 0)
        goto err2;
    if ( (result = pthread_cond_init(&rw->rw_condwriters, NULL)) != 0)
        goto err3;
    rw->rw_nwaitreaders = 0;
    rw->rw_nwaitwriters = 0;
    rw->rw_refcount = 0;
    rw->rw_magic = RW_MAGIC;

    return(0);

err3:
    pthread_cond_destroy(&rw->rw_condreaders);
err2:
    pthread_mutex_destroy(&rw->rw_mutex);
err1:
    return(result);            /**//* an errno value */
}

int  pthread_rwlock_destroy(pthread_rwlock_t  * rw)
{
    //检查参数是否有效
    if (rw->rw_magic != RW_MAGIC)
        return(EINVAL);
    if (rw->rw_refcount != 0 ||
        rw->rw_nwaitreaders != 0 || rw->rw_nwaitwriters != 0)
        return(EBUSY);

    pthread_mutex_destroy(&rw->rw_mutex);
    pthread_cond_destroy(&rw->rw_condreaders);
    pthread_cond_destroy(&rw->rw_condwriters);
    rw->rw_magic = 0;

    return(0);
}

int  pthread_rwlock_rdlock(pthread_rwlock_t  * rw)
{
    int        result;
    //检查参数是否有效
    if (rw->rw_magic != RW_MAGIC)
        return(EINVAL);
    //操作读写锁前，先给其互斥锁上锁
    if ( (result = pthread_mutex_lock(&rw->rw_mutex)) != 0)
        return(result);

        /**//* 4give preference to waiting writers */
    while (rw->rw_refcount < 0 || rw->rw_nwaitwriters > 0) 
{// rw_refcount 小于0（表示有写者持有读写锁），rw_nwaitwriters大于0表示有线程正等着获取读写锁的一个写入锁，则无法获取该读写锁的一个读出锁
        rw->rw_nwaitreaders++;
        result = pthread_cond_wait(&rw->rw_condreaders, &rw->rw_mutex);
        rw->rw_nwaitreaders--;
        if (result != 0)
            break;
    }
    if (result == 0)
        rw->rw_refcount++;        /**//* another reader has a read lock */

    pthread_mutex_unlock(&rw->rw_mutex);
    return (result);
}

int  pthread_rwlock_tryrdlock(pthread_rwlock_t  * rw)
{
    int        result;

    if (rw->rw_magic != RW_MAGIC)
        return(EINVAL);

    if ( (result = pthread_mutex_lock(&rw->rw_mutex)) != 0)
        return(result);

    if (rw->rw_refcount < 0 || rw->rw_nwaitwriters > 0)
        result = EBUSY;            /**//* held by a writer or waiting writers */
    else
        rw->rw_refcount++;        /**//* increment count of reader locks */

    pthread_mutex_unlock(&rw->rw_mutex);
    return(result);
}

int  pthread_rwlock_wrlock(pthread_rwlock_t  * rw)
{
    int        result;

    if (rw->rw_magic != RW_MAGIC)
        return(EINVAL);

    if ( (result = pthread_mutex_lock(&rw->rw_mutex)) != 0)
        return(result);

    while (rw->rw_refcount != 0) 
{//只要有读者持有读出锁或有一个写者持有唯一的写入锁，调用线程阻塞
        rw->rw_nwaitwriters++;
        result = pthread_cond_wait(&rw->rw_condwriters, &rw->rw_mutex);
        rw->rw_nwaitwriters--;
        if (result != 0)
            break;
    }
    if (result == 0)
        rw->rw_refcount = -1;

    pthread_mutex_unlock(&rw->rw_mutex);
    return(result);
}

int  pthread_rwlock_trywrlock(pthread_rwlock_t  * rw)
{
    int        result;

    if (rw->rw_magic != RW_MAGIC)
        return(EINVAL);

    if ( (result = pthread_mutex_lock(&rw->rw_mutex)) != 0)
        return(result);

    if (rw->rw_refcount != 0)
        result = EBUSY;            /**//* held by either writer or reader(s) */
    else
        rw->rw_refcount = -1;    /**//* available, indicate a writer has it */

    pthread_mutex_unlock(&rw->rw_mutex);
    return(result);
}

int  pthread_rwlock_unlock(pthread_rwlock_t  * rw)
{
    int        result;

    if (rw->rw_magic != RW_MAGIC)
        return(EINVAL);

    if ( (result = pthread_mutex_lock(&rw->rw_mutex)) != 0)
        return(result);

    if (rw->rw_refcount > 0)
        rw->rw_refcount--;            /**//* releasing a reader */
    else if (rw->rw_refcount == -1)
        rw->rw_refcount = 0;        /**//* releasing a reader */
    else
        err_dump("rw_refcount = %d", rw->rw_refcount);

        /**//* 4give preference to waiting writers over waiting readers */
    if (rw->rw_nwaitwriters > 0) {
        if (rw->rw_refcount == 0)
            result = pthread_cond_signal(&rw->rw_condwriters);
    } else if (rw->rw_nwaitreaders > 0)
        result = pthread_cond_broadcast(&rw->rw_condreaders);

    pthread_mutex_unlock(&rw->rw_mutex);
    return(result);
}

int  pthread_rwlock_unlock(pthread_rwlock_t  * rw)
{
    int        result;

    if (rw->rw_magic != RW_MAGIC)
        return(EINVAL);

    if ( (result = pthread_mutex_lock(&rw->rw_mutex)) != 0)
        return(result);

    if (rw->rw_refcount > 0)
        rw->rw_refcount--;            /**//* releasing a reader */
    else if (rw->rw_refcount == -1)
        rw->rw_refcount = 0;        /**//* releasing a writer */
    else
        err_dump("rw_refcount = %d", rw->rw_refcount);

        /**//* 4give preference to waiting writers over waiting readers */
        //写者优先，而只有一个写者
    if (rw->rw_nwaitwriters > 0) {
        if (rw->rw_refcount == 0)
            result = pthread_cond_signal(&rw->rw_condwriters);//通知等待的那个写者
    } else if (rw->rw_nwaitreaders > 0)
        result = pthread_cond_broadcast(&rw->rw_condreaders);//通知所有等待的读者

    pthread_mutex_unlock(&rw->rw_mutex);
    return(result);
}

这里的读出锁和写入锁函数都有一个问题，若调用线程阻塞在pthread_cond_wati调用上，并且随后此线程被取消了，则它会在还持有互斥锁的情况下终止，于是rw_nwaitreaders计数器的值会出错