Commit 448f6003 authored by unknown's avatar unknown
Browse files

Bug#25042 OPTIMIZE TABLE cause race condition in IO CACHE SHARE 

- The condition variable implementation "lost" a signal to
  WaitOnSingleObject when a semaphore was released.
- The signal could be consumed by a new call to pthread_cond_wait
  before all waiting threads had awoken.
- The new implementation of pthread_cond_* uses events
  instead of semaphores. It also uses an extra lock to protect entry
  into new cond wait before the broadcast has finished.


include/my_pthread.h:
  - New implementatin of pthread_cond_init. This version uses events
    instead of semaphores
mysys/my_wincond.c:
  - New implementatin of pthread_cond_init. This version uses events
    instead of semaphores
parent e5c1656e

include/my_pthread.h

+11 −5
Original line number Diff line number Diff line
@@ -80,11 +80,17 @@ typedef struct st_pthread_link { 


typedef struct {

  uint32 waiting;

#ifdef OS2

  HEV    semaphore;

#else

  HANDLE semaphore;

#endif

  CRITICAL_SECTION lock_waiting;

 

  enum {

    SIGNAL= 0,

    BROADCAST= 1,

    MAX_EVENTS= 2

  } EVENTS;



  HANDLE events[MAX_EVENTS];

  HANDLE broadcast_block_event;



} pthread_cond_t;



typedef int pthread_mutexattr_t;

mysys/my_wincond.c

+111 −38
Original line number Diff line number Diff line
@@ -28,26 +28,47 @@ 
int pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr)

{

  cond->waiting= 0;

  cond->semaphore=CreateSemaphore(NULL,0,0x7FFFFFFF,NullS);

  if (!cond->semaphore)

  InitializeCriticalSection(&cond->lock_waiting);

    

  cond->events[SIGNAL]= CreateEvent(NULL,  /* no security */

                                    FALSE, /* auto-reset event */

                                    FALSE, /* non-signaled initially */

                                    NULL); /* unnamed */



  /* Create a manual-reset event. */

  cond->events[BROADCAST]= CreateEvent(NULL,  /* no security */

                                       TRUE,  /* manual-reset */

                                       FALSE, /* non-signaled initially */

                                       NULL); /* unnamed */





  cond->broadcast_block_event= CreateEvent(NULL,  /* no security */

                                           TRUE,  /* manual-reset */

                                           TRUE,  /* signaled initially */

                                           NULL); /* unnamed */

  

  if( cond->events[SIGNAL] == NULL ||

      cond->events[BROADCAST] == NULL ||

      cond->broadcast_block_event == NULL )

    return ENOMEM;

  return 0;

}



int pthread_cond_destroy(pthread_cond_t *cond)

{

  return CloseHandle(cond->semaphore) ? 0 : EINVAL;

  DeleteCriticalSection(&cond->lock_waiting);



  if (CloseHandle(cond->events[SIGNAL]) == 0 ||

      CloseHandle(cond->events[BROADCAST]) == 0 ||

      CloseHandle(cond->broadcast_block_event) == 0)

    return EINVAL;

  return 0;

}





int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex)

{

  InterlockedIncrement(&cond->waiting);

  LeaveCriticalSection(mutex);

  WaitForSingleObject(cond->semaphore,INFINITE);

  InterlockedDecrement(&cond->waiting);

  EnterCriticalSection(mutex);

  return 0 ;

  return pthread_cond_timedwait(cond,mutex,NULL);

}
@@ -58,6 +79,8 @@ int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, 
  long timeout; 

  union ft64 now;



  if( abstime != NULL )

  {

    GetSystemTimeAsFileTime(&now.ft);



    /*
@@ -78,30 +101,80 @@ int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, 
    if (timeout > abstime->max_timeout_msec)

      timeout= abstime->max_timeout_msec;



  InterlockedIncrement(&cond->waiting);

  }

  else

  {

    /* No time specified; don't expire */

    timeout= INFINITE;

  }



  /* 

    Block access if previous broadcast hasn't finished.

    This is just for safety and should normally not

    affect the total time spent in this function.

  */

  WaitForSingleObject(cond->broadcast_block_event, INFINITE);



  EnterCriticalSection(&cond->lock_waiting);

  cond->waiting++;

  LeaveCriticalSection(&cond->lock_waiting);



  LeaveCriticalSection(mutex);

  result= WaitForSingleObject(cond->semaphore,timeout);

  InterlockedDecrement(&cond->waiting);



  result= WaitForMultipleObjects(2, cond->events, FALSE, timeout);

  

  EnterCriticalSection(&cond->lock_waiting);

  cond->waiting--;

  

  if (cond->waiting == 0 && result == (WAIT_OBJECT_0+BROADCAST))

  {

    /*

      We're the last waiter to be notified or to stop waiting, so

      reset the manual event. 

    */

    /* Close broadcast gate */

    ResetEvent(cond->events[BROADCAST]);

    /* Open block gate */

    SetEvent(cond->broadcast_block_event);

  }

  LeaveCriticalSection(&cond->lock_waiting);

  

  EnterCriticalSection(mutex);



  return result == WAIT_TIMEOUT ? ETIMEDOUT : 0;

}





int pthread_cond_signal(pthread_cond_t *cond)

{

  long prev_count;

  if (cond->waiting)

    ReleaseSemaphore(cond->semaphore,1,&prev_count);

  EnterCriticalSection(&cond->lock_waiting);

  

  if(cond->waiting > 0)

    SetEvent(cond->events[SIGNAL]);



  LeaveCriticalSection(&cond->lock_waiting);

  

  return 0;

}





int pthread_cond_broadcast(pthread_cond_t *cond)

{

  long prev_count;

  if (cond->waiting)

    ReleaseSemaphore(cond->semaphore,cond->waiting,&prev_count);

  EnterCriticalSection(&cond->lock_waiting);

  /*

     The mutex protect us from broadcasting if

     there isn't any thread waiting to open the

     block gate after this call has closed it.

   */

  if(cond->waiting > 0)

  {

    /* Close block gate */

    ResetEvent(cond->broadcast_block_event); 

    /* Open broadcast gate */

    SetEvent(cond->events[BROADCAST]);

  }



  LeaveCriticalSection(&cond->lock_waiting);  



  return 0;

}