Java 线程同步原理探析

1    public final native void wait(long timeout) throws InterruptedException;
2    public final native void notify();

1// java.base/share/native/libjava/Object.c
2static JNINativeMethod methods[] = {
3    {"hashCode",    "()I",                    (void *)&JVM_IHashCode},
4    {"wait",        "(J)V",                   (void *)&JVM_MonitorWait},
5    {"notify",      "()V",                    (void *)&JVM_MonitorNotify},
6    {"notifyAll",   "()V",                    (void *)&JVM_MonitorNotifyAll},
7    {"clone",       "()Ljava/lang/Object;",   (void *)&JVM_Clone},
8};

 1// LockSupport.java
 2    public static void unpark(Thread thread) {
 3        if (thread != null)
 4            UNSAFE.unpark(thread);
 5    }
 6    public static void park(Object blocker) {
 7        Thread t = Thread.currentThread();
 8        setBlocker(t, blocker);
 9        UNSAFE.park(false, 0L);
10        setBlocker(t, null);
11    }
12// Unsafe.java
13    /**
14     * Unblocks the given thread blocked on {@code park}, or, if it is
15     * not blocked, causes the subsequent call to {@code park} not to
16     * block.  Note: this operation is "unsafe" solely because the
17     * caller must somehow ensure that the thread has not been
18     * destroyed. Nothing special is usually required to ensure this
19     * when called from Java (in which there will ordinarily be a live
20     * reference to the thread) but this is not nearly-automatically
21     * so when calling from native code.
22     *
23     * @param thread the thread to unpark.
24     */
25    @HotSpotIntrinsicCandidate
26    public native void unpark(Object thread);
27
28    /**
29     * Blocks current thread, returning when a balancing
30     * {@code unpark} occurs, or a balancing {@code unpark} has
31     * already occurred, or the thread is interrupted, or, if not
32     * absolute and time is not zero, the given time nanoseconds have
33     * elapsed, or if absolute, the given deadline in milliseconds
34     * since Epoch has passed, or spuriously (i.e., returning for no
35     * "reason"). Note: This operation is in the Unsafe class only
36     * because {@code unpark} is, so it would be strange to place it
37     * elsewhere.
38     */
39    @HotSpotIntrinsicCandidate
40    public native void park(boolean isAbsolute, long time);
41
42// hotspot/share/prims/unsafe.cpp
43UNSAFE_ENTRY(void, Unsafe_Park(JNIEnv *env, jobject unsafe, jboolean isAbsolute, jlong time)) {
44  HOTSPOT_THREAD_PARK_BEGIN((uintptr_t) thread->parker(), (int) isAbsolute, time);
45  EventThreadPark event;
46
47  JavaThreadParkedState jtps(thread, time != 0);
48  thread->parker()->park(isAbsolute != 0, time);
49  if (event.should_commit()) {
50    post_thread_park_event(&event, thread->current_park_blocker(), time);
51  }
52  HOTSPOT_THREAD_PARK_END((uintptr_t) thread->parker());
53} UNSAFE_END
54    

 1// hotspot/share/runtime/park.hpp
 2class Parker : public os::PlatformParker
 3...
 4public:
 5  // For simplicity of interface with Java, all forms of park (indefinite,
 6  // relative, and absolute) are multiplexed into one call.
 7  void park(bool isAbsolute, jlong time);
 8  void unpark();
 9
