Небольшая мысль о синхронизации - далее

Java задняя часть Linux Безопасность

существуетНесколько способов реализации блокировок и принцип реализации фьютекса в Linux представлены в В Интернете есть много статей о реализации ReentrantLock, и эта статья кратко представит его реализацию на уровне Java, сосредоточив внимание на анализе того, как блокировать потоки после сбоя конкурирующих блокировок. Из-за ограниченного места содержание синхронизированного будет помещено в следующую статью.

Больше статей смотрите в личном блоге:GitHub.com/farmer Джон Брат…

Реализация блокировки Java

ReentrantLock — широко используемая реализация блокировки в jdk. Логика его реализации основана на AQS (большинство реализаций классов синхронизации в пакете juc основаны на AQS); далее я кратко представлю общий принцип AQS, детали его реализации и различные приложений, Подробный разбор будет написан позже.

AQS

AQS — это аббревиатура класса AbstractQueuedSynchronizer.java, ReentrantLock, CyclicBarrier и CountdownLatch в пакете JUC используют AQS.

Общий принцип таков:

  1. AQS поддерживает переменную int, называемую состоянием, и двусвязный список, состояние используется для представления состояния синхронизации, а двусвязный список хранит потоки, ожидающие блокировок.
  2. При блокировке сначала вызовите tryAcquire, чтобы попытаться получить блокировку.Если блокировка не удалась, вставьте поток в двусвязный список и вызовите метод LockSupport.park(), чтобы заблокировать текущий поток.
  3. Вызовите LockSupport.unpark(), чтобы вызвать поток первого узла в связанном списке, когда блокировка снята. Пробужденный поток снова пройдет процесс конкурирующих блокировок.

Метод tryAcquire является абстрактным методом, и конкретная реализация зависит от класса реализации.Разница между тем, что мы часто называем справедливыми блокировками и нечестными блокировками, заключается в реализации этого метода.

ReentrantLock

ReentrantLock делится на честную блокировку и нечестную блокировку, мы рассматриваем только честную блокировку. ReentrantLock.lock будет вызываться в ReentrantLock#FairSync.lock:

FairSync.java

  static final class FairSync extends Sync {
      
        final void lock() {
            acquire(1);
        }

        /**
         * Fair version of tryAcquire.  Don't grant access unless
         * recursive call or no waiters or is first.
         */
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }

AbstractQueuedSynchronizer.java


   public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

Вы можете видеть, что FairSync.lock вызывает AQSacquireметод, находясь вacquireпервый звонокtryAcquireПопытка получить блокировку возвращает true в следующих двух случаях:

  1. state==0 (представляет, что поток не удерживает блокировку), а очередь ожидания пуста (добросовестная реализация), и cas успешно изменяет состояние.
  2. Текущий поток получил блокировку, этот вызов重入

еслиtryAcquireзвонить в случае неудачиacquireQueuedЗаблокировать текущий поток.acquireQueuedв конце концов позвонитLockSupport.park()Заблокируйте ветку.

LockSupport.park

Лично для того, чтобы глубоко понять механизм блокировки, очень важным моментом является понимание того, как система блокирует потоки.

LockSupport.java

    public static void park(Object blocker) {
        Thread t = Thread.currentThread();
        setBlocker(t, blocker);
        UNSAFE.park(false, 0L);
        setBlocker(t, null);
    }

parkЗа этот блокирующий объект синхронизации отвечает параметр метода Blocker, при вызове AQS этим объектом является сам AQS. Мы знаем, что ключевое слово synchronized предназначено для указания объекта (если текущий объект или текущий класс воздействует на метод), тем же блокировщиком является объект, указанный Locksupport.

parkметод вызывает собственный методUNSAFE.park, первый параметр указывает, является ли второй параметр абсолютным временем, а второй параметр представляет максимальное время блокировки.

