线程同步的"手段"

Love The Way You Lie 2022-02-28 01:06 134阅读 0赞

### 文章目录 ###

*  进程的三种状态：运行，就绪，阻塞
 *  锁
 *   *   *  互斥量
         *  读写锁
         *  自旋锁
 *  条件变量
 *  屏障
 *  异步信号
 *  关于信号量
 *   *  1.内核信号量，由内核控制路径使用
     *  2.用户态信号量分为两种，一种为POSIX,另一种为 SYSTEM V
     *   *   *  POSIX 信号量与SYSTEM V信号量的对比
 *  POSIX信号量详解
 *   *   *  区别：在于创建和销毁的形式上，其他一样

# 进程的三种状态：运行，就绪，阻塞 #

![在这里插入图片描述][20190323213414519.png]

# 锁 #

### 互斥量 ###

在访问共享资源前对互斥量进行加锁,在访问完成后释放互斥量.

对互斥量进行加锁后,任何其他试图再次对互斥量加锁的线程都会被`阻塞直到当前线程释放该互斥锁.`

如果释放互斥量时有一个以上的线程阻塞,那么所有该锁上的阻塞线程都会变成可执行状态,第一个运行的线程就可以对互斥量加锁,其他线程就会看到互斥量依然锁着,只能回去再次等待它重新变为可用.

这种方式下,每次只有一个线程可以向前执行.

//初始化
    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_u014530704_article_details_77387536]　  
[https://blog.csdn.net/weed\_hz/article/details/8965733][https_blog.csdn.net_weed_hz_article_details_8965733]

[20190323213414519.png]: /images/20220228/3661f6e7f70d40368fecbfe06bd85a1c.png
[https_blog.csdn.net_u014530704_article_details_77387536]: https://blog.csdn.net/u014530704/article/details/77387536
[https_blog.csdn.net_weed_hz_article_details_8965733]: https://blog.csdn.net/weed_hz/article/details/8965733