利用interlockedexchangeadd实现用户态的互斥

folklore

(相当于用户态的Spinlock)，大家有什么想法，如何实现？
好像很难实现的样子啊，很早以前的问题了。题外话：不清楚为什么Linuz不使用Exchange而一定要使用Lock Inc这种方式，不知他怎么想的。

难不成使用：
long var=0;
interlockedexchangeadd(&var,1);
加
interlockedexchangeadd(&var,-1);

多使用了一次Interlocked，不但没有效率，不幸在激烈竞争时，效率更要指数下降，如果成千上万的线程（据说Linux支持线程了），可能成为“假死锁”(var很长很长时间无法回到0)；

OwnWaterloo

interlockedexchangeadd?
不是InterlockedCompareExchange?

folklore

谢谢回复

InterlockedCompareExchange
Linux下也没有这个函数，Windows有的，不过我不用～～

有的只是InterlockedExchageAddd，这个函数基本没有什么用.
PS我希望不要通过OS的API实现Mutex,这样有两个好处：1它更快（不管Windows或是Linux都一样，Call API要RUN成千上万个ＣＰＵ　Ｃｌｏｃｋ，只要看看上下文Switch的代价就无法接受了，更何况它还会建立内核对象）

OwnWaterloo

终于看明白了……
算法大致是这样：

1. 
   
2. #include <windows.h>
   
3. 
   
4. typedef struct {
   
5.       long enter;
   
6.       long leave;
   
7. } spinlock_t;
   
8. 
   
9. void spinlock_acquire(spinlock_t* self)
   
10. {
    
11.       long enter = _InterlockedExchangeAdd(&self->enter, 1);
    
12.       while (enter!=self->leave);
    
13. }
    
14. 
    
15. void spinlock_release(spinlock_t* self)
    
16. {
    
17.       ++self->leave;
    
18. }
    

复制代码

有2个计数器， enter和leave， 记录流量： 进入计数和离开计数。
显然， 进入条件就是进入计数与离开计数相同。

进入时， 原子地增加enter并获得原值。
如果发生竞争， 肯定有一个最幸运的家伙拿到和leave相同的原值， 并进入。

其他的按登记顺序依次等待。
最终离开时增加leave， 第2幸运的家伙就可以进入。


i486上long会回绕， 可以将计数器的取值看作一个"环"：

1. 
   
2. 
   
3.     0
   
4.   7    1
   
5. 6        2
   
6.   5    3
   
7.     4
   

复制代码

只要等待数不超过LONG_MAX即可。


linux(i486)中的实现没有使用两个long， 只用了32bit。
而且32bit都没有用完， 只用了ax。
将al当作leave， ah当作enter。

所以， 保险起见， 竞争者不能超过256。

OwnWaterloo

列出一些ref供参考……

首先是趴出的linux的源代码：

1. 
   
2. /* arch/x86/include/asm/alternative.h */
   
3. 
   
4. #ifdef CONFIG_SMP
   
5. #define LOCK_PREFIX_HERE                  \
   
6.       ".section .smp_locks,\"a\"\n"              \
   
7.       ".balign 4\n"                              \
   
8.       ".long 671f - .\n" /* offset */        \
   
9.       ".previous\n"                              \
   
10.       "671:"
    
11. 
    
12. #define LOCK_PREFIX LOCK_PREFIX_HERE "\n\tlock; "
    
13. 
    
14. #else /* ! CONFIG_SMP */
    
15. #define LOCK_PREFIX_HERE ""
    
16. #define LOCK_PREFIX ""
    
17. #endif
    
18. 
    
19. /* arch/x86/include/asm/spinlock_types.h */
    
20. typedef struct arch_spinlock {
    
21.       unsigned int slock;
    
22. } arch_spinlock_t;
    
23. 
    
24. /* arch/x86/include/asm/spinlock.h */
    
25. 
    
26. #ifdef CONFIG_X86_32
    
27. # define LOCK_PTR_REG "a"
    
28. # define REG_PTR_MODE "k"
    
29. #else
    
