sync0rw.cc

Go to the documentation of this file.

/*****************************************************************************

Copyright (c) 1995, 2011, Oracle and/or its affiliates. All Rights Reserved.
Copyright (c) 2008, Google Inc.

Portions of this file contain modifications contributed and copyrighted by
Google, Inc. Those modifications are gratefully acknowledged and are described
briefly in the InnoDB documentation. The contributions by Google are
incorporated with their permission, and subject to the conditions contained in
the file COPYING.Google.

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.,
51 Franklin Street, Suite 500, Boston, MA 02110-1335 USA

*****************************************************************************/

/**************************************************/
#include "sync0rw.h"
#ifdef UNIV_NONINL
#include "sync0rw.ic"
#endif

#include "os0thread.h"
#include "mem0mem.h"
#include "srv0srv.h"
#include "os0sync.h" /* for INNODB_RW_LOCKS_USE_ATOMICS */
#include "ha_prototypes.h"

/*
        IMPLEMENTATION OF THE RW_LOCK
        =============================
The status of a rw_lock is held in lock_word. The initial value of lock_word is
X_LOCK_DECR. lock_word is decremented by 1 for each s-lock and by X_LOCK_DECR
for each x-lock. This describes the lock state for each value of lock_word:

lock_word == X_LOCK_DECR:      Unlocked.
0 < lock_word < X_LOCK_DECR:   Read locked, no waiting writers.
                               (X_LOCK_DECR - lock_word) is the
                               number of readers that hold the lock.
lock_word == 0:                Write locked
-X_LOCK_DECR < lock_word < 0:  Read locked, with a waiting writer.
                               (-lock_word) is the number of readers
                               that hold the lock.
lock_word <= -X_LOCK_DECR:     Recursively write locked. lock_word has been
                               decremented by X_LOCK_DECR for the first lock
                               and the first recursive lock, then by 1 for
                               each recursive lock thereafter.
                               So the number of locks is:
                               (lock_copy == 0) ? 1 : 2 - (lock_copy + X_LOCK_DECR)

The lock_word is always read and updated atomically and consistently, so that
it always represents the state of the lock, and the state of the lock changes
with a single atomic operation. This lock_word holds all of the information
that a thread needs in order to determine if it is eligible to gain the lock
or if it must spin or sleep. The one exception to this is that writer_thread
must be verified before recursive write locks: to solve this scenario, we make
writer_thread readable by all threads, but only writeable by the x-lock holder.

The other members of the lock obey the following rules to remain consistent:

recursive:      This and the writer_thread field together control the
                behaviour of recursive x-locking.
                lock->recursive must be FALSE in following states:
                        1) The writer_thread contains garbage i.e.: the
                        lock has just been initialized.
                        2) The lock is not x-held and there is no
                        x-waiter waiting on WAIT_EX event.
                        3) The lock is x-held or there is an x-waiter
                        waiting on WAIT_EX event but the 'pass' value
                        is non-zero.
                lock->recursive is TRUE iff:
                        1) The lock is x-held or there is an x-waiter
                        waiting on WAIT_EX event and the 'pass' value
                        is zero.
                This flag must be set after the writer_thread field
                has been updated with a memory ordering barrier.
                It is unset before the lock_word has been incremented.
writer_thread:  Is used only in recursive x-locking. Can only be safely
                read iff lock->recursive flag is TRUE.
                This field is uninitialized at lock creation time and
                is updated atomically when x-lock is acquired or when
                move_ownership is called. A thread is only allowed to
                set the value of this field to it's thread_id i.e.: a
                thread cannot set writer_thread to some other thread's
                id.
waiters:        May be set to 1 anytime, but to avoid unnecessary wake-up
                signals, it should only be set to 1 when there are threads
                waiting on event. Must be 1 when a writer starts waiting to
                ensure the current x-locking thread sends a wake-up signal
                during unlock. May only be reset to 0 immediately before a
                a wake-up signal is sent to event. On most platforms, a
                memory barrier is required after waiters is set, and before
                verifying lock_word is still held, to ensure some unlocker
                really does see the flags new value.
event:          Threads wait on event for read or writer lock when another
                thread has an x-lock or an x-lock reservation (wait_ex). A
                thread may only wait on event after performing the following
                actions in order:
                   (1) Record the counter value of event (with os_event_reset).
                   (2) Set waiters to 1.
                   (3) Verify lock_word <= 0.
                (1) must come before (2) to ensure signal is not missed.
                (2) must come before (3) to ensure a signal is sent.
                These restrictions force the above ordering.
                Immediately before sending the wake-up signal, we should:
                   (1) Verify lock_word == X_LOCK_DECR (unlocked)
                   (2) Reset waiters to 0.
wait_ex_event:  A thread may only wait on the wait_ex_event after it has
                performed the following actions in order:
                   (1) Decrement lock_word by X_LOCK_DECR.
                   (2) Record counter value of wait_ex_event (os_event_reset,
                       called from sync_array_reserve_cell).
                   (3) Verify that lock_word < 0.
                (1) must come first to ensures no other threads become reader
                or next writer, and notifies unlocker that signal must be sent.
                (2) must come before (3) to ensure the signal is not missed.
                These restrictions force the above ordering.
                Immediately before sending the wake-up signal, we should:
                   Verify lock_word == 0 (waiting thread holds x_lock)
*/

