EMACS & 程序 编程点滴...

天下难事必作于易，天下大事必作于细

Quick Links

Lastupdated: 2010-01-04

并发编程

线程相关
可重入函数
volatile
Lock-Free 算法
通用原语
Semaphore（旗语/信号量，セマフォ）
协程
性能开销
与线程的区别
协程的实现
其他
longjmp时的资源泄露问题

TOP线程相关

TOP可重入函数

  • 可重入函数（reentrant, リエントラント）是线程安全函数, 其特征是函数中不能有共享资源的访问。
  • 显示可重入函数 ： 如果函数参数都是传值传递（没有指针），即所有的数据引用都是本地的自动栈变量（无静态或全局的变量），那么该函数是 显式的可重入函数（explicity reentrant）。
  • 隐式可重入函数 ： 如果函数参数是引用传递，并指向非共享数据的指针时，该函数为隐式可重入函数（implicity reentrant）。

TOPvolatile

  • 用volatile修饰的变量，不会被编译器优化。每次访问变量不由寄存器访问，而是直接从内存中读取。多线程编程中，为了防 止变量被其他线程改变而使用。
  • 将const volatile变量作为参数传递给某函数，函数中不能改变变量的值。但是该变量的值可以在别的线程中被改变。
  • 使用该修饰词将引起性能开销，但比使用CriticalSection，mutex等机制的开销小。

TOPLock-Free 算法

lock-free （锁无关）算法是一种不需要加锁来处理线程同步的算法。在 linux 和 windows 的内核代码中，经常用到 lock-free 算法。

与 lock-free 算法相似，也存在名为 wait-free （等待无关）的算法。lock-free 是指系统中总存在某个线程能够得以继续执行。wait-free 是指所有的线程等能往下执行，自身与其他的线程的运行没有关系，其自身的任何操作都在有限的时间，步骤内完成。所有的算法都可以使用 Lock-free 来实现，而 Wait-free 只能实现其中的一部分，或者说，Wait-free 的算法是Lock-free 的。

通常的基于锁的线程同步机制是，资源被加锁之后，其他的线程不能访问该资源，只有到加锁线程释放锁之后，才能访问。而wait-free 不是一直阻塞着请求资源的线程，当等待的线程数到达一定比例，在有限的时间内，资源像是在线程间轮询。正因为这样，利用 wait-free 的算法就不会产生死锁，优先级倒置等问题。

TOP通用原语

实际实现 lock-free 算法的时候，需要通用原语的支持。比如 CAS(Compare and Swap) 或者是LL/SC(Load-Linked/Store-Conditional) 命令。在 x86 上，已经支持了CAS。虽然没有支持 LL/SC 命令，但是可以用CAS来实现模拟。

另外，通用原语在 Java 中有专门的接口实现（java.util.concurrent.atomic），而在C语言中需要内联汇编来实现。

下面是 64-bit 环境下，使用 CAS 的代码。 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

static inline uint64_t cmpxchg64(void *ptr, uint64_t oldv, uint64_t newv)
{
        uint64_t out;

        uint64_t v = *((uint64_t *)ptr);

        // newline after `lock' for the work around of apple's gas(?) bug.
        __asm__ __volatile__(
                "lock\n cmpxchgq %2,%1"
                : "=a" (out), "+m" (*(volatile uint64_t *)ptr)
                : "q" (newv), "0" (oldv)
                : "cc");

        return out;
}

#define CAS64(ptr, oldv, newv) (cmpxchg64((ptr), (oldv), (newv)) == (oldv) ? 1 : 0)

32-bit 下的使用。

1
2
3
4
5
6
7
8

static inline uint32_t cmpxchg32(void *ptr, uint32_t oldv, uint32_t newv)
{
 asm volatile("lock\n cmpxchgl %1,%2" : "=a" (oldv) : "q" (newv), "m" (*(uint32_t *)ptr),"0" (oldv)
 : "memory");
 return oldv;
}

#define CAS32(ptr, oldv, newv) (cmpxchg32((ptr), (oldv), (newv)) == (oldv) ? 1 : 0)

