~drizzle-trunk/drizzle/development

/* 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 */

/*

  These functions handle keyblock cacheing for ISAM and MyISAM tables.

  One cache can handle many files.

  It must contain buffers of the same blocksize.

  init_key_cache() should be used to init cache handler.

  The free list (free_block_list) is a stack like structure.

  When a block is freed by free_block(), it is pushed onto the stack.

  When a new block is required it is first tried to pop one from the stack.

  If the stack is empty, it is tried to get a never-used block from the pool.

  If this is empty too, then a block is taken from the LRU ring, flushing it

  to disk, if neccessary. This is handled in find_key_block().

  With the new free list, the blocks can have three temperatures:

  hot, warm and cold (which is free). This is remembered in the block header

  by the enum BLOCK_TEMPERATURE temperature variable. Remembering the

  temperature is neccessary to correctly count the number of warm blocks,

  which is required to decide when blocks are allowed to become hot. Whenever

  a block is inserted to another (sub-)chain, we take the old and new

  temperature into account to decide if we got one more or less warm block.

  blocks_unused is the sum of never used blocks in the pool and of currently

  free blocks. blocks_used is the number of blocks fetched from the pool and

  as such gives the maximum number of in-use blocks at any time.

  Key Cache Locking

  =================

  All key cache locking is done with a single mutex per key cache:

  keycache->cache_lock. This mutex is locked almost all the time

  when executing code in this file (mf_keycache.c).

  However it is released for I/O and some copy operations.

  The cache_lock is also released when waiting for some event. Waiting

  and signalling is done via condition variables. In most cases the

  thread waits on its thread->suspend condition variable. Every thread

  has a my_thread_var structure, which contains this variable and a

  '*next' and '**prev' pointer. These pointers are used to insert the

  thread into a wait queue.

  A thread can wait for one block and thus be in one wait queue at a

  time only.

  Before starting to wait on its condition variable with

  pthread_cond_wait(), the thread enters itself to a specific wait queue

  with link_into_queue() (double linked with '*next' + '**prev') or

  wait_on_queue() (single linked with '*next').

  Another thread, when releasing a resource, looks up the waiting thread

  in the related wait queue. It sends a signal with

  pthread_cond_signal() to the waiting thread.

  NOTE: Depending on the particular wait situation, either the sending

  thread removes the waiting thread from the wait queue with

  unlink_from_queue() or release_whole_queue() respectively, or the waiting

  thread removes itself.

  There is one exception from this locking scheme when one thread wants

  to reuse a block for some other address. This works by first marking

  the block reserved (status= BLOCK_IN_SWITCH) and then waiting for all

  threads that are reading the block to finish. Each block has a

  reference to a condition variable (condvar). It holds a reference to

  the thread->suspend condition variable for the waiting thread (if such

  a thread exists). When that thread is signaled, the reference is

  cleared. The number of readers of a block is registered in

  block->hash_link->requests. See wait_for_readers() / remove_reader()

  for details. This is similar to the above, but it clearly means that

  only one thread can wait for a particular block. There is no queue in

  this case. Strangely enough block->convar is used for waiting for the

  assigned hash_link only. More precisely it is used to wait for all

  requests to be unregistered from the assigned hash_link.

  The resize_queue serves two purposes:

  1. Threads that want to do a resize wait there if in_resize is set.

     This is not used in the server. The server refuses a second resize

     request if one is already active. keycache->in_init is used for the

     synchronization. See set_var.cc.

  2. Threads that want to access blocks during resize wait here during

     the re-initialization phase.

  When the resize is done, all threads on the queue are signalled.

  Hypothetical resizers can compete for resizing, and read/write

  requests will restart to request blocks from the freshly resized

  cache. If the cache has been resized too small, it is disabled and

  'can_be_used' is false. In this case read/write requests bypass the

  cache. Since they increment and decrement 'cnt_for_resize_op', the

  next resizer can wait on the queue 'waiting_for_resize_cnt' until all

  I/O finished.

*/

