【问题标题】:Intel-64 and ia32 atomic operations acquire-release semantics and GCC 5+Intel-64 和 ia32 原子操作获取-释放语义和 GCC 5+
【发布时间】:2018-06-02 14:14:45
【问题描述】:

我正在研究我的 Haswell CPU 上的 Intel CPU 原子特性 (4/8 核 2.3-3.9ghz i7-4790M),我发现它真的很难 构建例如。可靠的 mutex_lock() 和 mutex_unlock() 例如 GCC 手册所建议的操作:

针对事务内存的 6.53 x86 特定内存模型扩展

x86 架构支持额外的内存排序标志来标记 锁定硬件锁省略的关键部分。这些必须 除了现有的内存模型之外,还指定了原子内在函数。

 '__ATOMIC_HLE_ACQUIRE'
 Start lock elision on a lock variable.  Memory model must be
 '__ATOMIC_ACQUIRE' or stronger.
 '__ATOMIC_HLE_RELEASE'
 End lock elision on a lock variable.  Memory model must be
 '__ATOMIC_RELEASE' or stronger.

当锁获取失败时,需要中止良好的性能 交易迅速。这可以通过'_mm_pause'来完成

 #include <immintrin.h> // For _mm_pause

 int lockvar;

 /* Acquire lock with lock elision */
 while (__atomic_exchange_n(&lockvar, 1, 
     __ATOMIC_ACQUIRE|__ATOMIC_HLE_ACQUIRE))
     _mm_pause(); /* Abort failed transaction */
 ...
 /* Free lock with lock elision */
 __atomic_store_n(&lockvar, 0, __ATOMIC_RELEASE|__ATOMIC_HLE_RELEASE);

因此,请阅读该内容和英特尔软件开发人员手册第 3 卷部分 8.1,“锁定原子操作”,特别是第 8.1.4 节, “LOCK 操作对内部处理器缓存的影响”, 首先让我实现了我的测试 mutex_lock() mutex_unlock() 喜欢:

... static inline attribute((always_inline,const)) bool ia64_has_clflush(void) { register unsigned int ebx=0; asm volatile ( "MOV $7, %%eax\n\t" "MOV $0, %%ecx\n\t" "CPUID\n\t" "MOV %%ebx, %0\n\t" : "=r" (ebx) : : "%eax", "%ecx", "%ebx" ); return ((ebx & (1U<<23)) ? true : false); }

#define _LD_SEQ_CST_ __ATOMIC_SEQ_CST
#define _ST_SEQ_CST_ __ATOMIC_SEQ_CST
#define _ACQ_SEQ_CST_ (__ATOMIC_SEQ_CST|__ATOMIC_HLE_ACQUIRE)
#define _REL_SEQ_CST_ (__ATOMIC_SEQ_CST|__ATOMIC_HLE_RELEASE)

static bool has_clflush=false;
static
void init_has_clflush(void)
{ has_clflush = ia64_has_clflush();
}
static
void init_has_clflush(void) __attribute__((constructor));

static inline __attribute__((always_inline))
void mutex_lock( register _Atomic int *ua )
{ // the SDM states that memory to be used as semaphores
  // should not be in the WB cache memory, but nearest we
  // can get to uncached memory is to explicitly un-cache it:
  if(has_clflush)
    asm volatile
    ( "CLFLUSHOPT (%0)"
      :: "r" (ua)
    );
    // why isn't the cache flush enough?
    else
      asm volatile
      ( "LFENCE" :: );
      register unsigned int x;
      x = __atomic_sub_fetch( ua, 1, _ACQ_SEQ_CST_);
      _mm_pause();
    if(has_clflush)
      asm volatile
      ( "CLFLUSHOPT (%0)"
       :: "r" (ua)
      );
    else
      asm volatile
      ( "SFENCE" :: );
  while((x = __atomic_load_n(ua,_LD_SEQ_CST_)) != 0)
    switch(syscall( SYS_futex, ua, FUTEX_WAIT, x, nullptr,nullptr,0))
    {case 0:
      break;
     case -1:
      switch( errno )
      { case EINTR:
        case EAGAIN:
         continue;
        default:
         fprintf(stderr,"Unexpected futex error: %d : '%s'.", errno,   
              strerror(errno));
        return;
      }
    }
  }

  static inline __attribute__((always_inline))
  void mutex_unlock( register _Atomic int *ua )
  { if(has_clflush)
      asm volatile
      ( "CLFLUSHOPT (%0)"
      :: "r" (ua)
      );
    else
      asm volatile( "LFENCE" :: );
    register unsigned int x;
    x = __atomic_add_fetch( ua, 1, _REL_SEQ_CST_);
    _mm_pause();
    if(has_clflush)
      asm volatile
      ( "CLFLUSHOPT (%0)"
        :: "r" (ua)
      );
    else
      asm volatile ( "SFENCE" :: );
    if(x == 0)
      while( (1 < syscall( SYS_futex, ua, FUTEX_WAKE, 1,
           nullptr,nullptr,0)) && (errno == EINTR));
  }