UNIV_INTERN rw_lock_stats_t     rw_lock_stats;

/* The global list of rw-locks */
UNIV_INTERN rw_lock_list_t      rw_lock_list;
UNIV_INTERN ib_mutex_t          rw_lock_list_mutex;

#ifdef UNIV_PFS_MUTEX
UNIV_INTERN mysql_pfs_key_t     rw_lock_list_mutex_key;
UNIV_INTERN mysql_pfs_key_t     rw_lock_mutex_key;
#endif /* UNIV_PFS_MUTEX */

#ifdef UNIV_SYNC_DEBUG
/* The global mutex which protects debug info lists of all rw-locks.
To modify the debug info list of an rw-lock, this mutex has to be
acquired in addition to the mutex protecting the lock. */

UNIV_INTERN ib_mutex_t          rw_lock_debug_mutex;

# ifdef UNIV_PFS_MUTEX
UNIV_INTERN mysql_pfs_key_t     rw_lock_debug_mutex_key;
# endif

/* If deadlock detection does not get immediately the mutex,
it may wait for this event */
UNIV_INTERN os_event_t          rw_lock_debug_event;
/* This is set to TRUE, if there may be waiters for the event */
UNIV_INTERN ibool               rw_lock_debug_waiters;

/******************************************************************/
static
rw_lock_debug_t*
rw_lock_debug_create(void);
/*======================*/
/******************************************************************/
static
void
rw_lock_debug_free(
/*===============*/
        rw_lock_debug_t* info);

/******************************************************************/
static
rw_lock_debug_t*
rw_lock_debug_create(void)
/*======================*/
{
        return((rw_lock_debug_t*) mem_alloc(sizeof(rw_lock_debug_t)));
}

/******************************************************************/
static
void
rw_lock_debug_free(
/*===============*/
        rw_lock_debug_t* info)
{
        mem_free(info);
}
#endif /* UNIV_SYNC_DEBUG */

/******************************************************************/
UNIV_INTERN
void
rw_lock_create_func(
/*================*/
        rw_lock_t*      lock,           
#ifdef UNIV_DEBUG
# ifdef UNIV_SYNC_DEBUG
        ulint           level,          
# endif /* UNIV_SYNC_DEBUG */
        const char*     cmutex_name,    
#endif /* UNIV_DEBUG */
        const char*     cfile_name,     
        ulint           cline)          
{
        /* If this is the very first time a synchronization object is
        created, then the following call initializes the sync system. */

#ifndef INNODB_RW_LOCKS_USE_ATOMICS
        mutex_create(rw_lock_mutex_key, rw_lock_get_mutex(lock),
                     SYNC_NO_ORDER_CHECK);

        lock->mutex.cfile_name = cfile_name;
        lock->mutex.cline = cline;

        ut_d(lock->mutex.cmutex_name = cmutex_name);
        ut_d(lock->mutex.ib_mutex_type = 1);
#else /* INNODB_RW_LOCKS_USE_ATOMICS */
# ifdef UNIV_DEBUG
        UT_NOT_USED(cmutex_name);
# endif
#endif /* INNODB_RW_LOCKS_USE_ATOMICS */

        lock->lock_word = X_LOCK_DECR;
        lock->waiters = 0;

        /* We set this value to signify that lock->writer_thread
        contains garbage at initialization and cannot be used for
        recursive x-locking. */
        lock->recursive = FALSE;
        /* Silence Valgrind when UNIV_DEBUG_VALGRIND is not enabled. */
        memset((void*) &lock->writer_thread, 0, sizeof lock->writer_thread);
        UNIV_MEM_INVALID(&lock->writer_thread, sizeof lock->writer_thread);

#ifdef UNIV_SYNC_DEBUG
        UT_LIST_INIT(lock->debug_list);

        lock->level = level;
#endif /* UNIV_SYNC_DEBUG */

        ut_d(lock->magic_n = RW_LOCK_MAGIC_N);

        lock->cfile_name = cfile_name;
        lock->cline = (unsigned int) cline;

        lock->count_os_wait = 0;
        lock->last_s_file_name = "not yet reserved";
        lock->last_x_file_name = "not yet reserved";
        lock->last_s_line = 0;
        lock->last_x_line = 0;
        lock->event = os_event_create();
        lock->wait_ex_event = os_event_create();

        mutex_enter(&rw_lock_list_mutex);

        ut_ad(UT_LIST_GET_FIRST(rw_lock_list) == NULL
              || UT_LIST_GET_FIRST(rw_lock_list)->magic_n == RW_LOCK_MAGIC_N);

        UT_LIST_ADD_FIRST(list, rw_lock_list, lock);

        mutex_exit(&rw_lock_list_mutex);
}

