多进程多线程环境下的同步机制性能测试

进行多进程、多线程编程时，难免会涉及到对同一块数据的访问，这时候我们需要使用各种锁来保证数据操作的正确性。正如《UNP》卷二所讲的一样，用于同步的锁主要分为两种：一、用于进程间同步的，有记录锁(record lock，对应于fcntl，还有flock,lockf等)；有信号量(system V semaphore, posix semaphore)；二、用于线程间同步的，有读写锁(pthread_rwlock_t)；互斥量(pthread_mutex_t)。对于支持互斥量和读写锁进程间共享的Linux来说，还可将互斥量与读写锁用于进程间的共享；而用于进程间同步的信号量，本身也可以用于线程间的同步，对于记录锁，情况有稍微复杂一些。由于Linux 2.4对线程的支持是通过进程模拟出来的，所以在2.4中，记录锁也可以用于线程间的同步，而在2.6中，线程不再需要通过进程来模拟，此时记录锁不能用于线程间的同步，即同一个进程中的不同线程可以同时获取写锁。特别需要强调的是，对于进程间同步的记录锁、信号量，如果由于进程、线程异常终止或退出而没有及时释放锁、信号量的话，内核会帮我们释放掉，这样的话就可以避免死锁。至于线程间同步的读写锁和互斥量，则需要我们自己去保证。
《UNP》卷二的附录A已经给出各种同步机制的性能数据，仍不够详细，特别是对于高并发情况下的测试，并且上面给出的例子并没有涉及到进程、线程调度，即当运行到某个线程或进程时，如果不切换出去，那这时候是保证只有该进程或线程拥有锁/信号量，为了同《UNP》卷二一致，测试代码也没有让线程sleep一小段时间从而使其发生调度，以下给出测试的源码及测试平台数据，不对之处敬请指出！
//---------count.h 计时器头文件
#ifndef COUNT_H
#define COUNT_H
#include <stdint.h>
class count
{
uint64_t m_beginTime;
uint64_t m_endTime;
public:
count();
~count();
};
#endif

//---------count.cpp 计时器实现,对于AMD，使用gettimeofday，对于intel，则使用rdtsc，暂不考虑其它硬件平台
#include "count.h"
#include <stdio.h>
#include <stdint.h>

#ifdef AMD
#include <sys/time.h>
#endif

count::count()
{
#ifndef AMD
uint32_t low = 0;
uint32_t high = 0;
asm volatile("rdtsc; mov %%edx, %0; mov %%eax, %1" : "=r"(low), "=r"(high));
m_beginTime = (uint64_t)high * (uint64_t)0xFFFFFFFF + high;
m_beginTime += low;
#else
struct timeval val = {0};
gettimeofday(&val, 0);
m_beginTime = (uint64_t)val.tv_sec * 1000000 + val.tv_usec;
#endif
}

count::~count()
{
#ifndef AMD
uint32_t low = 0;
uint32_t high = 0;
asm volatile("rdtsc; mov %%edx, %0; mov %%eax, %1" : "=r"(low), "=r"(high));
m_endTime= (uint64_t)high * (uint64_t)0xFFFFFFFF + high;
m_endTime += low;
#else
struct timeval val = {0};
gettimeofday(&val, 0);
m_endTime = (uint64_t)val.tv_sec * 1000000 + val.tv_usec;
#endif
printf("diff = %llu/n", m_endTime - m_beginTime);
}

//---------synchronization_performance_profile.cpp 测试程序
#include "count.h"
#include <pthread.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include <sys/types.h>
#include <sys/file.h>
#include <stdio.h>
#include <semaphore.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>

#define DBG(message) printf(#message " failed, reason = %s, file : %s, line :%d/n", strerror(errno), __FILE__, __LINE__)

struct sharedMemory{
int count; // 上锁后对读数器增加或减少
pthread_mutex_t mutex; // posix mutex
};

int g_lock_type = -1; // lock type
sharedMemory *g_critical_zone = 0;
int g_lockfd = -1; // for fcntl
int g_semid = -1; // for semaphore
int g_count = -1; // oper times

union semun {
int val; /* Value for SETVAL */
struct semid_ds *buf; /* Buffer for IPC_STAT, IPC_SET */
unsigned short *array; /* Array for GETALL, SETALL */
struct seminfo *__buf; /* Buffer for IPC_INFO (Linux specific) */
};

int lock()
{
static struct flock write_lock = {F_WRLCK, SEEK_SET, 0, 0};
switch(g_lock_type)
{
case 1:
{
int ret = fcntl(g_lockfd, F_SETLKW, &write_lock);
while (ret != 0) {
ret = fcntl(g_lockfd, F_SETLKW, &write_lock);
}
if (ret != 0) {
DBG(fcntl);
}
return ret;
}
case 2:
{
return pthread_mutex_lock(&g_critical_zone->mutex);
}
case 3:
{
static struct sembuf buf = {0, -1, SEM_UNDO};
return semop(g_semid, &buf, 1);
}
case 4:
{
return pthread_rwlock_wrlock(&g_critical_zone->rwlock);
}
default:
{
printf("unknown type = %d/n", g_type);
return -1;
break;
}
}
}

int unlock()
{
static struct flock un_lock = {F_UNLCK, SEEK_SET, 0, 0};
switch(g_lock_type)
{
case 1:
{
return fcntl(g_lockfd, F_SETLKW, &un_lock);
}
case 2:
{
return pthread_mutex_unlock(&g_critical_zone->mutex);
}
case 3:
{
static struct sembuf buf = {0, 1, SEM_UNDO};
return semop(g_semid, &buf, 1);
}
case 4:
{
return pthread_rwlock_unlock(&g_critical_zone->rwlock);
}
default:
{
printf("unknown type = %d/n", g_type);
return -1;
break;
}
}
}

