Skip to content

thread monitor

Jump to bottom
ceragon edited this page Mar 31, 2022 · 7 revisions

monitor

关系

两种用法。而且 wait 和 notify 无法单独使用。

  • 线程1：lock -> unlock | 线程2： lock -> unlock
  • 线程1：lock -> wait -> unlock | 线程2: lock -> notify/notifyAll -> unlock

结构体

class Monitor : public CHeapObj<mtSynchronizer> {
    SplitWord _LockWord ;
    Thread * volatile _owner;
    ParkEvent * volatile _EntryList ;
    ParkEvent * volatile _OnDeck ;
    volatile intptr_t _WaitLock [1] ; // _WaitSet 的锁
    ParkEvent * volatile  _WaitSet ;
    volatile bool     _snuck;
    char _name[MONITOR_NAME_LEN];
};

关于 SplitWord

  • SplitWord 构造允许我们将争用队列 (cxq) 与锁定字节放在一起。
  • 队列元素是 ParkEvents，它们始终在 256 字节地址上对齐 - ParkEvent 的最低有效字节始终为 0。
  • 将锁定字节与队列并置使我们能够轻松避免 lock() 中的竞争
  • 如果我们要为争用队列和锁定指示器使用两个完全独立的字段。
  • 具体来说，托管使我们免受竞争，即线程可能在锁持有者将外部锁放入 unlock() 快速路径后立即将自己排入 lock() 慢速路径。
  • Colocation 允许我们使用快速路径 unlock() 形式，该形式使用 A MEMBAR 而不是 CAS。
  • 在许多平台上，MEMBAR 的本地延迟低于 CAS。
  • 请参阅：+ http://blogs.sun.com/dave/entry/biased_locking_in_hotspot
  • 请注意，我们不使用单词撕裂经典的感觉。
  • lock() 快速路径将 CAS 锁定字，而 unlock() 快速路径将存储到位于锁定字内的锁定字节中。
  • 我们依赖于我们所有的参考平台都具有连贯和原子字节访问的事实。
  • 更准确地说，字节存储以安全、健全和预期的方式对包含字节的全字的 CAS、ST 和 LD 进行互操作。
  • 如果您将 HotSpot 移植到不是这种情况的平台，那么您需要从以下位置更改 unlock() 快速路径：
  • STB;MEMBAR #storeload;
  • LDN 到密码的全字 CAS。

lock&unlock

lock

void Monitor::lock() {
    this->lock(Thread::current());
}
void Monitor::lock(Thread * Self) {
    if (TryFast()) {
        // 说明加锁成功
 Exeunt:
        // _owner 修改为 Self
        set_owner(Self);
        return;
    }
    // 如果当前是 VM Thread，并且处在安全点开启状态，则直接抢占锁。
    bool can_sneak = Self->is_VM_thread() && SafepointSynchronize::is_at_safepoint();
    // 只要锁没有人在使用，就直接抢占
    if (can_sneak && _owner == NULL) {
        _snuck = true;
        // 由 VM_Thread 抢占锁
        goto Exeunt;
    }
    // 尝试自旋获取锁
    if (TrySpin(Self)) goto Exeunt;
    // 检查线程能否阻塞，如果是 VM Thread 则不允许阻塞
    check_block_state(Self);
    if (Self->is_Java_thread()) {
        // 当前是 java 线程
        // tbivm 用于记录当前线程的 PC 计数器的值
        ThreadBlockInVM tbivm((JavaThread *) Self);
        // 尝试进入 _OnDeck，进入失败就等待，直到被唤醒。如果在 _OnDeck 还加锁失败，就再等待，直到被唤醒
        ILock(Self);
    } else {
        ILock(Self);
    }
    goto Exeunt;
}

TryFast

尝试将最低有效位变为 1

