TCP/IP网络编程之多线程服务端的实现（二）

线程存在的问题和临界区

上一章TCP/IP网络编程之多线程服务端的实现（一）的thread4.c中，我们发现多线程对同一变量进行加减，最后的结果居然不是我们预料之内的。其实，如果多执行几次程序，会发现每次程序计算的结果都不一样。那么，造成这样的原因是什么呢？

现在，假设我们一个全局变量sum的值为99，我们创建两个线程，要对sum进行加1操作，那么理想情况下，sum的值应为101。那么要对sum加1并赋值给sum，我们可以简写为：sum+=1。于是多线程可以开始对sum进行操作，但因为是多线程，有可能在一个进程内，一个线程还没执行完，另外一个线程得到CPU时间开始执行。

所以，让我们把目光放回：sum+=1这行代码，这段代码其实有两个动作，第一个动作是sum+1，第二个是将之前相加的结果重新赋值给sum。那么，有两个线程并发给sum加1，有可能第一个线程执行完相加的结果，得到100，但在赋值之前便失去了CPU时间，轮到另外一个线程获得CPU时间，执行加1的操作，等到执行完相加的操作，第二个线程的CPU时间到头了，最后，两个线程都要执行赋值操作，最后我们看到sum只有100，而并不是我们预想中的101

线程同步

为了要解决这一问题，我们必须要求在从sum加1到完成sum的赋值这段临界区，只能有一个线程来完成。而就涉及到线程同步了，这里有两种技术可以实现线程同步，分别是“互斥量”（Mutex）和“信号量”（Semaphore）

互斥量

互斥量是“Mutual Exclusion”的简写，表示不允许多个线程同时访问。举个例子，临界区就好比是小房间里的取款机，现在大部分取款机都会装在一个小房间，当需要取款时一个一个人按顺序进入小房间，栓上门开始取款，取款完毕后离开小房间，让下一个人进来取款。把取款的人当做线程，如果多人同时进入小房间（临界区）那肯定会造成安全上的问题，如账号密码泄露。因此，门栓就是互斥量，当栓上门时，不允许其他人（线程）进入小房间（临界区）进行操作

#include <pthread.h>
int pthread_mutex_init(pthread_mutex_t * mutex,const pthread_mutexattr_t * attr);
int pthread_mutex_destroy(pthread_mutex_t * mutex);
//成功时返回0，失败时返回其他值

　　

• mutex：创建互斥量时传递保存互斥量的变量地址值，销毁时传递需要销毁的互斥量地址值
• attr：传递即将创建的互斥量属性，没有特别需要指定的属性时传递NULL

从上述函数声明中可以看出，为了创建相当于锁系统的互斥量，需要声明如下pthread_mutex_t型变量

pthread_mutex_t mutex;

　　

该变量的地址将传递给pthread_mutex_init函数，用来保存操作系统创建的互斥量（锁系统）。调用pthread_mutex_destroy函数时同样需要该信息，如果不需要配置特殊的互斥量属性，则向第二个参数传递NULL时，可以利用PTHREAD_MUTEX_INITALIZER宏进行如下声明：

pthread_mutex_t mutex = PTHREAD_MUTEX_INITALIZER;

　　

推荐使用pthread_mutex_init函数进行初始化，因为通过宏进行初始化时很难发现发生的错误。接下来介绍利用互斥量锁住或释放临界区时使用的函数

#include <pthread.h>
int pthread_mutex_lock(pthread_mutex_t * mutex);
int pthread_mutex_unlock(pthread_mutex_t * mutex);
//成功时返回0，失败时返回其他值

　　

通过函数名我们很容易理解函数的作用，进入临界区前调用pthread_mutex_lock，调用该函数时，如果发现有其他线程已进入临界区，则pthread_mutex_lock函数会陷入阻塞，直到进入里面的线程调用pthread_mutex_unlock函数退出临界区为止

mutex.c

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>
#define NUM_THREAD 100

void *thread_inc(void *arg);
void *thread_des(void *arg);

long long num = 0;
pthread_mutex_t mutex;

