~drizzle-trunk/drizzle/development : contents of drizzled/thr

~drizzle-trunk/drizzle/development : (revision 1807)
/* Copyright (C) 2000 MySQL AB

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/*
Read and write locks for Posix threads. All tread must acquire
all locks it needs through thr_multi_lock() to avoid dead-locks.
A lock consists of a master lock (THR_LOCK), and lock instances
(THR_LOCK_DATA).
Any thread can have any number of lock instances (read and write:s) on
any lock. All lock instances must be freed.
Locks are prioritized according to:

The current lock types are:

TL_READ	 		# Low priority read
TL_READ_WITH_SHARED_LOCKS
TL_READ_NO_INSERT	# Read without concurrent inserts
TL_WRITE_ALLOW_WRITE	# Write lock that allows other writers
TL_WRITE_ALLOW_READ	# Write lock, but allow reading
TL_WRITE_CONCURRENT_INSERT
			# Insert that can be mixed when selects
TL_WRITE		# High priority write
TL_WRITE_ONLY		# High priority write
			# Abort all new lock request with an error

Locks are prioritized according to:

WRITE_ALLOW_WRITE, WRITE_ALLOW_READ, WRITE_CONCURRENT_INSERT, WRITE_DELAYED,
WRITE_LOW_PRIORITY, READ, WRITE, READ_HIGH_PRIORITY and WRITE_ONLY

Locks in the same privilege level are scheduled in first-in-first-out order.

To allow concurrent read/writes locks, with 'WRITE_CONCURRENT_INSERT' one
should put a pointer to the following functions in the lock structure:
(If the pointer is zero (default), the function is not called)


The lock algorithm allows one to have one TL_WRITE_ALLOW_READ,
TL_WRITE_CONCURRENT_INSERT lock at the same time as multiple read locks.

*/

#include "config.h"
#include "drizzled/internal/my_sys.h"
#include "drizzled/internal/thread_var.h"
#include "drizzled/statistics_variables.h"

#include "drizzled/session.h"
#include "drizzled/current_session.h"

#include "thr_lock.h"
#include "drizzled/internal/m_string.h"
#include <errno.h>
#include <list>

#if TIME_WITH_SYS_TIME
# include <sys/time.h>
# include <time.h>
#else
# if HAVE_SYS_TIME_H
#  include <sys/time.h>
# else
#  include <time.h>
# endif
#endif

#include <drizzled/util/test.h>

#include <boost/interprocess/sync/lock_options.hpp>

using namespace std;

namespace drizzled
{

uint64_t table_lock_wait_timeout;
static enum thr_lock_type thr_upgraded_concurrent_insert_lock = TL_WRITE;


uint64_t max_write_lock_count= ~(uint64_t) 0L;

/*
** For the future (now the thread specific cond is alloced by my_pthread.c)
*/

static inline bool
thr_lock_owner_equal(THR_LOCK_OWNER *rhs, THR_LOCK_OWNER *lhs)
{
  return rhs == lhs;
}


	/* Initialize a lock */

void thr_lock_init(THR_LOCK *lock)
{
  lock->init();
  lock->read.last= &lock->read.data;
  lock->read_wait.last= &lock->read_wait.data;
  lock->write_wait.last= &lock->write_wait.data;
  lock->write.last= &lock->write.data;
}


void THR_LOCK_INFO::init()
{
  internal::st_my_thread_var *tmp= my_thread_var;
  thread= tmp->pthread_self;
  thread_id= tmp->id;
  n_cursors= 0;
}

	/* Initialize a lock instance */

void THR_LOCK_DATA::init(THR_LOCK *lock_arg, void *param_arg)
{
  lock= lock_arg;
  type= TL_UNLOCK;
  owner= NULL;                               /* no owner yet */
  status_param= param_arg;
  cond= NULL;
}


static inline bool
have_old_read_lock(THR_LOCK_DATA *data, THR_LOCK_OWNER *owner)
{
  for ( ; data ; data=data->next)
  {
    if (thr_lock_owner_equal(data->owner, owner))
      return true;					/* Already locked by thread */
  }
  return false;
}

static void wake_up_waiters(THR_LOCK *lock);


static enum enum_thr_lock_result wait_for_lock(struct st_lock_list *wait, THR_LOCK_DATA *data, bool in_wait_list)
{
  Session *session= current_session;
  internal::st_my_thread_var *thread_var= session->getThreadVar();

  boost::condition_variable *cond= &thread_var->suspend;
  enum enum_thr_lock_result result= THR_LOCK_ABORTED;
  bool can_deadlock= test(data->owner->info->n_cursors);