10// hotspot/os/posix/os_posix.hpp
11class PlatformParker : public CHeapObj<mtInternal> {
12 protected:
13  enum {
14    REL_INDEX = 0,
15    ABS_INDEX = 1
16  };
17  int _cur_index;  // which cond is in use: -1, 0, 1
18  pthread_mutex_t _mutex[1];
19  pthread_cond_t  _cond[2]; // one for relative times and one for absolute
20  ...
21};

 1struct semaphore {
 2     int val;
 3     thread_list waiting;  // List of threads waiting for semaphore
 4}
 5wait(semaphore Sem):    // Wait until > 0 then decrement
 6  // 这里用的是 while 而不是 if
 7  // 这是因为在 wait 过程中，其他线程还可能继续调用 wait
 8  while (Sem.val <= 0) {
 9    add this thread to Sem.waiting;
10    block(this thread);
11  }
12  Sem.val = Sem.val - 1;
13return;
14
15signal(semaphore Sem):// Increment value and wake up next thread
16     Sem.val = Sem.val + 1;
17     if (Sem.waiting is nonempty) {
18         remove a thread T from Sem.waiting;
19         wakeup(T);
20     }

 1emptySem = N
 2fullSem = 0
 3// Producer
 4while(whatever) {
 5    locally generate item
 6    wait(emptySem)
 7    fill empty buffer with item
 8    signal(fullSem)
 9}
10// Consumer
11while(whatever) {
12    wait(fullSem)
13    get item from full buffer
14    signal(emptySem)
15    use item
16}

1// Intialize: 
2sem_init(&theSem, 0, initialVal);
3// Wait: 
4sem_wait(&theSem);
5// Signal: 
6sem_post(&theSem);
7// Get the current value of the semaphore:       
8sem_getvalue(&theSem, &result);

1// initialize
2pthread_cond_init() 
3pthread_cond_wait(&theCV, &someLock);
4pthread_cond_signal(&theCV);
5pthread_cond_broadcast(&theCV);

 1pthread_mutex_t myLock;
 2pthread_cond_t myCV;
 3int count = 0;
 4
 5// Thread A
 6pthread_mutex_lock(&myLock);
 7while(count < 0) {
 8    pthread_cond_wait(&myCV, &myLock);
 9}
10pthread_mutex_unlock(&myLock);
11
12// Thread B
13
14pthread_mutex_lock(&myLock);
15count ++;
16while(count == 10) {
17    pthread_cond_signal(&myCV);
18}
19pthread_mutex_unlock(&myLock);

1    /**
2     * The number of nanoseconds for which it is faster to spin
3     * rather than to use timed park. A rough estimate suffices
4     * to improve responsiveness with very short timeouts.
5     */
6    static final long spinForTimeoutThreshold = 1000L;

 1// AQS 的 doAcquireNanos 方法节选
 2for (;;) {
 3    final Node p = node.predecessor();
 4    if (p == head && tryAcquire(arg)) {
 5        setHead(node);
 6        p.next = null; // help GC
 7        failed = false;
 8        return true;
 9    }
10    nanosTimeout = deadline - System.nanoTime();
11    if (nanosTimeout <= 0L)
12        return false;
13    if (shouldParkAfterFailedAcquire(p, node) &&
14        nanosTimeout > spinForTimeoutThreshold)
15        LockSupport.parkNanos(this, nanosTimeout);
16    if (Thread.interrupted())
17        throw new InterruptedException();
18}

 1    public void interrupt() {
 2        if (this != Thread.currentThread())
 3            checkAccess();
 4
 5        synchronized (blockerLock) {
 6            Interruptible b = blocker;
 7            if (b != null) {
 8                interrupt0();           // Just to set the interrupt flag
 9                b.interrupt(this);
10                return;
11            }
12        }
13        interrupt0();
14    }

 1    protected final void begin() {
 2        if (interruptor == null) {
 3            interruptor = new Interruptible() {
 4                    public void interrupt(Thread target) {
 5                        synchronized (closeLock) {
 6                            if (closed)
 7                                return;
 8                            closed = true;
 9                            interrupted = target;
10                            try {
11                                AbstractInterruptibleChannel.this.implCloseChannel();
12                            } catch (IOException x) { }
13                        }
14                    }};
15        }
16        blockedOn(interruptor);
17        Thread me = Thread.currentThread();
18        if (me.isInterrupted())
19            interruptor.interrupt(me);
20    }