#include "config.h"

#include "drizzled/my_error.h"

#include "drizzled/internal/my_sys.h"

#include "keycache.h"

#include "drizzled/internal/m_string.h"

#include "drizzled/internal/my_bit.h"

#include <errno.h>

#include <stdarg.h>

using namespace drizzled;

static void change_key_cache_param(KEY_CACHE *keycache, uint32_t division_limit,

                            uint32_t age_threshold);

/*

  Some compilation flags have been added specifically for this module

  to control the following:

  - not to let a thread to yield the control when reading directly

    from key cache, which might improve performance in many cases;

    to enable this add:

    #define SERIALIZED_READ_FROM_CACHE

  - to set an upper bound for number of threads simultaneously

    using the key cache; this setting helps to determine an optimal

    size for hash table and improve performance when the number of

    blocks in the key cache much less than the number of threads

    accessing it;

    to set this number equal to <N> add

      #define MAX_THREADS <N>

  - to substitute calls of pthread_cond_wait for calls of

    pthread_cond_timedwait (wait with timeout set up);

    this setting should be used only when you want to trap a deadlock

    situation, which theoretically should not happen;

    to set timeout equal to <T> seconds add

      #define KEYCACHE_TIMEOUT <T>

  Example of the settings:

    #define SERIALIZED_READ_FROM_CACHE

    #define MAX_THREADS   100

    #define KEYCACHE_TIMEOUT  1

*/

#define STRUCT_PTR(TYPE, MEMBER, a)                                           \

          (TYPE *) ((char *) (a) - offsetof(TYPE, MEMBER))

/* types of condition variables */

#define  COND_FOR_REQUESTED 0

#define  COND_FOR_SAVED     1

typedef pthread_cond_t KEYCACHE_CONDVAR;

/* descriptor of the page in the key cache block buffer */

struct st_keycache_page

{

  int file;               /* file to which the page belongs to  */

  internal::my_off_t filepos;       /* position of the page in the file   */

};

/* element in the chain of a hash table bucket */

struct st_hash_link

{

  struct st_hash_link *next, **prev; /* to connect links in the same bucket  */

  struct st_block_link *block;       /* reference to the block for the page: */

  int file;                         /* from such a file                     */

  internal::my_off_t diskpos;                  /* with such an offset                  */

  uint32_t requests;                     /* number of requests for the page      */

};

/* simple states of a block */

#define BLOCK_ERROR           1 /* an error occured when performing file i/o */

#define BLOCK_READ            2 /* file block is in the block buffer         */

#define BLOCK_IN_SWITCH       4 /* block is preparing to read new page       */

#define BLOCK_REASSIGNED      8 /* blk does not accept requests for old page */

#define BLOCK_IN_FLUSH       16 /* block is selected for flush               */

#define BLOCK_CHANGED        32 /* block buffer contains a dirty page        */

#define BLOCK_IN_USE         64 /* block is not free                         */

#define BLOCK_IN_EVICTION   128 /* block is selected for eviction            */

#define BLOCK_IN_FLUSHWRITE 256 /* block is in write to file                 */

#define BLOCK_FOR_UPDATE    512 /* block is selected for buffer modification */

/* page status, returned by find_key_block */

#define PAGE_READ               0

#define PAGE_TO_BE_READ         1

#define PAGE_WAIT_TO_BE_READ    2

/* block temperature determines in which (sub-)chain the block currently is */

enum BLOCK_TEMPERATURE { BLOCK_COLD /*free*/ , BLOCK_WARM , BLOCK_HOT };

/* key cache block */

struct st_block_link

{

  struct st_block_link

    *next_used, **prev_used;   /* to connect links in the LRU chain (ring)   */

  struct st_block_link

    *next_changed, **prev_changed; /* for lists of file dirty/clean blocks   */

  struct st_hash_link *hash_link; /* backward ptr to referring hash_link     */

  KEYCACHE_WQUEUE wqueue[2]; /* queues on waiting requests for new/old pages */

