/***************************************************************************** Copyright (c) 1995, 2009, Innobase Oy. 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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA *****************************************************************************/ /****************************************************** The read-write lock (for thread synchronization) Created 9/11/1995 Heikki Tuuri *******************************************************/ #include "sync0rw.h" #ifdef UNIV_NONINL #include "sync0rw.ic" #endif #include "os0thread.h" #include "mem0mem.h" #include "srv0srv.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 once for each lock, so the number of locks is: ((-lock_word) / X_LOCK_DECR) + 1 When lock_word <= -X_LOCK_DECR, we also know that lock_word % X_LOCK_DECR == 0: other values of lock_word are invalid. 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) */ /* number of spin waits on rw-latches, resulted during shared (read) locks */ UNIV_INTERN ib_int64_t rw_s_spin_wait_count = 0; UNIV_INTERN ib_int64_t rw_s_spin_round_count = 0; /* number of OS waits on rw-latches, resulted during shared (read) locks */ UNIV_INTERN ib_int64_t rw_s_os_wait_count = 0; /* number of unlocks (that unlock shared locks), set only when UNIV_SYNC_PERF_STAT is defined */ UNIV_INTERN ib_int64_t rw_s_exit_count = 0; /* number of spin waits on rw-latches, resulted during exclusive (write) locks */ UNIV_INTERN ib_int64_t rw_x_spin_wait_count = 0; UNIV_INTERN ib_int64_t rw_x_spin_round_count = 0; /* number of OS waits on rw-latches, resulted during exclusive (write) locks */ UNIV_INTERN ib_int64_t rw_x_os_wait_count = 0; /* number of unlocks (that unlock exclusive locks), set only when UNIV_SYNC_PERF_STAT is defined */ UNIV_INTERN ib_int64_t rw_x_exit_count = 0; /* The global list of rw-locks */ UNIV_INTERN rw_lock_list_t rw_lock_list; UNIV_INTERN mutex_t rw_lock_list_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 mutex_t rw_lock_debug_mutex; /* 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; /********************************************************************** Creates a debug info struct. */ static rw_lock_debug_t* rw_lock_debug_create(void); /*======================*/ /********************************************************************** Frees a debug info struct. */ static void rw_lock_debug_free( /*===============*/ rw_lock_debug_t* info); /********************************************************************** Creates a debug info struct. */ static rw_lock_debug_t* rw_lock_debug_create(void) /*======================*/ { return((rw_lock_debug_t*) mem_alloc(sizeof(rw_lock_debug_t))); } /********************************************************************** Frees a debug info struct. */ static void rw_lock_debug_free( /*===============*/ rw_lock_debug_t* info) { mem_free(info); } #endif /* UNIV_SYNC_DEBUG */ /********************************************************************** Creates, or rather, initializes an rw-lock object in a specified memory location (which must be appropriately aligned). The rw-lock is initialized to the non-locked state. Explicit freeing of the rw-lock with rw_lock_free is necessary only if the memory block containing it is freed. */ UNIV_INTERN void rw_lock_create_func( /*================*/ rw_lock_t* lock, /* in: pointer to memory */ #ifdef UNIV_DEBUG # ifdef UNIV_SYNC_DEBUG ulint level, /* in: level */ # endif /* UNIV_SYNC_DEBUG */ const char* cmutex_name, /* in: mutex name */ #endif /* UNIV_DEBUG */ const char* cfile_name, /* in: file name where created */ ulint cline) /* in: file line where created */ { /* 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_get_mutex(lock), SYNC_NO_ORDER_CHECK); lock->mutex.cfile_name = cfile_name; lock->mutex.cline = cline; # if defined UNIV_DEBUG && !defined UNIV_HOTBACKUP lock->mutex.cmutex_name = cmutex_name; lock->mutex.mutex_type = 1; # endif /* UNIV_DEBUG && !UNIV_HOTBACKUP */ #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; #ifdef INNODB_RW_LOCKS_USE_ATOMICS lock->s_waiters = 0; lock->x_waiters = 0; lock->wait_ex_waiters = 0; lock->writer = RW_LOCK_NOT_LOCKED; lock->writer_count = 0; lock->reader_count = 0; lock->writer_is_wait_ex = FALSE; #else lock->waiters = 0; #endif /* 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; #ifdef UNIV_SYNC_DEBUG UT_LIST_INIT(lock->debug_list); lock->level = level; #endif /* UNIV_SYNC_DEBUG */ 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; #ifdef INNODB_RW_LOCKS_USE_ATOMICS lock->s_event = os_event_create(NULL); lock->x_event = os_event_create(NULL); #else lock->event = os_event_create(NULL); #endif lock->wait_ex_event = os_event_create(NULL); mutex_enter(&rw_lock_list_mutex); if (UT_LIST_GET_LEN(rw_lock_list) > 0) { ut_a(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); } /********************************************************************** Calling this function is obligatory only if the memory buffer containing the rw-lock is freed. Removes an rw-lock object from the global list. The rw-lock is checked to be in the non-locked state. */ UNIV_INTERN void rw_lock_free( /*=========*/ rw_lock_t* lock) /* in: rw-lock */ { ut_ad(rw_lock_validate(lock)); #ifdef INNODB_RW_LOCKS_USE_ATOMICS ut_a(rw_lock_get_writer(lock) == RW_LOCK_NOT_LOCKED); ut_a(rw_lock_get_s_waiters(lock) == 0); ut_a(rw_lock_get_x_waiters(lock) == 0); ut_a(rw_lock_get_wx_waiters(lock) == 0); ut_a(rw_lock_get_reader_count(lock) == 0); #else ut_a(lock->lock_word == X_LOCK_DECR); #endif lock->magic_n = 0; #ifndef INNODB_RW_LOCKS_USE_ATOMICS mutex_free(rw_lock_get_mutex(lock)); #endif /* INNODB_RW_LOCKS_USE_ATOMICS */ mutex_enter(&rw_lock_list_mutex); #ifdef INNODB_RW_LOCKS_USE_ATOMICS os_event_free(lock->s_event); os_event_free(lock->x_event); #else os_event_free(lock->event); #endif os_event_free(lock->wait_ex_event); if (UT_LIST_GET_PREV(list, lock)) { ut_a(UT_LIST_GET_PREV(list, lock)->magic_n == RW_LOCK_MAGIC_N); } if (UT_LIST_GET_NEXT(list, lock)) { ut_a(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); } #ifdef UNIV_DEBUG /********************************************************************** Checks that the rw-lock has been initialized and that there are no simultaneous shared and exclusive locks. */ UNIV_INTERN ibool rw_lock_validate( /*=============*/ rw_lock_t* lock) { ut_a(lock); #ifdef INNODB_RW_LOCKS_USE_ATOMICS ut_a(lock->magic_n == RW_LOCK_MAGIC_N); ulint waiters = rw_lock_get_s_waiters(lock); ut_a(waiters == 0 || waiters == 1); waiters = rw_lock_get_x_waiters(lock); ut_a(waiters == 0 || waiters == 1); waiters = rw_lock_get_wx_waiters(lock); ut_a(waiters == 0 || waiters == 1); #else ulint waiters = rw_lock_get_waiters(lock); lint lock_word = lock->lock_word; ut_a(lock->magic_n == RW_LOCK_MAGIC_N); ut_a(waiters == 0 || waiters == 1); ut_a(lock_word > -X_LOCK_DECR ||(-lock_word) % X_LOCK_DECR == 0); #endif return(TRUE); } #endif /* UNIV_DEBUG */ /********************************************************************** Lock an rw-lock in shared mode for the current thread. If the rw-lock is locked in exclusive mode, or there is an exclusive lock request waiting, the function spins a preset time (controlled by SYNC_SPIN_ROUNDS), waiting for the lock, before suspending the thread. */ UNIV_INTERN void rw_lock_s_lock_spin( /*================*/ rw_lock_t* lock, /* in: pointer to rw-lock */ ulint pass, /* in: pass value; != 0, if the lock will be passed to another thread to unlock */ const char* file_name, /* in: file name where lock requested */ ulint line) /* in: line where requested */ { ulint index; /* index of the reserved wait cell */ ulint i = 0; /* spin round count */ ut_ad(rw_lock_validate(lock)); rw_s_spin_wait_count++; /* Count calls to this function */ lock_loop: /* Spin waiting for the writer field to become free */ #ifdef INNODB_RW_LOCKS_USE_ATOMICS while (i < SYNC_SPIN_ROUNDS && rw_lock_get_writer(lock) != RW_LOCK_NOT_LOCKED) { #else while (i < SYNC_SPIN_ROUNDS && lock->lock_word <= 0) { #endif if (srv_spin_wait_delay) { ut_delay(ut_rnd_interval(0, srv_spin_wait_delay)); } i++; } if (i == SYNC_SPIN_ROUNDS) { os_thread_yield(); } if (srv_print_latch_waits) { fprintf(stderr, "Thread %lu spin wait rw-s-lock at %p" " cfile %s cline %lu rnds %lu\n", (ulong) os_thread_pf(os_thread_get_curr_id()), (void*) lock, lock->cfile_name, (ulong) lock->cline, (ulong) i); } /* We try once again to obtain the lock */ if (TRUE == rw_lock_s_lock_low(lock, pass, file_name, line)) { rw_s_spin_round_count += i; return; /* Success */ } else { if (i < SYNC_SPIN_ROUNDS) { goto lock_loop; } rw_s_spin_round_count += i; sync_array_reserve_cell(sync_primary_wait_array, 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. */ #ifdef INNODB_RW_LOCKS_USE_ATOMICS rw_lock_set_s_waiter_flag(lock); #else rw_lock_set_waiter_flag(lock); #endif #ifdef INNODB_RW_LOCKS_USE_ATOMICS for (i = 0; i < 4; i++) { #endif if (TRUE == rw_lock_s_lock_low(lock, pass, file_name, line)) { sync_array_free_cell(sync_primary_wait_array, index); return; /* Success */ } #ifdef INNODB_RW_LOCKS_USE_ATOMICS } /* If wait_ex_waiter stalls, wakes it. */ if (lock->reader_count == 0 && os_compare_and_swap(&lock->wait_ex_waiters, 1, 0)) { os_event_set(lock->wait_ex_event); sync_array_object_signalled(sync_primary_wait_array); } #endif if (srv_print_latch_waits) { fprintf(stderr, "Thread %lu OS wait rw-s-lock at %p" " cfile %s cline %lu\n", os_thread_pf(os_thread_get_curr_id()), (void*) lock, lock->cfile_name, (ulong) lock->cline); } /* these stats may not be accurate */ lock->count_os_wait++; rw_s_os_wait_count++; sync_array_wait_event(sync_primary_wait_array, index); i = 0; goto lock_loop; } } /********************************************************************** This function is used in the insert buffer to move the ownership of an x-latch on a buffer frame to the current thread. The x-latch was set by the buffer read operation and it protected the buffer frame while the read was done. The ownership is moved because we want that the current thread is able to acquire a second x-latch which is stored in an mtr. This, in turn, is needed to pass the debug checks of index page operations. */ UNIV_INTERN void rw_lock_x_lock_move_ownership( /*==========================*/ rw_lock_t* lock) /* in: lock which was x-locked in the buffer read */ { ut_ad(rw_lock_is_locked(lock, RW_LOCK_EX)); #ifdef INNODB_RW_LOCKS_USE_ATOMICS lock->writer_thread = os_thread_get_curr_id(); lock->recursive = TRUE; #else rw_lock_set_writer_id_and_recursion_flag(lock, TRUE); #endif } /********************************************************************** Function for the next writer to call. Waits for readers to exit. The caller must have already decremented lock_word by X_LOCK_DECR.*/ UNIV_INLINE void rw_lock_x_lock_wait( /*================*/ rw_lock_t* lock, /* in: pointer to rw-lock */ #ifdef UNIV_SYNC_DEBUG ulint pass, /* in: pass value; != 0, if the lock will be passed to another thread to unlock */ #endif const char* file_name,/* in: file name where lock requested */ ulint line) /* in: line where requested */ { ulint index; ulint i = 0; 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_x_spin_round_count += i; i = 0; sync_array_reserve_cell(sync_primary_wait_array, lock, RW_LOCK_WAIT_EX, file_name, line, &index); /* 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_x_os_wait_count++; /* 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_primary_wait_array, 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_primary_wait_array, index); } } rw_x_spin_round_count += i; } /********************************************************************** Low-level function for acquiring an exclusive lock. */ UNIV_INLINE #ifdef INNODB_RW_LOCKS_USE_ATOMICS ulint #else ibool #endif rw_lock_x_lock_low( /*===============*/ /* out: RW_LOCK_NOT_LOCKED if did not succeed, RW_LOCK_EX if success. */ rw_lock_t* lock, /* in: pointer to rw-lock */ ulint pass, /* in: pass value; != 0, if the lock will be passed to another thread to unlock */ const char* file_name,/* in: file name where lock requested */ ulint line) /* in: line where requested */ { os_thread_id_t curr_thread = os_thread_get_curr_id(); #ifdef INNODB_RW_LOCKS_USE_ATOMICS retry_writer: /* try to lock writer */ if(__sync_lock_test_and_set(&(lock->writer),RW_LOCK_EX) == RW_LOCK_NOT_LOCKED) { /* success */ /* obtain RW_LOCK_WAIT_EX right */ lock->writer_thread = curr_thread; lock->recursive = pass ? FALSE : TRUE; lock->writer_is_wait_ex = TRUE; /* atomic operation may be safer about memory order. */ __sync_synchronize(); #ifdef UNIV_SYNC_DEBUG rw_lock_add_debug_info(lock, pass, RW_LOCK_WAIT_EX, file_name, line); #endif } if (!os_thread_eq(lock->writer_thread, curr_thread)) { return(RW_LOCK_NOT_LOCKED); } switch(rw_lock_get_writer(lock)) { case RW_LOCK_WAIT_EX: /* have right to try x-lock */ retry_x_lock: /* try x-lock */ if(__sync_sub_and_fetch(&(lock->lock_word), X_LOCK_DECR) == 0) { /* success */ lock->recursive = pass ? FALSE : TRUE; lock->writer_is_wait_ex = FALSE; __sync_fetch_and_add(&(lock->writer_count),1); #ifdef UNIV_SYNC_DEBUG rw_lock_remove_debug_info(lock, pass, RW_LOCK_WAIT_EX); 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 = line; /* Locking succeeded, we may return */ return(RW_LOCK_EX); } else if(__sync_fetch_and_add(&(lock->lock_word), X_LOCK_DECR) == 0) { /* retry x-lock */ goto retry_x_lock; } /* There are readers, we have to wait */ return(RW_LOCK_WAIT_EX); break; case RW_LOCK_EX: /* already have x-lock */ if (lock->recursive && (pass == 0)) { __sync_fetch_and_add(&(lock->writer_count),1); #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 = line; /* Locking succeeded, we may return */ return(RW_LOCK_EX); } return(RW_LOCK_NOT_LOCKED); break; default: /* RW_LOCK_NOT_LOCKED? maybe impossible */ goto retry_writer; } /* Locking did not succeed */ return(RW_LOCK_NOT_LOCKED); #else 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 { /* Decrement failed: relock or failed lock */ if (!pass && lock->recursive && os_thread_eq(lock->writer_thread, curr_thread)) { /* Relock */ lock->lock_word -= X_LOCK_DECR; } 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); #endif } /********************************************************************** NOTE! Use the corresponding macro, not directly this function! Lock an rw-lock in exclusive mode for the current thread. If the rw-lock is locked in shared or exclusive mode, or there is an exclusive lock request waiting, the function spins a preset time (controlled by SYNC_SPIN_ROUNDS), waiting for the lock before suspending the thread. If the same thread has an x-lock on the rw-lock, locking succeed, with the following exception: if pass != 0, only a single x-lock may be taken on the lock. NOTE: If the same thread has an s-lock, locking does not succeed! */ UNIV_INTERN void rw_lock_x_lock_func( /*================*/ rw_lock_t* lock, /* in: pointer to rw-lock */ ulint pass, /* in: pass value; != 0, if the lock will be passed to another thread to unlock */ const char* file_name,/* in: file name where lock requested */ ulint line) /* in: line where requested */ { ulint index; /* index of the reserved wait cell */ ulint i; /* spin round count */ ibool spinning = FALSE; #ifdef INNODB_RW_LOCKS_USE_ATOMICS ulint state = RW_LOCK_NOT_LOCKED; /* lock state acquired */ ulint prev_state = RW_LOCK_NOT_LOCKED; #endif ut_ad(rw_lock_validate(lock)); i = 0; lock_loop: #ifdef INNODB_RW_LOCKS_USE_ATOMICS prev_state = state; state = rw_lock_x_lock_low(lock, pass, file_name, line); lock_loop_2: if (state != prev_state) i=0; /* if progress, reset counter. */ if (state == RW_LOCK_EX) { #else if (rw_lock_x_lock_low(lock, pass, file_name, line)) { #endif rw_x_spin_round_count += i; return; /* Locking succeeded */ #ifdef INNODB_RW_LOCKS_USE_ATOMICS } else if (state == RW_LOCK_WAIT_EX) { if (!spinning) { spinning = TRUE; rw_x_spin_wait_count++; } /* Spin waiting for the reader count field to become zero */ while (i < SYNC_SPIN_ROUNDS && lock->lock_word != X_LOCK_DECR) { 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; } #endif } else { if (!spinning) { spinning = TRUE; rw_x_spin_wait_count++; } /* Spin waiting for the lock_word to become free */ while (i < SYNC_SPIN_ROUNDS #ifdef INNODB_RW_LOCKS_USE_ATOMICS && rw_lock_get_writer(lock) != RW_LOCK_NOT_LOCKED) { #else && lock->lock_word <= 0) { #endif 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_x_spin_round_count += i; if (srv_print_latch_waits) { fprintf(stderr, "Thread %lu spin wait rw-x-lock at %p" " cfile %s cline %lu rnds %lu\n", os_thread_pf(os_thread_get_curr_id()), (void*) lock, lock->cfile_name, (ulong) lock->cline, (ulong) i); } sync_array_reserve_cell(sync_primary_wait_array, lock, #ifdef INNODB_RW_LOCKS_USE_ATOMICS (state == RW_LOCK_WAIT_EX) ? RW_LOCK_WAIT_EX : RW_LOCK_EX, #else RW_LOCK_EX, #endif 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. */ #ifdef INNODB_RW_LOCKS_USE_ATOMICS if (state == RW_LOCK_WAIT_EX) { rw_lock_set_wx_waiter_flag(lock); } else { rw_lock_set_x_waiter_flag(lock); } #else rw_lock_set_waiter_flag(lock); #endif #ifdef INNODB_RW_LOCKS_USE_ATOMICS for (i = 0; i < 4; i++) { prev_state = state; state = rw_lock_x_lock_low(lock, pass, file_name, line); if (state == RW_LOCK_EX) { sync_array_free_cell(sync_primary_wait_array, index); return; /* Locking succeeded */ } else if (state != prev_state) { /* retry! */ sync_array_free_cell(sync_primary_wait_array, index); goto lock_loop_2; } } #else if (rw_lock_x_lock_low(lock, pass, file_name, line)) { sync_array_free_cell(sync_primary_wait_array, index); return; /* Locking succeeded */ } #endif if (srv_print_latch_waits) { fprintf(stderr, "Thread %lu OS wait for rw-x-lock at %p" " cfile %s cline %lu\n", os_thread_pf(os_thread_get_curr_id()), (void*) lock, lock->cfile_name, (ulong) lock->cline); } /* these stats may not be accurate */ lock->count_os_wait++; rw_x_os_wait_count++; sync_array_wait_event(sync_primary_wait_array, index); i = 0; goto lock_loop; } #ifdef UNIV_SYNC_DEBUG /********************************************************************** Acquires the debug mutex. We cannot use the mutex defined in sync0sync, because the debug mutex is also acquired in sync0arr while holding the OS mutex protecting the sync array, and the ordinary mutex_enter might recursively call routines in sync0arr, leading to a deadlock on the OS mutex. */ 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; } /********************************************************************** Releases the debug mutex. */ 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); } } /********************************************************************** Inserts the debug information for an rw-lock. */ UNIV_INTERN void rw_lock_add_debug_info( /*===================*/ rw_lock_t* lock, /* in: rw-lock */ ulint pass, /* in: pass value */ ulint lock_type, /* in: lock type */ const char* file_name, /* in: file where requested */ ulint line) /* in: line where requested */ { 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); } } /********************************************************************** Removes a debug information struct for an rw-lock. */ UNIV_INTERN void rw_lock_remove_debug_info( /*======================*/ rw_lock_t* lock, /* in: rw-lock */ ulint pass, /* in: pass value */ ulint lock_type) /* in: 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 /********************************************************************** Checks if the thread has locked the rw-lock in the specified mode, with the pass value == 0. */ UNIV_INTERN ibool rw_lock_own( /*========*/ /* out: TRUE if locked */ rw_lock_t* lock, /* in: rw-lock */ ulint lock_type) /* in: lock type: RW_LOCK_SHARED, RW_LOCK_EX */ { 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 */ /********************************************************************** Checks if somebody has locked the rw-lock in the specified mode. */ UNIV_INTERN ibool rw_lock_is_locked( /*==============*/ /* out: TRUE if locked */ rw_lock_t* lock, /* in: rw-lock */ ulint lock_type) /* in: lock type: RW_LOCK_SHARED, RW_LOCK_EX */ { 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 /******************************************************************* Prints debug info of currently locked rw-locks. */ UNIV_INTERN void rw_lock_list_print_info( /*====================*/ FILE* file) /* in: file where to print */ { 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); #ifdef INNODB_RW_LOCKS_USE_ATOMICS if (rw_lock_get_s_waiters(lock)) { fputs(" s_waiters for the lock exist", file); } if (rw_lock_get_x_waiters(lock)) { fputs(" x_waiters for the lock exist", file); } if (rw_lock_get_wx_waiters(lock)) { fputs(" wait_ex_waiters for the lock exist", file); } putc('\n', file); #else if (rw_lock_get_waiters(lock)) { fputs(" Waiters for the lock exist\n", file); } else { putc('\n', file); } #endif info = UT_LIST_GET_FIRST(lock->debug_list); while (info != NULL) { rw_lock_debug_print(info); info = UT_LIST_GET_NEXT(list, info); } } #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); } /******************************************************************* Prints debug info of an rw-lock. */ UNIV_INTERN void rw_lock_print( /*==========*/ rw_lock_t* lock) /* in: rw-lock */ { rw_lock_debug_t* info; fprintf(stderr, "-------------\n" "RW-LATCH INFO\n" "RW-LATCH: %p ", (void*) lock); #ifndef INNODB_RW_LOCKS_USE_ATOMICS mutex_enter(&(lock->mutex)); #endif if (lock->lock_word != X_LOCK_DECR) { #ifdef INNODB_RW_LOCKS_USE_ATOMICS if (rw_lock_get_s_waiters(lock)) { fputs(" s_waiters for the lock exist", stderr); } if (rw_lock_get_x_waiters(lock)) { fputs(" x_waiters for the lock exist", stderr); } if (rw_lock_get_wx_waiters(lock)) { fputs(" wait_ex_waiters for the lock exist", stderr); } putc('\n', stderr); #else if (rw_lock_get_waiters(lock)) { fputs(" Waiters for the lock exist\n", stderr); } else { putc('\n', stderr); } #endif info = UT_LIST_GET_FIRST(lock->debug_list); while (info != NULL) { rw_lock_debug_print(info); info = UT_LIST_GET_NEXT(list, info); } } #ifndef INNODB_RW_LOCKS_USE_ATOMICS mutex_exit(&(lock->mutex)); #endif } /************************************************************************* Prints info of a debug struct. */ UNIV_INTERN void rw_lock_debug_print( /*================*/ rw_lock_debug_t* info) /* in: debug struct */ { ulint rwt; rwt = info->lock_type; fprintf(stderr, "Locked: thread %ld file %s line %ld ", (ulong) os_thread_pf(info->thread_id), info->file_name, (ulong) info->line); if (rwt == RW_LOCK_SHARED) { fputs("S-LOCK", stderr); } else if (rwt == RW_LOCK_EX) { fputs("X-LOCK", stderr); } else if (rwt == RW_LOCK_WAIT_EX) { fputs("WAIT X-LOCK", stderr); } else { ut_error; } if (info->pass != 0) { fprintf(stderr, " pass value %lu", (ulong) info->pass); } putc('\n', stderr); } /******************************************************************* Returns the number of currently locked rw-locks. Works only in the debug version. */ 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 */