const intptr_t _LBIT = 1;
int Monitor::TryFast() {
    // 如果 &_LockWord.FullWord 指向的内容是0，则更改为1。
    // 返回 _LockWord.FullWord 的旧值
    intptr_t v = Atomic::cmpxchg(_LBIT, &_LockWord.FullWord, (intptr_t)0);  // agro ...
    // 旧值为0，说明加锁成功
    if (v == 0) return 1;

    for (;;) {
        // 如果 v的最低位是1，则说明已经被其他线程加锁了
        if ((v & _LBIT) != 0) return 0;
        // 说明最低位是 0。再次尝试让最低位变为 1
        const intptr_t u = Atomic::cmpxchg(v|_LBIT, &_LockWord.FullWord, v);
        // u 是旧值，如果旧值和期望值相等，说明加锁成功
        if (v == u) return 1;
        // 重试
        v = u;
    }
}

TrySpin

int Monitor::TrySpin(Thread *const Self) {
    // 如果可以加锁成功，就返回成功
    if (TryLock()) return 1;
    // 只要不是多核系统，就返回失败（单核没必要自旋）
    if (!os::is_MP()) return 0;

    int Probes = 0;
    int Delay = 0;
    // 自旋尝试次数
    int SpinMax = 20;
    for (;;) {
        intptr_t v = _LockWord.FullWord;
        if ((v & _LBIT) == 0) {
            // 当前没有加锁，尝试 cas 获取锁。方法是将最低有效位置为1
            if (Atomic::cmpxchg(v | _LBIT, &_LockWord.FullWord, v) == v) {
                // 只要旧值 == 期望值，就说嘛加锁成功
                return 1;
            }
            continue;
        }
        // cpu 休息一下
        SpinPause();
        // 自旋次数 +1
        ++Probes;
        // 是否超过自旋次数
        if (Probes > SpinMax) return 0;
        // 0x7 == 111。就是对8取余
        if ((Probes & 0x7) == 0) {
            // 0x7FF = 111_1111_1111
            // 第一次：1
            // 第二次: 11
            // 第三次: 111
            Delay = ((Delay << 1) | 1) & 0x7FF;
            // CONSIDER: Delay += 1 + (Delay/4); Delay &= 0x7FF ;
        }
        // 如果 OFFPROC 中止旋转，请考虑检查 _owner 的 schedctl 状态。 
        // 如果所有者是OFFPROC，那么锁定将不会及时被释放，这表明旋转不会富有成效或盈利。
        // 暂停“延迟”时间单位 - 当前实现中的迭代。 
        // 避免在停滞时产生一致性流量。
        // 可能的延迟方式：PAUSE、SLEEP、MEMBAR #sync、MEMBAR #halt、wr %g0、%asi、gethrtime、rdstick、rdtick、rdtsc 等。
        // 请注意，在 Niagara 级系统上，我们希望在 自旋循环。 
        // N1 和弟兄们在 xbar 上将 L1$ 写入 L2$。 
        // 此外，它们没有像传统 SPARC 处理器那样的 W$。 
        // 我们目前使用 Marsaglia Shift-Xor RNG 循环。
        
        // 看不懂，总之就是延迟一定的时间再循环
        if (Self != NULL) {
            jint rv = Self->rng[0];
            for (int k = Delay; --k >= 0;) {
                rv = MarsagliaXORV(rv);
                if (SafepointMechanism::poll(Self)) return 0;
            }
            Self->rng[0] = rv;
        } else {
            Stall(Delay);
        }
    }
}

TryLock

int Monitor::TryLock() {
    intptr_t v = _LockWord.FullWord;
    for (;;) {
        // 说明已经被加锁了 
        if ((v & _LBIT) != 0) return 0;
        // 尝试将最低位置为1
        const intptr_t u = Atomic::cmpxchg(v | _LBIT, &_LockWord.FullWord, v);
        // 返回的 _LockWord.FullWord 旧值与预期值相等，说明加锁成功
        if (v == u) return 1;
        // 重试
        v = u;
    }
}

ILock

