system V IPC 中的进程锁低效

    系统中使用了 System V IPC 方式的进程锁，今天在压测的时候发现在单进程的情况下，依然有很多semop的系统调用，在单进程程的情况下，不存在锁冲突，经过了futex优化，应该不会有semop系统调用的，就仔细研究了一把这个问题，最终发现 System V IPC下的锁是没有经过 futex优化的，不管有没有锁争用，都必然有semop的系统调用（加锁和解锁各一次），而经过futex优化的NPTL实现在无冲突的时候是不会导致系统调用的，直接在应用层就解决了，两者在性能上差异较大。

    为了验证这个差异，特意写了个简单的程序做了一个测试，每个测试线程均完成100万次的简单运算，多个线程之间共享一个锁，测试代码如下：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <pthread.h>
#include <semaphore.h>
#include <sys/time.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>

int g_iLockType = -1;
int g_iSemID = -1;

sem_t g_sem;


long g_result = 1;
int createSemMutex(key_t iKey)
{
    #if defined(__GNU_LIBRARY__) && !defined(_SEM_SEMUN_UNDEFINED)
    /* union semun is defined by including <sys/sem.h> */
    #else
    /* according to X/OPEN we have to define it ourselves */
    union semun
    {
         int val;                  /* value for SETVAL */
         struct semid_ds *buf;     /* buffer for IPC_STAT, IPC_SET */
         unsigned short *array;    /* array for GETALL, SETALL */
                                   /* Linux specific part: */
         struct seminfo *__buf;    /* buffer for IPC_INFO */
    };
    #endif

    int  iSemID;
    union semun arg;
    u_short array[2] = { 0, 0 };

    //生成信号量集, 包含两个信号量
    if ( (iSemID = semget( iKey, 2, IPC_CREAT | IPC_EXCL | 0666)) != -1 )
    {
        arg.array = &array[0];

        //将所有信号量的值设置为0
        if ( semctl( iSemID, 0, SETALL, arg ) == -1 )
        {
            return -1;
        }
    }
    else
    {
        //信号量已经存在
        if ( errno != EEXIST )
        {
            return -2;
        }

        //连接信号量
        if ( (iSemID = semget( iKey, 2, 0666 )) == -1 )
        {
            return -3;
        }
    }

    return iSemID;
}


int SemMutexLock(int iSemID)
{
    struct sembuf sops[3] = { {0, 0, SEM_UNDO}, {1, 0, SEM_UNDO}, {0, 1, SEM_UNDO} };
    size_t nsops = 3;

    int ret = -1;

    do
    {
        ret=semop(iSemID,&sops[0],nsops);

    } while ((ret == -1) &&(errno==EINTR));

    return ret;
}


int SemMutexUnLock(int iSemID)
{
    struct sembuf sops[1] = { {0, -1, SEM_UNDO} };
    size_t nsops = 1;

    int ret = -1;

    do
    {
        ret=semop(iSemID,&sops[0],nsops);

    } while ((ret == -1) &&(errno==EINTR));

    return ret;
}

void* threadFunc(void* ptr)
{
    int iThread = (long)(ptr);
    printf("thread %d start.\n",iThread);
    if (g_iLockType == 1)
    {
        for (int i = 0 ; i < 1000000 ; ++i)
        {
            if (0 != SemMutexLock(g_iSemID))
            {
                printf("SemMutexLock fail in thread %d.\n",iThread);
                return NULL;
            }

            for (int j = 0 ; j < 100 ; ++j)
            {
                int tmp = iThread * i * i * 997 % (i + 7);
                g_result = iThread * tmp * tmp * 997 % (i + 31);
                g_result = g_result * 1997 % 91;
            }


            if (0 != SemMutexUnLock(g_iSemID))
            {
                printf("SemMutexUnLock fail in thread %d.\n",iThread);
                return NULL;
            }
        }
    }
    else
    {
        for (int i = 0 ; i < 1000000 ; ++i)
        {
            if (0 != sem_wait(&g_sem))
            {
                printf("sem_wait fail in thread %d.\n",iThread);
                return NULL;
            }

            for (int j = 0 ; j < 100 ; ++j)
            {
                int tmp = iThread * i * i * 997 % (i + 7);
                g_result = iThread * tmp * tmp * 997 % (i + 31);
                g_result = g_result * 1997 % 91;
            }


            if (0 != sem_post(&g_sem))
            {
                printf("sem_post fail in thread %d.\n",iThread);
                return NULL;
            }
        }
    }
    return NULL;
}

int main(int argc, char* argv[])
{
    if (argc != 3)
    {
        printf("usuage:%s lockType threadCount\n",argv[0]);
        return -1;
    }

    g_iLockType = atoi(argv[1]);
    int iThreadCount =     atoi(argv[2]);
    printf("locktype=%d ,threadCount=%d\n",g_iLockType,iThreadCount);
    key_t semKey = ftok("sem_mutex_nptl_test",100);
    g_iSemID = createSemMutex(semKey);
    if (g_iSemID < 0)
    {
        printf("createSemMutex fail.return %d\n",g_iSemID);
        return 2;
    }

    if (0 != sem_init(&g_sem, 1, 1))
    {
        printf("sem_init fail.\n");
        return 3;
    }

    pthread_t pt[1000];

    struct timeval start;
    struct timeval end;

    gettimeofday(&start,NULL);
    for (int i = 0 ; i < iThreadCount ; ++i)
    {
        if (0 != pthread_create(pt+i,0,threadFunc,(void*)i))
        {
            printf("create thread %d fail.",i);
            return 3;
        }
    }

    for (int i = 0 ; i < iThreadCount ; ++i)
    {
        pthread_join(pt[i],NULL);
    }

    gettimeofday(&end,NULL);

    printf("time cost = %f ms\n", ((end.tv_sec - start. tv_sec) * 1000000 + (end.tv_usec - start.tv_usec)) / 1000.0);
    return 0;
}

测试结果如下：

锁类型           竞争线程数          耗时(毫秒)

System V IPC          1               9593.426

NPTL                  1               9161.838

System V IPC          2               24292.156

NPTL                  2               21991.775

System V IPC          3               38749.863

NPTL                  3               33610.102

System V IPC          4               50919.030

NPTL                  4               45871.873

System V IPC          5               61978.562

NPTL                  5               56515.495

System V IPC          10              126402.621

NPTL                  10              112752.429

 

    通过strace观察，发现NPTL的锁是没有semop调用的，只有futex_*** 调用，性能上NPTL比System V IPC 高出10%以上，具体的性能差异与加解锁之间的业务代码的复杂程度相关，建议在使用进程锁的时候，选用NPTL的锁实现替代旧的System V IPC，以提升效率。