  uint32_t requests;          /* number of requests for the block                */

  unsigned char *buffer;           /* buffer for the block page                       */

  uint32_t offset;            /* beginning of modified data in the buffer        */

  uint32_t length;            /* end of data in the buffer                       */

  uint32_t status;            /* state of the block                              */

  enum BLOCK_TEMPERATURE temperature; /* block temperature: cold, warm, hot */

  uint32_t hits_left;         /* number of hits left until promotion             */

  uint64_t last_hit_time; /* timestamp of the last hit                      */

  KEYCACHE_CONDVAR *condvar; /* condition variable for 'no readers' event    */

};

KEY_CACHE dflt_key_cache_var;

KEY_CACHE *dflt_key_cache= &dflt_key_cache_var;

#define FLUSH_CACHE         2000            /* sort this many blocks at once */

static int flush_all_key_blocks(KEY_CACHE *keycache);

static void wait_on_queue(KEYCACHE_WQUEUE *wqueue,

                          pthread_mutex_t *mutex);

static void release_whole_queue(KEYCACHE_WQUEUE *wqueue);

static void free_block(KEY_CACHE *keycache, BLOCK_LINK *block);

#define KEYCACHE_HASH(f, pos)                                                 \

(((uint32_t) ((pos) / keycache->key_cache_block_size) +                          \

                                     (uint32_t) (f)) & (keycache->hash_entries-1))

#define FILE_HASH(f)                 ((uint) (f) & (CHANGED_BLOCKS_HASH-1))

#ifdef KEYCACHE_TIMEOUT

static int keycache_pthread_cond_wait(pthread_cond_t *cond,

                                      pthread_mutex_t *mutex);

#else

#define  keycache_pthread_cond_wait pthread_cond_wait

#endif

#define keycache_pthread_mutex_lock pthread_mutex_lock

#define keycache_pthread_mutex_unlock pthread_mutex_unlock

#define keycache_pthread_cond_signal pthread_cond_signal

static inline uint32_t next_power(uint32_t value)

{

  return my_round_up_to_next_power(value) << 1;

}

/*

  Initialize a key cache

  SYNOPSIS

    init_key_cache()

    keycache			pointer to a key cache data structure

    key_cache_block_size	size of blocks to keep cached data

    use_mem                 	total memory to use for the key cache

    division_limit		division limit (may be zero)

    age_threshold		age threshold (may be zero)

  RETURN VALUE

    number of blocks in the key cache, if successful,

    0 - otherwise.

  NOTES.

    if keycache->key_cache_inited != 0 we assume that the key cache

    is already initialized.  This is for now used by myisamchk, but shouldn't

    be something that a program should rely on!

    It's assumed that no two threads call this function simultaneously

    referring to the same key cache handle.

*/

int init_key_cache(KEY_CACHE *keycache, uint32_t key_cache_block_size,

		   size_t use_mem, uint32_t division_limit,

		   uint32_t age_threshold)