int main(int argc, char *argv[])
{
    pthread_t thread_id[NUM_THREAD];
    int i;

    pthread_mutex_init(&mutex, NULL);

    for (i = 0; i < NUM_THREAD; i++)
    {
        if (i % 2)
            pthread_create(&(thread_id[i]), NULL, thread_inc, NULL);
        else
            pthread_create(&(thread_id[i]), NULL, thread_des, NULL);
    }

    for (i = 0; i < NUM_THREAD; i++)
        pthread_join(thread_id[i], NULL);

    printf("result: %lld 
", num);
    pthread_mutex_destroy(&mutex);
    return 0;
}

void *thread_inc(void *arg)
{
    int i;
    pthread_mutex_lock(&mutex);
    for (i = 0; i < 50000000; i++)
        num += 1;
    pthread_mutex_unlock(&mutex);
    return NULL;
}
void *thread_des(void *arg)
{
    int i;
    for (i = 0; i < 50000000; i++)
    {
        pthread_mutex_lock(&mutex);
        num -= 1;
        pthread_mutex_unlock(&mutex);
    }
    return NULL;
}

　　

• 第11行：声明了保存互斥量读取值的变量，之所以声明全局变量是因为thread_inc函数和thread_des函数都需要访问互斥量
• 第32行：销毁互斥量，不需要互斥量时应销毁
• 第39、42行：实际临界区只是第41行，但此处连同第40行的循环语句一起用作临界区，调用了lock、unlock函数
• 第50、52行：通过lock、unlock函数围住对应于临界区的第51行语句

编译mutex.c并运行

# gcc mutex.c -D_REENTRANT -o mutex -lpthread
# ./mutex 
result: 0 

　　

从运行结果可以看出，已解TCP/IP网络编程之多线程服务端的实现（一）中thread4.c的问题。但确认运行结果需要等待比较长的时间，因为互斥量lock、unlock函数的调用过程耗时较久。首先 ，分析一下thread_inc函数的同步过程

void *thread_inc(void *arg)
{
    int i;
    pthread_mutex_lock(&mutex);
    for (i = 0; i < 50000000; i++)
        num += 1;
    pthread_mutex_unlock(&mutex);
    return NULL;
}

　　

 以上临界区划分范围较大，但可以最大限度减少互斥量lock、unlock函数的调用次数，上述示例中，thread_des函数比thread_inc函数多调用49,999,999次互斥量lock、unlock函数；但是thread_inc相比于thread_des也不是全无缺点，因为当循环完成之前，不允许任何线程访问 　　

#include <semaphore.h>
int sem_init(sem_t * sem, int pshared, unsigned int value); 
int sem_destroy(sem_t * sem); 
//成功时返回0，失败时返回其他值

　　

• sem：创建信号量时传递保存信号量的变量地址值，销毁时传递需要销毁的信号量变量地址值
• pshared：传递其他值时，创建可由多个进程共享的信号量；传递0时，创建只允许一个进程内部使用的信号量，我们需要完成同一进程内的线程同步，故传0
• value：指定新创建的信号量初始值

上述函数的pshared参数超出我们关注的范围，故默认向其传递0。稍后讲解通过value参数初始化的信号量值是多少，接下来介绍信号量中相当于互斥量lock、unlock的函数

#include <semaphore.h>
int sem_wait(sem_t *sem); 
int sem_post(sem_t *sem); 

　　

•  sem：传递保存信号量读取值的变量地址值，传递给sem_post时信号量加1，传递给sem_wait时信号量减1

调用sem_init函数时，操作系统将创建信号量对象，此对象中记录着“信号量值”（整数）。该值在调用sem_post函数时加1，调用sem_wait函数时减1。但信号量的值不能小于0。因此，在信号量为0的情况下调用sem_wait函数时，调用函数的线程将进入阻塞状态，如果此时有其他线程函数调用sem_post函数，信号量的值将变为1，而原本阻塞的线程可以将该信号量重新减为0并跳出阻塞状态。实际上就是通过这种特性完成临界区的同步操作，可以通过如下形式同步临界区

sem_wait(&sem);
//临界区开始
//……
//临界区结束
sem_post(&sem);

　　

上述代码结构中，调用sem_wait函数进入临界区的线程在调用sem_post函数前不允许其他线程进入临界区。信号量的值在0和1之间跳转。因此，具有这种特性的机制称为“二进制信号量”

semaphore.c

#include <stdio.h>
#include <pthread.h>
#include <semaphore.h>

void *read(void *arg);
void *accu(void *arg);
static sem_t sem_one;
static sem_t sem_two;
static int num;

int main(int argc, char *argv[])
{
    pthread_t id_t1, id_t2;
    sem_init(&sem_one, 0, 0);
    sem_init(&sem_two, 0, 1);

    pthread_create(&id_t1, NULL, read, NULL);
    pthread_create(&id_t2, NULL, accu, NULL);

    pthread_join(id_t1, NULL);
    pthread_join(id_t2, NULL);

    sem_destroy(&sem_one);
    sem_destroy(&sem_two);
    return 0;
}

void *read(void *arg)
{
    int i;
    for (i = 0; i < 5; i++)
    {
        fputs("Input num: ", stdout);

        sem_wait(&sem_two);
        scanf("%d", &num);
        sem_post(&sem_one);
    }
    return NULL;
}
void *accu(void *arg)
{
    int sum = 0, i;
    for (i = 0; i < 5; i++)
    {
        sem_wait(&sem_one);
        sum += num;
        sem_post(&sem_two);
    }
    printf("Result: %d 
", sum);
    return NULL;
}

　　

• 第14、15行：生成两个信号量，一个信号量的值为0，另一个为1
• 第35、48行：利用信号量变量sem_two调用wait函数和post函数，这是为了防止在调用accu函数的线程还未取走数据的情况下，调用read函数的线程覆盖原值
• 第37、46行：利用信号量变量sem_one调用wait和post函数，这是为了防止调用read函数的线程写入新值之前，accu函数再取走旧的数据

编译semaphore.c并运行

# gcc semaphore.c -D_REENTRANT -o semaphore -lpthread
# ./semaphore 
Input num: 1
Input num: 2
Input num: 3
Input num: 4
Input num: 5
Result: 15 

　　

线程的销毁和多线程并发服务端的实现

Linux线程并不是在首次调用的线程main函数返回时自动销毁，所以用如下两种方法之一加以明确，否则由线程创建的内存空间将一直存在

• 调用pthread_join函数
• 调用pthread_detach函数

之前调用过pthread_join函数，调用该函数时，不仅会等待线程终止，还会引导线程销毁。但该函数的问题是，线程终止前，调用该函数的线程将进入阻塞状态。因此，通常通过如下函数调用引导线程销毁

#include <pthread.h>
int pthread_detach(pthread_t thread);//成功时返回0，失败时返回其他值

　　

• thread：终止的同时需要销毁的线程ID

调用上述函数不会引起线程终止或进入阻塞状态，可以通过该函数引导销毁线程创建的内存空间。调用该函数后不能针对相应线程调用pthread_join函数，需要注意一下。

多线程并发服务端的实现

chat_serv.c

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <pthread.h>

#define BUF_SIZE 100
#define MAX_CLNT 256

void *handle_clnt(void *arg);
void send_msg(char *msg, int len);
void error_handling(char *msg);

int clnt_cnt = 0;
int clnt_socks[MAX_CLNT];
pthread_mutex_t mutx;

int main(int argc, char *argv[])
{
    int serv_sock, clnt_sock;
    struct sockaddr_in serv_adr, clnt_adr;
    int clnt_adr_sz;
    pthread_t t_id;
    if (argc != 2) {
        printf("Usage : %s <port>
", argv[0]);
        exit(1);
    }

    pthread_mutex_init(&mutx, NULL);
    serv_sock = socket(PF_INET, SOCK_STREAM, 0);

    memset(&serv_adr, 0, sizeof(serv_adr));
    serv_adr.sin_family = AF_INET;
    serv_adr.sin_addr.s_addr = htonl(INADDR_ANY);
    serv_adr.sin_port = htons(atoi(argv[1]));

    if (bind(serv_sock, (struct sockaddr *)&serv_adr, sizeof(serv_adr)) == -1)
        error_handling("bind() error");
    if (listen(serv_sock, 5) == -1)
        error_handling("listen() error");

    while (1)
    {
        clnt_adr_sz = sizeof(clnt_adr);
        clnt_sock = accept(serv_sock, (struct sockaddr *)&clnt_adr, &clnt_adr_sz);

        pthread_mutex_lock(&mutx);
        clnt_socks[clnt_cnt++] = clnt_sock;
        pthread_mutex_unlock(&mutx);

        pthread_create(&t_id, NULL, handle_clnt, (void *)&clnt_sock);
        pthread_detach(t_id);
        printf("Connected client IP: %s 
", inet_ntoa(clnt_adr.sin_addr));
    }
    close(serv_sock);
    return 0;
}

void *handle_clnt(void *arg)
{
    int clnt_sock = *((int *)arg);
    int str_len = 0, i;
    char msg[BUF_SIZE];

    while ((str_len = read(clnt_sock, msg, sizeof(msg))) != 0)
        send_msg(msg, str_len);

    pthread_mutex_lock(&mutx);
    for (i = 0; i < clnt_cnt; i++) // remove disconnected client
    {
        if (clnt_sock == clnt_socks[i])
        {
            while (i++ < clnt_cnt - 1)
                clnt_socks[i] = clnt_socks[i + 1];
            break;
        }
    }
    clnt_cnt--;
    pthread_mutex_unlock(&mutx);
    close(clnt_sock);
    return NULL;
}
void send_msg(char *msg, int len) // send to all
{
    int i;
    pthread_mutex_lock(&mutx);
    for (i = 0; i < clnt_cnt; i++)
        write(clnt_socks[i], msg, len);
    pthread_mutex_unlock(&mutx);
}
void error_handling(char *msg)
{
    fputs(msg, stderr);
    fputc('
', stderr);
    exit(1);
}

　　

• 第17、18行：用于管理接入的客户端套接字的变量和数组，访问这两个变量的代码将构成临界区
• 第51行：每当有新连接时，将相关信息写入变量clnt_cnt和clnt_socks
• 第54行：创建线程向新接入的客户端提供服务，由该线程执行第62行定义的函数
• 第55行：调用pthread_detach函数从内存中完全销毁已终止的线程
• 第86行： 该函数负责向所连接的客户端发送信息

chat_clnt.c

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <pthread.h>

#define BUF_SIZE 100
#define NAME_SIZE 20

void *send_msg(void *arg);
void *recv_msg(void *arg);
void error_handling(char *msg);

char name[NAME_SIZE] = "[DEFAULT]";
char msg[BUF_SIZE];

int main(int argc, char *argv[])
{
    int sock;
    struct sockaddr_in serv_addr;
    pthread_t snd_thread, rcv_thread;
    void *thread_return;
    if (argc != 4) {
        printf("Usage : %s <IP> <port> <name>
", argv[0]);
        exit(1);
    }

    sprintf(name, "[%s]", argv[3]);
    sock = socket(PF_INET, SOCK_STREAM, 0);

    memset(&serv_addr, 0, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_addr.s_addr = inet_addr(argv[1]);
    serv_addr.sin_port = htons(atoi(argv[2]));

    if (connect(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) == -1)
        error_handling("connect() error");

    pthread_create(&snd_thread, NULL, send_msg, (void *)&sock);
    pthread_create(&rcv_thread, NULL, recv_msg, (void *)&sock);
    pthread_join(snd_thread, &thread_return);
    pthread_join(rcv_thread, &thread_return);
    close(sock);
    return 0;
}

void *send_msg(void *arg) // send thread main
{
    int sock = *((int *)arg);
    char name_msg[NAME_SIZE + BUF_SIZE];
    while (1)
    {
        fgets(msg, BUF_SIZE, stdin);
        if (!strcmp(msg, "q
") || !strcmp(msg, "Q
"))
        {
            close(sock);
            exit(0);
        }
        sprintf(name_msg, "%s %s", name, msg);
        write(sock, name_msg, strlen(name_msg));
    }
    return NULL;
}

void *recv_msg(void *arg) // read thread main
{
    int sock = *((int *)arg);
    char name_msg[NAME_SIZE + BUF_SIZE];
    int str_len;
    while (1)
    {
        str_len = read(sock, name_msg, NAME_SIZE + BUF_SIZE - 1);
        if (str_len == -1)
            return (void *)-1;
        name_msg[str_len] = 0;
        fputs(name_msg, stdout);
    }
    return NULL;
}

void error_handling(char *msg)
{
    fputs(msg, stderr);
    fputc('
', stderr);
    exit(1);
}

　　 

编译chat_serv.c并运行 

# gcc chat_serv.c -D_REENTRANT -o chat_serv -lpthread
# ./chat_serv 8500
Connected client IP: 127.0.0.1 
Connected client IP: 127.0.0.1 

　　

编译chat_clnt.c并运行

# gcc chat_clnt.c -D_REENTRANT -o chat_clnt -lpthread
# ./chat_clnt 127.0.0.1 8500 Sam
Hi everyone~
[Sam] Hi everyone~
[Amy] Hi Sam!
Hello Amy!
[Sam] Hello Amy!

　　

# ./chat_clnt 127.0.0.1 8500 Amy
[Sam] Hi everyone~
Hi Sam!
[Amy] Hi Sam!
[Sam] Hello Amy!