现在,有趣的是关键的 mutex_lock() 减法和 mutex_unlock() 加法操作以指令结束:

互斥锁:

# 61 "intel_lock1.c" 1
    CLFLUSHOPT (%rbx)
# 0 "" 2
#NO_APP
.L7:
    lock xacquire subl  $1, lck(%rip)
    rep nop
    cmpb    $0, has_clflush(%rip)
    je  .L8
#APP
# 72 "intel_lock1.c" 1
    CLFLUSHOPT (%rbx)
# 0 "" 2

互斥锁:

#APP
# 98 "intel_lock1.c" 1
    CLFLUSHOPT (%rbx)
# 0 "" 2
#NO_APP
.L24:
    movl    $1, %eax
    lock xacquire xaddl %eax, lck(%rip)
    rep nop
    addl    $1, %eax
    cmpb    $0, has_clflush(%rip)
    je  .L25
#APP
# 109 "intel_lock1.c" 1
    CLFLUSHOPT (%rbx)
# 0 "" 2
#NO_APP

但是这个实现似乎需要 LFENCE / SFENCE 可靠地运行(CLFLUSHOPT 是不够的),否则 两个线程最终都可能在 futex() 中死锁 锁定值是相同的 -1 。

我无法通过阅读英特尔文档看到它是如何 可能会发生两个线程进入指令 顺序:

# %rbx == $lck
CLFLUSHOPT (%rbx)
lock xacquire subl  $1, lck(%rip)
rep nop

如果 *lck 为 0 ,则在 *lck 中都可以得到结果 '-1' ; 肯定一个线程必须得到-1,另一个线程必须得到-2?

但 strace 说不是:

