java同步机制之LockSupport::park/unpark实现溯源-part4

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class LockSupport {
  // Disables the current thread for thread scheduling purposes unless the permit is available.
  public static void park(Object blocker) {
        Thread t = Thread.currentThread();
        setBlocker(t, blocker);
        UNSAFE.park(false, 0L);
        setBlocker(t, null);
    }
  
  public static void unpark(Thread thread) {
        if (thread != null)
            U.unpark(thread);
  }
  
  private static void setBlocker(Thread t, Object arg) {
        UNSAFE.putObject(t, parkBlockerOffset, arg);
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// These are the methods for 1.8.0
// 方法名映射
static JNINativeMethod methods_18[] = {
    ...
    ...
    {CC "park",               CC "(ZJ)V",                  FN_PTR(Unsafe_Park)},
    {CC "unpark",             CC "(" OBJ ")V",               FN_PTR(Unsafe_Unpark)}
};

UNSAFE_ENTRY(void, Unsafe_Park(JNIEnv *env, jobject unsafe, jboolean isAbsolute, jlong time))
  UnsafeWrapper("Unsafe_Park");
  EventThreadPark event;
  ...
  JavaThreadParkedState jtps(thread, time != 0);
  // 重要方法
  thread->parker()->park(isAbsolute != 0, time);
	...
UNSAFE_END

1
2
3
4
5
6
7
8
9
10
11
12
13
UNSAFE_ENTRY(void, Unsafe_Unpark(JNIEnv *env, jobject unsafe, jobject jthread))
  UnsafeWrapper("Unsafe_Unpark");
  Parker* p = NULL;
  ...
  java_thread = JNIHandles::resolve_non_null(jthread);
  // jvm线程
  JavaThread* thr = java_lang_Thread::thread(java_thread);
  p = thr->parker();
  if (p != NULL) {
    // 关键点
    p->unpark();
  }
UNSAFE_END

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// Class hierarchy
// - Thread
//   - NamedThread
//     - VMThread
//     - ConcurrentGCThread
//     - WorkerThread
//       - GangWorker
//       - GCTaskThread
//   - JavaThread
//   - WatcherThread

class Thread: public ThreadShadow {
  
  protected:
  // OS data associated with the thread
  OSThread* _osthread;  // Platform-specific thread information
  
  ......
}

class JavaThread: public Thread {
	oop          _threadObj; // The Java level thread object
	
	// JSR166 per-thread parker
	private:
  	Parker*    _parker;  
  
  ......
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class Parker : public os::PlatformParker {
private:
  // 重要变量，通过0/1表示授权，决定是否阻塞
  volatile int _counter ;
  JavaThread * AssociatedWith ; // Current association

public:
  Parker() : PlatformParker() {
    _counter       = 0 ;
    AssociatedWith = NULL ;
  }

public:
  void park(bool isAbsolute, jlong time);
  void unpark();
  ...
  ...
};

1
2
3
4
5
6
7
8
9
class PlatformParker : public CHeapObj<mtInternal> {
    int _cur_index;  // which cond is in use: -1, 0, 1
    // 锁
    pthread_mutex_t _mutex [1] ;
    // 条件变量
    pthread_cond_t  _cond  [2] ; // one for relative times and one for abs.
    ...
    ...
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
void Parker::park(bool isAbsolute, jlong time) {
  // 原_counter不为零，有权限，不需等待
  // _counter 被设置为0
  if (Atomic::xchg(0, &_counter) > 0) return;
  
	// 加锁
  if (Thread::is_interrupted(thread, false) || pthread_mutex_trylock(_mutex) != 0) {
    return;
  }

	......
	
  if (time == 0) {
  	// 等待
    status = pthread_cond_wait (&_cond[_cur_index], _mutex) ;
  } else {
    // 计时等待
    status = os::Linux::safe_cond_timedwait (&_cond[_cur_index], _mutex, &absTime) ;
  }
  
  ......
  
	// 从pthread_cond_wait return表明接收到signal，_counter=1
	// 授权已经使用一次，作废
  _counter = 0 ;
  status = pthread_mutex_unlock(_mutex) ;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
void Parker::unpark() {
  // 整体按照加锁->通知->解锁的顺序
  int s, status ;
  status = pthread_mutex_lock(_mutex);
  s = _counter;
  _counter = 1;
  if (s < 1) {
    // thread might be parked
    if (_cur_index != -1) {
      // thread is definitely parked
      if (WorkAroundNPTLTimedWaitHang) {
        status = pthread_cond_signal (&_cond[_cur_index]);
        status = pthread_mutex_unlock(_mutex);
      } else {
        int index = _cur_index;
        status = pthread_mutex_unlock(_mutex);
        status = pthread_cond_signal (&_cond[index]);
      }
    } else {
      pthread_mutex_unlock(_mutex);
    }
  } else {
    pthread_mutex_unlock(_mutex);
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
#include <stdio.h>
...

int main() {
    pthread_cond_t cond;
    pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

    if (initializeCond(&cond)) {
        return -1;
    }
    ...
    
    struct timespec start, end, deadline;
    if (pthread_mutex_lock(&lock)) {
        return -1;
    }
    clock_gettime(CLOCK_TYPE, &start);
    addNanos(&start, &deadline, deltaNanos);

    int iteration = 0;
    while (!isDeadlineReached(&deadline)) {
        // pthread_cond_timedwait(&cond, &lock, &deadline);
    	  pthread_cond_wait(&cond, &lock);
        iteration++;;
    }
		...
    clock_gettime(CLOCK_TYPE, &end);
    return 0;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public class Object {
  // Causes the current thread to wait until it is awakened,
  // typically by being notified or interrupted
	public final void wait() throws InterruptedException {
        wait(0L);
  }
  
  public final native void wait(long timeoutMillis) throws InterruptedException;
  
  // Wakes up a single thread that is waiting on this object's monitor. 
  // If any threads are waiting on this object, one of them is chosen to be awakened. 
  // The choice is arbitrary and occurs at the discretion of the implementation. 
  // A thread waits on an object's monitor by calling one of the wait methods.
  // 
  // The awakened thread will not be able to proceed until the current thread 
  // relinquishes the lock on this object. The awakened thread will compete in the 
  // usual manner with any other threads that might be actively competing to synchronize 
  // on this object; for example, the awakened thread enjoys no reliable privilege or
  // disadvantage in being the next thread to lock this object.
  // 
  // This method should only be called by a thread that is the owner of this object's
  // monitor. A thread becomes the owner of the object's monitor in one of three ways:
  public final native void notify();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
// Give up the CPU for one scheduling round.
void yield(void)
{
  acquire(&ptable.lock);  //DOC: yieldlock
  myproc()->state = RUNNABLE;
  sched();
  release(&ptable.lock);
}

// Wait for a child process to exit and return its pid.
int wait(void)
{
  ...
  for(;;){
    for(p = ptable.proc; p < &ptable.proc[NPROC]; p++){
      if(p->state == ZOMBIE){
        // 回收子进程资源
        pid = p->pid;
        p->killed = 0;
        ...
        p->state = UNUSED;
        return pid;
      }
    }

    sleep(curproc, &ptable.lock);
  }
}

void sleep(void *chan, struct spinlock *lk)
{
  ...
  // Go to sleep.
  p->chan = chan;
  p->state = SLEEPING;

  sched();

  // Tidy up.
  p->chan = 0;
  ...
}

1
2
3
4
5
6
7
8
9
10
11
// Enter scheduler
void sched(void)
{
  int intena;
  struct proc *p = myproc();
  ...
  intena = mycpu()->intena;
  // 切换当前进程上下文，调度器上下文，进入调度器
  swtch(&p->context, mycpu()->scheduler);
  mycpu()->intena = intena;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# Context switch
#
#   void swtch(struct context **old, struct context *new);
# 
# Save the current registers on the stack, creating
# a struct context, and save its address in *old.
# Switch stacks to new and pop previously-saved registers.

.globl swtch
swtch:
  movl 4(%esp), %eax
  movl 8(%esp), %edx

  # Save old callee-saved registers
  pushl %ebp
  pushl %ebx
  pushl %esi
  pushl %edi

  # Switch stacks
  movl %esp, (%eax)
  movl %edx, %esp

  # Load new callee-saved registers
  popl %edi
  popl %esi
  popl %ebx
  popl %ebp
  ret