Его реализация выглядит следующим образом, сохраняется только код ядра, а полный код можно посмотреть в unsafe.cpp.

 Unsafe_Park(JNIEnv *env, jobject unsafe, jboolean isAbsolute, jlong time){
 ...
 thread->parker()->park(isAbsolute != 0, time);
 ...
 }

Метод park находится в os_linux.cpp (реализация других операционных систем находится в os_xxx)

void Parker::park(bool isAbsolute, jlong time) {
  
  ...
  //获得当前线程
  Thread* thread = Thread::current();
  assert(thread->is_Java_thread(), "Must be JavaThread");
  JavaThread *jt = (JavaThread *)thread;

 //如果当前线程被设置了interrupted标记，则直接返回
  if (Thread::is_interrupted(thread, false)) {
    return;
  }
 
  if (time > 0) {
  //unpacktime中根据isAbsolute的值来填充absTime结构体，isAbsolute为true时，time代表绝对时间且单位是毫秒，否则time是相对时间且单位是纳秒
  //absTime.tvsec代表了对于时间的秒
  //absTime.tv_nsec代表对应时间的纳秒
    unpackTime(&absTime, isAbsolute, time);
  }

	//调用mutex trylock方法
    if (Thread::is_interrupted(thread, false) || pthread_mutex_trylock(_mutex) != 0) {
    return;
  }
	
 	//_counter是一个许可的数量，跟ReentrantLock里定义的许可变量基本都是一个原理。 unpack方法调用时会将_counter赋值为1。
 	//_counter>0代表已经有人调用了unpark，所以不用阻塞
  int status ;
  if (_counter > 0)  { // no wait needed
    _counter = 0;
    //释放mutex锁
    status = pthread_mutex_unlock(_mutex);
    return;
  }

//设置线程状态为CONDVAR_WAIT
  OSThreadWaitState osts(thread->osthread(), false /* not Object.wait() */);
 ...
 //等待
 _cur_index = isAbsolute ? ABS_INDEX : REL_INDEX;
 pthread_cond_timedwait(&_cond[_cur_index], _mutex,  &absTime);
 
 ...
  //释放mutex锁
  status = pthread_mutex_unlock(_mutex) ;
  
  
}

parkметод с использованием POSIXpthread_cond_timedwaitСпособ блокирования нити, вызовpthread_cond_timedwaitЗамок должен быть получен в первую очередь, поэтомуparkОсновной процесс:

  1. перечислитьpthread_mutex_trylockПопытаться получить блокировку и вернуться напрямую, если получение блокировки не удалось.
  2. перечислитьpthread_cond_timedwaitждать
  3. перечислитьpthread_mutex_unlockразблокировать замок

Кроме того, прежде чем заблокировать текущий поток, он вызоветOSThreadWaitStateконструктор устанавливает состояние потока вCONDVAR_WAIT, перечисление состояния потока в JVM выглядит следующим образом

  enum ThreadState {
  ALLOCATED,                    // Memory has been allocated but not initialized
  INITIALIZED,                  // The thread has been initialized but yet started
  RUNNABLE,                     // Has been started and is runnable, but not necessarily running
  MONITOR_WAIT,                 // Waiting on a contended monitor lock
  CONDVAR_WAIT,                 // Waiting on a condition variable
  OBJECT_WAIT,                  // Waiting on an Object.wait() call
  BREAKPOINTED,                 // Suspended at breakpoint
  SLEEPING,                     // Thread.sleep()
  ZOMBIE                        // All done, but not reclaimed yet
};

Время ожидания Linux

Из вышеизложенного мы можем знать, что метод LockSupport.park в конечном итоге используется POSIX.pthread_cond_timedwaitРеализованный метод. Теперь мы поближе познакомимсяpthread_mutex_trylock,pthread_cond_timedwait,pthread_mutex_unlockКак эти методы реализованы.

Соответствующий код в системе Linux находится в библиотеке glibc.