{

  uint32_t blocks, hash_links;

  size_t length;

  int error;

  assert(key_cache_block_size >= 512);

  if (keycache->key_cache_inited && keycache->disk_blocks > 0)

  {

    return(0);

  }

  keycache->global_cache_w_requests= keycache->global_cache_r_requests= 0;

  keycache->global_cache_read= keycache->global_cache_write= 0;

  keycache->disk_blocks= -1;

  if (! keycache->key_cache_inited)

  {

    keycache->key_cache_inited= 1;

    /*

      Initialize these variables once only.

      Their value must survive re-initialization during resizing.

    */

    keycache->in_resize= 0;

    keycache->resize_in_flush= 0;

    keycache->cnt_for_resize_op= 0;

    keycache->waiting_for_resize_cnt.last_thread= NULL;

    keycache->in_init= 0;

    pthread_mutex_init(&keycache->cache_lock, MY_MUTEX_INIT_FAST);

    keycache->resize_queue.last_thread= NULL;

  }

  keycache->key_cache_mem_size= use_mem;

  keycache->key_cache_block_size= key_cache_block_size;

  blocks= (uint32_t) (use_mem / (sizeof(BLOCK_LINK) + 2 * sizeof(HASH_LINK) +

                              sizeof(HASH_LINK*) * 5/4 + key_cache_block_size));

  /* It doesn't make sense to have too few blocks (less than 8) */

  if (blocks >= 8)

  {

    for ( ; ; )

    {

      /* Set my_hash_entries to the next bigger 2 power */

      if ((keycache->hash_entries= next_power(blocks)) < blocks * 5/4)

        keycache->hash_entries<<= 1;

      hash_links= 2 * blocks;

#if defined(MAX_THREADS)

      if (hash_links < MAX_THREADS + blocks - 1)

        hash_links= MAX_THREADS + blocks - 1;

#endif

      while ((length= (ALIGN_SIZE(blocks * sizeof(BLOCK_LINK)) +

		       ALIGN_SIZE(hash_links * sizeof(HASH_LINK)) +

		       ALIGN_SIZE(sizeof(HASH_LINK*) *

                                  keycache->hash_entries))) +

	     ((size_t) blocks * keycache->key_cache_block_size) > use_mem)

        blocks--;

      /* Allocate memory for cache page buffers */

      if ((keycache->block_mem= (unsigned char *)malloc((size_t) blocks * keycache->key_cache_block_size)))

      {

        /*

	  Allocate memory for blocks, hash_links and hash entries;

	  For each block 2 hash links are allocated

        */

        if ((keycache->block_root= (BLOCK_LINK*) malloc(length)))

          break;

        free(keycache->block_mem);

        keycache->block_mem= 0;

      }

      if (blocks < 8)

      {

        errno= ENOMEM;

        my_error(EE_OUTOFMEMORY, MYF(0), blocks * keycache->key_cache_block_size);

        goto err;

      }

      blocks= blocks / 4*3;

    }

    keycache->blocks_unused= blocks;

    keycache->disk_blocks= (int) blocks;

    keycache->hash_links= hash_links;

    keycache->hash_root= (HASH_LINK**) ((char*) keycache->block_root +

				        ALIGN_SIZE(blocks*sizeof(BLOCK_LINK)));

    keycache->hash_link_root= (HASH_LINK*) ((char*) keycache->hash_root +

				            ALIGN_SIZE((sizeof(HASH_LINK*) *

							keycache->hash_entries)));

    memset(keycache->block_root, 0,

           keycache->disk_blocks * sizeof(BLOCK_LINK));

    memset(keycache->hash_root, 0,

           keycache->hash_entries * sizeof(HASH_LINK*));

    memset(keycache->hash_link_root, 0,

           keycache->hash_links * sizeof(HASH_LINK));

    keycache->hash_links_used= 0;

    keycache->free_hash_list= NULL;

    keycache->blocks_used= keycache->blocks_changed= 0;

    keycache->global_blocks_changed= 0;

    keycache->blocks_available=0;		/* For debugging */

    /* The LRU chain is empty after initialization */

    keycache->used_last= NULL;

    keycache->used_ins= NULL;

    keycache->free_block_list= NULL;

    keycache->keycache_time= 0;

    keycache->warm_blocks= 0;

    keycache->min_warm_blocks= (division_limit ?

				blocks * division_limit / 100 + 1 :

				blocks);

    keycache->age_threshold= (age_threshold ?

			      blocks * age_threshold / 100 :

			      blocks);

    keycache->can_be_used= 1;

    keycache->waiting_for_hash_link.last_thread= NULL;

    keycache->waiting_for_block.last_thread= NULL;

    memset(keycache->changed_blocks, 0,

           sizeof(keycache->changed_blocks[0]) * CHANGED_BLOCKS_HASH);

    memset(keycache->file_blocks, 0,

           sizeof(keycache->file_blocks[0]) * CHANGED_BLOCKS_HASH);

  }

  else

  {

    /* myisam_key_buffer_size is specified too small. Disable the cache. */

    keycache->can_be_used= 0;

  }

  keycache->blocks= keycache->disk_blocks > 0 ? keycache->disk_blocks : 0;

  return((int) keycache->disk_blocks);

err:

  error= errno;

  keycache->disk_blocks= 0;

  keycache->blocks=  0;

  if (keycache->block_mem)

  {

    free(keycache->block_mem);

    keycache->block_mem= NULL;

  }

  if (keycache->block_root)

  {

    free((unsigned char*) keycache->block_root);

    keycache->block_root= NULL;

  }

  errno= error;

  keycache->can_be_used= 0;

  return(0);

}

