readme.cv

pthread source code,you can compile directly

too
   * This sync.level supports _timedwait and cancellation
   */
  else
    {
      result = pthread_mutex_unlock(&(cv->mtxUnblockLock));
    }

  return(result);

}                               /* ptw32_cond_unblock */

int
pthread_cond_wait (pthread_cond_t * cond,
                   pthread_mutex_t * mutex)
{
  /* The NULL abstime arg means INFINITE waiting. */
  return(ptw32_cond_timedwait(cond, mutex, NULL));
}                               /* pthread_cond_wait */


int
pthread_cond_timedwait (pthread_cond_t * cond,
                        pthread_mutex_t * mutex,
                        const struct timespec *abstime)
{
  if (abstime == NULL)
    {(abstime
      return EINVAL;
    }

  return(ptw32_cond_timedwait(cond, mutex, abstime));
}                               /* pthread_cond_timedwait */


int
pthread_cond_signal (pthread_cond_t * cond)
{
  /* The '0'(FALSE) unblockAll arg means unblock ONE waiter. */
  return(ptw32_cond_unblock(cond, 0));
}                               /* pthread_cond_signal */

int
pthread_cond_broadcast (pthread_cond_t * cond)
{
  /* The '1'(TRUE) unblockAll arg means unblock ALL waiters. */
  return(ptw32_cond_unblock(cond, 1));
}                               /* pthread_cond_broadcast */




[e-mail] on 17.01.2001 01:00:57

Please respond to [e-mail]

To:   [e-mail]
cc:   [e-mail]
Subject:  win32 conditions: sem+counter+event = broadcast_deadlock +
      spur.wakeup/unfairness/incorrectness ??







Hi,

Problem 1: broadcast_deadlock

It seems that current implementation does not provide "atomic"
broadcasts. That may lead to "nested" broadcasts... and it seems
that nested case is not handled correctly -> producing a broadcast
DEADLOCK as a result.

Scenario:

N (>1) waiting threads W1..N are blocked (in _wait) on condition's
semaphore.

Thread B1 calls pthread_cond_broadcast, which results in "releasing" N
W threads via incrementing semaphore counter by N (stored in
cv->waiters) BUT cv->waiters counter does not change!! The caller
thread B1 remains blocked on cv->waitersDone event (auto-reset!!) BUT
condition is not protected from starting another broadcast (when called
on another thread) while still waiting for the "old" broadcast to
complete on thread B1.

M (>=0, <N) W threads are fast enough to go thru their _wait call and
decrement cv->waiters counter.

L (N-M) "late" waiter W threads are a) still blocked/not returned from
their semaphore wait call or b) were preempted after sem_wait but before
lock( &cv->waitersLock ) or c) are blocked on cv->waitersLock.

cv->waiters is still > 0 (= L).

Another thread B2 (or some W thread from M group) calls
pthread_cond_broadcast and gains access to counter... neither a) nor b)
prevent thread B2 in pthread_cond_broadcast from gaining access to
counter and starting another broadcast ( for c) - it depends on
cv->waitersLock scheduling rules: FIFO=OK, PRTY=PROBLEM,... )

That call to pthread_cond_broadcast (on thread B2) will result in
incrementing semaphore by cv->waiters (=L) which is INCORRECT (all
W1..N were in fact already released by thread B1) and waiting on
_auto-reset_ event cv->waitersDone which is DEADLY WRONG (produces a
deadlock)...

All late W1..L threads now have a chance to complete their _wait call.
Last W_L thread sets an auto-reselt event cv->waitersDone which will
release either B1 or B2 leaving one of B threads in a deadlock.

Problem 2: spur.wakeup/unfairness/incorrectness

It seems that:

a) because of the same problem with counter which does not reflect the
actual number of NOT RELEASED waiters, the signal call may increment
a semaphore counter w/o having a waiter blocked on it. That will result
in (best case) spurious wake ups - performance degradation due to
unnecessary context switches and predicate re-checks and (in worth case)
unfairness/incorrectness problem - see b)

