Linux进程同步

目录

Linux进程同步调研...
1

背景...
1

Futex.
1

进/线程利用futex同步...
2

性能对比...
3

参考文档...
5

背景

超级负载均衡需要在线程/进程之前共享数据，这些数据需要支持多写多读，因此需要考虑同步。

本文调研了linux下的mutex与semaphone。

目前线上2.6.9内核下，mutex与semaphone都是基于Futex。

Futex

Futex是由HubertusFranke（IBM Thomas J. Watson研究中心），Matthew Kirkwood，Ingo Molnar (Red Hat)和Rusty Russell (IBM Linux科技中心)设计并维护的。最初的Futex的支持是从Linux2.5.7开始的，但是以上述语义有些不同，当前的语义是从Linux2.5.40获得的。

Futex 由一块能够被多个进程共享的内存空间（一个对齐后的整型变量）组成；这个整型变量的值能够通过汇编语言调用CPU提供的原子操作指令来增加或减少，并且一个进程可以等待直到那个值变成正数。Futex 的操作几乎全部在应用程序空间完成；只有当操作结果不一致从而需要仲裁时，才需要进入操作系统内核空间执行。这种机制允许使用 futex 的锁定原语有非常高的执行效率：由于绝大多数的操作并不需要在多个进程之间进行仲裁，所以绝大多数操作都可以在应用程序空间执行，而不需要使用（相对高代价的）内核系统调用。

Futex是一种用户态和内核态混合机制，所以需要两个部分合作完成，linux上提供了sys_futex系统调用，对进程竞争情况下的同步处理提供支持。

其原型和系统调用号为

#include <linux/futex.h>
#include <sys/time.h>
int futex (int *uaddr, int op,int val, const struct timespec *timeout,int *uaddr2, int val3);
#define __NR_futex 240
虽然参数有点长，其实常用的就是前面三个，后面的timeout大家都能理解，其他的也常被ignore。

uaddr就是用户态下共享内存的地址，里面存放的是一个对齐的整型计数器。

op存放着操作类型。定义的有5中，这里我简单的介绍一下两种，剩下的感兴趣的自己去man futex

FUTEX_WAIT: 原子性的检查uaddr中计数器的值是否为val,如果是则让进程休眠，直到FUTEX_WAKE或者超时(time-out)。也就是把进程挂到uaddr相对应的等待队列上去。

FUTEX_WAKE: 最多唤醒val个等待在uaddr上进程。

在futex的系统调用中，FUTEX_WAIT和FUTEX_WAKE只是用来挂起或者唤醒进程，这部分工作只能在内核态下完成。futex同步机制还包括用户态下的操作（区分futex同步机制和futex系统调用）。

进/线程利用futex同步

进程或者线程都可以利用futex来进行同步。

对于线程，情况比较简单，因为线程共享虚拟内存空间，虚拟地址就可以唯一的标识出futex变量，即线程用同样的虚拟地址来访问futex变量。

对于进程，情况相对复杂，因为进程有独立的虚拟内存空间，只有通过mmap()让它们共享一段地址空间来使用futex变量。每个进程用来访问futex的虚拟地址可以是不一样的，只要系统知道所有的这些虚拟地址都映射到同一个物理内存地址，并用物理内存地址来唯一标识futex变量。

mutex的实现原理：

pthread_mutex_lock:
atomic_dec(pthread_mutex_t.value);
if(pthread_mutex_t.value!=0)
futex(WAIT)
else
success

pthread_mutex_unlock:
atomic_inc(pthread_mutex_t.value);
if(pthread_mutex_t.value!=1)
futex(WAKEUP)
else
success

信号量sem_t的实现原理：

sem_wait(sem_t *sem)
{
for (;;) {

if(atomic_decrement_if_positive(sem->count))
break;

futex_wait(&sem->count, 0)
}
}

sem_post(sem_t *sem)
{
n = atomic_increment(sem->count);
// Pass the new value of sem->count
futex_wake(&sem->count, n + 1);
}

对比，pthread_mutex_unlock()和sem_post()的实现，我们发现一个不同点，sem_post()无论如何都会调用futex_wake()，进行系统调用。但是pthread_mutex_unlock()却符合futex的初衷，只有在需要仲裁的时候才调用futex_wake()。

通过strace跟踪也能印证，使用semaphone，就算没有出现互斥，在unlock，一定会调用futex(0x7fbffff490,FUTEX_WAKE, 1)。

如下所示列出了unlock附近的strace结果：

mutex的strace监控：

……
brk(0) = 0x502000
brk(0x523000) = 0x523000
exit_group(0) = ?

semaphone的strace监控：

……
brk(0) = 0x502000
brk(0x523000) = 0x523000
futex(0x7fbffff490,FUTEX_WAKE, 1) = 0
exit_group(0) = ?

性能对比

性能测试说明：

构造了三个临界区（按照我们的应用情况）：

int lock_test1() { // 执行1条语句
TbAutoLock _lock(lock);
int id = g_shm_id;
return id;
}