void Monitor::ILock(Thread * Self) {
    // 调用快速获取锁的方法，尝试将锁的最低位置为 1
    if (TryFast()) {
        // 成功
 Exeunt:        
        return;
    }
    // 线程的 _MutexEvent
    ParkEvent * const ESelf = Self->_MutexEvent;
    // 再次自旋获取锁，尽量避免进入等待
    if (TrySpin(Self)) goto Exeunt;
    // 重置 event
    ESelf->reset();
    // 指令和编译器双重屏障
    OrderAccess::fence();
    // 最后一次尝试加锁。如果加锁成功，则返回 1.否则将自己记录为最后一个尝试获得锁的线程
    if (AcquireOrPush(ESelf)) goto Exeunt;
    // 加入编译器屏障，从内存中加载 _OnDeck 的值，只要值不是自己就循环
    while (OrderAccess::load_acquire(&_OnDeck) != ESelf) {
        // 进入等待
        ParkCommon(ESelf, 0);
    }
    // 说明自己在 _OnDeck 的位置了
    for (;;) {
        // 再次自旋加锁
        if (TrySpin(Self)) break;
        // 还是失败，再次等待
        ParkCommon(ESelf, 0);
    }
    // 既然自己拿到了锁，则把 _OnDeck 让出来
    _OnDeck = NULL;
    goto Exeunt;
}

AcquireOrPush

inline int Monitor::AcquireOrPush(ParkEvent *ESelf) {
    intptr_t v = _LockWord.FullWord;
    for (;;) {
        if ((v & _LBIT) == 0) {
            // 无锁，所以又一次尝试加锁
            const intptr_t u = Atomic::cmpxchg(v | _LBIT, &_LockWord.FullWord, v);
            if (u == v) return 1;        // indicate acquired
            v = u;
        } else {
            // 有锁
            // 此时锁信息存储的就是 ParkEvent 的指针，在当前线程的 ParkEvent 中记录这个指针信息
            ESelf->ListNext = (ParkEvent *) (v & ~_LBIT);
            // 尝试把信息替换成自己的信息，因为自己是最后一个进来的
            const intptr_t u = Atomic::cmpxchg(intptr_t(ESelf) | _LBIT, &_LockWord.FullWord, v);
            // 替换成功就返回 0
            if (u == v) return 0;        // indicate pushed onto cxq
            // 替换失败了？重试一次
            v = u;
        }
    }
}

ParkCommon

park() 详看 ParkEvent.park

static int ParkCommon(ParkEvent *ev, jlong timo) {
    int err = OS_OK;
    if (0 == timo) {
        // 无超时等待
        ev->park();
    } else {
        err = ev->park(timo);
    }
    return err;
}

unlock

void Monitor::unlock() {
    // 释放锁
    set_owner(NULL);
    if (_snuck) {
        // 如果锁是被强占的，则置为 false
        _snuck = false;
        return;
    }
    IUnlock(false);
}

IUnlock