b) neither signal nor broadcast prevent other threads - "new waiters"
(and in the case of signal, the caller thread as well) from going into
_wait and overtaking "old" waiters (already released but still not returned
from sem_wait on condition's semaphore). Win semaphore just [API DOC]:
"Maintains a count between zero and some maximum value, limiting the number
of threads that are simultaneously accessing a shared resource." Calling
ReleaseSemaphore does not imply (at least not documented) that on return
from ReleaseSemaphore all waiters will in fact become released (returned
from their Wait... call) and/or that new waiters calling Wait... afterwards
will become less importance. It is NOT documented to be an atomic release
of
waiters... And even if it would be there is still a problem with a thread
being preempted after Wait on semaphore and before Wait on cv->waitersLock
and scheduling rules for cv->waitersLock itself
(??WaitForMultipleObjects??)
That may result in unfairness/incorrectness problem as described
for SetEvent impl. in "Strategies for Implementing POSIX Condition
Variables
on Win32": http://www.cs.wustl.edu/~schmidt/win32-cv-1.html

Unfairness -- The semantics of the POSIX pthread_cond_broadcast function is
to wake up all threads currently blocked in wait calls on the condition
variable. The awakened threads then compete for the external_mutex. To
ensure
fairness, all of these threads should be released from their
pthread_cond_wait calls and allowed to recheck their condition expressions
before other threads can successfully complete a wait on the condition
variable.

Unfortunately, the SetEvent implementation above does not guarantee that
all
threads sleeping on the condition variable when cond_broadcast is called
will
acquire the external_mutex and check their condition expressions. Although
the Pthreads specification does not mandate this degree of fairness, the
lack of fairness can cause starvation.

To illustrate the unfairness problem, imagine there are 2 threads, C1 and
C2,
that are blocked in pthread_cond_wait on condition variable not_empty_ that
is guarding a thread-safe message queue. Another thread, P1 then places two
messages onto the queue and calls pthread_cond_broadcast. If C1 returns
from
pthread_cond_wait, dequeues and processes the message, and immediately
waits
again then it and only it may end up acquiring both messages. Thus, C2 will
never get a chance to dequeue a message and run.

The following illustrates the sequence of events:

1.   Thread C1 attempts to dequeue and waits on CV non_empty_
2.   Thread C2 attempts to dequeue and waits on CV non_empty_
3.   Thread P1 enqueues 2 messages and broadcasts to CV not_empty_
4.   Thread P1 exits
5.   Thread C1 wakes up from CV not_empty_, dequeues a message and runs
6.   Thread C1 waits again on CV not_empty_, immediately dequeues the 2nd
        message and runs
7.   Thread C1 exits
8.   Thread C2 is the only thread left and blocks forever since
        not_empty_ will never be signaled

Depending on the algorithm being implemented, this lack of fairness may
yield
concurrent programs that have subtle bugs. Of course, application
developers
should not rely on the fairness semantics of pthread_cond_broadcast.
However,
there are many cases where fair implementations of condition variables can
simplify application code.

Incorrectness -- A variation on the unfairness problem described above
occurs
when a third consumer thread, C3, is allowed to slip through even though it
was not waiting on condition variable not_empty_ when a broadcast occurred.

To illustrate this, we will use the same scenario as above: 2 threads, C1
and
C2, are blocked dequeuing messages from the message queue. Another thread,
P1
then places two messages onto the queue and calls pthread_cond_broadcast.
C1
returns from pthread_cond_wait, dequeues and processes the message. At this
time, C3 acquires the external_mutex, calls pthread_cond_wait and waits on
the events in WaitForMultipleObjects. Since C2 has not had a chance to run
yet, the BROADCAST event is still signaled. C3 then returns from
WaitForMultipleObjects, and dequeues and processes the message in the
queue.
Thus, C2 will never get a chance to dequeue a message and run.

The following illustrates the sequence of events:

1.   Thread C1 attempts to dequeue and waits on CV non_empty_
2.   Thread C2 attempts to dequeue and waits on CV non_empty_
3.   Thread P1 enqueues 2 messages and broadcasts to CV not_empty_
4.   Thread P1 exits
5.   Thread C1 wakes up from CV not_empty_, dequeues a message and runs
6.   Thread C1 exits
7.   Thread C3 waits on CV not_empty_, immediately dequeues the 2nd
        message and runs
8.   Thread C3 exits
9.   Thread C2 is the only thread left and blocks forever since
        not_empty_ will never be signaled

In the above case, a thread that was not waiting on the condition variable
when a broadcast occurred was allowed to proceed. This leads to incorrect
semantics for a condition variable.


COMMENTS???

regards,
alexander.

-----------------------------------------------------------------------------

Subject: RE: FYI/comp.programming.threads/Re: pthread_cond_*
     implementation questions
Date: Wed, 21 Feb 2001 11:54:47 +0100
From: [e-mail]
To: [e-mail]
CC: [e-mail], Thomas Pfaff <[e-mail]>,
     Nanbor Wang <[e-mail]>

Hi Louis,

generation number 8..

had some time to revisit timeouts/spurious wakeup problem..
found some bugs (in 7.b/c/d) and something to improve
(7a - using IPC semaphores but it should speedup Win32
version as well).

regards,
alexander.

---------- Algorithm 8a / IMPL_SEM,UNBLOCK_STRATEGY == UNBLOCK_ALL ------
given:
semBlockLock - bin.semaphore
semBlockQueue - semaphore
mtxExternal - mutex or CS
mtxUnblockLock - mutex or CS
nWaitersGone - int
nWaitersBlocked - int
nWaitersToUnblock - int

wait( timeout ) {

  [auto: register int result          ]     // error checking omitted
  [auto: register int nSignalsWasLeft ]
  [auto: register int nWaitersWasGone ]

  sem_wait( semBlockLock );
  nWaitersBlocked++;
  sem_post( semBlockLock );

  unlock( mtxExternal );
  bTimedOut = sem_wait( semBlockQueue,timeout );

  lock( mtxUnblockLock );
  if ( 0 != (nSignalsWasLeft = nWaitersToUnblock) ) {
    if ( bTimeout ) {                       // timeout (or canceled)
      if ( 0 != nWaitersBlocked ) {
        nWaitersBlocked--;
      }
      else {
        nWaitersGone++;                     // count spurious wakeups
      }
    }
    if ( 0 == --nWaitersToUnblock ) {
      if ( 0 != nWaitersBlocked ) {
        sem_post( semBlockLock );           // open the gate
        nSignalsWasLeft = 0;                // do not open the gate below
again
      }
      else if ( 0 != (nWaitersWasGone = nWaitersGone) ) {
        nWaitersGone = 0;
      }
    }
  }
  else if ( INT_MAX/2 == ++nWaitersGone ) { // timeout/canceled or spurious
semaphore :-)
    sem_wait( semBlockLock );
    nWaitersBlocked -= nWaitersGone;        // something is going on here -
test of timeouts? :-)
    sem_post( semBlockLock );
    nWaitersGone = 0;
  }
  unlock( mtxUnblockLock );

  if ( 1 == nSignalsWasLeft ) {
    if ( 0 != nWaitersWasGone ) {
      // sem_adjust( -nWaitersWasGone );
      while ( nWaitersWasGone-- ) {
        sem_wait( semBlockLock );          // better now than spurious
later
      }
    }
    sem_post( semBlockLock );              // open the gate
  }

  lock( mtxExternal );

  return ( bTimedOut ) ? ETIMEOUT : 0;
}

