Skip to content

thread semaphore

Jump to bottom
ceragon edited this page Mar 29, 2022 · 4 revisions

jvm 的信号量实现

结构体

class Semaphore : public CHeapObj<mtSynchronizer> {
    SemaphoreImpl _impl;
};
class PosixSemaphore : public CHeapObj<mtInternal> {
    sem_t _semaphore;
};
typedef PosixSemaphore SemaphoreImpl;
define __SIZEOF_SEM_T	32
typedef union
{
    char __size[__SIZEOF_SEM_T];
    long int __align;
} sem_t;
struct new_sem
{
    uint64_t data;
    int private;
    int pad;
};

初始化

PosixSemaphore::PosixSemaphore(uint value) {
    int ret = sem_init(&_semaphore, 0, value);
}
int __new_sem_init (sem_t *sem, int pshared, unsigned int value) {
    struct new_sem *isem = (struct new_sem *) sem;
    isem->data = value;
    isem->private = (pshared == PTHREAD_PROCESS_PRIVATE
		   ? FUTEX_PRIVATE : FUTEX_SHARED);
}
versioned_symbol (libc, __new_sem_init, sem_init, GLIBC_2_34);

释放信号量

void PosixSemaphore::signal(uint count) {
  for (uint i = 0; i < count; i++) {
    int ret = sem_post(&_semaphore);
  }
}
int __new_sem_post (sem_t *sem) {
    struct new_sem *isem = (struct new_sem *) sem;
    int private = isem->private;
//    uint64_t d = atomic_load_relaxed (&isem->data);
    // __atomic_load_n + __ATOMIC_RELAXED，告诉编译器强制从内存读取 &isem->data
    uint64_t d = __atomic_load_n((&isem->data), __ATOMIC_RELAXED);
    do
    {
        // 当前信号量超过了最大上限
        if ((d & SEM_VALUE_MASK) == SEM_VALUE_MAX)
        {
            __set_errno (EOVERFLOW);
            return -1;
        }
    } // 期望是 d这个值，然后与 d + 1 交换，d 会被修改为旧值
//    while (!atomic_compare_exchange_weak_release (&isem->data, &d, d + 1));
    while (!__atomic_compare_exchange_n((&isem->data), (&d), (d + 1), 1, __ATOMIC_RELEASE, __ATOMIC_RELAXED));
    // SEM_NWAITERS_SHIFT = 32，右移 32位。 如果 +1 成功，则判断高32 有没有值
    // 如果条件成立，就会进入内核态去唤醒等待中的线程
    if ((d >> SEM_NWAITERS_SHIFT) > 0) {
        // 高32位有值，说明有人在等待，唤醒队列
        // 小端机器，SEM_VALUE_OFFSET = 0
        futex_wake (((unsigned int *) &isem->data) + SEM_VALUE_OFFSET, 1, private);
    }
    return 0;
}
static __always_inline void futex_wake (unsigned int* futex_word, int processes_to_wake, int private)
{
//    int res = lll_futex_wake (futex_word, processes_to_wake, private);
//    int res = lll_futex_syscall (4, futex_word, __lll_private_flag (FUTEX_WAKE, private), processes_to_wake, 0);
    int res = INTERNAL_SYSCALL (futex, 4, futex_word, __lll_private_flag (FUTEX_WAKE, private), processes_to_wake, 0);
    if (res >= 0)
        return;
    // 代码省略
}

等待获取信号量

void PosixSemaphore::wait() {
    int ret;
    do {
        ret = sem_wait(&_semaphore);
        // EINTR = 4
        // ret == 0 表示成功。 ret == EAGAIN == 11，表示需要重试，但也是成功。
        // 所以这里的判断条件要这么写
    } while (ret != 0 && errno == EINTR);
}
  • return: 0 表示获取到信号量， EAGAIN==11 表示需要重试(当前值与目标值不一致)。
int __new_sem_wait (sem_t *sem) {
    // 我们需要在这里检查我们是否需要对取消请求采取行动，
    // 因为 POSIX 指定取消点在 sem_wait 和 sem_timedwait 中“应发生”，
    // 这也意味着无论是否阻塞它们都需要检查这一点（与“可能发生”功能不同）。
    __pthread_testcancel ();
    // 第二个参数传0，表示只尝试一次 cas 
    if (__new_sem_wait_fast ((struct new_sem *) sem, 0) == 0)
        // 如果获取到信号量，直接返回
        return 0;
    else
        return __new_sem_wait_slow64 ((struct new_sem *) sem, CLOCK_REALTIME, NULL);
}