/*

  Resize a key cache

  SYNOPSIS

    resize_key_cache()

    keycache     	        pointer to a key cache data structure

    key_cache_block_size        size of blocks to keep cached data

    use_mem			total memory to use for the new key cache

    division_limit		new division limit (if not zero)

    age_threshold		new age threshold (if not zero)

  RETURN VALUE

    number of blocks in the key cache, if successful,

    0 - otherwise.

  NOTES.

    The function first compares the memory size and the block size parameters

    with the key cache values.

    If they differ the function free the the memory allocated for the

    old key cache blocks by calling the end_key_cache function and

    then rebuilds the key cache with new blocks by calling

    init_key_cache.

    The function starts the operation only when all other threads

    performing operations with the key cache let her to proceed

    (when cnt_for_resize=0).

*/

int resize_key_cache(KEY_CACHE *keycache, uint32_t key_cache_block_size,

		     size_t use_mem, uint32_t division_limit,

		     uint32_t age_threshold)

{

  int blocks;

  if (!keycache->key_cache_inited)

    return(keycache->disk_blocks);

  if(key_cache_block_size == keycache->key_cache_block_size &&

     use_mem == keycache->key_cache_mem_size)

  {

    change_key_cache_param(keycache, division_limit, age_threshold);

    return(keycache->disk_blocks);

  }

  keycache_pthread_mutex_lock(&keycache->cache_lock);

  /*

    We may need to wait for another thread which is doing a resize

    already. This cannot happen in the MySQL server though. It allows

    one resizer only. In set_var.cc keycache->in_init is used to block

    multiple attempts.

  */

  while (keycache->in_resize)

  {

    wait_on_queue(&keycache->resize_queue, &keycache->cache_lock);

  }

  /*

    Mark the operation in progress. This blocks other threads from doing

    a resize in parallel. It prohibits new blocks to enter the cache.

    Read/write requests can bypass the cache during the flush phase.

  */

  keycache->in_resize= 1;

  /* Need to flush only if keycache is enabled. */

  if (keycache->can_be_used)

  {

    /* Start the flush phase. */

    keycache->resize_in_flush= 1;

    if (flush_all_key_blocks(keycache))

    {

      /* TODO: if this happens, we should write a warning in the log file ! */

      keycache->resize_in_flush= 0;

      blocks= 0;

      keycache->can_be_used= 0;

      goto finish;

    }

    /* End the flush phase. */

    keycache->resize_in_flush= 0;

  }

  /*

    Some direct read/write operations (bypassing the cache) may still be

    unfinished. Wait until they are done. If the key cache can be used,

    direct I/O is done in increments of key_cache_block_size. That is,

    every block is checked if it is in the cache. We need to wait for

    pending I/O before re-initializing the cache, because we may change

    the block size. Otherwise they could check for blocks at file

    positions where the new block division has none. We do also want to

    wait for I/O done when (if) the cache was disabled. It must not

    run in parallel with normal cache operation.

  */

  while (keycache->cnt_for_resize_op)

    wait_on_queue(&keycache->waiting_for_resize_cnt, &keycache->cache_lock);

  /*

    Free old cache structures, allocate new structures, and initialize

    them. Note that the cache_lock mutex and the resize_queue are left

    untouched. We do not lose the cache_lock and will release it only at

    the end of this function.

  */

  end_key_cache(keycache, 0);			/* Don't free mutex */

  /* The following will work even if use_mem is 0 */

  blocks= init_key_cache(keycache, key_cache_block_size, use_mem,

			 division_limit, age_threshold);

finish:

  /*

    Mark the resize finished. This allows other threads to start a

    resize or to request new cache blocks.

  */

  keycache->in_resize= 0;

  /* Signal waiting threads. */

  release_whole_queue(&keycache->resize_queue);

  keycache_pthread_mutex_unlock(&keycache->cache_lock);

  return(blocks);

}