int init()
{
key_t key = ftok("/home/xjx/Profile/IPC/test.cpp", 1);
if (key == -1) {
DBG(ftok);
return -1;
}
// prepare shared memory and posix mutex
int shmid = shmget(key, 64, IPC_EXCL|IPC_CREAT|0666);
if (shmid == -1 && errno == EEXIST) {
shmid = shmget(key, 0, 0666);
if (shmid == -1) {
DBG(shmget);
return -1;
}
else
{
g_critical_zone = (sharedMemory*)shmat(shmid, 0, 0666);
if (g_critical_zone == 0) {
DBG(shmat);
return -1;
}
}
}
else
{
g_critical_zone = (sharedMemory*)shmat(shmid, 0, 0666);
if (g_critical_zone == 0) {
DBG(shmat);
return -1;
}
pthread_mutexattr_t attr = {0};
int ret = pthread_mutexattr_init(&attr);
if (ret != 0) {
DBG(pthread_mutexatr_init);
return -1;
}
ret = pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
if (ret != 0) {
DBG(pthread_mutexattr_setpshared);
return -1;
}

ret = pthread_mutex_init(&g_critical_zone->mutex, &attr);
if (ret != 0) {
DBG(pthread_mutex_init);
return -1;
}
{
pthread_rwlockattr_t attr = {0};
int ret = pthread_rwlockattr_init(&attr);
if (ret != 0) {
DBG(pthread_rwlockattr_init);
return -1;
}
ret = pthread_rwlockattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
if (ret != 0) {
DBG(pthread_rwlockattr_setpshared);
return -1;
}

ret = pthread_rwlock_init(&g_critical_zone->rwlock, &attr);
if (ret != 0) {
DBG(pthread_rwlock_init);
return -1;
}
}
}

// prepare lockfd for fcntl
g_lockfd = open("/home/xjx/Profile/IPC/test.cpp", 0666);
if (g_lockfd == -1) {
DBG(open);
return -1;
}

// prepare semid for semaphore
g_semid = semget(key, 1, IPC_EXCL|IPC_CREAT|0666);
if (g_semid == -1) {
g_semid = semget(key, 0, 0666);
if (g_semid == -1) {
DBG(semget);
return -1;
}
}
else
{
union semun buf = {0};
buf.val = 1;
int ret = semctl(g_semid, 0, SETVAL, buf);
if (ret != 0) {
DBG(semctl);
return -1;
}
}
return 0;
}

void* thread_inc(void *arg)
{
for (int ix = 0; ix < g_count; ix++) {
lock();
++g_critical_zone->count;
unlock();
}
return (void*)0;
}

void* thread_dec(void *arg)
{
for (int ix = 0; ix < g_count; ix++) {
lock();
--g_critical_zone->count;
unlock();
}
return (void*)0;
}

int main(int argc, char *argv[])
{
if (argc != 4) {
printf("Usage : sync_test type count thread_count/n");
printf("number : /n 1.fcntl/n 2.posix mutex/n 3.system V semaphore/n 4.posix rwlock");
return -1;
}
g_lock_type = atoi(argv[1]);
g_count = atoi(argv[2]);
int thread_count = atoi(argv[3]);
int ret = init();
if (ret != 0) {
errno = 0;
DBG("init failed");
return -1;
}
pthread_t *pthread_array = (pthread_t*)malloc(sizeof(pthread_t)*thread_count*2);
if (pthread_array == 0) {
DBG(malloc);
return -1;
}
pthread_attr_t attr = {0};
ret = pthread_attr_init(&attr);
if (ret != 0) {
DBG(pthread_attr_init);
return -1;
}

count tick_count;
for (int ix = 0; ix < thread_count; ix++) {
int ret = pthread_create(&pthread_array[ix], &attr, thread_inc, 0);
if (ret != 0) {
DBG(pthread_create);
return -1;
}
}
for (int ix = thread_count; ix < thread_count*2; ix++) {
int ret = pthread_create(&pthread_array[ix], &attr, thread_dec, 0);
if (ret != 0) {
DBG(pthread_create);
return -1;
}
}
void *result = 0;
for (int ix = 0; ix < thread_count*2; ix++) {
int ret = pthread_join(pthread_array[ix], &result);
if (ret != 0) {
DBG(pthread_join);
return -1;
}
}
printf("result = %d/n", g_critical_zone->count);
return 0;
}

操作系统为CentOS 5.0
Linux localhost.localdomain 2.6.18-8.el5 #1 SMP Thu Mar 15 19:57:35 EDT 2007 i686 athlon i386 GNU/Linux
//---------结果
单进程多线程情况下,每个线程操作1000次
线程个数 1 10 100 200 300
文件锁 9197 81314 778531 1560966 2285197
互斥锁 2932 10822 86007 166860 243554
信号量 8882 290273 2085485 8141024 18869807
读写锁 2623 11511 88334 172875 256680

多进程多线程(40)情况下，每个线程操作1000次
进程个数 10 100
文件锁 3665876 18718567
互斥锁 444270 3785420
信号量 137378859 463111685
读写锁 559855 7772213
结论：不管是在多进程还是多线程环境下，进程间用的同步机制的性能低于线程间的同步机制的性能，特别是信号量，随着进程、线程个数的增加，性能恶化严重，而记录锁的性能则相对平稳。互斥量在这几种同步机制中性能最好，但由于线程间的同步机制需要考虑到其健壮性(进程、线程异常退出时要释放未释放的锁)，需要做其它特殊处理，增加了复杂度，在实际的应用中，需要根据实际情况进行取舍。