pthread_mutex_trylock

Сначала посмотрите на реализацию trylock, код в glibcpthread_mutex_trylock.cВ файле много кода для этого метода, смотрим только основной код

//pthread_mutex_t是posix中的互斥锁结构体
int
__pthread_mutex_trylock (mutex)
     pthread_mutex_t *mutex;
{
  int oldval;
  pid_t id = THREAD_GETMEM (THREAD_SELF, tid);
switch (__builtin_expect (PTHREAD_MUTEX_TYPE (mutex),
			    PTHREAD_MUTEX_TIMED_NP))
    {
    
    case PTHREAD_MUTEX_ERRORCHECK_NP:
    case PTHREAD_MUTEX_TIMED_NP:
    case PTHREAD_MUTEX_ADAPTIVE_NP:
      /* Normal mutex.  */
      if (lll_trylock (mutex->__data.__lock) != 0)
	break;

      /* Record the ownership.  */
      mutex->__data.__owner = id;
      ++mutex->__data.__nusers;

      return 0;
    }
    
} 
 //以下代码在lowlevellock.h中  
   #define __lll_trylock(futex) \
  (atomic_compare_and_exchange_val_acq (futex, 1, 0) != 0)
  #define lll_trylock(futex) __lll_trylock (&(futex))

мьютекс используется по умолчаниюPTHREAD_MUTEX_NORMALтипа (сPTHREAD_MUTEX_TIMED_NPтакой же); Так сначала позвонитlll_trylockметод,lll_trylockНа самом деле это операция cas, если mutex->__data.__lock==0, то измените его на 1 и верните 0, иначе верните 1.

В случае успеха измените владельца мьютекса на текущий поток.

pthread_mutex_unlock

pthread_mutex_unlock.c

int
internal_function attribute_hidden
__pthread_mutex_unlock_usercnt (mutex, decr)
     pthread_mutex_t *mutex;
     int decr;
{
    if (__builtin_expect (type, PTHREAD_MUTEX_TIMED_NP)
      == PTHREAD_MUTEX_TIMED_NP)
    {
      /* Always reset the owner field.  */
    normal:
      mutex->__data.__owner = 0;
      if (decr)
	/* One less user.  */
	--mutex->__data.__nusers;

      /* Unlock.  */
      lll_unlock (mutex->__data.__lock, PTHREAD_MUTEX_PSHARED (mutex));
      return 0;
    }
 }

pthread_mutex_unlockОчистить владельца в мьютексе и вызватьlll_unlockметод

lowlevellock.h

 

 #define __lll_unlock(futex, private)					      \
  ((void) ({								      \
    int *__futex = (futex);						      \
    int __val = atomic_exchange_rel (__futex, 0);			      \
									      \
    if (__builtin_expect (__val > 1, 0))				      \
      lll_futex_wake (__futex, 1, private);				      \
  }))
#define lll_unlock(futex, private) __lll_unlock(&(futex), private)


#define lll_futex_wake(ftx, nr, private)				\
({									\
   DO_INLINE_SYSCALL(futex, 3, (long) (ftx),				\
		     __lll_private_flag (FUTEX_WAKE, private),		\
		     (int) (nr));					\
   _r10 == -1 ? -_retval : _retval;					\
})

lll_unlockДелится на два этапа:

  1. Установите фьютекс на 0 и получите значение перед установкой (операция в пользовательском режиме)
  2. Если значение предыдущего Futex> 1, указывает на наличие конфликтов блокировки, что означает, что существует резьбовые вызовыFUTEX_WAITспит, поэтому, вызвав системную функциюFUTEX_WAKEразбудить спящую нить

FUTEX_WAKEВ предыдущей статье суть механизма фьютекса заключалась в попытке изменить переменную int (операция пользовательского режима) с помощью cas при получении блокировки.Если исходное значение целого числа равно 0, модификация прошла успешно, и поток получает блокировку, иначе будет Текущий поток помещается в очередь ожидания, и потоки в очереди ожидания не будут планироваться системой (операция в режиме ядра).