/*

  Increment counter blocking resize key cache operation

*/

static inline void inc_counter_for_resize_op(KEY_CACHE *keycache)

{

  keycache->cnt_for_resize_op++;

}

/*

  Decrement counter blocking resize key cache operation;

  Signal the operation to proceed when counter becomes equal zero

*/

static inline void dec_counter_for_resize_op(KEY_CACHE *keycache)

{

  if (!--keycache->cnt_for_resize_op)

    release_whole_queue(&keycache->waiting_for_resize_cnt);

}

/*

  Change the key cache parameters

  SYNOPSIS

    change_key_cache_param()

    keycache			pointer to a key cache data structure

    division_limit		new division limit (if not zero)

    age_threshold		new age threshold (if not zero)

  RETURN VALUE

    none

  NOTES.

    Presently the function resets the key cache parameters

    concerning midpoint insertion strategy - division_limit and

    age_threshold.

*/

static void change_key_cache_param(KEY_CACHE *keycache, uint32_t division_limit,

			    uint32_t age_threshold)

{

  keycache_pthread_mutex_lock(&keycache->cache_lock);

  if (division_limit)

    keycache->min_warm_blocks= (keycache->disk_blocks *

				division_limit / 100 + 1);

  if (age_threshold)

    keycache->age_threshold=   (keycache->disk_blocks *

				age_threshold / 100);

  keycache_pthread_mutex_unlock(&keycache->cache_lock);

  return;

}

/*

  Remove key_cache from memory

  SYNOPSIS

    end_key_cache()

    keycache		key cache handle

    cleanup		Complete free (Free also mutex for key cache)

  RETURN VALUE

    none

*/

void end_key_cache(KEY_CACHE *keycache, bool cleanup)

{

  if (!keycache->key_cache_inited)

    return;

  if (keycache->disk_blocks > 0)

  {

    if (keycache->block_mem)

    {

      free(keycache->block_mem);

      keycache->block_mem= NULL;

      free((unsigned char*) keycache->block_root);

      keycache->block_root= NULL;

    }

    keycache->disk_blocks= -1;

    /* Reset blocks_changed to be safe if flush_all_key_blocks is called */

    keycache->blocks_changed= 0;

  }

  if (cleanup)

  {

    pthread_mutex_destroy(&keycache->cache_lock);

    keycache->key_cache_inited= keycache->can_be_used= 0;

  }

  return;

} /* end_key_cache */

/*

  Link a thread into double-linked queue of waiting threads.

  SYNOPSIS

    link_into_queue()

      wqueue              pointer to the queue structure

      thread              pointer to the thread to be added to the queue

  RETURN VALUE

    none

  NOTES.

    Queue is represented by a circular list of the thread structures

    The list is double-linked of the type (**prev,*next), accessed by

    a pointer to the last element.

*/

static void link_into_queue(KEYCACHE_WQUEUE *wqueue,

                            internal::st_my_thread_var *thread)

{

  internal::st_my_thread_var *last;

  assert(!thread->next && !thread->prev);

  if (! (last= wqueue->last_thread))

  {

    /* Queue is empty */

    thread->next= thread;

    thread->prev= &thread->next;

  }

  else

  {

    thread->prev= last->next->prev;

    last->next->prev= &thread->next;

    thread->next= last->next;

    last->next= thread;

  }

  wqueue->last_thread= thread;

}