  if (!in_wait_list)
  {
    (*wait->last)=data;				/* Wait for lock */
    data->prev= wait->last;
    wait->last= &data->next;
  }

  current_global_counters.locks_waited++;

  /* Set up control struct to allow others to abort locks */
  thread_var->current_mutex= data->lock->native_handle();
  thread_var->current_cond=  &thread_var->suspend;
  data->cond= &thread_var->suspend;;

  while (!thread_var->abort || in_wait_list)
  {
    boost::mutex::scoped_lock scoped(*data->lock->native_handle(), boost::adopt_lock_t());

    if (can_deadlock)
    {
      boost::xtime xt; 
      xtime_get(&xt, boost::TIME_UTC); 
      xt.sec += table_lock_wait_timeout; 
      if (not cond->timed_wait(scoped, xt))
      {
        result= THR_LOCK_WAIT_TIMEOUT;
        scoped.release();
        break;
      }
    }
    else
    {
      cond->wait(scoped);
    }
    /*
      We must break the wait if one of the following occurs:
      - the connection has been aborted (!thread_var->abort), but
        this is not a delayed insert thread (in_wait_list). For a delayed
        insert thread the proper action at shutdown is, apparently, to
        acquire the lock and complete the insert.
      - the lock has been granted (data->cond is set to NULL by the granter),
        or the waiting has been aborted (additionally data->type is set to
        TL_UNLOCK).
      - the wait has timed out (rc == ETIMEDOUT)
      Order of checks below is important to not report about timeout
      if the predicate is true.
    */
    if (data->cond == NULL)
    {
      scoped.release();
      break;
    }
    scoped.release();
  }
  if (data->cond || data->type == TL_UNLOCK)
  {
    if (data->cond)                             /* aborted or timed out */
    {
      if (((*data->prev)=data->next))		/* remove from wait-list */
	data->next->prev= data->prev;
      else
	wait->last=data->prev;
      data->type= TL_UNLOCK;                    /* No lock */
      wake_up_waiters(data->lock);
    }
  }
  else
  {
    result= THR_LOCK_SUCCESS;
  }
  data->lock->unlock();

  /* The following must be done after unlock of lock->mutex */
  boost::mutex::scoped_lock scopedLock(thread_var->mutex);
  thread_var->current_mutex= NULL;
  thread_var->current_cond= NULL;
  return(result);
}


static enum enum_thr_lock_result thr_lock(THR_LOCK_DATA *data, THR_LOCK_OWNER *owner, enum thr_lock_type lock_type)
{
  THR_LOCK *lock= data->lock;
  enum enum_thr_lock_result result= THR_LOCK_SUCCESS;
  struct st_lock_list *wait_queue;
  THR_LOCK_DATA *lock_owner;