__new_sem_wait_fast

static int __new_sem_wait_fast (struct new_sem *sem, int definitive_result) {
    // 原子从内存加载一个值
    uint64_t d = atomic_load_relaxed (&sem->data);
    do
    {
        // SEM_VALUE_MASK 对应的是 0xFFFFFFFF
        if ((d & SEM_VALUE_MASK) == 0) // == 0 说明低32位都是0。也就是信号量不足,必须等待
            break;
        // 原子地将 data - 1，且此时低32位不为0，
        if (atomic_compare_exchange_weak_acquire(&sem->data, &d, d - 1))
            // 由于操作之前 data > 0，所以走到这里就说明获取到了信号量
            return 0;
    }
    while (definitive_result);
    return -1;
}

__new_sem_wait_slow64

  • return: 0 表示获取到信号量， EAGAIN==11 表示需要重试(当前值与目标值不一致)。
static int
__attribute__ ((noinline))
__new_sem_wait_slow64 (struct new_sem *sem, clockid_t clockid,
		       const struct __timespec64 *abstime)
{
    int err = 0;
    // SEM_NWAITERS_SHIFT = 32。相当于拿 data 的高32位来记录当前有几个等待者。
    uint64_t d = atomic_fetch_add_relaxed (&sem->data, (uint64_t) 1 << SEM_NWAITERS_SHIFT);
    pthread_cleanup_push (__sem_wait_cleanup, sem);
    for (;;) {
        // SEM_VALUE_MASK = 0xFFFFFFFF，如果信号量为0
        if ((d & SEM_VALUE_MASK) == 0) {
            // 进入等待状态。 
            // err = 0 表示获取到信号量， err = EAGAIN = 11 表示需要重试(当前值与目标值不一致)。
            err = do_futex_wait (sem, clockid, abstime);
            // 遇到错误。超时 | | 栈溢出
            if (err == ETIMEDOUT || err == EINTR || err == EOVERFLOW) {
                __set_errno (err);
                err = -1;
                /* Stop being registered as a waiter.  */
                // 取消等待中的状态
                atomic_fetch_add_relaxed (&sem->data,
                -((uint64_t) 1 << SEM_NWAITERS_SHIFT));
                break;
            }
            // 由于被唤醒了，大概率是信号量不为0了，所以再次尝试从内存获取一下 data 值
            d = atomic_load_relaxed (&sem->data);
        } else {
            // 尝试获取一下锁，由于高32 + 1 和 低 32 + 1。所以通过参数3就可以还原回去。
            if (atomic_compare_exchange_weak_acquire (&sem->data, 
                                  &d, d - 1 - ((uint64_t) 1 << SEM_NWAITERS_SHIFT)))
	        {
                err = 0;
                break;
            }
        }
    }
    // 此操作会将高32位的等待获取信号量的个数 -1
    pthread_cleanup_pop (0);
    return err;
}
static void __sem_wait_cleanup (void *arg) {
    struct new_sem *sem = (struct new_sem *) arg;
    // SEM_NWAITERS_SHIFT = 32，就是将高32的值 -1
    atomic_fetch_add_relaxed (&sem->data, -((uint64_t) 1 << SEM_NWAITERS_SHIFT));
}

do_futex_wait

  • return: 0 表示获取到信号量， EAGAIN==11 表示需要重试(当前值与目标值不一致)。
static int
__attribute__ ((noinline))
do_futex_wait (struct new_sem *sem, clockid_t clockid,
	       const struct __timespec64 *abstime)
{
    int err;
    // clockid = CLOCK_REALTIME， abstime = NULL
    err = __futex_abstimed_wait_cancelable64 (
      (unsigned int *) &sem->data + SEM_VALUE_OFFSET, 0,
      clockid, abstime,
      sem->private);
    return err;
}
  • return: 0 表示获取到信号量， EAGAIN==11 表示需要重试(当前值与目标值不一致)。
