原子操作耗时对比

　　早前写的原子测试demo，一直挂在心头准备来篇介绍文章。

　　今天在编译服务器中找了半天还是没找到，最后总算在个人PC中找到了，再不做总结的话可能哪天真会不小心误删了。

　　代码已上传到这里，有需要的可以拿去测试（本人的demo大多基于Android4.4进行编译和调试，在其他版本上可能出现编译问题）。

1. 背景

　　多个线程访问同一个资源时，出现并发访问（读和写）问题，如果对资源的管理不合理，可能出现跟预期不同的结果。

　　例如，多个线程同时对某个全局变量进行自增操作，这个自增操作至少涉及三条指令：

• 从内存单元读入寄存器
• 在寄存器中对变量操作（加/减1）
• 把新值写回到内存单元

　　如果一个线程执行上面三个步骤过程中，其他线程进入，则会打断前一个的执行动作，导致结果不为预期值。

　　因此，需要将这几个动作作为原子操作，来加以保护。

2. Android平台下的几种锁使用（以自增为例）

　　2.1. cutils中实现的锁android_atomic_inc()

　　　　其实是对android_atomic_add()的一层封装，参考这里。

　　　　

　　　　 

　　2.2. bionic中实现的锁__atomic_inc()

　　　　其实是对__sync_fetch_and_add()的一层封装，参考这里。

　　　　

　　2.3. bionic中实现的互斥锁pthread_mutex_lock()

　　　　其实现参考这个文件。

　　　　

　　　　这个与前两个使用场景有些差异，前面两个主要用于对某个内存变量进行原子操作，这个除具有这个功能外，还可以保护临界资源，

　　使先拿到锁的线程先运行，只有当释放锁后其他线程才能访问。

3. 性能对比

　　这几种锁的性能如何呢？参考demo，接下来分几种场景我对其进行对比。

源码如下：

#define LOG_TAG "AtomicTester"
#include <utils/Log.h>

#include <sys/atomics.h>

#include <utils/Atomic.h>

#define USE_ATOMIC_TYPE     1   //1-cutils; 2-bionic; 3-others(mutex)
#define THREAD_COUNT        10
#define ITERATION_COUNT     500000

#define TEST_BIONIC
#ifdef TEST_BIONIC
extern int __atomic_cmpxchg(int old, int _new, volatile int *ptr);
extern int __atomic_swap(int _new, volatile int *ptr);
extern int __atomic_dec(volatile int *ptr);
extern int __atomic_inc(volatile int *ptr);
#endif

static pthread_mutex_t waitLock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t waitCond = PTHREAD_COND_INITIALIZER;


#define TEST_MUTEX // not define it will cause wrong result when multi-threads ops the same var.

#ifdef TEST_MUTEX
static pthread_mutex_t incLock = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t decLock = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t addLock = PTHREAD_MUTEX_INITIALIZER;
#endif

static volatile int threadsStarted = 0;

/* results */
static int incTest = 0;
static int decTest = 0;
static int addTest = 0;
static int andTest = 0;
static int orTest = 0;
static int casTest = 0;
static int failingCasTest = 0;

/*
 * Get a relative time value.
 */
static int64_t getRelativeTimeNsec()
{
#define HAVE_POSIX_CLOCKS
#ifdef HAVE_POSIX_CLOCKS
    struct timespec now;
    clock_gettime(CLOCK_MONOTONIC, &now);
    return (int64_t) now.tv_sec*1000000000LL + now.tv_nsec;
#else
    struct timeval now;
    gettimeofday(&now, NULL);
    return (int64_t) now.tv_sec*1000000000LL + now.tv_usec * 1000LL;
#endif
}


/*
 * Non-atomic implementations, for comparison.
 *
 * If these get inlined the compiler may figure out what we're up to and
 * completely elide the operations.
 */
#define USE_NOINLINE 0
#if USE_NOINLINE == 1
static void incr() __attribute__((noinline));
static void decr() __attribute__((noinline));
static void add(int addVal) __attribute__((noinline));
static int compareAndSwap(int oldVal, int newVal, int* addr) __attribute__((noinline));
#else
static void incr() __attribute__((always_inline));
static void decr() __attribute__((always_inline));
static void add(int addVal) __attribute__((always_inline));
static int compareAndSwap(int oldVal, int newVal, int* addr) __attribute__((always_inline));
#endif