Существует три значения переменной фьютекса: 0 означает, что текущая блокировка простаивает, 1 означает, что есть поток, удерживающий текущую блокировку, и 2 означает, что есть конфликт блокировок. Значение фьютекса инициализируется равным 0; при вызове try_lock оно будет изменено на 1 с помощью операции cas (см. функцию trylock выше); при вызове try_locklll_lock, если нет конфликта блокировок, измените его на 1, иначе измените на 2.

#define __lll_lock(futex, private)					      \
  ((void) ({								      \
    int *__futex = (futex);						      \
    if (__builtin_expect (atomic_compare_and_exchange_bool_acq (__futex,      \
								1, 0), 0))    \
      {									      \
	if (__builtin_constant_p (private) && (private) == LLL_PRIVATE)	      \
	  __lll_lock_wait_private (__futex);				      \
	else								      \
	  __lll_lock_wait (__futex, private);				      \
      }									      \
  }))
#define lll_lock(futex, private) __lll_lock (&(futex), private)

void
__lll_lock_wait_private (int *futex)
{
//第一次进来的时候futex==1,所以不会走这个if
  if (*futex == 2)
    lll_futex_wait (futex, 2, LLL_PRIVATE);
//在这里会把futex设置成2，并调用futex_wait让当前线程等待
  while (atomic_exchange_acq (futex, 2) != 0)
    lll_futex_wait (futex, 2, LLL_PRIVATE);
}

pthread_cond_timedwait

pthread_cond_timedwaitИспользуется для блокировки потоков, реализации ожидания потока, код в glibcpthread_cond_timedwait.cВ файле код длинный, можно просто сначала пройтись по нему, прочитать следующий разбор и еще раз прочитать код