/*

  Unlink a thread from double-linked queue of waiting threads

  SYNOPSIS

    unlink_from_queue()

      wqueue              pointer to the queue structure

      thread              pointer to the thread to be removed from the queue

  RETURN VALUE

    none

  NOTES.

    See NOTES for link_into_queue

*/

static void unlink_from_queue(KEYCACHE_WQUEUE *wqueue,

                                     internal::st_my_thread_var *thread)

{

  assert(thread->next && thread->prev);

  if (thread->next == thread)

    /* The queue contains only one member */

    wqueue->last_thread= NULL;

  else

  {

    thread->next->prev= thread->prev;

    *thread->prev=thread->next;

    if (wqueue->last_thread == thread)

      wqueue->last_thread= STRUCT_PTR(internal::st_my_thread_var, next,

                                      thread->prev);

  }

  thread->next= NULL;

  thread->prev= NULL;

}

/*

  Add a thread to single-linked queue of waiting threads

  SYNOPSIS

    wait_on_queue()

      wqueue            Pointer to the queue structure.

      mutex             Cache_lock to acquire after awake.

  RETURN VALUE

    none

  NOTES.

    Queue is represented by a circular list of the thread structures

    The list is single-linked of the type (*next), accessed by a pointer

    to the last element.

    The function protects against stray signals by verifying that the

    current thread is unlinked from the queue when awaking. However,

    since several threads can wait for the same event, it might be

    necessary for the caller of the function to check again if the

    condition for awake is indeed matched.

*/

static void wait_on_queue(KEYCACHE_WQUEUE *wqueue,

                          pthread_mutex_t *mutex)

{

  internal::st_my_thread_var *last;

  internal::st_my_thread_var *thread= my_thread_var;

  /* Add to queue. */

  assert(!thread->next);

  assert(!thread->prev); /* Not required, but must be true anyway. */

  if (! (last= wqueue->last_thread))

    thread->next= thread;

  else

  {

    thread->next= last->next;

    last->next= thread;

  }

  wqueue->last_thread= thread;

  /*

    Wait until thread is removed from queue by the signalling thread.

    The loop protects against stray signals.

  */

  do

  {

    keycache_pthread_cond_wait(&thread->suspend, mutex);

  }

  while (thread->next);

}

/*

  Remove all threads from queue signaling them to proceed

  SYNOPSIS

    release_whole_queue()

      wqueue            pointer to the queue structure

  RETURN VALUE

    none

  NOTES.

    See notes for wait_on_queue().

    When removed from the queue each thread is signaled via condition

    variable thread->suspend.

*/

static void release_whole_queue(KEYCACHE_WQUEUE *wqueue)

{

  internal::st_my_thread_var *last;

  internal::st_my_thread_var *next;

  internal::st_my_thread_var *thread;

  /* Queue may be empty. */

  if (!(last= wqueue->last_thread))

    return;

  next= last->next;

  do

  {

    thread=next;

    /* Signal the thread. */

    keycache_pthread_cond_signal(&thread->suspend);

    /* Take thread from queue. */

    next=thread->next;

    thread->next= NULL;

  }

  while (thread != last);

  /* Now queue is definitely empty. */

  wqueue->last_thread= NULL;

}

/*

  Unlink a block from the chain of dirty/clean blocks

*/

static void unlink_changed(BLOCK_LINK *block)

{

  assert(block->prev_changed && *block->prev_changed == block);

  if (block->next_changed)

    block->next_changed->prev_changed= block->prev_changed;

  *block->prev_changed= block->next_changed;

  block->next_changed= NULL;

  block->prev_changed= NULL;

}

/*

  Link a block into the chain of dirty/clean blocks

*/

static void link_changed(BLOCK_LINK *block, BLOCK_LINK **phead)

{

  assert(!block->next_changed);

  assert(!block->prev_changed);

  block->prev_changed= phead;

  if ((block->next_changed= *phead))

    (*phead)->prev_changed= &block->next_changed;

  *phead= block;

}