/******************************************************************/
UNIV_INTERN
void
rw_lock_free_func(
/*==============*/
        rw_lock_t*      lock)   
{
#ifndef INNODB_RW_LOCKS_USE_ATOMICS
        ib_mutex_t*     mutex;
#endif /* !INNODB_RW_LOCKS_USE_ATOMICS */

        ut_ad(rw_lock_validate(lock));
        ut_a(lock->lock_word == X_LOCK_DECR);

        mutex_enter(&rw_lock_list_mutex);

#ifndef INNODB_RW_LOCKS_USE_ATOMICS
        mutex = rw_lock_get_mutex(lock);
#endif /* !INNODB_RW_LOCKS_USE_ATOMICS */

        os_event_free(lock->event);

        os_event_free(lock->wait_ex_event);

        ut_ad(UT_LIST_GET_PREV(list, lock) == NULL
              || UT_LIST_GET_PREV(list, lock)->magic_n == RW_LOCK_MAGIC_N);
        ut_ad(UT_LIST_GET_NEXT(list, lock) == NULL
              || UT_LIST_GET_NEXT(list, lock)->magic_n == RW_LOCK_MAGIC_N);

        UT_LIST_REMOVE(list, rw_lock_list, lock);

        mutex_exit(&rw_lock_list_mutex);

        ut_d(lock->magic_n = 0);

#ifndef INNODB_RW_LOCKS_USE_ATOMICS
        /* We have merely removed the rw_lock from the list, the memory
        has not been freed. Therefore the pointer to mutex is valid. */
        mutex_free(mutex);
#endif /* !INNODB_RW_LOCKS_USE_ATOMICS */
}

#ifdef UNIV_DEBUG
/******************************************************************/
UNIV_INTERN
ibool
rw_lock_validate(
/*=============*/
        rw_lock_t*      lock)   
{
        ulint   waiters;
        lint    lock_word;

        ut_ad(lock);

        waiters = rw_lock_get_waiters(lock);
        lock_word = lock->lock_word;

        ut_ad(lock->magic_n == RW_LOCK_MAGIC_N);
        ut_ad(waiters == 0 || waiters == 1);
        ut_ad(lock_word > -(2 * X_LOCK_DECR));
        ut_ad(lock_word <= X_LOCK_DECR);

        return(TRUE);
}
#endif /* UNIV_DEBUG */