int
int
__pthread_cond_timedwait (cond, mutex, abstime)
     pthread_cond_t *cond;
     pthread_mutex_t *mutex;
     const struct timespec *abstime;
{
  struct _pthread_cleanup_buffer buffer;
  struct _condvar_cleanup_buffer cbuffer;
  int result = 0;

  /* Catch invalid parameters.  */
  if (abstime->tv_nsec < 0 || abstime->tv_nsec >= 1000000000)
    return EINVAL;

  int pshared = (cond->__data.__mutex == (void *) ~0l)
		? LLL_SHARED : LLL_PRIVATE;

  //1.获得cond锁
  lll_lock (cond->__data.__lock, pshared);

  //2.释放mutex锁
  int err = __pthread_mutex_unlock_usercnt (mutex, 0);
  if (err)
    {
      lll_unlock (cond->__data.__lock, pshared);
      return err;
    }

  /* We have one new user of the condvar.  */
  //每执行一次wait(pthread_cond_timedwait/pthread_cond_wait)，__total_seq就会+1
  ++cond->__data.__total_seq;
  //用来执行futex_wait的变量
  ++cond->__data.__futex;
  //标识该cond还有多少线程在使用，pthread_cond_destroy需要等待所有的操作完成
  cond->__data.__nwaiters += 1 << COND_NWAITERS_SHIFT;

  /* Remember the mutex we are using here.  If there is already a
     different address store this is a bad user bug.  Do not store
     anything for pshared condvars.  */
  //保存mutex锁
  if (cond->__data.__mutex != (void *) ~0l)
    cond->__data.__mutex = mutex;

  /* Prepare structure passed to cancellation handler.  */
  cbuffer.cond = cond;
  cbuffer.mutex = mutex;

  /* Before we block we enable cancellation.  Therefore we have to
     install a cancellation handler.  */
  __pthread_cleanup_push (&buffer, __condvar_cleanup, &cbuffer);

  /* The current values of the wakeup counter.  The "woken" counter
     must exceed this value.  */
  //记录futex_wait前的__wakeup_seq（为该cond上执行了多少次sign操作+timeout次数）和__broadcast_seq（代表在该cond上执行了多少次broadcast）
  unsigned long long int val;
  unsigned long long int seq;
  val = seq = cond->__data.__wakeup_seq;
  /* Remember the broadcast counter.  */
  cbuffer.bc_seq = cond->__data.__broadcast_seq;

  while (1)
    {
      //3.计算要wait的相对时间
      struct timespec rt;
      {
#ifdef __NR_clock_gettime
	INTERNAL_SYSCALL_DECL (err);
	int ret;
	ret = INTERNAL_VSYSCALL (clock_gettime, err, 2,
				(cond->__data.__nwaiters
				 & ((1 << COND_NWAITERS_SHIFT) - 1)),
				&rt);
# ifndef __ASSUME_POSIX_TIMERS
	if (__builtin_expect (INTERNAL_SYSCALL_ERROR_P (ret, err), 0))
	  {
	    struct timeval tv;
	    (void) gettimeofday (&tv, NULL);

	    /* Convert the absolute timeout value to a relative timeout.  */
	    rt.tv_sec = abstime->tv_sec - tv.tv_sec;
	    rt.tv_nsec = abstime->tv_nsec - tv.tv_usec * 1000;
	  }
	else
# endif
	  {
	    /* Convert the absolute timeout value to a relative timeout.  */
	    rt.tv_sec = abstime->tv_sec - rt.tv_sec;
	    rt.tv_nsec = abstime->tv_nsec - rt.tv_nsec;
	  }
#else
	/* Get the current time.  So far we support only one clock.  */
	struct timeval tv;
	(void) gettimeofday (&tv, NULL);

	/* Convert the absolute timeout value to a relative timeout.  */
	rt.tv_sec = abstime->tv_sec - tv.tv_sec;
	rt.tv_nsec = abstime->tv_nsec - tv.tv_usec * 1000;
#endif
      }
      if (rt.tv_nsec < 0)
	{
	  rt.tv_nsec += 1000000000;
	  --rt.tv_sec;
	}
   /*---计算要wait的相对时间 end---- */

  //是否超时
      /* Did we already time out?  */
      if (__builtin_expect (rt.tv_sec < 0, 0))
	{
    //被broadcast唤醒，这里疑问的是，为什么不需要判断__wakeup_seq？
	  if (cbuffer.bc_seq != cond->__data.__broadcast_seq)
	    goto bc_out;

	  goto timeout;
	}

      unsigned int futex_val = cond->__data.__futex;

      //4.释放cond锁，准备wait
      lll_unlock (cond->__data.__lock, pshared);

      /* Enable asynchronous cancellation.  Required by the standard.  */
      cbuffer.oldtype = __pthread_enable_asynccancel ();

      //5.调用futex_wait
      /* Wait until woken by signal or broadcast.  */
      err = lll_futex_timed_wait (&cond->__data.__futex,
				  futex_val, &rt, pshared);

      /* Disable asynchronous cancellation.  */
      __pthread_disable_asynccancel (cbuffer.oldtype);


      //6.重新获得cond锁，因为又要访问&修改cond的数据了
      lll_lock (cond->__data.__lock, pshared);

      //__broadcast_seq值发生改变，代表发生了有线程调用了广播
      if (cbuffer.bc_seq != cond->__data.__broadcast_seq)
	goto bc_out;

     //判断是否是被sign唤醒的，sign会增加__wakeup_seq
     //第二个条件cond->__data.__woken_seq != val的意义在于
    //可能两个线程A、B在wait，一个线程调用了sign导致A被唤醒，这时B因为超时被唤醒
    //对于B线程来说，执行到这里时第一个条件也是满足的，从而导致上层拿到的result不是超时
    //所以这里需要判断下__woken_seq（即该cond已经被唤醒的线程数）是否等于__wakeup_seq（sign执行次数+timeout次数）
      val = cond->__data.__wakeup_seq;
      if (val != seq && cond->__data.__woken_seq != val)
	break;

      /* Not woken yet.  Maybe the time expired?  */
      if (__builtin_expect (err == -ETIMEDOUT, 0))
	{
	timeout:
	  /* Yep.  Adjust the counters.  */
	  ++cond->__data.__wakeup_seq;
	  ++cond->__data.__futex;

	  /* The error value.  */
	  result = ETIMEDOUT;
	  break;
	}
    }

  //一个线程已经醒了所以这里__woken_seq +1
  ++cond->__data.__woken_seq;

 bc_out:
  //
  cond->__data.__nwaiters -= 1 << COND_NWAITERS_SHIFT;

  /* If pthread_cond_destroy was called on this variable already,
     notify the pthread_cond_destroy caller all waiters have left
     and it can be successfully destroyed.  */
  if (cond->__data.__total_seq == -1ULL
      && cond->__data.__nwaiters < (1 << COND_NWAITERS_SHIFT))
    lll_futex_wake (&cond->__data.__nwaiters, 1, pshared);

 //9.cond数据修改完毕，释放锁
  lll_unlock (cond->__data.__lock, pshared);

  /* The cancellation handling is back to normal, remove the handler.  */
  __pthread_cleanup_pop (&buffer, 0);

 //10.重新获得mutex锁
  err = __pthread_mutex_cond_lock (mutex);

  return err ?: result;
}