  data->next=0;
  data->cond=0;					/* safety */
  data->type=lock_type;
  data->owner= owner;                           /* Must be reset ! */
  lock->lock();
  if ((int) lock_type <= (int) TL_READ_NO_INSERT)
  {
    /* Request for READ lock */
    if (lock->write.data)
    {
      /* We can allow a read lock even if there is already a write lock
	 on the table in one the following cases:
	 - This thread alread have a write lock on the table
	 - The write lock is TL_WRITE_ALLOW_READ or TL_WRITE_DELAYED
           and the read lock is TL_READ_HIGH_PRIORITY or TL_READ
         - The write lock is TL_WRITE_CONCURRENT_INSERT or TL_WRITE_ALLOW_WRITE
	   and the read lock is not TL_READ_NO_INSERT
      */

      if (thr_lock_owner_equal(data->owner, lock->write.data->owner) ||
	  (lock->write.data->type <= TL_WRITE_CONCURRENT_INSERT &&
	   (((int) lock_type <= (int) TL_READ_WITH_SHARED_LOCKS) ||
	    (lock->write.data->type != TL_WRITE_CONCURRENT_INSERT &&
	     lock->write.data->type != TL_WRITE_ALLOW_READ))))
      {						/* Already got a write lock */
	(*lock->read.last)=data;		/* Add to running FIFO */
	data->prev=lock->read.last;
	lock->read.last= &data->next;
	if (lock_type == TL_READ_NO_INSERT)
	  lock->read_no_write_count++;
        current_global_counters.locks_immediate++;
	goto end;
      }
      if (lock->write.data->type == TL_WRITE_ONLY)
      {
	/* We are not allowed to get a READ lock in this case */
	data->type=TL_UNLOCK;
        result= THR_LOCK_ABORTED;               /* Can't wait for this one */
	goto end;
      }
    }
    else if (!lock->write_wait.data ||
	     lock->write_wait.data->type <= TL_WRITE_DEFAULT ||
	     have_old_read_lock(lock->read.data, data->owner))
    {						/* No important write-locks */
      (*lock->read.last)=data;			/* Add to running FIFO */
      data->prev=lock->read.last;
      lock->read.last= &data->next;
      if (lock_type == TL_READ_NO_INSERT)
	lock->read_no_write_count++;
      current_global_counters.locks_immediate++;
      goto end;
    }
    /*
      We're here if there is an active write lock or no write
      lock but a high priority write waiting in the write_wait queue.
      In the latter case we should yield the lock to the writer.
    */
    wait_queue= &lock->read_wait;
  }
  else						/* Request for WRITE lock */
  {
    if (lock_type == TL_WRITE_CONCURRENT_INSERT)
      data->type=lock_type= thr_upgraded_concurrent_insert_lock;

    if (lock->write.data)			/* If there is a write lock */
    {
      if (lock->write.data->type == TL_WRITE_ONLY)
      {
        /* Allow lock owner to bypass TL_WRITE_ONLY. */
        if (!thr_lock_owner_equal(data->owner, lock->write.data->owner))
        {
          /* We are not allowed to get a lock in this case */
          data->type=TL_UNLOCK;
          result= THR_LOCK_ABORTED;               /* Can't wait for this one */
          goto end;
        }
      }

      /*
	The following test will not work if the old lock was a
	TL_WRITE_ALLOW_WRITE, TL_WRITE_ALLOW_READ or TL_WRITE_DELAYED in
	the same thread, but this will never happen within MySQL.
      */
      if (thr_lock_owner_equal(data->owner, lock->write.data->owner) ||
	  (lock_type == TL_WRITE_ALLOW_WRITE &&
	   !lock->write_wait.data &&
	   lock->write.data->type == TL_WRITE_ALLOW_WRITE))
      {
	/*
          We have already got a write lock or all locks are
          TL_WRITE_ALLOW_WRITE
        */

	(*lock->write.last)=data;	/* Add to running fifo */
	data->prev=lock->write.last;
	lock->write.last= &data->next;
        current_global_counters.locks_immediate++;
	goto end;
      }
    }
    else
    {
      if (!lock->write_wait.data)
      {						/* no scheduled write locks */
        bool concurrent_insert= 0;
	if (lock_type == TL_WRITE_CONCURRENT_INSERT)
        {
          concurrent_insert= 1;
        }

	if (!lock->read.data ||
	    (lock_type <= TL_WRITE_CONCURRENT_INSERT &&
	     ((lock_type != TL_WRITE_CONCURRENT_INSERT &&
	       lock_type != TL_WRITE_ALLOW_WRITE) ||
	      !lock->read_no_write_count)))
	{
	  (*lock->write.last)=data;		/* Add as current write lock */
	  data->prev=lock->write.last;
	  lock->write.last= &data->next;
          current_global_counters.locks_immediate++;
	  goto end;
	}
      }
    }
    wait_queue= &lock->write_wait;
  }
  /*
    Try to detect a trivial deadlock when using cursors: attempt to
    lock a table that is already locked by an open cursor within the
    same connection. lock_owner can be zero if we succumbed to a high
    priority writer in the write_wait queue.
  */
  lock_owner= lock->read.data ? lock->read.data : lock->write.data;
  if (lock_owner && lock_owner->owner->info == owner->info)
  {
    result= THR_LOCK_DEADLOCK;
    goto end;
  }

  /* Can't get lock yet;  Wait for it */
  return(wait_for_lock(wait_queue, data, 0));
end:
  lock->unlock();

