网络编程学习——线程（三）

摘要: 通过《UNIX网络编程卷1：套接字联网API》学习网络编程

[code=plain]/* Lock a mutex.  */
extern int pthread_mutex_lock (pthread_mutex_t *__mutex)
__THROWNL __nonnull ((1));

/* Unlock a mutex.  */
extern int pthread_mutex_unlock (pthread_mutex_t *__mutex)
__THROWNL __nonnull ((1));

如果试图上锁已被另外某个线程锁住一个互斥锁，本线程将被阻塞，直到该互斥锁被解锁为止。

如果某个互斥锁变量是静态分配的，我们就必须把它初始化为常值PTHREAD_MUTEX_INITIALIZER。如果我们在共享内存区中分配一个互斥锁，那么必须通过调用pthread_mutex_init函数在运行时把它初始化

下面是前面程序的改正版本，它使用单个互斥锁保护由两个线程共同访问的计数器。

[code=plain]#include <pthread.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>

#define NLOOP 5000

int counter; // incrementrd by threads
pthread_mutex_t couter_mutex = PTHREAD_MUTEX_INITIALIZER;

void *doit( void * );

int main( int argc, char **argv )
{
pthread_t tidA, tidB;

pthread_create( &tidA, NULL, &doit, NULL );
pthread_create( &tidB, NULL, &doit, NULL );

// wait for both threads to terminate
pthread_join( tidA, NULL );
pthread_join( tidB, NULL );

exit( 0 );
}

void * doit( void *vptr )
{
int i, val;

// Each thread fetches,prints,and increments the counter NLOOP times
// The value of the counter should increase monotonically
for( i = 0; i < NLOOP; i++ )
{
pthread_mutex_lock( &couter_mutex );

val = counter;
printf( " %d: %d\n ",( int ) pthread_self(), val + 1 );
counter = val + 1;

pthread_mutex_unlock( &couter_mutex );
}

return( NULL );
}

我们声明一个名为counter_mutex的互斥锁，线程在操作counter变量之前必须锁住该互斥锁。无论何时运行这个程序，其输出总是正确的；计数器值被单调地递增，所显示的最终值总是10000.

7
3ff0
条件变量

互斥锁适用于防止访问某个共享变量，但是我们需要另外某种在等待某个条件发生期间能让我们进入睡眠的东西。

我们把Web客户程序的Solaris的thr_join替换成pthread_join。然而在知道某个线程已经终止之前，我们无法调用这个Pthread函数。我们首先声明一个计量已终止线程数，并使用一个互斥锁保护它。

[code=plain]int ndone；// number of terminated threads
pthread_mutex_t ndone_mutex = PTHREAD_MUTEX_INITIALIZER;

我们接着要求每个线程在即将终止之前警慎使用关联的互斥锁递增这个计数器。

[code=plain]void * do_get_read( void *vptr )
{
...
pthread_mutex_lock( &ndone_mutex );
ndone++;
pthread_mutex_unlock( &ndone_mutex );

return( fptr ); // terminate thread
}

主循环需要一次又一次地锁住这个互斥锁以便检查是否有任何线程终止了。

[code=plain]while( nlefttoread > 0 )
{
while( nconn < maxcoon && nlefttoconn > 0 )
{
// find a file to read
...
}
// see if one of the threads is done
pthread_mutex_lock( &ndone_mutex );
if( ndone > 0 )
{
for( i = 0; i < nfiles; i++ )
{
if( file[ i ].f_flags & F_DONE )
{
pthread_join( file[ i ].f_tid, ( void ** ) &fptr );
// update file[i] for terminated thread
...
}
}
}
pthread_mutex_unlock( &ndone_mutex );
}

如此编写主循环尽管正确，却意味着主循环永远不进入睡眠，它就是不断地循环，每次循环回来检查一下ndone。这种方法称为轮询（polling），相当浪费CPU时间。

我们需要一个让主循环进入睡眠，直到某个线程通知它有事可做才醒来的方法。条件变量（condition variable）结合互斥锁能够提供这个功能，呼出所提供互斥机制，条件变量提供信号机制。

按照Pthread，条件变量是类型为pthread_cond_t的变量。以下两个函数使用条件变量。

[code=plain]#include <pthread.h>