Хотя приведенный выше код прокомментирован, я считаю, что большинство людей не могут понять его с первого раза. Кратко разберемся, в приведенном выше коде две блокировки, одна блокировка мьютекса, а другая блокировка условного прохода. Кроме того, при вызовеpthread_cond_timedwaitнужно вызывать до и послеpthread_mutex_lock(&mutex);иpthread_mutex_unlock(&mutex);Добавить/разблокировать мьютекс.

следовательноpthread_cond_timedwaitИспользование условно делится на несколько процессов:

  1. Добавить блокировку MUTEX (вpthread_cond_timedwaitдо звонка)
  2. добавить условный замок
  3. снять блокировку мьютекса
  4. Изменить данные условия
  5. снять блокировку состояния
  6. Выполнить futex_wait.
  7. Восстановить блокировку
  8. Сравните данные cond, чтобы определить, пробуждается ли текущий поток нормально или по тайм-ауту, и нужно ли ему снова ждать.
  9. Изменить данные условия
  10. Конденсаторный замок
  11. Восстановить блокировку мьютекса
  12. снять блокировку мьютекса (вpthread_cond_timedwaitПосле звонка)

Увидев это, у вас может возникнуть несколько вопросов: Зачем вам два замка? Какова роль блокировки мьютекса и условной блокировки?

блокировка мьютекса

Прежде чем говорить о роли блокировок мьютексов, давайте рассмотрим использование Object.wait в Java. Object.wait должен использоваться в синхронизированном блоке. Представьте, если вы сможете запустить Object.wait без синхронизации, в чем будет проблема?

Object condObj=new Object();
voilate int flag = 0;
public void waitTest(){
	if(flag == 0){
		condObj.wait();
	}
}
public void notifyTest(){
	flag=1;
	condObj.notify();
}

Как и в приведенном выше коде, поток A вызывает waitTest, в это время flag==0, поэтому он готов вызвать метод ожидания для сна.В это время поток B начинает выполняться, вызывает notifyTest, чтобы установить флаг в 1, и вызывает метод уведомления.Примечание: в это время поток A все еще не вызывается, поэтому notfiy не пробуждает ни один поток. Затем поток A продолжает выполняться, вызывая метод ожидания для сна, и тогда никто не будет будить поток A, поток A будет ждать вечно!