/******************************************************************/
UNIV_INTERN
void
rw_lock_s_lock_spin(
/*================*/
        rw_lock_t*      lock,   
        ulint           pass,   
        const char*     file_name, 
        ulint           line)   
{
        ulint           index;  /* index of the reserved wait cell */
        ulint           i = 0;  /* spin round count */
        sync_array_t*   sync_arr;
        size_t          counter_index;

        /* We reuse the thread id to index into the counter, cache
        it here for efficiency. */

        counter_index = (size_t) os_thread_get_curr_id();

        ut_ad(rw_lock_validate(lock));

        rw_lock_stats.rw_s_spin_wait_count.add(counter_index, 1);
lock_loop:

        /* Spin waiting for the writer field to become free */
        while (i < SYNC_SPIN_ROUNDS && lock->lock_word <= 0) {
                if (srv_spin_wait_delay) {
                        ut_delay(ut_rnd_interval(0, srv_spin_wait_delay));
                }

                i++;
        }

        if (i == SYNC_SPIN_ROUNDS) {
                os_thread_yield();
        }

        /* We try once again to obtain the lock */
        if (TRUE == rw_lock_s_lock_low(lock, pass, file_name, line)) {
                rw_lock_stats.rw_s_spin_round_count.add(counter_index, i);

                return; /* Success */
        } else {

                if (i < SYNC_SPIN_ROUNDS) {
                        goto lock_loop;
                }

                rw_lock_stats.rw_s_spin_round_count.add(counter_index, i);

                sync_arr = sync_array_get();

                sync_array_reserve_cell(
                        sync_arr, lock, RW_LOCK_SHARED,
                        file_name, line, &index);

                /* Set waiters before checking lock_word to ensure wake-up
                signal is sent. This may lead to some unnecessary signals. */
                rw_lock_set_waiter_flag(lock);

                if (TRUE == rw_lock_s_lock_low(lock, pass, file_name, line)) {
                        sync_array_free_cell(sync_arr, index);
                        return; /* Success */
                }

                /* these stats may not be accurate */
                lock->count_os_wait++;
                rw_lock_stats.rw_s_os_wait_count.add(counter_index, 1);

                sync_array_wait_event(sync_arr, index);

                i = 0;
                goto lock_loop;
        }
}

/******************************************************************/
UNIV_INTERN
void
rw_lock_x_lock_move_ownership(
/*==========================*/
        rw_lock_t*      lock)   
{
        ut_ad(rw_lock_is_locked(lock, RW_LOCK_EX));

        rw_lock_set_writer_id_and_recursion_flag(lock, TRUE);
}

/******************************************************************/
UNIV_INLINE
void
rw_lock_x_lock_wait(
/*================*/
        rw_lock_t*      lock,   
#ifdef UNIV_SYNC_DEBUG
        ulint           pass,   
#endif
        const char*     file_name,
        ulint           line)   
{
        ulint           index;
        ulint           i = 0;
        sync_array_t*   sync_arr;
        size_t          counter_index;

        /* We reuse the thread id to index into the counter, cache
        it here for efficiency. */

        counter_index = (size_t) os_thread_get_curr_id();

        ut_ad(lock->lock_word <= 0);

        while (lock->lock_word < 0) {
                if (srv_spin_wait_delay) {
                        ut_delay(ut_rnd_interval(0, srv_spin_wait_delay));
                }
                if(i < SYNC_SPIN_ROUNDS) {
                        i++;
                        continue;
                }

                /* If there is still a reader, then go to sleep.*/
                rw_lock_stats.rw_x_spin_round_count.add(counter_index, i);

                sync_arr = sync_array_get();

                sync_array_reserve_cell(
                        sync_arr, lock, RW_LOCK_WAIT_EX,
                        file_name, line, &index);

                i = 0;

                /* Check lock_word to ensure wake-up isn't missed.*/
                if (lock->lock_word < 0) {

                        /* these stats may not be accurate */
                        lock->count_os_wait++;
                        rw_lock_stats.rw_x_os_wait_count.add(counter_index, 1);

                        /* Add debug info as it is needed to detect possible
                        deadlock. We must add info for WAIT_EX thread for
                        deadlock detection to work properly. */
#ifdef UNIV_SYNC_DEBUG
                        rw_lock_add_debug_info(lock, pass, RW_LOCK_WAIT_EX,
                                               file_name, line);
#endif

                        sync_array_wait_event(sync_arr, index);
#ifdef UNIV_SYNC_DEBUG
                        rw_lock_remove_debug_info(
                                lock, pass, RW_LOCK_WAIT_EX);
#endif
                        /* It is possible to wake when lock_word < 0.
                        We must pass the while-loop check to proceed.*/
                } else {
                        sync_array_free_cell(sync_arr, index);
                }
        }
        rw_lock_stats.rw_x_spin_round_count.add(counter_index, i);
}