同样，还可以使用 cmpchg8b 命令。

1
2
3
4
5
6
7
8
9
10
11
12
13

static inline int cas64(volatile double_var_t *addr,
                        unsigned long old1,
                        unsigned long old2,
                        unsigned long new1,
                        unsigned long new2)
{
  char result;
  __asm__ __volatile__("lock\n cmpxchg8b %0; setz %1"
                       : "=m"(*addr), "=q"(result)
                       : "m"(*addr), "d" (old1), "a" (old2),
                         "c" (new1), "b" (new2) : "memory");
  return (int) result;
}

另外，在 x86_64 环境下还有 cmpxchg16b （128-bit）的通用原语命令。其他处理器的通用原语命令可以参考这里。

TOPSemaphore（旗语/信号量，セマフォ）

P操作

如果想使用资源的时候使用P操作(确保资源)； 如果信号量S>0，将信号量S的值减1; 如果S==0，表示别的线程正在使用该资源，当前线程置为等待状态，排入等待队列。

V操作

如果资源使用完毕，使用V操作释放资源； 将信号量S的值加1，即S=S+1；释放等待队列中第一个等待信号量的进程。（所以V操作需要管理等待队列）

如果信号量只是0或者1，那么称为 二进制信号量。

TOP协程

TOP性能开销

先来看看各种程序粒度的性能开销。

coroutine

其中Generator可以是带有状态的Callback，或者对象化的CallBack。所以，它们的性能消耗基本相同。

TOP与线程的区别

  • 线程和协程都是用异步来模拟同步，线程是由操作系统模拟，而协程由用户级调度器模拟
  • 协程是非抢占式的，当一个协程正在运行时，不能在外部终止它，只能在协程内部调用专用的yield函数，退出协程上下文。所以基于协程的程序中只要有一个协程被阻塞，程序将终止

TOP协程的实现

实现协程主要利用了标准C的 setjmp，longjmp。longjmp时保存当前协程上下文状态，setjmp时恢复指定的上下文状态。即使 在不支持这两个函数的系统下，只要保存了当前寄存器的状态，在需要的时候恢复即可。

这里给出了一个实现方式。

TOP其他

TOPlongjmp时的资源泄露问题

中断发生的时候，需要调用 setjmp 来记住当前的上下文环境，然后跳到中断的服务程序中去。中断程序结束后会调用 longjmp 返回到原先的程序上下文。但是在一个不恰当的时刻，到来的信号结合一个返回程序上下文的 longjmp 调用，可能 会导致资源泄露（文件没有关闭或者临时内存没有释放）。就是说在调用 longjmp 之前，需保证中断程序释放了所持的资源。 使用临界区可以解决这个问题。1

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

int mutex = 0;            /* if set, signals set "sigflags" */
int sigflags = 0;         /* if set, signals received while mutex set */

// 进入临界区，延迟信号处理
#define SPL1() mutex++

// 退出临界区
#define SPL0()                                                          \
    if (--mutex == 0) {                                                 \
        // 处理未决信号                                                 \
        if (sigflags & (1 << (SIGHUP - 1))) handle_hup(SIGHUP);         \
        if (sigflags & (1 << (SIGINT - 1))) handle_int(SIGINT);         \
}

// 信号处理器
void signal_int(int signo)
{
    if (mutex)
        sigflags |= (1 << (signo - 1)); // 如果处理互斥区域，只是设置标志，延迟处理
    else
        handle_int(signo);              // 未设互斥量，立即对中断进行处理
}

// 立即或延迟处理中断信号
void handle_int(int signo)
{
    [...]
    // 中断标志复位
    sigflags &= ~(1 << (signo - 1));
    // 跳回到命令循环
    longjmp(env, -1);
}

void clear_active_list()
{
    // 进入临界区
    SPL1();

    active_size = active_last = active_ptr = active_ndx = 0;
    free(active_list);
    active_list = NULL;

    // 退出临界区
    SPL0();
}

1. 参考《Code Reading : The Open Source Perspective》

© www.yifeiyang.net
net tracking stats