int lock_test2() { // 执行1000条语句
TbAutoLock _lock(lock);
int sum = 0;
for(int i = 0; i < 1000; i++) {
sum += i;
}
return sum;
}

long long lock_test3() { // 执行10万条语句
TbAutoLock _lock(lock);
long long sum = 0;
for(int i = 0; i < 100000; i++) {
sum += i;
}
return sum;
}
然后：

for(int i=0; i<100; i++) {
for(int j=0; j<1000; j++) {
lock_test1();
}
for(int j=0; j<100; j++) {
lock_test2();
}
for(int j=0; j<1000; j++) {
lock_test1();
}
for(int j=0; j<10; j++) {
lock_test3();
}
}
启动100个进程，测试这三种临界区。

执行时间 & 对CPU的消耗如下表：

	Mutex	Semaphone
执行时间	34575129 us 约34s	43529735 us 约43s
cpu_id	80	57
cpu_us	13	11
cpu_sy	7	32

可以看出：

l Mutex性能更优，cpu_sy明显低于Semaphone，这和之前结论一致；

l 综合cpu占用，在我们的应用下，Mutex的性能是Semaphone的两倍。

关于pthread_mutex用于进程模型

pthread_mutex用于进程模型的条件：

l 所在共享内存上分配；

l 设置锁的状态为：PTHREAD_PROCESS_SHARED

后面测试代码中有完整示例

参考文档

Linux中的线程同步机制(一二三)：

http://blog.csdn.net/Javadino/archive/2008/09/06/2891385.aspx

http://blog.csdn.net/Javadino/archive/2008/09/06/2891388.aspx

http://blog.csdn.net/Javadino/archive/2008/09/06/2891399.aspx

关于信号量与线程互斥锁的区别与实现：

http://blog.csdn.net/borefo/archive/2009/11/21/4840284.aspx

相关测试代码

class TbLock {
public:
static const char LOCK_TYPE_THREAD = 1;
static const char LOCK_TYPE_PROCESS = 2;
protected:
char _type;
public:
virtual ~TbLock() {};
virtual int lock() {
return 0;
};
virtual int unlock() {
return 0;
};
};

class TbLockMutex : public TbLock{
private:
pthread_mutex_t *_mutex_p;
public:
TbLockMutex(pthread_mutex_t *mutex_p, chartype) {
_mutex_p = mutex_p;
_type = type;
if (type == LOCK_TYPE_PROCESS) {
pthread_mutexattr_t mat;
pthread_mutexattr_init(&mat);
pthread_mutexattr_setpshared(&mat,PTHREAD_PROCESS_SHARED);
pthread_mutex_init(_mutex_p,&mat);
} else {
pthread_mutex_init(_mutex_p, NULL);
}
}
~TbLockMutex() {
pthread_mutex_destroy(_mutex_p);
}
int lock() {
return pthread_mutex_lock(_mutex_p);
}
int unlock() {
return pthread_mutex_unlock(_mutex_p);
}
};

class TbLockSema : public TbLock{
private:
sem_t * _sem_p;
public:
TbLockSema(sem_t * sem_p, char type) {
_sem_p = sem_p;
_type = type;
if (type == LOCK_TYPE_PROCESS) {
sem_init(_sem_p, 1, 1);
} else {
sem_init(_sem_p, 0, 1);
}
}
~TbLockSema() {
sem_destroy(_sem_p);
}
int lock() {
return sem_wait(_sem_p);
}
int unlock() {
return sem_post(_sem_p);
}
};

class TbLockFactory {
private:
char * _lockBuf;
uint _lockBufSize;
uint _curLockNum;
char _type;
static const uint LOCK_SIZE = sizeof(pthread_mutex_t)> sizeof(sem_t) ? sizeof(pthread_mutex_t) : sizeof(sem_t);

public:

TbLockFactory(char * lockBuf, intlockBufSize, char type):_lockBuf(lockBuf), _lockBufSize(lockBufSize),_type(type) {
_curLockNum = 0;
}

TbLock * getMutexLock() {
pthread_mutex_t *mutex = NULL;
if ((_curLockNum+1)*LOCK_SIZE <_lockBufSize) {
mutex = (pthread_mutex_t*)(_lockBuf+ _curLockNum*LOCK_SIZE);
_curLockNum++;
return new TbLockMutex(mutex,_type);
}
return NULL;
}

TbLock * getSemaLock() {
sem_t * sem = NULL;
if ((_curLockNum+1)*LOCK_SIZE <_lockBufSize) {
sem = (sem_t *)(_lockBuf +_curLockNum*LOCK_SIZE);
_curLockNum++;
return new TbLockSema(sem, _type);
}
return NULL;
}
};

class TbAutoLock {
TbLock * _lock;
public:
TbAutoLock(TbLock * lock){
_lock = lock;
_lock->lock();
}
~TbAutoLock(){
_lock->unlock();
}
};