signal(bAll) {

  [auto: register int result         ]
  [auto: register int nSignalsToIssue]

  lock( mtxUnblockLock );

  if ( 0 != nWaitersToUnblock ) { // the gate is closed!!!
    if ( 0 == nWaitersBlocked ) { // NO-OP
      return unlock( mtxUnblockLock );
    }
    if (bAll) {
      nWaitersToUnblock += nSignalsToIssue=nWaitersBlocked;
      nWaitersBlocked = 0;
    }
    else {
      nSignalsToIssue = 1;
      nWaitersToUnblock++;
      nWaitersBlocked--;
    }
  }
  else if ( nWaitersBlocked > nWaitersGone ) { // HARMLESS RACE CONDITION!
    sem_wait( semBlockLock ); // close the gate
    if ( 0 != nWaitersGone ) {
      nWaitersBlocked -= nWaitersGone;
      nWaitersGone = 0;
    }
    if (bAll) {
      nSignalsToIssue = nWaitersToUnblock = nWaitersBlocked;
      nWaitersBlocked = 0;
    }
    else {
      nSignalsToIssue = nWaitersToUnblock = 1;
      nWaitersBlocked--;
    }
  }
  else { // NO-OP
    return unlock( mtxUnblockLock );
  }

  unlock( mtxUnblockLock );
  sem_post( semBlockQueue,nSignalsToIssue );
  return result;
}