int
__futex_abstimed_wait_cancelable64 (unsigned int* futex_word,
                                    unsigned int expected, clockid_t clockid,
                                    const struct __timespec64* abstime,
                                    int private)
{
    return __futex_abstimed_wait_common (futex_word, expected, clockid,
                                       abstime, private, true);
}
  • return: 0 表示获取到信号量， EAGAIN==11 表示需要重试(当前值与目标值不一致)。
static int
__futex_abstimed_wait_common (unsigned int* futex_word,
                              unsigned int expected, clockid_t clockid,
                              const struct __timespec64* abstime,
                              int private, bool cancel)
{
    int err;
    unsigned int clockbit;
    // 由于 clockid == CLOCK_REALTIME，所以 clockbit = FUTEX_CLOCK_REALTIME 
    clockbit = (clockid == CLOCK_REALTIME) ? FUTEX_CLOCK_REALTIME : 0;
//    int op = __lll_private_flag (FUTEX_WAIT_BITSET | clockbit, private);
    int op = ((FUTEX_WAIT_BITSET | clockbit) | FUTEX_PRIVATE_FLAG) ^ (private);
    err = __futex_abstimed_wait_common64 (futex_word, expected, op, abstime,
					private, cancel);
    switch (err)
    {
        case 0:
        case -EAGAIN:
        case -EINTR:
        case -ETIMEDOUT:
        case -EINVAL:
        case -EOVERFLOW:  /* Passed absolute timeout uses 64 bit time_t type, but
                             underlying kernel does not support 64 bit time_t futex
                             syscalls.  */
          return -err;

        case -EFAULT: /* Must have been caused by a glibc or application bug.  */
        case -ENOSYS: /* Must have been caused by a glibc bug.  */
        /* No other errors are documented at this time.  */
        default:
          futex_fatal_error ();
    }
}
  • return: 0 表示获取到信号量， EAGAIN==11 表示需要重试(当前值与目标值不一致)。
static int
__futex_abstimed_wait_common64 (unsigned int* futex_word,
                                unsigned int expected, int op,
                                const struct __timespec64* abstime,
                                int private, bool cancel)
{
    if (cancel)
        // cancel 为 true
        // 发起一次linux 系统调用，系统函数是 futex_time64。且等待 futex_word 指向的值是 op
        return INTERNAL_SYSCALL_CANCEL (futex_time64, futex_word, op, expected,
				    abstime, NULL /* Unused.  */,
				    FUTEX_BITSET_MATCH_ANY);
    else
        return INTERNAL_SYSCALL_CALL (futex_time64, futex_word, op, expected,
				  abstime, NULL /* Ununsed.  */,
				  FUTEX_BITSET_MATCH_ANY);
}

附录

GCC 规定的 C/C++ 内存模型

  • MEMMODEL_RELAXED = 0: 最宽松的模式，memory_order_relaxed没有happens-before的约束，编译器和处理器可以对memory access做任何的re-order，因此另外的线程不能对其做任何的假设，这种模式下能做的唯一保证，就是一旦线程读到了变量var的最新值，那么这个线程将再也见不到var修改之前的值了。
  • MEMMODEL_CONSUME = 1: 是一个相比Acquire/Release更加宽松的内存模型，对非依赖的变量也去除了happens-before的限制，减少了所需同步的数据量，可以加快执行的速度。相当于 happen-before
  • MEMMODEL_ACQUIRE = 2: 保证在这个操作之后的memory accesses不会重排到这个操作之前去，但是这个操作之前的memory accesses可能会重排到这个操作之后去。
  • MEMMODEL_RELEASE = 3: 保证在这个操作之前的memory accesses不会重排到这个操作之后去，但是这个操作之后的memory accesses可能会重排到这个操作之前去。
  • MEMMODEL_ACQ_REL = 4:
  • MEMMODEL_SEQ_CST = 5: 所有memory accesses操作的重排不允许跨域这个操作，同时这个限制是双向的。
  • MEMMODEL_LAST = 6,

线程模型

CMS GC 的线程模型

G1 GC 的线程模型

G1 GC 相关

G1 垃圾收集器简介

G1 Young GC 启动条件解析

G1 GC 启动并发标记的解析

G1 Mixed GC 启动条件解析

Clone this wiki locally