strace: Process 11978 attached with 2 threads
[pid 11979] futex(0x60209c, FUTEX_WAIT, 4294967295, NULL <unfinished ...>
[pid 11978] futex(0x60209c, FUTEX_WAIT, 4294967295, NULL^C

这是死锁的情况。我哪里做错了?

请任何英特尔 CPU 锁定和缓存专家解释一下 同一个未缓存位置 *lck 的两个原子如何递减或递增 这两者 断言#LOCK 总线信号(独占总线访问)和 XACQUIRE 最终能在 *lck 中得到相同的结果吗?

我认为这就是#LOCK 前缀(和 HLE)的目的? 我尝试过不使用 HLE,而只使用 __ATOMIC_SEQ_CST 进行所有访问, (这只是添加了 LOCK 前缀,而不是 XACQUIRE)但它没有区别 - 没有 {L,S}FENCE-es 仍然会导致死锁。

我读过 Ulrich Drepper 的优秀论文 [ Futexes are Tricky ] :http://www.akkadia.org/drepper/futex.pdf ,但他提出了 仅将硬编码常量写入的互斥体实现 锁内存。我明白为什么了。很难 让互斥锁与服务员数量或任何数量可靠地工作 对锁定值进行的一种算术运算。 有没有人找到方法来做可靠的锁定算术 这样结果适合锁/信号量 x86_64 Linux 上的价值?最有兴趣讨论它们...

所以在调查了 HLE 和 CLFLUSH 几条死胡同之后, 我能够做到的唯一有效的锁定/解锁版本 到达使用硬编码常量和 __atomic_compare_exchange_n - 测试程序的完整源代码,它增加了一个计数器 (无锁定)直到收到 + / 退出信号, 位于:

工作示例:intel_lock3.c

[]:https://drive.google.com/open?id=1ElB0qmwcDMxy9NBYkSXVxljj5djITYxa

enum LockStatus
{ LOCKED_ONE_WAITER = -1
, LOCKED_NO_WAITERS = 0
, UNLOCKED=1
};

static inline __attribute__((always_inline))
bool mutex_lock( register _Atomic int *ua )
{ register int x;
  int cx;
 lock_superceded:
  x  = __atomic_load_n( ua, _LD_SEQ_CST_ );
  cx = x;
  x = (x == UNLOCKED)
       ? LOCKED_NO_WAITERS
       : LOCKED_ONE_WAITER;
  if (! __atomic_compare_exchange_n
      ( ua, &cx, x, false, _ACQ_SEQ_CST_,  _ACQ_SEQ_CST_) )
    goto lock_superceded;
  if( x == LOCKED_ONE_WAITER )
  { do{
    switch(syscall( SYS_futex, ua, FUTEX_WAIT, x, nullptr,nullptr,0))
    {case 0:
      break;
     case -1:
      switch( errno )
      { case EINTR:
         return false;
        case EAGAIN:
          break;
        default:
          fprintf(stderr,"Unexpected futex WAIT error: %d : '%s'.",
                  errno, strerror(errno));
          return false;
       }
    }
    x = __atomic_load_n(ua,_LD_SEQ_CST_);
    } while(x < 0);
  }
  return true;
}

static inline __attribute__((always_inline))
bool mutex_unlock( register _Atomic int *ua )
{ register int x;
  int cx;
 unlock_superceded:
  x  = __atomic_load_n( ua, _LD_SEQ_CST_ );
  cx = x;
  x = (x == LOCKED_ONE_WAITER)
       ? LOCKED_NO_WAITERS
       : UNLOCKED;
  if (! __atomic_compare_exchange_n
       ( ua, &cx, x, false, _ACQ_SEQ_CST_,  _ACQ_SEQ_CST_) )
    goto unlock_superceded;
    if(x == LOCKED_NO_WAITERS)
    { while((1 < 
             syscall( SYS_futex, ua, FUTEX_WAKE, 1, nullptr,nullptr,0))
         ||( UNLOCKED != __atomic_load_n( ua, _LD_SEQ_CST_ ))
         ) // we were a waiter, so wait for locker to unlock !
      { if( errno != 0 )
          switch(errno)
          {case EINTR:
            return false;
           case EAGAIN:
            break;
           default:
            fprintf(stderr,
                  "Unexpected futex WAKE error: %d : '%s'.", 
                  errno, strerror(errno));
            return false;
          }
      }
   }
   return true;
 }

 Build & Test (GCC 7.3.1 & 6.4.1 & 5.4.0) used:
 $ gcc -std=gnu11 -march=x86-64 -mtune=native -D_REENTRANT \
   -pthread -Wall -Wextra -O3 -o intel_lock3 intel_lock3.c

 $ ./intel_lock3
 # wait a couple of seconds and press ^C
 ^C59362558

使用算术破解的版本:

https://drive.google.com/open?id=10yLrohdKLZT4p3G1icFHdjF5eHY68Yws

用例如编译:

$ gcc -std=gnu11 -march=x86_64 -mtune=native -O3 -Wall -Wextra 
  -o intel_lock2 intel_lock2.c
$ ./intel_lock2
# wait a couple of seconds and press ^C
$ ./intel_lock2
^Cwas locked!
446

它不应该打印“被锁定!”并且在 几秒钟应该超过一个计数,打印 最后,@ 5e8 : 5x10^8 ,而不是 446。

用strace运行显示有两个线程阻塞 等待-1的锁值变为0:

$ strace -f -e trace=futex ./intel_lock2
strace: Process 14481 attached
[pid 14480] futex(0x602098, FUTEX_WAIT, 4294967295, NULL <unfinished ...>
[pid 14481] futex(0x602098, FUTEX_WAKE, 1 <unfinished ...>
[pid 14480] <... futex resumed> )       = -1 EAGAIN (Resource temporarily
                                          unavailable)
[pid 14481] <... futex resumed> )       = 0
[pid 14480] futex(0x602098, FUTEX_WAKE, 1 <unfinished ...>
[pid 14481] futex(0x602098, FUTEX_WAIT, 4294967295, NULL <unfinished ...>
[pid 14480] <... futex resumed> )       = 0
[pid 14481] <... futex resumed> )       = -1 EAGAIN (Resource temporarily
                                          unavailable)
[pid 14480] futex(0x602098, FUTEX_WAIT, 4294967295, NULL <unfinished ...>
[pid 14481] futex(0x602098, FUTEX_WAIT, 4294967295, NULL^C <unfinished  
...>
[pid 14480] <... futex resumed> )       = ? ERESTARTSYS (To be restarted 
if SA_RESTART is set)
strace: Process 14480 detached
strace: Process 14481 detached
was locked!
7086

$

通常,WAIT 应该安排在 WAKE 之前,但不知何故 GCC 将内存排序语义解释为 WAKE 总是在任何 WAIT 之前被安排;但即使那样 发生,代码应该只是延迟,并且永远不会结束 两个线程在进入 futex(...FUTEX_WAIT..) 时获得 -1 lck 值。

几乎相同的算法在锁定值上使用算术总是 当两个线程都获得 (-1,-1) 时出现死锁 - 注意,从未见过 -2 值 通过任何线程:

static inline __attribute__((always_inline))
bool mutex_lock( register _Atomic volatile int *ua )
{ register int x;
  x = __atomic_add_fetch( ua, -1, _ACQ_SEQ_);
  if( x < 0 )
  { do{
    // here you can put:
    // if( x == -2) { .. NEVER REACHED! }
    switch(syscall( SYS_futex, ua, FUTEX_WAIT, x, nullptr,nullptr,0))
    {case 0:
      break;
     case -1:
      switch( errno )
      { case EINTR:
         return false; // interrupted - user wants to exit?
        case EAGAIN:
          break;
        default:
          fprintf(stderr,"Unexpected futex WAIT error: %d : '%s'.",
                  errno, strerror(errno));
          return false;
       }
    }
    x = __atomic_load_n(ua,_LD_SEQ_);
    } while(x < 0);
  }
  return true;
}

static inline __attribute__((always_inline))
bool mutex_unlock( register _Atomic volatile int *ua )
{ register int x;
  x = __atomic_add_fetch( ua, 1, _REL_SEQ_);
  if(x == 0) // there was ONE waiter
     while(  (1 < 
             syscall( SYS_futex, ua, FUTEX_WAKE, 1, nullptr,nullptr,0)
             )
           ||(1 < __atomic_load_n(ua, _LD_SEQ_)
             ) // wait for first locker to unlock
           ) 
     { if( errno != 0 )
         switch(errno)
         {case EINTR:
           return false;
          case EAGAIN:
           break;
          default:
           fprintf(stderr,"Unexpected futex WAKE error: %d : '%s'.", 
                  errno, strerror(errno));
           return false;
         }
       }
     return true;
   }

所以,我想如果算术运算是 预期,即。被序列化和原子化,那么上面 代码不会死锁;算术应该生成 与中使用的 LockStatus 枚举值相同的数字 工作示例。

但是算术出了点问题,现在产生 说明:

互斥锁:

movl    $-1, %eax
lock xaddl  %eax, (%rdx)

互斥锁:

movl    $1, %eax
lock xaddl  %eax, (%rdx)

代码本身没有插入栅栏,但每个 __atomic_store_n(ua,...) 都会生成一个。

AFAICS,没有产生该代码的有效时间表 在两个线程中获得相同的 -1 值。

所以我的结论是在算术上使用 intel LOCK 前缀 指令不安全并在用户模式下引入错误行为 Linux x86_64 gcc 编译程序 - 仅限 将常量值从文本存储器写入数据存储器是 Intel Haswell i7-4790M 平台上的原子顺序排序 使用 gcc 和 Linux, 并且此类平台上的算术不能通过使用以下任何组合来实现原子和顺序排序 HLE / XACQUIRE、锁定前缀或 FENCE 指令。

我的预感是分支预测在某种程度上失败了,并且 添加额外的算术运算/未能执行 此平台上的算术运算,带有 LOCK 前缀断言 以及不同物理内核上的多个线程。 因此,所有带有 LOCK 前缀的算术运算都被断言 是可疑的,应该避免。

【问题讨论】:

  • asm("lfence") 没有 "memory" 破坏器来阻止编译器重新排序内存操作是不安全的。此外,如果您不使用 NT 存储或 WC 内存,lfencesfence 对正确性没有影响。如果它恰好使您的代码工作,那只是因为额外的延迟。顺便说一句,对齐地址上的 lock 前缀不会导致总线锁定,只会导致该行的缓存锁定。 IDK 为什么要在锁上使用clflushopt。这将使它变慢而不会获得正确性。存储缓冲区已经使操作尽快可见。
  • ia64_has_clflush(void) 名称错误:IA64 是安腾。 64 位 x86 称为 x86-64。或者就叫它x86_has_clflush。或者最好不要使用它。
  • WTF?不,原子操作在 WB 内存上工作得非常好,并且以这种方式最有效。缓存是连贯的,因此原子更新 L1d 中的一行会使更新对系统中的所有其他观察者来说是原子的。 (即所有其他核心)。 Can num++ be atomic for 'int num'?。有关使用 C11 原子的计数信号量,请参阅 C & low-level semaphore implementation。 (不使用 futex 系统调用,只是一个纯用户空间实现,没有回退到操作系统睡眠/等待,但展示了原子如何工作)
  • 我使用 'ia64' 来表示“英特尔酷睿 64 位架构”,而不是安腾。那你就错了;不要那样做:P IA64 在 Intel CPU 的上下文中已经具有特定的技术含义。有效术语为 Intel64、x86-64 和 amd64。或者只是 x86,因为您的函数也可以在 32 位 x86 上编译和工作。
  • Re: 映射 WC 内存:显然在 Linux 下的用户空间是可能的:how to map memory as USWC under windows/linux?。但就像我说的,你绝对不想要这个。正常的locked 操作在 WB 内存上高效工作,根本无需回写到 DRAM,而且 HLE 还被设计为在 WB 内存中的用户空间互斥体上高效工作。

标签: linux gcc x86-64 atomic futex


【解决方案1】:

lock subl $1, (%rdi)lock xaddl %eax, (%rdx) 在所有情况下都是 100% 原子的,即使指针未对齐(但在这种情况下要慢得多),并且是完整的内存屏障。在可缓存内存上,不会有任何外部#LOCK 总线信号;内部实现只是将高速缓存行锁定在运行 locked 指令的内核中的 MESI 的 M 状态。详情请见Can num++ be atomic for 'int num'?

如果您的测试发现它不是原子的,则说明您的硬件已损坏或您的测试已损坏。发现死锁告诉您设计中存在错误,而不是您的原子原始构建块不是原子的。您可以通过使用两个线程来增加一个共享计数器来非常轻松地测试原子增量,并注意不会丢失任何计数。与您使用addl $1, shared(%rip) 而不使用lock 不同,您会看到丢失的计数。

此外,lfencesfencepause 在正常情况下(没有 NT 存储,仅使用 WB(回写)内存)对正确性没有影响。如果您的任何 fence / clflush 东西有帮助,那只是在某处添加额外的延迟,这可能会使该线程在您的测试中总是输掉比赛,而不是实际上使其安全。 mfence 是唯一重要的栅栏,它阻止 StoreLoad 重新排序和存储转发效果。 (这就是为什么 gcc 使用它作为实现 seq-cst 存储的一部分)。

在您考虑搞乱 HLE/事务性内存之前,获得一个可以正常工作的基本版本。


获取锁的第一个版本中的竞争条件

x = __atomic_sub_fetch( ua, 1, _ACQ_SEQ_CST_); 是原子的,只有一个线程的lock sub 可以将ua0 更改为-1 并得到x=-1 从那里

但是您没有使用sub_fetch 结果,您正在使用
while((x = __atomic_load_n(ua,_LD_SEQ_CST_)) != 0) 进行另一次加载

如果第一个线程在lock sub 和第二个线程中的负载之间锁定然后解锁,那么另一个线程可以看到ua=-1

之所以称为sub_fetch,是因为它以原子方式返回旧值,并以原子方式修改内存中的值。您丢弃 sub_fetch 结果的事实是它可以编译为 lock sub 的原因,而不是 lock xadd 与保存 -1 的寄存器。

(或者智能编译器可以将其编译为 lock sub 并检查 ZF,因为您可以从 lock sub 设置的标志中判断该值何时变为非零或负数。)


请参阅C & low-level semaphore implementation 以获取不回退到操作系统辅助睡眠/唤醒的简单信号量。它在负载时旋转,直到我们看到大于 0 的值,然后尝试使用 C11 fetch_add(-1) 获取锁。

但如果它在与另一个线程的竞争中输了,它会撤消减量。

这可能是一个糟糕的设计;最好使用lock cmpxchg 尝试减量,这样失败的线程就不必撤消它们的减量。


我没有使用过 HLE,但我认为这个错误也会破坏您的 HLE 锁定。

您不需要 SFENCE、LFENCE 或 CLFLUSH[OPT] 或任何东西。 lock xadd 已经是一个完整的内存屏障,它本身是 100% 原子的,适用于任何内存类型(包括 WB)。

如果您认为 SDM 表示您应该避免 WB 内存用于互斥体/信号量,那么您可能误读了它。


您在唤醒期间还有一个可能导致死锁的竞争窗口

mutex_lock 中的这段代码看起来损坏/容易竞争

x = __atomic_sub_fetch( ua, 1, _ACQ_SEQ_CST_);  // ok, fine
_mm_pause();   // you don't want a pause on the fast path.

if( x < 0 )   // just make this a while(x<0) loop
do {
   futex(..., FUTEX_WAIT, ...);

   x = __atomic_load_n(ua,_LD_SEQ_CST_);        // races with lock sub in other threads.
} while(x < 0);

给定线程 A 在 futexlck == -1 中休眠(如果可能?):

  • 线程B解锁,产生lck == 0,调用futex(FUTEX_WAKE)
  • 线程A唤醒,futex在lck仍为0时返回
  • 其他一些线程(B 或第三个线程)进入mutex_lock 并运行__atomic_sub_fetch( ua, 1, _ACQ_SEQ_CST_);,离开lck == -1
  • 线程 A 在其循环底部运行 x = __atomic_load_n(ua,_LD_SEQ_CST_); 并看到 -1

现在您有 2 个线程卡在 futex 等待循环中,实际上没有线程获得互斥锁/进入临界区。


我认为如果你的设计依赖于在 futex 返回后进行加载

fwait()the futex(2) man page 中的示例显示它在futex 返回后返回,不再加载。

futex() 是一个原子比较和阻塞操作。如果一个线程正在等待锁而第三个线程试图获取它,您的设计会将您的计数器值更改为-1。因此,您的设计可能适用于 2 个线程,但不适用于 3 个线程。

使用原子 CAS 进行递减可能是个好主意,因此您永远不会真正将 lck 更改为 -1 或更低,futex 可以保持阻塞状态。

那么,如果您可以指望它只唤醒 1,那么您是否也可以相信它的返回值意味着您确实拥有锁而没有容易竞争的单独负载。我想。

【讨论】:

  • 不是最新代码问题的答案。见讨论。
  • 解决上面评论中提出的问题:`如果一个线程进入mutex_lock(intel_lock2.c算法使用版本)并且锁值为0,则表示另一个线程拥有锁,并且锁值变为-1;但是有一个待定的增量`我只专注于让 1 个生产者和 1 个消费者线程在这里工作的关键部分。 `
  • @"Peter Cordes" : RE: >o 线程 A 唤醒,futex 在 lck 仍为 0 时返回,futex 不能这样做,因为 lck 值在 entry 时为 0; futex(WAIT) 仅在其监视的指针已更改值并且另一个线程执行 futex(WAKE) 时返回给等待者。
  • @"Peter Cordes" : RE >o 其他线程(B 或第三线程)进入 mutex_lock 并运行 _atomic_sub_fetch(ua, 1, _ACQ_SEQ_CST);,离开 lck == -1
  • @"Peter Cordes" : RE >o 其他线程(B 或第三线程)进入 mutex_lock 并运行 _atomic_sub_fetch(ua, 1, _ACQ_SEQ_CST);,离开 lck == -1 但是没有第三个线程!对于 lck 为 0,在这种情况下 B 必须是允许继续其临界区的线程,它只能以调用 unlock() 结束。所以 A 进来,发现 lck 值已经变成负数,然后进入 futex 直到 B 调用 unlock 并将 lck 设置为 0 。没有其他线程可以进来并在这里搞砸数学!
【解决方案2】:

最新的例子 intel_lock2.c 程序在

https://drive.google.com/open?id=10yLrohdKLZT4p3G1icFHdjF5eHY68Yws

现在可以与最新的 intel_lock3.c 程序在

https://drive.google.com/open?id=1ElB0qmwcDMxy9NBYkSXVxljj5djITYxa

现在有一个版本可以保持准确的否定服务员 计数,并使用锁定算术,在:

intel_lock4.c:https://drive.google.com/open?id=1kNOppMtobNHU0lfkfWTh8auXvRcbZfhO

unlock_mutex() 例程,如果有服务员,必须等待每个 现有的服务员解锁,这样当它返回时,互斥锁是 没有上锁,没有服务员。它可以通过 自旋锁定 + sched_yield() 等待锁定值变为 1, 或者它可以使用另一个 futex 调用。所以原来的储物柜,当它 进入 mutex_unlock(),负责确保每个 现有服务员唤醒并解锁互斥锁。

之前这个答案包含:

但仍有奇怪之处:如果任一进程是 ptrace-ed() strace 或使用 '-g3' 而不是 '-O3' 编译,它现在体验 一个“不一致”—— IE。不一致的临界区修改值。这不会发生 如果程序不是 ptrace-d 并且使用 -O3 编译。

请参阅下面的讨论。为了 GCC 的内置 __atomic* 函数 要工作,必须使用任何 -O$x 标志调用 GCC 的优化阶段 在编译期间指定足以启用正确操作 __atomic* 的内置函数。

mutex_lock() / 解锁例程的最终最佳版本:

static inline __attribute__((always_inline))
bool mutex_lock( register _Atomic volatile int *ua )
// lock the mutex value pointed to by 'ua';
// can return false if operation was interrupted ( a signal received ).
{ register int x;
  // lock_again:
  x = __atomic_add_fetch( ua, -1, _ACQ_SEQ_);
  while( x < 0 )
  { switch(syscall( SYS_futex, ua, FUTEX_WAIT, x, nullptr,nullptr,0))
    {case 0:
      break;
     case -1:
      switch( errno )
      { case EINTR:
         return false;
        case EAGAIN:
          break;
        default:
          // this has never been observed to happen, but in any 
          // production implementation
          // should be replaced by some kind of 
          // 'throw( exception )' statement:
          fprintf(stderr,"Unexpected futex WAIT error: %d : '%s'.",
                  errno, strerror(errno));
          return false;
       }
    }
    x = __atomic_load_n(ua,_LD_SEQ_);
  }
  return true;
}

static inline __attribute__((always_inline))
bool mutex_unlock( register _Atomic volatile int *ua )
// unlock: returns false only if interrupted, else returns true
// only when the mutex pointed to by *ua has been unlocked and 
// has no waiters.
{
#ifdef _WITH_UWAIT_
  static int has_unlock_waiter = 0;
#endif
  register int x;
  x = __atomic_add_fetch( ua, 1, _REL_SEQ_);
  if(x < 1) // there was at least ONE waiter, 
            // so we are the original locker
  { while(1 < syscall( SYS_futex, ua, FUTEX_WAKE, 1, nullptr,nullptr,0))
    { if( errno != 0 )
        switch(errno)
        {case EINTR:
          return false;
         case EAGAIN:
          break;
         default:
           // never observed to happen - should be a throw()
          fprintf(stderr,"Unexpected futex WAKE error: %d : '%s'.", 
                  errno, strerror(errno));
          return false;
        }
    }
#ifdef _WITH_UWAIT_
// this is strictly unnecessary, and can be replaced by use of
// sched_yield() (see below), but it
// makes the situation clearer:
// unlock :
    // so we have woken a waiter; wait for that waiter to 
    // actually unlock before returning -
    // by definition, when that waiter enters mutex_unlock() 
    // (AND IT MUST!!), it will not
    // enter the clause containing this code unless there is more than
    // one other waiter., in which case we want to continue until there
    // are no waiters.
    while(1 > (x = __atomic_load_n( ua, _LD_SEQ_ )))
    { __atomic_store_n(&has_unlock_waiter, 1, _ST_SEQ_);
      if( (-1 == 
          syscall( SYS_futex, ua, FUTEX_WAIT, x, nullptr,nullptr,0)
          ) && (errno == EINTR)
        ) return false;
    }
    if( __atomic_load_n(&has_unlock_waiter, _ST_SEQ_) )
      __atomic_store_n(&has_unlock_waiter, 0, _ST_SEQ_);
#else
// The same result is actually achieved by this loop:
    while(1 > (x = __atomic_load_n(ua, _LD_SEQ_)))
      sched_yield();
#endif
    // we do need to wait for the waiting locker to unlock 
    // before proceeding, else
    // mutex_lock could be reentered with lck < 0 and deadlock 
    // would result.
#ifdef _WITH_UWAIT_
  }else if( (x==1) && __atomic_load_n(&has_unlock_waiter, _ST_SEQ_) )
  { // so we're the waiter that a previous unlock woke up 
    // and is waiting for - it now needs to be woken:
    while(1 < syscall( SYS_futex, ua, FUTEX_WAKE, 1, nullptr,nullptr,0))
    { if( errno != 0 )
        switch(errno)
        {case EINTR:  // no, we cannot let user try to unlock again, since modification of lock value succeeded.
         case EAGAIN:
          break;
         default:
          fprintf(stderr,"Unexpected futex WAKE error: %d : '%s'.", errno, strerror(errno));
          return false;
        }
    }
  }
#else
  }
#endif
  return true;
}

测试:

$ gcc -std=gnu11 -pthread -D_WITH_UWAIT_ -O3 -o il2 il2.c
$ ./il2
^C20906015
$ gcc -std=gnu11 -pthread -O3 -o il2 il2.c
$ ./il2
^C45851541

('^C'表示同时按+键)。

现在所有版本都不会死锁并且可以使用:

$ strace -f -e trace=futex ./{intel_lock2 OR intel_lock3 OR intel_lock4} 

我试图 strace 一个“-g”(仅)编译版本并得到一个不一致 - 如果还使用了任何“-O”标志,则不会发生这种情况。

【讨论】:

  • 如果您在解锁后需要调用sched_yield 以避免竞争条件,那么您的代码并不真正安全。几乎总是隐藏比赛,只会让发现错误变得更加困难,至少在系统不是很重的时候是这样。
  • 当然,但它指出了问题的原因:锁定算法和编译器围绕退出条件变量的原子存储插入的“mfence”-es 的组合导致了处理器相信解锁必须始终“发生在”锁定之前,因此解锁最终会占用 CPU 并且锁定不会被安排。但至少我现在看到两个版本都存在同样的问题。我现在更感兴趣的是,没有使用 -O3 编译的版本是互斥的;如果编译了-gX,则不能实现互斥。
  • 你可以用-O3 -g编译。添加调试元数据与优化并不相互排斥。 (但在调试优化代码时,您有时会在尝试检查局部变量时看到“已优化”,或者在全局变量中看到陈旧的值,因为调试信息无法跟踪寄存器中的变量。)
  • 这是对问题的回答还是只是对您的调查的更新,还是什么?
  • 是的,有一个错误 - 我在临界区之外测试翻转 (var 0} 编译版本中,该测试没有被触发,在 -O0 编译版本中,它触发了。很抱歉造成混乱。固定版本在编译时使用或不使用任何 '-O$x' 标志都有效,位于:drive.google.com/open?id=1kNOppMtobNHU0lfkfWTh8auXvRcbZfhO
猜你喜欢
  • 2011-06-16
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2014-04-12
  • 1970-01-01
  • 1970-01-01
  • 2011-05-25
  • 1970-01-01
相关资源
最近更新 更多