  return(result);
}


static void free_all_read_locks(THR_LOCK *lock, bool using_concurrent_insert)
{
  THR_LOCK_DATA *data= lock->read_wait.data;

  /* move all locks from read_wait list to read list */
  (*lock->read.last)=data;
  data->prev=lock->read.last;
  lock->read.last=lock->read_wait.last;

  /* Clear read_wait list */
  lock->read_wait.last= &lock->read_wait.data;

  do
  {
    boost::condition_variable *cond= data->cond;
    if ((int) data->type == (int) TL_READ_NO_INSERT)
    {
      if (using_concurrent_insert)
      {
	/*
	  We can't free this lock;
	  Link lock away from read chain back into read_wait chain
	*/
	if (((*data->prev)=data->next))
	  data->next->prev=data->prev;
	else
	  lock->read.last=data->prev;
	*lock->read_wait.last= data;
	data->prev= lock->read_wait.last;
	lock->read_wait.last= &data->next;
	continue;
      }
      lock->read_no_write_count++;
    }
    data->cond= NULL;				/* Mark thread free */
    cond->notify_one();
  } while ((data=data->next));
  *lock->read_wait.last=0;
  if (!lock->read_wait.data)
    lock->write_lock_count=0;
}

/* Unlock lock and free next thread on same lock */

static void thr_unlock(THR_LOCK_DATA *data)
{
  THR_LOCK *lock=data->lock;
  enum thr_lock_type lock_type=data->type;
  lock->lock();

  if (((*data->prev)=data->next))		/* remove from lock-list */
    data->next->prev= data->prev;
  else if (lock_type <= TL_READ_NO_INSERT)
    lock->read.last=data->prev;
  else
    lock->write.last=data->prev;
  if (lock_type >= TL_WRITE_CONCURRENT_INSERT)
  { }
  else
  { }
  if (lock_type == TL_READ_NO_INSERT)
    lock->read_no_write_count--;
  data->type=TL_UNLOCK;				/* Mark unlocked */
  wake_up_waiters(lock);
  lock->unlock();
}


/**
  @brief  Wake up all threads which pending requests for the lock
          can be satisfied.

  @param  lock  Lock for which threads should be woken up

*/

static void wake_up_waiters(THR_LOCK *lock)
{
  THR_LOCK_DATA *data;
  enum thr_lock_type lock_type;

  if (!lock->write.data)			/* If no active write locks */
  {
    data=lock->write_wait.data;
    if (!lock->read.data)			/* If no more locks in use */
    {
      /* Release write-locks with TL_WRITE or TL_WRITE_ONLY priority first */
      if (data &&
	  (!lock->read_wait.data || lock->read_wait.data->type <= TL_READ_WITH_SHARED_LOCKS))
      {
	if (lock->write_lock_count++ > max_write_lock_count)
	{
	  /* Too many write locks in a row;  Release all waiting read locks */
	  lock->write_lock_count=0;
	  if (lock->read_wait.data)
	  {
	    free_all_read_locks(lock,0);
	    goto end;
	  }
	}
	for (;;)
	{
	  if (((*data->prev)=data->next))	/* remove from wait-list */
	    data->next->prev= data->prev;
	  else
	    lock->write_wait.last=data->prev;
	  (*lock->write.last)=data;		/* Put in execute list */
	  data->prev=lock->write.last;
	  data->next=0;
	  lock->write.last= &data->next;

	  {
            boost::condition_variable *cond= data->cond;
	    data->cond= NULL;				/* Mark thread free */
            cond->notify_one(); /* Start waiting thred */
	  }
	  if (data->type != TL_WRITE_ALLOW_WRITE ||
	      !lock->write_wait.data ||
	      lock->write_wait.data->type != TL_WRITE_ALLOW_WRITE)
	    break;
	  data=lock->write_wait.data;		/* Free this too */
	}
	if (data->type >= TL_WRITE)
          goto end;
	/* Release possible read locks together with the write lock */
      }
      if (lock->read_wait.data)
	free_all_read_locks(lock,
			    data &&
			    (data->type == TL_WRITE_CONCURRENT_INSERT ||
			     data->type == TL_WRITE_ALLOW_WRITE));
    }
    else if (data &&
	     (lock_type=data->type) <= TL_WRITE_CONCURRENT_INSERT &&
	     ((lock_type != TL_WRITE_CONCURRENT_INSERT &&
	       lock_type != TL_WRITE_ALLOW_WRITE) ||
	      !lock->read_no_write_count))
    {
      do {
        boost::condition_variable *cond= data->cond;
	if (((*data->prev)=data->next))		/* remove from wait-list */
	  data->next->prev= data->prev;
	else
	  lock->write_wait.last=data->prev;
	(*lock->write.last)=data;		/* Put in execute list */
	data->prev=lock->write.last;
	lock->write.last= &data->next;
	data->next=0;				/* Only one write lock */
	data->cond= NULL;				/* Mark thread free */
        cond->notify_one(); /* Start waiting thread */
      } while (lock_type == TL_WRITE_ALLOW_WRITE &&
	       (data=lock->write_wait.data) &&
	       data->type == TL_WRITE_ALLOW_WRITE);
      if (lock->read_wait.data)
	free_all_read_locks(lock,
			    (lock_type == TL_WRITE_CONCURRENT_INSERT ||
			     lock_type == TL_WRITE_ALLOW_WRITE));
    }
    else if (!data && lock->read_wait.data)
    {
      free_all_read_locks(lock,0);
    }
  }
end:
  return;
}


/*
** Get all locks in a specific order to avoid dead-locks
** Sort acording to lock position and put write_locks before read_locks if
** lock on same lock.
*/


#define LOCK_CMP(A,B) ((unsigned char*) (A->lock) - (uint32_t) ((A)->type) < (unsigned char*) (B->lock)- (uint32_t) ((B)->type))

static void sort_locks(THR_LOCK_DATA **data,uint32_t count)
{
  THR_LOCK_DATA **pos,**end,**prev,*tmp;

  /* Sort locks with insertion sort (fast because almost always few locks) */

  for (pos=data+1,end=data+count; pos < end ; pos++)
  {
    tmp= *pos;
    if (LOCK_CMP(tmp,pos[-1]))
    {
      prev=pos;
      do {
	prev[0]=prev[-1];
      } while (--prev != data && LOCK_CMP(tmp,prev[-1]));
      prev[0]=tmp;
    }
  }
}


enum enum_thr_lock_result
thr_multi_lock(THR_LOCK_DATA **data, uint32_t count, THR_LOCK_OWNER *owner)
{
  THR_LOCK_DATA **pos,**end;
  if (count > 1)
    sort_locks(data,count);
  /* lock everything */
  for (pos=data,end=data+count; pos < end ; pos++)
  {
    enum enum_thr_lock_result result= thr_lock(*pos, owner, (*pos)->type);
    if (result != THR_LOCK_SUCCESS)
    {						/* Aborted */
      thr_multi_unlock(data,(uint32_t) (pos-data));
      return(result);
    }
  }
  /*
    Ensure that all get_locks() have the same status
    If we lock the same table multiple times, we must use the same
    status_param!
  */
#if !defined(DONT_USE_RW_LOCKS)
  if (count > 1)
  {
    THR_LOCK_DATA *last_lock= end[-1];
    pos=end-1;
    do
    {
      pos--;
      last_lock=(*pos);
    } while (pos != data);
  }
#endif
  return(THR_LOCK_SUCCESS);
}