30. # define LOCK_PTR_REG "D"
    
31. # define REG_PTR_MODE "q"
    
32. #endif
    
33. 
    
34. #if defined(CONFIG_X86_32) && \
    
35.       (defined(CONFIG_X86_OOSTORE) || defined(CONFIG_X86_PPRO_FENCE))
    
36. /*
    
37. * On PPro SMP or if we are using OOSTORE, we use a locked operation to unlock
    
38. * (PPro errata 66, 92)
    
39. */
    
40. # define UNLOCK_LOCK_PREFIX LOCK_PREFIX
    
41. #else
    
42. # define UNLOCK_LOCK_PREFIX
    
43. #endif
    
44. 
    
45. 
    
46.       #define  __always_inline
    
47.       #define static
    
48. 
    
49. static __always_inline void __ticket_spin_lock(arch_spinlock_t *lock)
    
50. {
    
51.       short inc = 0x0100;
    
52. 
    
53.       asm volatile (
    
54.             LOCK_PREFIX "xaddw %w0, %1\n"
    
55.             "1:\t"
    
56.             "cmpb %h0, %b0\n\t"
    
57.             "je 2f\n\t"
    
58.             "rep ; nop\n\t"
    
59.             "movb %1, %b0\n\t"
    
60.             /* don't need lfence here, because loads are in-order */
    
61.             "jmp 1b\n"
    
62.             "2:"
    
63.             : "+Q" (inc), "+m" (lock->slock)
    
64.             :
    
65.       : "memory", "cc");
    
66. }
    
67. 
    
68. static __always_inline int __ticket_spin_trylock(arch_spinlock_t *lock)
    
69. {
    
70.       int tmp, new;
    
71. 
    
72.       asm volatile("movzwl %2, %0\n\t"
    
73.             "cmpb %h0,%b0\n\t"
    
74.             "leal 0x100(%" REG_PTR_MODE "0), %1\n\t"
    
75.             "jne 1f\n\t"
    
76.             LOCK_PREFIX "cmpxchgw %w1,%2\n\t"
    
77.             "1:"
    
78.             "sete %b1\n\t"
    
79.             "movzbl %b1,%0\n\t"
    
80.             : "=&a" (tmp), "=&q" (new), "+m" (lock->slock)
    
81.             :
    
82.       : "memory", "cc");
    
83. 
    
84.       return tmp;
    
85. }
    
86. 
    
87. static __always_inline void __ticket_spin_unlock(arch_spinlock_t *lock)
    
88. {
    
89.       asm volatile(UNLOCK_LOCK_PREFIX "incb %0"
    
90.             : "+m" (lock->slock)
    
91.             :
    
92.       : "memory", "cc");
    
93. }
    

复制代码

编译后的汇编代码：

1. 
   
2. 00000000 <___ticket_spin_lock>:
   
3.    0:        8b 54 24 04                  mov    edx,DWORD PTR [esp+0x4]
   
4.    4:        b8 00 01 00 00               mov    eax,0x100
   
5.    9:        66 0f c1 02                  xadd   WORD PTR [edx],ax
   
6.    d:        38 e0                        cmp    al,ah
   
7.    f:        74 06                        je     17 <___ticket_spin_lock+0x17>
   
8.   11:        f3 90                        pause  
   
9.   13:        8a 02                        mov    al,BYTE PTR [edx]
   
10.   15:        eb f6                        jmp    d <___ticket_spin_lock+0xd>
    
11.   17:        c3                           ret   
    
12. 
    
13. 00000018 <___ticket_spin_trylock>:
    
14.   18:        8b 54 24 04                  mov    edx,DWORD PTR [esp+0x4]
    
15.   1c:        0f b7 02                     movzx  eax,WORD PTR [edx]
    
16.   1f:        38 e0                        cmp    al,ah
    
17.   21:        8d 88 00 01 00 00            lea    ecx,[eax+0x100]
    
18.   27:        75 04                        jne    2d <___ticket_spin_trylock+0x15>
    