void Monitor::IUnlock(bool RelaxAssert) {
    // 原子的将 _LockWord 的低8位设为0
    OrderAccess::release_store(&_LockWord.Bytes[_LSBINDEX], jbyte(0));
    // 指令屏障 + 编译器屏障。将此线程的所有修改刷新到CPU缓存，保证可见性。
    OrderAccess::storeload();
    // 获取当前 _OnDeck 的等待者
    ParkEvent * const w = _OnDeck;
    if (w != NULL) {
        // 既然有等待者，且 ParkEvent 对应的最低位是0，则将 w 唤醒
//        if ((UNS(w) & _LBIT) == 0) w->unpark();
        if ((uintptr_t(w) & _LBIT) == 0) w->unpark();
        return;
    }
    // 锁的当前值
    intptr_t cxq = _LockWord.FullWord;
    // cxq == 0或1 则为真 | _EntryList 为空则为真
    if (((cxq & ~_LBIT) | UNS(_EntryList)) == 0) {
        // 说明没人等待，或等待队列为空
        return;
    }
    if (cxq & _LBIT) {
        // 已经有线程获取到锁了，即便唤醒等待者也白搭
        return;
    }
  Succession:
    if (!Atomic::replace_if_null((ParkEvent*)_LBIT, &_OnDeck)) {
        // _OnDeck 不为空，所以不管了
        return;
    }
    // 如果 _OnDeck 为空，则把 1 置为 _OnDeck 的值
    // List 为当前等待者
    ParkEvent * List = _EntryList;
    if (List != NULL) {
        // 有人在等待，唤醒一个
      WakeOne:
        ParkEvent * const w = List;
        _EntryList = w->ListNext;
        // 将 w 放到 _OnDeck 上
        OrderAccess::release_store(&_OnDeck, w);
        // 双屏障
        OrderAccess::storeload();
        cxq = _LockWord.FullWord;
        // 判断当前是否已经上锁了
        if (cxq & _LBIT) return;
        // 没上锁就把 w 唤醒吧
        w->unpark();
        return;
    }
    // 没有等待着
    cxq = _LockWord.FullWord;
    // 但是有人等待
    if ((cxq & ~_LBIT) != 0) {
        for (;;) {
            // 被上锁了
            if (cxq & _LBIT) goto Punt;
            // _LockWord 是 cxq，则将 _LockWord 最低位置为1，并返回旧值
            const intptr_t vfy = Atomic::cmpxchg(cxq & _LBIT, &_LockWord.FullWord, cxq);
            // 相等说明修改成功
            if (vfy == cxq) break;
            // 重试
            cxq = vfy;
        }
        // 获取到等待者，并将它唤醒
        _EntryList = List = (ParkEvent *)(cxq & ~_LBIT);
        goto WakeOne;
    }
  Punt:  
    // 当前已经上锁了，或者没有人等待
    _OnDeck = NULL;
    // 双屏障
    OrderAccess::storeload();
    cxq = _LockWord.FullWord;
    if ((cxq & ~_LBIT) != 0 && (cxq & _LBIT) == 0) {
        // 有人在等待，且没有人上锁。
        goto Succession;
    }
    return;
}

wait&notify

wait

wait 方法会在两个地方陷入等待状态。

  1. 获取 _WaitLock 时陷入等待。
  2. 成功获取了 _WaitLock，等待被 notify 唤醒。
bool Monitor::wait(bool no_safepoint_check, long timeout,
                   bool as_suspend_equivalent) {
    Thread * const Self = Thread::current();
    int wait_status;
    set_owner(NULL);
    if (no_safepoint_check) {
        wait_status = IWait(Self, timeout);
    } else {
        JavaThread *jt = (JavaThread *)Self;
        ThreadBlockInVM tbivm(jt);
        OSThreadWaitState osts(Self->osthread(), false /* not Object.wait() */);
        if (as_suspend_equivalent) {
            jt->set_suspend_equivalent();
        }
        wait_status = IWait(Self, timeout);
        if (as_suspend_equivalent && jt->handle_special_suspend_equivalent_condition()) {
            IUnlock(true);
            jt->java_suspend_self();
            ILock(Self);
        }
    }
    set_owner(Self);
    return wait_status != 0;
}

IWait