/******************************************************************/
UNIV_INLINE
ibool
rw_lock_x_lock_low(
/*===============*/
        rw_lock_t*      lock,   
        ulint           pass,   
        const char*     file_name,
        ulint           line)   
{
        if (rw_lock_lock_word_decr(lock, X_LOCK_DECR)) {

                /* lock->recursive also tells us if the writer_thread
                field is stale or active. As we are going to write
                our own thread id in that field it must be that the
                current writer_thread value is not active. */
                ut_a(!lock->recursive);

                /* Decrement occurred: we are writer or next-writer. */
                rw_lock_set_writer_id_and_recursion_flag(
                        lock, pass ? FALSE : TRUE);

                rw_lock_x_lock_wait(lock,
#ifdef UNIV_SYNC_DEBUG
                                    pass,
#endif
                                    file_name, line);

        } else {
                os_thread_id_t  thread_id = os_thread_get_curr_id();

                /* Decrement failed: relock or failed lock */
                if (!pass && lock->recursive
                    && os_thread_eq(lock->writer_thread, thread_id)) {
                        /* Relock */
                        if (lock->lock_word == 0) {
                                lock->lock_word -= X_LOCK_DECR;
                        } else {
                                --lock->lock_word;
                        }

                } else {
                        /* Another thread locked before us */
                        return(FALSE);
                }
        }
#ifdef UNIV_SYNC_DEBUG
        rw_lock_add_debug_info(lock, pass, RW_LOCK_EX, file_name, line);
#endif
        lock->last_x_file_name = file_name;
        lock->last_x_line = (unsigned int) line;

        return(TRUE);
}

/******************************************************************/
UNIV_INTERN
void
rw_lock_x_lock_func(
/*================*/
        rw_lock_t*      lock,   
        ulint           pass,   
        const char*     file_name,
        ulint           line)   
{
        ulint           i;      
        ulint           index;  
        sync_array_t*   sync_arr;
        ibool           spinning = FALSE;
        size_t          counter_index;

        /* We reuse the thread id to index into the counter, cache
        it here for efficiency. */

        counter_index = (size_t) os_thread_get_curr_id();

        ut_ad(rw_lock_validate(lock));
#ifdef UNIV_SYNC_DEBUG
        ut_ad(!rw_lock_own(lock, RW_LOCK_SHARED));
#endif /* UNIV_SYNC_DEBUG */

        i = 0;

lock_loop:

        if (rw_lock_x_lock_low(lock, pass, file_name, line)) {
                rw_lock_stats.rw_x_spin_round_count.add(counter_index, i);

                return; /* Locking succeeded */

        } else {

                if (!spinning) {
                        spinning = TRUE;

                        rw_lock_stats.rw_x_spin_wait_count.add(
                                counter_index, 1);
                }

                /* Spin waiting for the lock_word to become free */
                while (i < SYNC_SPIN_ROUNDS
                       && lock->lock_word <= 0) {
                        if (srv_spin_wait_delay) {
                                ut_delay(ut_rnd_interval(0,
                                                         srv_spin_wait_delay));
                        }

                        i++;
                }
                if (i == SYNC_SPIN_ROUNDS) {
                        os_thread_yield();
                } else {
                        goto lock_loop;
                }
        }

        rw_lock_stats.rw_x_spin_round_count.add(counter_index, i);

        sync_arr = sync_array_get();

        sync_array_reserve_cell(
                sync_arr, lock, RW_LOCK_EX, file_name, line, &index);

        /* Waiters must be set before checking lock_word, to ensure signal
        is sent. This could lead to a few unnecessary wake-up signals. */
        rw_lock_set_waiter_flag(lock);

        if (rw_lock_x_lock_low(lock, pass, file_name, line)) {
                sync_array_free_cell(sync_arr, index);
                return; /* Locking succeeded */
        }

        /* these stats may not be accurate */
        lock->count_os_wait++;
        rw_lock_stats.rw_x_os_wait_count.add(counter_index, 1);

        sync_array_wait_event(sync_arr, index);

        i = 0;
        goto lock_loop;
}

#ifdef UNIV_SYNC_DEBUG
/******************************************************************/
UNIV_INTERN
void
rw_lock_debug_mutex_enter(void)
/*===========================*/
{
loop:
        if (0 == mutex_enter_nowait(&rw_lock_debug_mutex)) {
                return;
        }

        os_event_reset(rw_lock_debug_event);

        rw_lock_debug_waiters = TRUE;

        if (0 == mutex_enter_nowait(&rw_lock_debug_mutex)) {
                return;
        }

        os_event_wait(rw_lock_debug_event);

        goto loop;
}