  /* free all locks */

void thr_multi_unlock(THR_LOCK_DATA **data,uint32_t count)
{
  THR_LOCK_DATA **pos,**end;

  for (pos=data,end=data+count; pos < end ; pos++)
  {
    if ((*pos)->type != TL_UNLOCK)
      thr_unlock(*pos);
  }
  return;
}

/*
  Abort all threads waiting for a lock. The lock will be upgraded to
  TL_WRITE_ONLY to abort any new accesses to the lock
*/

void THR_LOCK::abort_locks()
{
  boost::mutex::scoped_lock scopedLock(mutex);

  for (THR_LOCK_DATA *local_data= read_wait.data; local_data ; local_data= local_data->next)
  {
    local_data->type= TL_UNLOCK;			/* Mark killed */
    /* It's safe to signal the cond first: we're still holding the mutex. */
    local_data->cond->notify_one();
    local_data->cond= NULL;				/* Removed from list */
  }
  for (THR_LOCK_DATA *local_data= write_wait.data; local_data ; local_data= local_data->next)
  {
    local_data->type= TL_UNLOCK;
    local_data->cond->notify_one();
    local_data->cond= NULL;
  }
  read_wait.last= &read_wait.data;
  write_wait.last= &write_wait.data;
  read_wait.data= write_wait.data=0;
  if (write.data)
    write.data->type=TL_WRITE_ONLY;
}


/*
  Abort all locks for specific table/thread combination

  This is used to abort all locks for a specific thread
*/

bool THR_LOCK::abort_locks_for_thread(uint64_t thread_id_arg)
{
  bool found= false;

  boost::mutex::scoped_lock scopedLock(mutex);
  for (THR_LOCK_DATA *local_data= read_wait.data; local_data ; local_data= local_data->next)
  {
    if (local_data->owner->info->thread_id == thread_id_arg)
    {
      local_data->type= TL_UNLOCK;			/* Mark killed */
      /* It's safe to signal the cond first: we're still holding the mutex. */
      found= true;
      local_data->cond->notify_one();
      local_data->cond= 0;				/* Removed from list */

      if (((*local_data->prev)= local_data->next))
	local_data->next->prev= local_data->prev;
      else
	read_wait.last= local_data->prev;
    }
  }
  for (THR_LOCK_DATA *local_data= write_wait.data; local_data ; local_data= local_data->next)
  {
    if (local_data->owner->info->thread_id == thread_id_arg)
    {
      local_data->type= TL_UNLOCK;
      found= true;
      local_data->cond->notify_one();
      local_data->cond= NULL;

      if (((*local_data->prev)= local_data->next))
	local_data->next->prev= local_data->prev;
      else
	write_wait.last= local_data->prev;
    }
  }
  wake_up_waiters(this);

  return found;
}

} /* namespace drizzled */