int Monitor::IWait(Thread * Self, jlong timo) {
    // Phases:
    // 1. 在 WaitSet 上将 Self 加入队列 - 目前在前面
    // 2. unlock - 放下外层的锁
    // 3. 等待通知或超时
    // 4. lock - reentry - reacquire 外层锁
    ParkEvent * const ESelf = Self->_MutexEvent;
    ESelf->Notified = 0;
    // _event = 0
    ESelf->reset();
    // 双屏障
    OrderAccess::fence();
    // 将自身添加到 WaitSet
    //  理想情况下，只有外锁的持有者会操纵 WaitSet——也就是说，外锁会隐式地保护 WaitSet。
    // 但是如果 wait() 中的线程遇到超时，它将需要在成为锁的所有者之前从 WaitSet 中取出自己。
    // 我们需要出队，因为作为线程代理的 ParkEvent 不能同时驻留在 WaitSet 和 EntryList|cxq 上。
    // 也就是说，WaitSet 上的线程不能允许竞争锁，直到它设法将其 ParkEvent 与 WaitSet 解除链接。
    // 因此需要WaitLock。 WaitLock 上的争用是最小的。
    //  另一种可行的方法是将另一个 ParkEvent，“WaitEvent”添加到线程类。 
    // WaitSet 将由WaitEvents 组成。只有外部锁的所有者会操纵 WaitSet。
    // 然后，wait() 中的线程可以竞争外部锁，然后，如果有必要，只有在获得外部锁之后，才能将自己与 WaitSet 断开链接。
    // 更准确地说，不会有 WaitLock。 wait() 中的线程会将其 WaitEvent 排入 WaitSet 队列；释放外锁；
    // 等待通知或超时；重新获得内锁；然后，如果需要，将自身与 WaitSet 取消链接。
    //  或者，ParkEvent 中的第二组列表链接字段可能就足够了。一组用于 WaitSet，一组用于 EntryList。
    // 我们还可以将 ParkEvent 解构为“纯”事件，并添加一个新的不朽/TSM“ListElement”类来引用 ParkEvents。
    // 在这种情况下，我们可以在 WaitSet 上拥有一个 ListElement，在 EntryList 上拥有另一个，两者都指向同一个纯事件。
    
    // 将 Thread 的 _MuxEvent 置入 _WaitLock 即可加锁成功
    Thread::muxAcquire(_WaitLock, "wait:WaitLock:Add");
    // ESelf = _MutexEvent。 _WaitSet 是当前的等待者
    // 这两行代码的作用是，将当前等待者排到自己后面，将自己作为当前等待者
    ESelf->ListNext = _WaitSet;
    _WaitSet = ESelf;
    // 尝试将 _WaitLock 从1置为0，成功则释放 _MuxEvent 成功
    // 失败则唤醒下一个等待者(调用 _MuxEvent.unpark)，并选择一个新的等待者
    Thread::muxRelease(_WaitLock);
    // 释放外层的锁，如果有等待者，调用  _MutexEvent.unpark()
    // 这里必须释放，否则 Notify 的线程会 lock 等待
    IUnlock(true);
    for (;;) {
        // 已经收到了通知，退出
        if (ESelf->Notified) break;
        // 进入等待，直到被唤醒。调用的是 _MutexEvent.park()
        // 注意：这里有三个情况被唤醒。
        // 1. timo 超时
        // 2. 有一个不调用 wait 的线程获取了锁，然后释放了锁，此时把 Self 唤醒
        // 3. 有一个 notify 的线程获取了锁，调用完 notify 后，释放了锁，此时把 Self 唤醒
        int err = ParkCommon(ESelf, timo);
        // 如果是情况1 和 情况3，都会跳出循环。而情况2会重新 Park
        if (err == OS_TIMEOUT) break;
    }
    // 双屏障
    OrderAccess::fence();
    // 自己在等待队列中吗
    int WasOnWaitSet = 0;
    // 如果是超时唤醒的，Notified 还会是 0
    if (ESelf->Notified == 0) {
        // 再次获取 _WaitLock，但这次是从等待队列中移除
        Thread::muxAcquire(_WaitLock, "wait:WaitLock:remove");
        // 再次确认没有受到通知
        if (ESelf->Notified == 0) {
            // p ： 当前等待者
            ParkEvent * p = _WaitSet;
            ParkEvent * q = NULL;
            // 从 _WaitSet 中通过遍历找到自身
            while (p != NULL && p != ESelf) {
                q = p;
                p = p->ListNext;
            }
            // 说明自己当前处于等待的首位
            if (p == _WaitSet) {      // found at head
                _WaitSet = p->ListNext;
            } else {                  // found in interior
                // 通过这种方式，将自己从队列中移除
                q->ListNext = p->ListNext;
            }
            // 自己是在等待队列中被唤醒的，说白了就是超时且没有notify 被唤醒的
            WasOnWaitSet = 1;
        }
        // 释放 _WaitLock，并唤醒等待者
        Thread::muxRelease(_WaitLock);
    }
    if (WasOnWaitSet) {
        // 这里可以理解为，调用了 wait(time) 方法，在 time 到时间后就走到了这里
        ILock(Self);
    } else {
        // 不在等待队列，说明自己是被其他线程调用 notify 后唤醒的。
        for (;;) {
            // 如果自己刚好处于 _OnDeck 的位置，则尝试自旋加锁，对 _LockWord 加锁
            if (OrderAccess::load_acquire(&_OnDeck) == ESelf && TrySpin(Self)) break;
            // 不成功就陷入等待，调用的是 _MutexEvent.park()
            ParkCommon(ESelf, 0);
        }
        // 由于加锁成功，所以自己要从 _OnDeck 上离开
        _OnDeck = NULL;
    }
    return WasOnWaitSet != 0;
}

