文章目录
进程的三种状态：运行，就绪，阻塞
锁
互斥量
在访问共享资源前对互斥量进行加锁,在访问完成后释放互斥量.
对互斥量进行加锁后,任何其他试图再次对互斥量加锁的线程都会被阻塞直到当前线程释放该互斥锁.
如果释放互斥量时有一个以上的线程阻塞,那么所有该锁上的阻塞线程都会变成可执行状态,第一个运行的线程就可以对互斥量加锁,其他线程就会看到互斥量依然锁着,只能回去再次等待它重新变为可用.
这种方式下,每次只有一个线程可以向前执行.
//初始化
1.pthread_mutex_init
2.PTHREAD_MUTEX_INITIALIZER (只适用于静态分配的互斥量)
struct foo {
int f_count;
pthread_mutex_t f_lock;
int f_id;
/* ... more stuff here ... */
};
struct foo *
foo_alloc(int id) /* allocate the object */
{
struct foo *fp;
if ((fp = malloc(sizeof(struct foo))) != NULL) {
fp->f_count = 1;
fp->f_id = id;
if (pthread_mutex_init(&fp->f_lock, NULL) != 0) {
free(fp);
return(NULL);
}
/* ... continue initialization ... */
}
return(fp);
}
void
foo_hold(struct foo *fp) /* add a reference to the object */
{
pthread_mutex_lock(&fp->f_lock);
fp->f_count++;
pthread_mutex_unlock(&fp->f_lock);
}
在运用的过程中需要避免死锁（1.同一个线程加锁两次！2.互相请求）
锁的时候一直Ａ－＞Ｂ（相同的顺序加锁），不要有相反的顺序出现
pthread_mutex_timedlock() 超时返回ETIMEOUT
int pthread_mutex_timedlock(pthread_mutex_t *restrict mutex,
const struct timespec *restrict abstime);
读写锁
-
读写锁(reader-writer lock)与互斥量类似,不过读写锁允许更高的并行性.
-
互斥量要么是锁住状态,要么就是不加锁状态,而且一次只有一个线程可以对其加锁.
-
读写锁可以有3中状态:读
模式下加锁,写模式下加锁,不加锁状态.一次只有一个线程可以占有写模式的读写锁,但是多个线程可以同时占有读模式的读写锁. -
当读写锁是写加锁状态时,在这个锁被解锁之前,所有试图对这个锁加锁的线程都会被阻塞.
-
当读写锁在读加锁状态时,所有试图以读模式对它进行加锁的线程都可以得以访问权,但是任何希望以写模式对此锁进行加锁的线程都会阻塞,直到所有线程释放他们的读锁为止.
适用于读＞写的场景
struct job {
struct job *j_next;
struct job *j_prev;
pthread_t j_id; /* tells which thread handles this job */
/* ... more stuff here ... */
};
struct queue {
struct job *q_head;
struct job *q_tail;
pthread_rwlock_t q_lock;
};
/*
* Initialize a queue.
*/
int
queue_init(struct queue *qp)
{
int err;
qp->q_head = NULL;
qp->q_tail = NULL;
err = pthread_rwlock_init(&qp->q_lock, NULL);
if (err != 0)
return(err);
/* ... continue initialization ... */
return(0);
}
/*
* Insert a job at the head of the queue.
*/
void
job_insert(struct queue *qp, struct job *jp)
{
pthread_rwlock_wrlock(&qp->q_lock); //写锁
jp->j_next = qp->q_head;
jp->j_prev = NULL;
if (qp->q_head != NULL)
qp->q_head->j_prev = jp;
else
qp->q_tail = jp; /* list was empty */
qp->q_head = jp;
pthread_rwlock_unlock(&qp->q_lock); //解除写锁
}
/*
* Append a job on the tail of the queue.
*/
void
job_append(struct queue *qp, struct job *jp)
{
pthread_rwlock_wrlock(&qp->q_lock);
jp->j_next = NULL;
jp->j_prev = qp->q_tail;
if (qp->q_tail != NULL)
qp->q_tail->j_next = jp;
else
qp->q_head = jp; /* list was empty */
qp->q_tail = jp;
pthread_rwlock_unlock(&qp->q_lock);
}
/*
* Remove the given job from a queue.
*/
void
job_remove(struct queue *qp, struct job *jp)
{
pthread_rwlock_wrlock(&qp->q_lock);
if (jp == qp->q_head) {
qp->q_head = jp->j_next;
if (qp->q_tail == jp)
qp->q_tail = NULL;
else
jp->j_next->j_prev = jp->j_prev;
} else if (jp == qp->q_tail) {
qp->q_tail = jp->j_prev;
jp->j_prev->j_next = jp->j_next;
} else {
jp->j_prev->j_next = jp->j_next;
jp->j_next->j_prev = jp->j_prev;
}
pthread_rwlock_unlock(&qp->q_lock);
}
/*
* Find a job for the given thread ID.
*/
struct job *
job_find(struct queue *qp, pthread_t id)
{
struct job *jp;
if (pthread_rwlock_rdlock(&qp->q_lock) != 0) //读锁
return(NULL);
for (jp = qp->q_head; jp != NULL; jp = jp->j_next)
if (pthread_equal(jp->j_id, id))
break;
pthread_rwlock_unlock(&qp->q_lock);
return(jp);
}
超时的读写锁：
int pthread_rwlock_timedrdlock(pthread_rwlock_t *restrict rwlock,
const struct timespec *restrict tsptr);
int pthread_rwlock_timedwrlock(pthread_rwlock_t *restrict rwlock,
const struct timespec *restrict tsptr);
//超时返回`ETIMEOUT`
自旋锁
自旋锁和互斥量类似，但是它不能通过睡眠使进程阻塞，而是在获取锁之前一直处于忙等状态（整个时间片可能都在等这个锁）．
- 互斥锁：线程会从sleep（加锁）-> 上下文的切换,调度—>running（解锁）
- 自旋锁：线程在整个时间片中一直是running(加锁——>解锁)，死循环检测锁的标志位，机制不复杂。
适用情况：锁被持有的时间短，而且线程并不希望在重新调度上花费太多的成本．
int pthread_spin_init(pthread_spinlock_t *lock, int pshared);
int pthread_spin_destory(pthread_spinlock_t *lock);
// 返回值:成功返回0;否则返回错误编号
// pshared: 表示进程共享属性,表明自旋锁是如何获取的
//PTHREAD_PROCESS_SHARED :自旋锁能被可以访问锁底层内存的线程所获取,即便那些线程属于不同的进程.
// PTHREAD_PROCESS_PRIVATE: 自旋锁只能被初始化该锁的进程内部的线程访问.
int pthread_spin_lock(pthread_spinlock_t *lock);
int ptrhead_spin_trylock(pthread_spinlock_t *lock);
int pthread_spin_unlock(pthread_spinlock_t *lock);
// 返回值：成功返回０；否则返回错误编号
条件变量
利用线程间共享全局变量进行同步的机制，因为是全局变量，所以要与互斥锁搭配使用
- 条件本身是由互斥量保护的．线程在改变条件状态之前必须首先锁住互斥量．
- 使用条件变量之前，必须先对它初始化．
#include <pthread.h>
struct msg {
struct msg *m_next;
/* ... more stuff here ... */
};
struct msg *workq;
pthread_cond_t qready = PTHREAD_COND_INITIALIZER;
pthread_mutex_t qlock = PTHREAD_MUTEX_INITIALIZER;
void
process_msg(void)
{
struct msg *mp;
for (;;) {
pthread_mutex_lock(&qlock); //先锁住，然后
while (workq == NULL)
pthread_cond_wait(&qready, &qlock);//在等待条件变量发生时，会解锁
//pthread_cond_wait函数返回又会解锁
mp = workq;
workq = mp->m_next;
pthread_mutex_unlock(&qlock);
/* now process the message mp */
}
}
void
pushqueue_msg(struct msg *mp)
{
pthread_mutex_lock(&qlock);
mp->m_next = workq;
workq = mp;
pthread_mutex_unlock(&qlock);
pthread_cond_signal(&qready);
}
屏障
-
屏障（barrier）是用户协调多个线程并行工作的同步机制.屏障允许每个线程等待,直到所有的合作线程都达到某一点,然后从改点继续执行.
-
pthread_join 就是一种屏障,允许一个线程等待,直到另一个线程退出.
-
屏障（barrier）的概念更广,它们允许任意数量的线程等待,知道所有的线程完成处理工作,而线程不需要退出.所有线程到达屏障后可以接着工作.
对于任一线程,pthread_barrier_wait 函数返回了 PTHREAD_BARRIER_SERIAL_THREAD. 剩余的线程看到的返回值为0.这使得一个线程可以作为主线程,它可以工作在其他所有线程已完成的工作结果上.
#include "apue.h"
#include <pthread.h>
#include <limits.h>
#include <sys/time.h>
#define NTHR 8 /* number of threads */
#define NUMNUM 8000000L /* number of numbers to sort */
#define TNUM (NUMNUM/NTHR) /* number to sort per thread */
long nums[NUMNUM];
long snums[NUMNUM];
pthread_barrier_t b;
#ifdef SOLARIS
#define heapsort qsort
#else
extern int heapsort(void *, size_t, size_t,
int (*)(const void *, const void *));
#endif
/*
* Compare two long integers (helper function for heapsort)
*/
int
complong(const void *arg1, const void *arg2)
{
long l1 = *(long *)arg1;
long l2 = *(long *)arg2;
if (l1 == l2)
return 0;
else if (l1 < l2)
return -1;
else
return 1;
}
/*
* Worker thread to sort a portion of the set of numbers.
*/
void *
thr_fn(void *arg)
{
long idx = (long)arg;
heapsort(&nums[idx], TNUM, sizeof(long), complong);
pthread_barrier_wait(&b); //未到达8就休眠，到达就唤醒所有线程
/*
* Go off and perform more work ...
*/
return((void *)0);
}
/*
* Merge the results of the individual sorted ranges.
*/
void
merge()
{
long idx[NTHR];
long i, minidx, sidx, num;
for (i = 0; i < NTHR; i++)
idx[i] = i * TNUM;
for (sidx = 0; sidx < NUMNUM; sidx++) {
num = LONG_MAX;
for (i = 0; i < NTHR; i++) {
if ((idx[i] < (i+1)*TNUM) && (nums[idx[i]] < num)) {
num = nums[idx[i]];
minidx = i;
}
}
snums[sidx] = nums[idx[minidx]];
idx[minidx]++;
}
}
int
main()
{
unsigned long i;
struct timeval start, end;
long long startusec, endusec;
double elapsed;
int err;
pthread_t tid;
/*
* Create the initial set of numbers to sort.
*/
srandom(1);
for (i = 0; i < NUMNUM; i++)
nums[i] = random();
/*
* Create 8 threads to sort the numbers.
*/
gettimeofday(&start, NULL);
pthread_barrier_init(&b, NULL, NTHR+1);
//设定线程数目＝8，也就是说只有达到8+1个线程(包含了主线程)，所有线程才会唤醒
for (i = 0; i < NTHR; i++) {
err = pthread_create(&tid, NULL, thr_fn, (void *)(i * TNUM));
if (err != 0)
err_exit(err, "can't create thread");
}
pthread_barrier_wait(&b);
merge();
gettimeofday(&end, NULL);
/*
* Print the sorted list.
*/
startusec = start.tv_sec * 1000000 + start.tv_usec;
endusec = end.tv_sec * 1000000 + end.tv_usec;
elapsed = (double)(endusec - startusec) / 1000000.0;
printf("sort took %.4f seconds\n", elapsed);
for (i = 0; i < NUMNUM; i++)
printf("%ld\n", snums[i]);
exit(0);
}
异步信号
多个线程接受异步信号，则只有一个被选中．
// pthread_kill - send a signal to a thread
int pthread_kill(pthread_t thread, int sig);
// sigwait - wait for a signal
int sigwait(const sigset_t *set, int *sig);//等待信号集中的任意信号到达
`可以用来统计信号到达的次数`
关于信号量
一种软件中断，提供了一种处理异步事件的方法，也是进程间唯一的异步通信方式
1.内核信号量，由内核控制路径使用
Linux内核的信号量在概念和原理上与用户态的System V的IPC机制信号量是一样的，但是它绝不可能在内核之外使用，因此它与System V的IPC机制信号量毫不相干。
信号量在创建时需要设置一个初始值，表示同时可以有几个任务可以访问该信号量保护的共享资源，初始值为1就变成互斥锁（Mutex），即同时只能有一个任务可以访问信号量保护的共享资源。
一个任务要想访问共享资源，首先必须得到信号量，获取信号量的操作将把信号量的值减1，若当前信号量的值为负数，表明无法获得信号量，该任务必须挂起在该信号量的等待队列等待该信号量可用；若当前信号量的值为非负数，表示可以获得信号量，因而可以立刻访问被该信号量保护的共享资源。当任务访问完被信号量保护的共享资源后，必须释放信号量，释放信号量通过把信号量的值加1实现，如果信号量的值为非正数，表明有任务等待当前信号量，因此它也唤醒所有等待该信号量的任务
内核信号量的相关函数
初始化：
void sema_init (struct semaphore *sem, int val);
void init_MUTEX (struct semaphore *sem); //将sem的值置为1，表示资源空闲
void init_MUTEX_LOCKED (struct semaphore *sem); //将sem的值置为0，表示资源忙
申请内核信号量所保护的资源：
void down(struct semaphore * sem); // 可引起睡眠
int down_interruptible(struct semaphore * sem); // down_interruptible能被信号打断
int down_trylock(struct semaphore * sem); // 非阻塞函数，不会睡眠。无法锁定资源则马上返回
释放内核信号量所保护的资源：
void up(struct semaphore * sem);
2.用户态信号量分为两种，一种为POSIX,另一种为 SYSTEM V
POSIX 信号量与SYSTEM V信号量的对比
- 1.对POSIX来说，信号量是个非负整数。常用于线程间同步。而SYSTEM V信号量则是一个或多个信号量的集合，它对应的是一个信号量结构体，这个结构体是为SYSTEM V IPC服务的，信号量只不过是它的一部分。常用于进程间同步。
- 2．POSIX信号量的引用头文件是“<semaphore.h>”，而SYSTEM V信号量的引用头文件是“<sys/sem.h>”。
- 3．从使用的角度，System V信号量是复杂的，而Posix信号量是简单。比如，POSIX信号量的创建和初始化或PV操作就很非常方便。
POSIX信号量详解
分为两种：有名信号量和无名信号量
区别：在于创建和销毁的形式上，其他一样
- 无名信号量：信号量的值只存在于内存中，常用于多线程间的同步，同时也用于相关进程间的同步．意思就是
单个进程间使用 - 有名信号量：有名信号量一般保存在/dev/shm/ 目录下。像文件一样存储在文件系统中。
多个进程间通信使用
创建和销毁：
- 无名信号量：像变量一样使用sem_init，销毁：sem_destory
- 有名信号量：sem_open ， 销毁：sem_close
具体使用例子：
-
sem_wait(或sem_trywait)相当于P操作，即申请资源。
-
sem_post相当于V操作，释放资源。
-
无名：
使用无名信号量在同一进程中对线程进行同步
#include <stdio.h>
#include <pthread.h>
#include <semaphore.h>//信号量头文件
#include <unistd.h>
sem_t sem_id1,sem_id2;
int global_var;
void * pt1_func(void *argc)
{
while(1){
sem_wait(&sem_id1);
global_var =0;
printf("func 1 global_var = %d\n",global_var);
sleep(2);
sem_post(&sem_id2);
}
}
void *pt2_func(void *argc)
{
while(1){
sem_wait(&sem_id2);//Ｐ操作 -1，如果sem ==0,则阻塞，直到sem>0,然后减一并返回
global_var = 1;
printf("func 2 global_var = %d\n",global_var);
sleep(2);
sem_post(&sem_id1);//Ｖ操作 +1
}
}
int main(void)
{
pthread_t pt1;
pthread_t pt2;
/*
int sem_init(sem_t *sem, int pshared, unsigned int value);
1. pshared > 0 ,表示用于多个相关进程间的同步， pshared ==0,同一进程的多个线程间的同步
2. value 信号量初始值
*/
sem_init(&sem_id1,0,1); //初始植为1
sem_init(&sem_id2,0,0); //初始植为0
pthread_create(&pt1,NULL,pt1_func,NULL);
pthread_create(&pt2,NULL,pt2_func,NULL);
pthread_join(pt1,NULL);
pthread_join(pt2,NULL);
printf("main..\n");
return 0;
}
- 有名：
使用有名信号量的同步两个不相关的进程：
进程1
//程序一
/*NameSemWrite_1.c*/
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <pthread.h>
#include <semaphore.h>
#define FILENAME "write_name_sem_file"
int main()
{
sem_t *sem_1 = NULL;
sem_t *sem_2 = NULL;
sem_1 = sem_open("write_name_sem_1", O_CREAT | O_RDWR, 0666, 1); //信号量值为 1
sem_2 = sem_open("write_name_sem_2", O_CREAT | O_RDWR, 0666, 0); //信号量值为 0
if ((sem_1 == SEM_FAILED) | (sem_2 == SEM_FAILED))
{
perror("sem_open");
exit(-1);
}
int count = 0;
int fd = open(FILENAME, O_RDWR | O_CREAT | O_APPEND, 0777);
if (fd < 0)
{
perror("open");
}
char write_buf[] = "1";
int i = 0;
while (1)
{
/*信号量减一，P 操作*/
sem_wait(sem_1);
for (i = 0; i < 5; i++)
{
if (write(fd, write_buf, sizeof(write_buf)))
{
perror("write");
}
count++;
}
printf("in child....and count = %d\n", count);
/*信号量加一，V 操作*/
sem_post(sem_2);
sleep(2);
}
sem_close(sem_1);
sem_unlink("write_name_sem_1");
sem_close(sem_2);
sem_unlink("write_name_sem_2");
return 0;
}
进程2
//程序二
/* NameSemWrite_2.c*/
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <pthread.h>
#include <semaphore.h>
#define FILENAME "write_name_sem_file"
int main()
{
int count = 0;
char write_buf[] = "2";
int i = 0;
sem_t *sem_1 = NULL;
sem_t *sem_2 = NULL;
sem_1 = sem_open("write_name_sem_1", O_CREAT | O_RDWR, 0666, 1); //信号量值为 1
sem_2 = sem_open("write_name_sem_2", O_CREAT | O_RDWR, 0666, 0); //信号量值为 1
if ((sem_1 == SEM_FAILED) | (sem_2 == SEM_FAILED))
{
perror("sem_open");
exit(-1);
}
int fd = open(FILENAME, O_RDWR | O_CREAT | O_APPEND, 0777);
if (fd < 0)
{
perror("open");
}
while (1)
{
/*信号量减一，P 操作*/
sem_wait(sem_2);
for (i = 0; i < 5; i++)
{
if (write(fd, write_buf, sizeof(write_buf)))
{
perror("write");
}
count++;
}
printf("in father.... count = %d\n", count);
/*信号量加一，V 操作*/
sem_post(sem_1);
sleep(2);
}
return 0;
}
参考：
https://blog.csdn.net/u014530704/article/details/77387536
https://blog.csdn.net/weed_hz/article/details/8965733