19.   29:        66 0f b1 0a                  cmpxchg WORD PTR [edx],cx
    
20.   2d:        0f 94 c1                     sete   cl
    
21.   30:        0f b6 c1                     movzx  eax,cl
    
22.   33:        c3                           ret   
    
23. 
    
24. 00000034 <___ticket_spin_unlock>:
    
25.   34:        8b 44 24 04                  mov    eax,DWORD PTR [esp+0x4]
    
26.   38:        fe 00                        inc    BYTE PTR [eax]
    
27.   3a:        c3                           ret   
    
28.   3b:        90                           nop   
    

复制代码

可以看出， trylock还是使用了cmpxchg ， 也就是InterlockedCompareExchange。

最后是xadd 和cmpxchg

OwnWaterloo

再帖一个用InterlockedCompareExchange实现的：

1. 
   
2. #include <windows.h>
   
3. 
   
4. typedef struct {
   
5.       long owner;
   
6. } spinlock_t;
   
7. enum { invalid_owner=0 };
   
8. 
   
9. int spinlock_acquire_try_(spinlock_t* self, long id)
   
10. {
    
11.       return _InterlockedCompareExchange(&self->owner, id, invalid_owner)
    
12.              == invalid_owner;
    
13. }
    
14. 
    
15. void spinlock_acquire(spinlock_t* self)
    
16. {
    
17.       long id = GetCurrentThreadId();
    
18.       while (!spinlock_acquire_try_(self, id));
    
19. }
    
20. 
    
21. int spinlock_acquire_try(spinlock_t* self)
    
22. {
    
23.       long id = GetCurrentThreadId();
    
24.       return spinlock_acquire_try_(self, id);
    
25. }
    
26. 
    
27. void spinlock_release(spinlock_t* self)
    
28. {
    
29.       self->owner = invalid_owner;
    
30. }
    

复制代码

每个线程以id号作为标识。lock保存持有者的id。
Windows保证线程id不会为0， 所以用0作为特殊所持者， 表示锁可获取。

尝试进入时， 原子地比较id与owner。
若相等（可获取）， 则将id写入owner， 以阻止其他线程获得锁。
若owner的原值为invalid_owner， 说明自己正好就是获得锁的线程。

进入时， 不断尝试直到成功。

离开时， 将锁标识为可获取。

这种算法依赖于：
1. 能获取线程id
2. id是不超过long的整数
3. 存在一个被保留的id

当时是一个同学申请场外援助时想出的， 题目给的就是compare_and_swap。
实现后觉得至少在windows上还行。
pthread_t好像不保证是整数， 也没说有保留id……
也没考虑过完全不依赖os api或者线程库……
然后也没继续深入了……

所以2楼会有那样的问题……

folklore

回复 4# OwnWaterloo
    十分感谢！以前在哪个角落看过这个代码，那时感到实现得很XX，没在意，现在想想，的确有其巧妙处。
巧妙点在于，它按先后顺序给竞争者排队了。

下面的代码则无法做到这一点：

1. 
   
2. volatile long spinLock=0;
   
3. while(InterlockedExchage(&spinLock,1)){}
   
4. Intermutex codes..
   
5. spinLock=0;
   

复制代码

folklore

再帖一个用InterlockedCompareExchange实现的：每个线程以id号作为标识。lock保存持有者的id。
Windows保证 ...
OwnWaterloo 发表于 2010-08-01 04:51

    谢谢，唯一的ID可以自已实现的～～，直接使用InerlockedAdd即可

OwnWaterloo

回复 8# folklore

这个怎么做？？？

folklore

struct {
  long enter;
  long leave;
} uspin{enter:0;leave:0};

long thid=InterlockedExchageAdd(uspin.enter);  //此外本身产生了一个唯一的ID
while(thid==thid.leave){}
!@#$%^&*
uspin.leave++;

另以上代码因为是自已在用户态实现的，故没有AX的16Bit问题，可以允许2^32=4G个线程同时竞争（这是不可能的，因为uspin是本身应用于快速Acquire/Release的Case)。