线程存在的问题和临界区
上一章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 \n", 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 \n", 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>\n", 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 \n", 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('\n', 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>\n", 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\n") || !strcmp(msg, "Q\n"))
{
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('\n', 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!