/******************************************************************/
UNIV_INTERN
void
rw_lock_debug_mutex_exit(void)
/*==========================*/
{
        mutex_exit(&rw_lock_debug_mutex);

        if (rw_lock_debug_waiters) {
                rw_lock_debug_waiters = FALSE;
                os_event_set(rw_lock_debug_event);
        }
}

/******************************************************************/
UNIV_INTERN
void
rw_lock_add_debug_info(
/*===================*/
        rw_lock_t*      lock,           
        ulint           pass,           
        ulint           lock_type,      
        const char*     file_name,      
        ulint           line)           
{
        rw_lock_debug_t*        info;

        ut_ad(lock);
        ut_ad(file_name);

        info = rw_lock_debug_create();

        rw_lock_debug_mutex_enter();

        info->file_name = file_name;
        info->line      = line;
        info->lock_type = lock_type;
        info->thread_id = os_thread_get_curr_id();
        info->pass      = pass;

        UT_LIST_ADD_FIRST(list, lock->debug_list, info);

        rw_lock_debug_mutex_exit();

        if ((pass == 0) && (lock_type != RW_LOCK_WAIT_EX)) {
                sync_thread_add_level(lock, lock->level,
                                      lock_type == RW_LOCK_EX
                                      && lock->lock_word < 0);
        }
}

/******************************************************************/
UNIV_INTERN
void
rw_lock_remove_debug_info(
/*======================*/
        rw_lock_t*      lock,           
        ulint           pass,           
        ulint           lock_type)      
{
        rw_lock_debug_t*        info;

        ut_ad(lock);

        if ((pass == 0) && (lock_type != RW_LOCK_WAIT_EX)) {
                sync_thread_reset_level(lock);
        }

        rw_lock_debug_mutex_enter();

        info = UT_LIST_GET_FIRST(lock->debug_list);

        while (info != NULL) {
                if ((pass == info->pass)
                    && ((pass != 0)
                        || os_thread_eq(info->thread_id,
                                        os_thread_get_curr_id()))
                    && (info->lock_type == lock_type)) {

                        /* Found! */
                        UT_LIST_REMOVE(list, lock->debug_list, info);
                        rw_lock_debug_mutex_exit();

                        rw_lock_debug_free(info);

                        return;
                }

                info = UT_LIST_GET_NEXT(list, info);
        }

        ut_error;
}
#endif /* UNIV_SYNC_DEBUG */

#ifdef UNIV_SYNC_DEBUG
/******************************************************************/
UNIV_INTERN
ibool
rw_lock_own(
/*========*/
        rw_lock_t*      lock,           
        ulint           lock_type)      
{
        rw_lock_debug_t*        info;

        ut_ad(lock);
        ut_ad(rw_lock_validate(lock));

        rw_lock_debug_mutex_enter();

        info = UT_LIST_GET_FIRST(lock->debug_list);

        while (info != NULL) {

                if (os_thread_eq(info->thread_id, os_thread_get_curr_id())
                    && (info->pass == 0)
                    && (info->lock_type == lock_type)) {

                        rw_lock_debug_mutex_exit();
                        /* Found! */

                        return(TRUE);
                }

                info = UT_LIST_GET_NEXT(list, info);
        }
        rw_lock_debug_mutex_exit();

        return(FALSE);
}
#endif /* UNIV_SYNC_DEBUG */

/******************************************************************/
UNIV_INTERN
ibool
rw_lock_is_locked(
/*==============*/
        rw_lock_t*      lock,           
        ulint           lock_type)      
{
        ibool   ret     = FALSE;

        ut_ad(lock);
        ut_ad(rw_lock_validate(lock));

        if (lock_type == RW_LOCK_SHARED) {
                if (rw_lock_get_reader_count(lock) > 0) {
                        ret = TRUE;
                }
        } else if (lock_type == RW_LOCK_EX) {
                if (rw_lock_get_writer(lock) == RW_LOCK_EX) {
                        ret = TRUE;
                }
        } else {
                ut_error;
        }

        return(ret);
}