/* Wait for condition variable COND to be signaled or broadcast.
MUTEX is assumed to be locked before.

This function is a cancellation point and therefore not marked with
__THROW.  */
extern int pthread_cond_wait (pthread_cond_t *__restrict __cond,
pthread_mutex_t *__restrict __mutex)
__nonnull ((1, 2));

/* Wake up one thread waiting for condition variable COND.  */
extern int pthread_cond_signal (pthread_cond_t *__cond)
__THROWNL __nonnull ((1));

第二个函数的名字中“signal”一词并不指称Unix的SIGxxx信号。

解释这些函数最容易的方法是举例说明。回到我们的Web客户程序例子，现在我们给计数器ndone同时关联一个条件变量和一个互斥锁。

[code=plain]int ndone;
pthread_mutex_t ndone_mutex = PTHREAD_MUTEX_INIALIZER;
pthread_cond_t ndone_cond = PTHREAD_COND_INITIALIZER;

通过在持有该互斥锁期间递增该计数器并发送信号到该条件变量，一个线程通知主循环自身即将终止。

[code=plain]pthread_mutex_lock( &ndone_mutex );
ndone++;
pthread_cond_signal( &ndone_cond );
pthread_mutex_unlock( &ndone_mutex );

主循环阻塞在pthread_cond_wait调用中，等待某个即将终止的线程发送信号到与ndone关联的条件变量。

