thread mux

Thread mux

对于 Thread 的 _MuxEvent 进行 acquire 和 release

Thread 相关锁

class Thread: public ThreadShadow {
    ParkEvent * _ParkEvent;                     // for synchronized()
    ParkEvent * _SleepEvent;                    // for Thread.sleep
    ParkEvent * _MutexEvent;                    // for native internal Mutex/Monitor
    ParkEvent * _MuxEvent;                      // for low-level muxAcquire-muxRelease
};

结构体

class ParkEvent : public os::PlatformEvent {
    ParkEvent * volatile ListNext ;
    volatile intptr_t OnList ;
};

class PlatformEvent : public CHeapObj<mtSynchronizer> {
    volatile int _event;  // Event count/permit: -1, 0 or 1
    void reset() { _event = 0; }
};

muxAcquire

Self->park(); 详看 ParkEvent.park

const intptr_t LOCKBIT = 1;
void Thread::muxAcquire(volatile intptr_t * Lock, const char * LockName) {
    // 期望 Lock 的值是0，然后置为1。返回 Lock 的旧值 w
    intptr_t w = Atomic::cmpxchg(LOCKBIT, Lock, (intptr_t) 0);
    // 旧值为0，加锁成功
    if (w == 0) return;
    // 旧值不为0，但是最低有效位是 0，且能把 w 的最低位置为1，也算是加锁成功
    if ((w & LOCKBIT) == 0 && Atomic::cmpxchg(w | LOCKBIT, Lock, w) == w) {
        return;
    }
    ParkEvent *const Self = Thread::current()->_MuxEvent;
    for (;;) {
        // 如果是多核心，its = 101
        int its = (os::is_MP() ? 100 : 0) + 1;

        // Optional spin phase: spin-then-park strategy
        while (--its >= 0) {
            w = *Lock;
            // 尝试加锁
            if ((w & LOCKBIT) == 0 && Atomic::cmpxchg(w | LOCKBIT, Lock, w) == w) {
                // 加锁成功
                return;
            }
        }
        // _event = 0
        Self->reset();
        // 在 OnList 中保存 Lock 的指针
        Self->OnList = intptr_t(Lock);
        // 双屏障
        OrderAccess::fence();
        for (;;) {
            w = *Lock;
            if ((w & LOCKBIT) == 0) {
                // 无锁，尝试加锁
                if (Atomic::cmpxchg(w | LOCKBIT, Lock, w) == w) {
                    // 加锁成功， 清除 OnList 记录
                    Self->OnList = 0;   // hygiene - allows stronger asserts
                    return;
                }
                // 自旋重试
                continue;      // Interference -- *Lock changed -- Just retry
            }
            // w 除了最低有效位，其余位实际上是某个等待中的线程的 ParkEvent*
            // 所以此步骤会使得当前等待中的线程排在自己后面
            Self->ListNext = (ParkEvent *) (w & ~LOCKBIT);
            // 将自己的 ParkEvent* 最低位置为1，并尝试替换 Lock
            if (Atomic::cmpxchg(intptr_t(Self) | LOCKBIT, Lock, w) == w) break;
        }
        // 如果自己还在队列中，就陷入等待。释放锁的线程负责把 OnList 置为 0
        while (Self->OnList != 0) {
            // 陷入等待，直到被唤醒
            Self->park();
        }
    }
}

muxRelease

void Thread::muxRelease(volatile intptr_t *Lock) {
    for (;;) {
        // 期望 Lock 是 1，将 Lock 置为 0。返回 Lock 旧值
        const intptr_t w = Atomic::cmpxchg((intptr_t) 0, Lock, LOCKBIT);
        // 旧值 == 1，修改成功
        if (w == LOCKBIT) return;
        // 旧值不为1，则有线程在等待。获取等待线程的 ParkEvent*
        ParkEvent *const List = (ParkEvent *) (w & ~LOCKBIT);
        // 之所以这里获取的是 ListNext，是因为遵循的后到先得的原则，让先进入等待的线程继续等待
        ParkEvent *const nxt = List->ListNext;
        // 尝试把 Lock 置为下一个等待者的 ParkEvent
        if (Atomic::cmpxchg(intptr_t(nxt), Lock, w) != w) {
            // 操作失败，重试
            continue;
        }
        // 本次的目的就是唤醒最后一个进入等待的等待者，所以将它的 OnList 标记为0
        List->OnList = 0;
        // 双屏障
        OrderAccess::fence();
        // 唤醒等待者
        List->unpark();
        return;
    }
}