notify

notify 主要作用就是设置 _WaitSet 的唤醒标记

bool Monitor::notify() {
    // 没有等待者，无需唤醒
    if (_WaitSet == NULL) return true;
    // 因为下面涉及修改 _WaitSet，所以先加锁。
    Thread::muxAcquire(_WaitLock, "notify:WaitLock");
    // _WaitSet 对应的是 _MutexEvent，所以 nfy 是 _MutexEvent
    ParkEvent * nfy = _WaitSet;
    // double check
    if (nfy != NULL) {
        // nfy->ListNext 其实是前一个等待者（看wait方法就知道了）
        // 将 nfy->ListNext 作为下一个唤醒者
        _WaitSet = nfy->ListNext;
        for (;;) {
            const intptr_t v = _LockWord.FullWord;
            // (v & ~_LBIT) 表示当前等待者
            nfy->ListNext = (ParkEvent *)(v & ~_LBIT);
            // 这一步的目的是帮 nfy 获取锁。
            if (Atomic::cmpxchg(intptr_t(nfy)|_LBIT, &_LockWord.FullWord, v) == v) break;
        }
        // 双屏障
        OrderAccess::fence();
        // 唤醒标记
        nfy->Notified = 1;
    }
    // 已经唤醒了一个等待者，由于Acquire 用的是 _MuxEvent，所以如果有等待者，调用的是 _MuxEvent.unpark()
    Thread::muxRelease(_WaitLock);
    return true;
}

notifyAll

bool Monitor::notify_all() {
    while (_WaitSet != NULL) notify();
    return true;
}

通用方法

OrderAccess

release_store

  • temp = *p
  • *p = v
  • v = temp
template <typename T, typename D>
inline void OrderAccess::release_store(volatile D* p, T v) {
  StoreImpl<T, D, PlatformOrderedStore<sizeof(D), RELEASE_X> >()(v, p);
}

template<size_t byte_size, ScopedFenceType type>
struct OrderAccess::PlatformOrderedStore {
  template <typename T>
  void operator()(T v, volatile T* p) const {
    ordered_store<T, type>(p, v);
  }
};

template<>
struct OrderAccess::PlatformOrderedStore<8, RELEASE_X_FENCE>
{
  template <typename T>
  void operator()(T v, volatile T* p) const {
      // 交换 (p), v 
    __asm__ volatile (  "xchgq (%2), %0"
                      : "=r" (v)
                      : "0" (v), "r" (p)
                      : "memory");
  }
};

storeload

inline void OrderAccess::storeload()  { fence();            }
inline void OrderAccess::fence() {
    // 原子地将立即数0添加到 rsp 寄存器寻址后的地址上。等于一个空操作，目的是为了禁止 CPU 指令在这里乱序执行。
    __asm__ volatile ("lock; addl $0,0(%%rsp)" : : : "cc", "memory");
    // 编译器屏障
    compiler_barrier();
}
static inline void compiler_barrier() {
  __asm__ volatile ("" : : : "memory");
}

fence

双屏障

inline void OrderAccess::fence() {
    __asm__ volatile ("lock; addl $0,0(%%rsp)" : : : "cc", "memory");
    compiler_barrier();
}

load_acquire