Object condObj=new Object();
voilate int flag = 0;
public void waitTest(){
	synchronized(condObj){
		if(flag == 0){
			condObj.wait();
		}
	}
	
}
public void notifyTest(){
	synchronized(condObj){
		flag=1;
		condObj.notify();
	}
}

В случае защиты от блокировки при вызове condObj.wait флаг должен быть равен 0, и не будет проблемы ожидания все время.

назадpthread_cond_timedwait, причина, по которой необходимо добавить блокировку мьютекса, очевидна:Гарантирует атомность ждать и его состояние ожидания

либо glibcpthread_cond_timedwait/pthread_cond_signalили java слойObject.wait/Object.notify, JDK AQSCondition.await/Condition.signal, все механизмы Condition могут использоваться только в заблокированной среде.Фундаментальная причина заключается в том, чтобы гарантировать, что переменная условия не будет изменена, когда поток спит.

Обратите внимание на время снятия блокировки мьютекса, просмотрите приведенное выше.pthread_cond_timedwaitПроцесс снятия блокировки мьютекса на шаге 2 и последующего вызоваfutex_waitДля гибернации, зачем освобождать блокировку мьютекса перед гибернацией? Причина также очень проста: если вы начинаете спать, не освобождая блокировку мьютекса, другие потоки никогда не смогут вызвать сигнальный метод, чтобы разбудить спящий поток (поскольку перед вызовом сигнального метода необходимо получить блокировку мьютекса).

Повторное получение блокировки мьютекса на шаге 10 после пробуждения потока необходимо для сохранения семантики блокировки (подумайте, что произойдет, если блокировка мьютекса не будет повторно получена).

конд замок

Роль условной блокировки на самом деле очень проста: гарантировать объектcond->dataбезопасность резьбы. существуетpthread_cond_timedwaitнужно изменитьcond->dataданные, такие как добавление __total_seq (сколько ожиданий было выполнено для этого условия), добавление __nwaiters (сколько потоков все еще ожидают этого условия), все из которых изменяются и доступныcond->dataКогда нужно добавить cond lock.

Чего я здесь не понял, так это того, что использование блокировок мьютексов также может гарантироватьcond->dataМодификация является потокобезопасной, если блокировка мьютекса освобождается позже. Зачем сначала освобождать мьютекс, а затем повторно получать условие, чтобы обеспечить потокобезопасность? Это делается для того, чтобы мьютекс не блокировал слишком большой диапазон?

Как разбудить спящую нить

Код для пробуждения спящего потока относительно прост, в основном это вызов lll_futex_wake.


int
__pthread_cond_signal (cond)
     pthread_cond_t *cond;
{
  int pshared = (cond->__data.__mutex == (void *) ~0l)
		? LLL_SHARED : LLL_PRIVATE;

  //因为要操作cond的数据，所以要加锁
  lll_lock (cond->__data.__lock, pshared);

  /* Are there any waiters to be woken?  */
  if (cond->__data.__total_seq > cond->__data.__wakeup_seq)
    {
      //__wakeup_seq为执行sign与timeout次数的和
      ++cond->__data.__wakeup_seq;
      ++cond->__data.__futex;

       ...
		//唤醒wait的线程
      lll_futex_wake (&cond->__data.__futex, 1, pshared);
    }

  /* We are done.  */
  lll_unlock (cond->__data.__lock, pshared);

  return 0;
}

End

В этой статье кратко представлен принцип реализации ReentrantLock для Java и базовая реализация LockSupport.park.pthread_cond_timedwaitПодробно анализируются механизмы.

После прочтения этой статьи у вас могут остаться вопросы: Реализация синхронизированной блокировки такая же, как у ReentrantLock? В чем разница между принципом спящего потока Thread.sleep/Object.wait и LockSupport.park? Как реализован фьютекс уровня ядра Linux?

На эти вопросы будут даны ответы один за другим в следующих статьях, так что ждите с нетерпением~

Категории