static void incr()
{
#ifdef TEST_MUTEX
    pthread_mutex_lock(&incLock);
#endif
    incTest++;
#ifdef TEST_MUTEX
    pthread_mutex_unlock(&incLock);
#endif
}
static void decr()
{
#ifdef TEST_MUTEX
    pthread_mutex_lock(&decLock);
#endif
    decTest--;
#ifdef TEST_MUTEX
    pthread_mutex_unlock(&decLock);
#endif
}
static void add(int32_t addVal)
{
#ifdef TEST_MUTEX
    pthread_mutex_lock(&addLock);
#endif
    addTest += addVal;
#ifdef TEST_MUTEX
    pthread_mutex_unlock(&addLock);
#endif
}
static int compareAndSwap(int32_t oldVal, int32_t newVal, int32_t* addr)
{
#ifdef TEST_MUTEX
    pthread_mutex_lock(&addLock);
#endif
    if (*addr == oldVal) {
        *addr = newVal;
#ifdef TEST_MUTEX
    pthread_mutex_unlock(&addLock);
#endif
        return 0;
    }
#ifdef TEST_MUTEX
    pthread_mutex_unlock(&addLock);
#endif
    return 1;
}

/*
 * Exercise several of the atomic ops.
 */
static void doAtomicTest(int num)
{
    int addVal = (num & 0x01) + 1;

    int i;
    for (i = 0; i < ITERATION_COUNT; i++) {
        if (USE_ATOMIC_TYPE == 1) {         //cutils, duration=865ms on ARM_CPU_PART_CORTEX_A53
            android_atomic_inc(&incTest); // wrapper for android_atomic_add(1, addr)
            android_atomic_dec(&decTest);
            android_atomic_add(addVal, &addTest);

            int val;
            do {
                val = casTest;
            } while (android_atomic_release_cas(val, val+3, &casTest) != 0);
            //} while (android_atomic_cmpxchg(val, val+3, &casTest) != 0);
            do {
                val = casTest;
            } while (android_atomic_acquire_cas(val, val-1, &casTest) != 0); // same with release_cas
        } else if (USE_ATOMIC_TYPE == 2) {  //bionic, duration=650ms
            __atomic_inc(&incTest); // wrapper for __sync_fetch_and_add(addr, 1), which is gcc built-in API
            __atomic_dec(&decTest);
            __sync_fetch_and_add(&addTest, addVal);

            int val;
            do {
                val = casTest;
            } while (android_atomic_release_cas(val, val+3, &casTest) != 0);
            do {
                val = casTest;
            } while (android_atomic_acquire_cas(val, val-1, &casTest) != 0);
        } else {                            //mutex, noinline_duration=3500ms, inline_duration=28000ms
            //usleep(1); // if no sleep, result may be right; when have sleep, result must be wrong!
            incr();
            decr();
            add(addVal);

            int val;
            do {
                val = casTest;
            } while (compareAndSwap(val, val+3, &casTest) != 0);
            do {
                val = casTest;
            } while (compareAndSwap(val, val-1, &casTest) != 0);
        }
    }
}

/*
 * Entry point for multi-thread test.
 */
static void* atomicTest(void* arg)
{
    pthread_mutex_lock(&waitLock);
    threadsStarted++;
    ALOGD("thread[%d] wait to run...", (int)arg);
    pthread_cond_wait(&waitCond, &waitLock);
    pthread_mutex_unlock(&waitLock);
    doAtomicTest((int) arg);

    return NULL;
}