template <typename T>
inline T OrderAccess::load_acquire(const volatile T* p) {
  return LoadImpl<T, PlatformOrderedLoad<sizeof(T), X_ACQUIRE> >()(p);
}
template<size_t byte_size, ScopedFenceType type>
struct OrderAccess::PlatformOrderedLoad {
  template <typename T>
  T operator()(const volatile T* p) const {
    return ordered_load<T, type>(p);
  }
};
template <typename FieldType, ScopedFenceType FenceType>
inline FieldType OrderAccess::ordered_load(const volatile FieldType* p) {
  ScopedFence<FenceType> f((void*)p);
  return Atomic::load(p);
}

Atomic

xchg

template<typename T, typename D>
inline D Atomic::xchg(T exchange_value, volatile D* dest, atomic_memory_order order) {
  return XchgImpl<T, D>()(exchange_value, dest, order);
}

template<typename T>
struct Atomic::XchgImpl<
  T, T,
  typename EnableIf<IsIntegral<T>::value || IsRegisteredEnum<T>::value>::type>
{
  T operator()(T exchange_value, T volatile* dest, atomic_memory_order order) const {
    // Forward to the platform handler for the size of T.
    return PlatformXchg<sizeof(T)>()(exchange_value, dest, order);
  }
};

template<>
template<typename T>
inline T Atomic::PlatformXchg<8>::operator()(T exchange_value, T volatile* dest,
                                             atomic_memory_order order) const {
  STATIC_ASSERT(8 == sizeof(T));
  // temp = *dest，*dest = exchange_value, exchange_value = temp
  __asm__ __volatile__ ("xchgq (%2),%0"
                        : "=r" (exchange_value)
                        : "0" (exchange_value), "r" (dest)
                        : "memory");
  return exchange_value;
}

replace_if_null

template<typename T, typename D>
inline bool Atomic::replace_if_null(T* value, D* volatile* dest,
                                    atomic_memory_order order) {
  D* expected_null = NULL;
  // 期望 *dest 的值是 NULL，然后替换为 value，此时返回真，反之为假
  return expected_null == cmpxchg(value, dest, expected_null, order);
}

template<typename T, typename D, typename U>
inline D Atomic::cmpxchg(T exchange_value,
                         D volatile* dest,
                         U compare_value,
                         atomic_memory_order order) {
  return CmpxchgImpl<T, D, U>()(exchange_value, dest, compare_value, order);
}

template<typename T>
struct Atomic::CmpxchgImpl<
  T, T, T,
  typename EnableIf<IsIntegral<T>::value || IsRegisteredEnum<T>::value>::type>
{
  T operator()(T exchange_value, T volatile* dest, T compare_value,
               atomic_memory_order order) const {
    return PlatformCmpxchg<sizeof(T)>()(exchange_value,
                                        dest,
                                        compare_value,
                                        order);
  }
};

template<>
template<typename T>
inline T Atomic::PlatformCmpxchg<8>::operator()(T exchange_value,
                                                T volatile* dest,
                                                T compare_value,
                                                atomic_memory_order /* order */) const {
  STATIC_ASSERT(8 == sizeof(T));
  // 比较并交换
  // if (*dest == compare_value) { *dest = exchange_value; rax 保持不变 }
  // if (*dest != compare_value) { rax = *dest; }
  __asm__ __volatile__ ("lock cmpxchgq %1,(%3)"
                        : "=a" (exchange_value)
                        : "r" (exchange_value), "a" (compare_value), "r" (dest)
                        : "cc", "memory");
  // 返回 rax 寄存器的内容
  return exchange_value;
}

load

在 linux_x86 下，就是直接返回 *dest 指针的内容

template<typename T>
inline T Atomic::load(const volatile T* dest) {
  return LoadImpl<T, PlatformLoad<sizeof(T)> >()(dest);
}
template<size_t byte_size>
struct Atomic::PlatformLoad {
  template<typename T>
  T operator()(T const volatile* dest) const {
    STATIC_ASSERT(sizeof(T) <= sizeof(void*)); // wide atomics need specialization
    return *dest;
  }
};

线程模型

CMS GC 的线程模型

G1 GC 的线程模型

G1 GC 相关

G1 垃圾收集器简介

G1 Young GC 启动条件解析

G1 GC 启动并发标记的解析

G1 Mixed GC 启动条件解析

Clone this wiki locally