[code=plain]while( nlefttoread > 0 )
{
while( nconn < maxcoon && nlefttoconn > 0 )
{
// find a file to read
...
}
// wait for one of the threads to terminate
pthread_mutex_lock( &ndone_mutex );
while( ndone == 0 )
pthread_cond_wait( &ndone_cond, &ndone_mutex );

for( i = 0; i < nfile; i++ )
{
if( file[ i ].f_flags & F_DONE )
{
pthread_join( file[ i ].f_tid, ( void ** ) &fptr );
// update file[i] for terminated thread
}
}
pthread_mutex_unlock( &ndone_mutex );

注意，主循环仍然只是在持有互斥锁期间检查ndon变量。然后，如果发现无事可做，那就调用pthread_cond_wait。该函数把调用线程投入睡眠并释放调用线程持有的互斥锁。此外，当调用线程后来从pthread_cond_wait返回时（其他某个线程发送信号到与ndone关联的条件变量之后），该线程再次持有该互斥锁。

为什么每个条件变量都要关联一个互斥锁呢？因为“条件”通常是线程之间共享的某个变量的值。允许不同线程设置和测试该变量要求有一个与该变量关联的互斥锁。举例来说，要是刚才给出的例子代码中我们没用互斥锁，那么主循环将如下测试变量ndone。

[code=plain]// waitfor one of the threads to terminate
while( ndone == 0 )
pthread_cond_wait( &ndone_cond, &ndone_mutex );

这里存在如此可能性：主线程外最后一个线程在主循环测试ndone==0之后但在调用pthread_cond_wait之前递增ndone。如果发生这样的情形，最后那个“信号”就丢失了，造成主循环永远在pthread_cond_wait调用中，等待永远不再发生的某事再次出现。

同样的理由要求pthread_cond_wait被调用时其所关联的互斥锁必须是上锁的，该函数作为单个原子操作解锁并把调用线程投入睡眠也是出于这个理由。要是该函数不先解锁该互斥锁，到返回时再给它上锁，调用线程就不得不事先解锁事后上锁，测试变量ndone的代码变为：

[code=plain]  // wait for one of the threads to terminate
pthread_mutex_lock( &ndone_mutex );
while( ndone == 0 )
{
pthread_mutex_lock( &ndone_mutex );
pthread_cond_wait( &ndone_cond, &ndone_mutex );
pthread_mutex_unlock( &ndone_mutex );
}

然而这里再次存在如此可能性：主线程外最后一个线程在主线程调用pthread_mutex_unlock和pthread_cond_wait之间终止并递增ndone的值。

pthread_cond_signal通常唤醒等在相应条件变量上的单个线程。有时候一个线程知道自己应该唤醒多个线程，这种情况下它可以调用pthread_cond_broadcast唤醒等在相应条件变量上的所有线程。

[code=plain]#include <pthread.h>

/* Wake up all threads waiting for condition variables COND.  */
extern int pthread_cond_broadcast (pthread_cond_t *__cond)
__THROWNL __nonnull ((1));

/* Wait for condition variable COND to be signaled or broadcast until
ABSTIME.  MUTEX is assumed to be locked before.  ABSTIME is an
absolute time specification; zero is the beginning of the epoch
(00:00:00 GMT, January 1, 1970).

This function is a cancellation point and therefore not marked with
__THROW.  */
extern int pthread_cond_timedwait (pthread_cond_t *__restrict __cond,
pthread_mutex_t *__restrict __mutex,
const struct timespec *__restrict __abstime)

pthread_cond_timewait允许线程设置一个阻塞时间的限制。__abstime是一个timespec结构，指定该函数必须返回时刻的系统时间，即使到时候相应条件变量尚未收到信号。如果发生这样的超时，那就返回ETIME错误。

这个时间是一个绝对时间（absolute time），而不是一个时间增量（time delta）。也就是说abstime参数是函数应该返回时刻的系统时间——从1970年1月1日UTC时间以来的秒数和纳秒数。这一点不同于selcet和pselcet，他们指定的是从调用时刻开始到函数应该返回时刻的秒数和微妙数（对于pselect为纳秒数）。通常采用的过程是：调用gettimeofday获取当前时间（作为一个timeval结构），把它复制到一个timespec结构中，再加上期望的时间限制。例如：

[code=plain]struct timeval tv；
struct timespec ts；

if( gettimeofday( &tv, NULL ) < 0 )
err_sys( " gettimeofday error " );
ts.tv_sec = tv.tv_sec + 5; // 5 seconds in future
ts.tv_nsec = tv.tv_usec * 1000; // microsec to nanosec

pthread_cond_timewait( ..., &ts );

使用绝对时间取代增量时间的优点是，如果该函数过早返回（可能因为捕获了某个信号），那么不必改动timespec结构参数的内容就可以再次调用该函数，缺点是首次调用该函数之前不得不调用gettimeofday。

8 Web客户与同时连接（续）

我们现在重新编写Web客户程序，把其中对于Solaris之thr_join函数的调用替换成调用pthread_join。这么一来必须明确指定等待哪个线程。为了做到这一点，我们要使用条件变量。

全局变量的唯一变动是增加一个新标志和一个条件变量。

[code=plain]#define F_JOINED 8  // main has pthread_join'ed

int ndone;  // number of terminated threads
pthread_mutex_t ndone_mutex = PTHREAD_MUTEX_INIALIZER;
pthread_cond_t ndone_cond = PTHREAD_COND_INITIALIZER;

do_get_read函数的唯一变动是在本线程终止之前递增ndone并通知主循环。

[code=plain]printf( " end-of-file on %s\n ", fptr->f_name );
close( fd );

pthread_mutex_lock( &ndone_mutex );
fptr->f_flags = F_DONE;  // clears F_READING
ndone++;
pthread_cond_signal( &ndone_cond );
pthread_mutex_unlock( &ndone_mutex );

return( fptr ); // terminate thread

大多数变动发生在主循环中，下面是主循环的新版本。

[code=plain]while( nlefttoread > 0 )
{
// 这段代码没有变动
while( nconn < maxcoon && nlefttoconn > 0 )
{
// find a file to read
for( i = 0; i < nfiles; i++ )
if( file[ i ].f_flag == 0 )
break;
if( i == nfiles )
err_quit( " nlefttoconn == %d but nothing found ", nlefttoconn );

file[ i ].f_tid = tid;
nconn++;
nlefttoconn--;
}

// 为了等待某个线程终止，我们等待ndone变为非0。这个测试必须在锁住所关联期间进行。睡眠由
// pthread_ond_wait执行
// wait for one of the threads to terminate
pthread_mutex_lock( &ndone_mutex );
while( ndone == 0 )
pthread_cond_wait( &ndone_cond, &ndone_mutex );

//&n
3ff0
bsp;当发现某个线程终止时，我们便利所有file结构找出这个线程，在调用pthread_join，然后设置新的F_JOINED标志。
for( i = 0; i < nfile; i++ )
{
if( file[ i ].f_flags & F_DONE )
{
pthread_join( file[ i ].f_tid, ( void ** ) &fptr );
// update file[i] for terminated thread
if( &file[ i ] = fptr )
err_quit( " file[ i ] != fptr " );
fptr->f_flags = F_JOINED; // clears F_DONE
ndone--;
nconn--;
nlefttoread--;
printf( " thread %d for %s done\n ", fptr->f_tid, fptr->f_name );
}
}
pthread_mutex_unlock( &ndone_mutex );
}

exit( 0 );
}