static void doBionicTest()
{
#ifdef TEST_BIONIC
    /*
     * Quick function test on the bionic ops.
     */
    int prev;
    int initVal = 7;
    prev = __atomic_inc(&initVal);
    __atomic_inc(&initVal);
    __atomic_inc(&initVal);
    ALOGD("bionic __atomic_inc: %d -> %d
", prev, initVal);
    prev = __atomic_dec(&initVal);
    __atomic_dec(&initVal);
    __atomic_dec(&initVal);
    ALOGD("bionic __atomic_dec: %d -> %d
", prev, initVal);
    prev = __atomic_swap(27, &initVal);
    ALOGD("bionic __atomic_swap: %d -> %d
", prev, initVal);
    int swap_ret = __atomic_cmpxchg(27, 72, &initVal);
    ALOGD("bionic __atomic_cmpxchg: %d (%d)
", initVal, swap_ret);
#endif
}

int main(void)
{
    ALOGW("-------------AtomicTester begin--------------");

    pthread_t threads[THREAD_COUNT];
    void *(*startRoutine)(void*) = atomicTest;
    int64_t startWhen, endWhen;

#if defined(__ARM_ARCH__)
    ALOGD("__ARM_ARCH__ is %d", __ARM_ARCH__);
#endif
#if defined(ANDROID_SMP)
    ALOGD("ANDROID_SMP is %d", ANDROID_SMP);
#endif
    ALOGD("Creating threads...");

    int i;
    for (i = 0; i < THREAD_COUNT; i++) {
        void* arg = (void*) i;
        if (pthread_create(&threads[i], NULL, startRoutine, arg) != 0) {
            ALOGD("thread create failed");
        }
    }

    /* wait for all the threads to reach the starting line */
    while (1) {
        pthread_mutex_lock(&waitLock);
        if (threadsStarted == THREAD_COUNT) {
            ALOGD("Starting test!");
            startWhen = getRelativeTimeNsec();
            pthread_cond_broadcast(&waitCond);
            pthread_mutex_unlock(&waitLock);
            break;
        }
        pthread_mutex_unlock(&waitLock);
        usleep(100000);
    }

    for (i = 0; i < THREAD_COUNT; i++) {
        void* retval;
        if (pthread_join(threads[i], &retval) != 0) {
            ALOGE("thread join (%d) failed
", i);
        }
    }

    endWhen = getRelativeTimeNsec();
    if (USE_ATOMIC_TYPE == 3) {
        ALOGD("test pthread_mutex_t with %s", USE_NOINLINE==1 ? "noinline":"always_inline");
    }
    ALOGD("TYPE=%d, All threads stopped, duration time is %.6fms
",
        USE_ATOMIC_TYPE, (endWhen - startWhen) / 1000000.0);

    /*
     * Show results; expecting:
     *
     * incTest = 5000000
     * decTest = -5000000
     * addTest = 7500000
     * casTest = 10000000
     */
    ALOGD("incTest = %d
", incTest);
    ALOGD("decTest = %d
", decTest);
    ALOGD("addTest = %d
", addTest);
    ALOGD("casTest = %d
", casTest);

    doBionicTest();

    ALOGW("-------------AtomicTester end--------------");

    return 0;
}

　　测试手段：10个线程，同时操作同一个全局变量，操作次数为50k次，测算耗时。

测试时，通过宏#define USE_ATOMIC_TYPE 1 来控制测试方法的选择。

　　3.1. 使用cutils实现的原子操作函数：

　　

　　3.2. 使用bionic中实现的原子操作函数（其实是gcc build-in api）：

　　

　　3.3. 使用常用的pthread_mutex_t来保护全局变量：

　　　　测试inline：（无函数调用开销，速度快）

　　

　　　　测试no-inline：（有函数调用开销，速度稍慢）

　　

　　3.4. 不使用任何锁来保护：（结果可能出错）

　　

　　3.5. 不使用任何锁来保护+sleep：（结果一定出错）

　　

　　3.6. pthread_mutex_lock+sleep：

　　

　　经以上对比，可以得到，速度方面：__atomic_inc() > android_atomic_inc() >  pthread_mutex_lock()

　　其他补充：

• 编译器gcc的build-in实现__sync_fetch_and_add()性能最好。
• 测试机器硬件平台为Cortex-A53。
• inline函数因为少了函数调用开销，速度较快，但会导致代码膨胀，因为其是在“预-编-汇-链”步骤中的预处理阶段进行了代码展开。
• 对变量自增操作，如果不加锁或不用原子操作，结果仍有可能正确，毕竟自增操作的几条指令被别的线程打断的概率非常低（另外一个极端的猜测是——在一个cpu时间片周期内，就做完了50k的自增，这样不被打断时结果就不出错了）。 但是，不加锁并且sleep，被打断的概率非常大，参考3.5的测试结果，导致结果出错（其实解释起来也有点牵强，在sleep将要执行前，自增操作的几条指令已经全部完成）。
• 加sleep()会导致耗时大大增加，因为其休眠时长不精确，再扩大50k倍后，导致耗时非常大。但是，usleep(1)不精确，即使扩大5倍后，单个线程执行耗时为：5us*50k=250ms，最后10个线程累加为：250ms*10=2.5s，但是3.5/3.6的测试显示总耗时大约为：39s，因此猜测得出下条的结论：
• sleep是线程被调用时，让出系统资源，放弃当前cpu时间片并阻塞指定时间，让其他线程可以占用cpu。因此，耗时增加如此之多，大概就是大部分时间耗在了线程上下文切换上，每自增一次，线程抢占一次。但是，这个分析可能有误，因为top看cpu占用时，远低于100%，如果不sleep的话，基本上就是100%。后面需要辅助用其他手段来求证。。。
• 关于cpu切换时间片的一点说明，时间片设得太短会导致过多的进程切换，降低了CPU效率；而设得太长又可能引起对短的交互请求的响应变差。将时间片设为100毫秒通常是一个比较合理的折衷。