#ifdef UNIV_SYNC_DEBUG
/***************************************************************/
UNIV_INTERN
void
rw_lock_list_print_info(
/*====================*/
        FILE*   file)           
{
        rw_lock_t*      lock;
        ulint           count           = 0;
        rw_lock_debug_t* info;

        mutex_enter(&rw_lock_list_mutex);

        fputs("-------------\n"
              "RW-LATCH INFO\n"
              "-------------\n", file);

        lock = UT_LIST_GET_FIRST(rw_lock_list);

        while (lock != NULL) {

                count++;

#ifndef INNODB_RW_LOCKS_USE_ATOMICS
                mutex_enter(&(lock->mutex));
#endif
                if (lock->lock_word != X_LOCK_DECR) {

                        fprintf(file, "RW-LOCK: %p ", (void*) lock);

                        if (rw_lock_get_waiters(lock)) {
                                fputs(" Waiters for the lock exist\n", file);
                        } else {
                                putc('\n', file);
                        }

                        rw_lock_debug_mutex_enter();
                        info = UT_LIST_GET_FIRST(lock->debug_list);
                        while (info != NULL) {
                                rw_lock_debug_print(file, info);
                                info = UT_LIST_GET_NEXT(list, info);
                        }
                        rw_lock_debug_mutex_exit();
                }
#ifndef INNODB_RW_LOCKS_USE_ATOMICS
                mutex_exit(&(lock->mutex));
#endif

                lock = UT_LIST_GET_NEXT(list, lock);
        }

        fprintf(file, "Total number of rw-locks %ld\n", count);
        mutex_exit(&rw_lock_list_mutex);
}

/***************************************************************/
UNIV_INTERN
void
rw_lock_print(
/*==========*/
        rw_lock_t*      lock)   
{
        rw_lock_debug_t* info;

        fprintf(stderr,
                "-------------\n"
                "RW-LATCH INFO\n"
                "RW-LATCH: %p ", (void*) lock);

#ifndef INNODB_RW_LOCKS_USE_ATOMICS
        /* We used to acquire lock->mutex here, but it would cause a
        recursive call to sync_thread_add_level() if UNIV_SYNC_DEBUG
        is defined.  Since this function is only invoked from
        sync_thread_levels_g(), let us choose the smaller evil:
        performing dirty reads instead of causing bogus deadlocks or
        assertion failures. */
#endif
        if (lock->lock_word != X_LOCK_DECR) {

                if (rw_lock_get_waiters(lock)) {
                        fputs(" Waiters for the lock exist\n", stderr);
                } else {
                        putc('\n', stderr);
                }

                rw_lock_debug_mutex_enter();
                info = UT_LIST_GET_FIRST(lock->debug_list);
                while (info != NULL) {
                        rw_lock_debug_print(stderr, info);
                        info = UT_LIST_GET_NEXT(list, info);
                }
                rw_lock_debug_mutex_exit();
        }
}

/*********************************************************************/
UNIV_INTERN
void
rw_lock_debug_print(
/*================*/
        FILE*                   f,      
        rw_lock_debug_t*        info)   
{
        ulint   rwt;

        rwt       = info->lock_type;

        fprintf(f, "Locked: thread %lu file %s line %lu  ",
                (ulong) os_thread_pf(info->thread_id), info->file_name,
                (ulong) info->line);
        if (rwt == RW_LOCK_SHARED) {
                fputs("S-LOCK", f);
        } else if (rwt == RW_LOCK_EX) {
                fputs("X-LOCK", f);
        } else if (rwt == RW_LOCK_WAIT_EX) {
                fputs("WAIT X-LOCK", f);
        } else {
                ut_error;
        }
        if (info->pass != 0) {
                fprintf(f, " pass value %lu", (ulong) info->pass);
        }
        putc('\n', f);
}

/***************************************************************/
UNIV_INTERN
ulint
rw_lock_n_locked(void)
/*==================*/
{
        rw_lock_t*      lock;
        ulint           count           = 0;

        mutex_enter(&rw_lock_list_mutex);

        lock = UT_LIST_GET_FIRST(rw_lock_list);

        while (lock != NULL) {

                if (lock->lock_word != X_LOCK_DECR) {
                        count++;
                }

                lock = UT_LIST_GET_NEXT(list, lock);
        }

        mutex_exit(&rw_lock_list_mutex);

        return(count);
}
#endif /* UNIV_SYNC_DEBUG */