---------- Algorithm 8b / IMPL_SEM,UNBLOCK_STRATEGY == UNBLOCK_ONEBYONE
------
given:
semBlockLock - bin.semaphore
semBlockQueue - bin.semaphore
mtxExternal - mutex or CS
mtxUnblockLock - mutex or CS
nWaitersGone - int
nWaitersBlocked - int
nWaitersToUnblock - int

wait( timeout ) {

  [auto: register int result          ]     // error checking omitted
  [auto: register int nWaitersWasGone ]
  [auto: register int nSignalsWasLeft ]

  sem_wait( semBlockLock );
  nWaitersBlocked++;
  sem_post( semBlockLock );

  unlock( mtxExternal );
  bTimedOut = sem_wait( semBlockQueue,timeout );

  lock( mtxUnblockLock );
  if ( 0 != (nSignalsWasLeft = nWaitersToUnblock) ) {
    if ( bTimeout ) {                       // timeout (or canceled)
      if ( 0 != nWaitersBlocked ) {
        nWaitersBlocked--;
        nSignalsWasLeft = 0;                // do not unblock next waiter
below (already unblocked)
      }
      else {
        nWaitersGone = 1;                   // spurious wakeup pending!!
      }
    }
    if ( 0 == --nWaitersToUnblock &&
      if ( 0 != nWaitersBlocked ) {
        sem_post( semBlockLock );           // open the gate
        nSignalsWasLeft = 0;                // do not open the gate below
again
      }
      else if ( 0 != (nWaitersWasGone = nWaitersGone) ) {
        nWaitersGone = 0;
      }
    }
  }
  else if ( INT_MAX/2 == ++nWaitersGone ) { // timeout/canceled or spurious
semaphore :-)
    sem_wait( semBlockLock );
    nWaitersBlocked -= nWaitersGone;        // something is going on here -
test of timeouts? :-)
    sem_post( semBlockLock );
    nWaitersGone = 0;
  }
  unlock( mtxUnblockLock );

  if ( 1 == nSignalsWasLeft ) {
    if ( 0 != nWaitersWasGone ) {
      // sem_adjust( -1 );
      sem_wait( semBlockQueue );           // better now than spurious
later
    }
    sem_post( semBlockLock );              // open the gate
  }
  else if ( 0 != nSignalsWasLeft ) {
    sem_post( semBlockQueue );             // unblock next waiter
  }

  lock( mtxExternal );

  return ( bTimedOut ) ? ETIMEOUT : 0;
}

signal(bAll) {

  [auto: register int result ]

  lock( mtxUnblockLock );

  if ( 0 != nWaitersToUnblock ) { // the gate is closed!!!
    if ( 0 == nWaitersBlocked ) { // NO-OP
      return unlock( mtxUnblockLock );
    }
    if (bAll) {
      nWaitersToUnblock += nWaitersBlocked;
      nWaitersBlocked = 0;
    }
    else {
      nWaitersToUnblock++;
      nWaitersBlocked--;
    }
    unlock( mtxUnblockLock );
  }
  else if ( nWaitersBlocked > nWaitersGone ) { // HARMLESS RACE CONDITION!
    sem_wait( semBlockLock ); // close the gate
    if ( 0 != nWaitersGone ) {
      nWaitersBlocked -= nWaitersGone;
      nWaitersGone = 0;
    }
    if (bAll) {
      nWaitersToUnblock = nWaitersBlocked;
      nWaitersBlocked = 0;
    }
    else {
      nWaitersToUnblock = 1;
      nWaitersBlocked--;
    }
    unlock( mtxUnblockLock );
    sem_post( semBlockQueue );
  }
  else { // NO-OP
    unlock( mtxUnblockLock );
  }

  return result;
}

---------- Algorithm 8c / IMPL_EVENT,UNBLOCK_STRATEGY == UNBLOCK_ONEBYONE
---------
given:
hevBlockLock - auto-reset event
hevBlockQueue - auto-reset event