diff --recursive -u innodb_plugin-1.0.1/btr/btr0cur.c innodb_plugin-1.0.1-new/btr/btr0cur.c --- innodb_plugin-1.0.1/btr/btr0cur.c 2008-03-17 07:19:04.000000000 -0700 +++ innodb_plugin-1.0.1-new/btr/btr0cur.c 2008-06-02 16:53:35.876937000 -0700 @@ -371,7 +371,7 @@ #ifdef UNIV_SEARCH_PERF_STAT info->n_searches++; #endif - if (btr_search_latch.writer == RW_LOCK_NOT_LOCKED + if (rw_lock_get_writer(&btr_search_latch) == RW_LOCK_NOT_LOCKED && latch_mode <= BTR_MODIFY_LEAF && info->last_hash_succ && !estimate #ifdef PAGE_CUR_LE_OR_EXTENDS diff --recursive -u innodb_plugin-1.0.1/btr/btr0sea.c innodb_plugin-1.0.1-new/btr/btr0sea.c --- innodb_plugin-1.0.1/btr/btr0sea.c 2008-02-29 13:45:22.000000000 -0800 +++ innodb_plugin-1.0.1-new/btr/btr0sea.c 2008-06-02 16:53:35.881937000 -0700 @@ -773,8 +773,8 @@ rw_lock_s_lock(&btr_search_latch); } - ut_ad(btr_search_latch.writer != RW_LOCK_EX); - ut_ad(btr_search_latch.reader_count > 0); + ut_ad(rw_lock_get_writer(&btr_search_latch) != RW_LOCK_EX); + ut_ad(rw_lock_get_reader_count(&btr_search_latch) > 0); rec = ha_search_and_get_data(btr_search_sys->hash_index, fold); diff --recursive -u innodb_plugin-1.0.1/buf/buf0buf.c innodb_plugin-1.0.1-new/buf/buf0buf.c --- innodb_plugin-1.0.1/buf/buf0buf.c 2008-04-12 04:04:28.000000000 -0700 +++ innodb_plugin-1.0.1-new/buf/buf0buf.c 2008-09-10 12:01:11.400393000 -0700 @@ -2060,8 +2060,8 @@ ibool success; if (rw_latch == RW_S_LATCH) { - success = rw_lock_s_lock_func_nowait(&(block->lock), - file, line); + success = rw_lock_s_lock_nowait(&(block->lock), + file, line); fix_type = MTR_MEMO_PAGE_S_FIX; } else { ut_ad(rw_latch == RW_X_LATCH); @@ -2175,8 +2175,8 @@ buf_block_get_page_no(block))); if (rw_latch == RW_S_LATCH) { - success = rw_lock_s_lock_func_nowait(&(block->lock), - file, line); + success = rw_lock_s_lock_nowait(&(block->lock), + file, line); fix_type = MTR_MEMO_PAGE_S_FIX; } else { success = rw_lock_x_lock_func_nowait(&(block->lock), @@ -2288,8 +2288,8 @@ ut_ad(!ibuf_inside() || (mode == BUF_KEEP_OLD)); if (rw_latch == RW_S_LATCH) { - success = rw_lock_s_lock_func_nowait(&(block->lock), - file, line); + success = rw_lock_s_lock_nowait(&(block->lock), + file, line); fix_type = MTR_MEMO_PAGE_S_FIX; } else { success = rw_lock_x_lock_func_nowait(&(block->lock), @@ -2366,7 +2366,7 @@ mutex_exit(&block->mutex); fix_type = MTR_MEMO_PAGE_S_FIX; - success = rw_lock_s_lock_func_nowait(&block->lock, file, line); + success = rw_lock_s_lock_nowait(&block->lock, file, line); if (!success) { /* Let us try to get an X-latch. If the current thread diff --recursive -u innodb_plugin-1.0.1/handler/ha_innodb.cc innodb_plugin-1.0.1-new/handler/ha_innodb.cc --- innodb_plugin-1.0.1/handler/ha_innodb.cc 2008-05-06 16:29:53.000000000 -0700 +++ innodb_plugin-1.0.1-new/handler/ha_innodb.cc 2008-09-10 12:29:41.438699000 -0700 @@ -506,6 +506,10 @@ (char*) &export_vars.innodb_dblwr_pages_written, SHOW_LONG}, {"dblwr_writes", (char*) &export_vars.innodb_dblwr_writes, SHOW_LONG}, + {"have_atomic_builtins", + (char*) &export_vars.innodb_have_atomic_builtins, SHOW_BOOL}, + {"heap_enabled", + (char*) &export_vars.innodb_heap_enabled, SHOW_BOOL}, {"log_waits", (char*) &export_vars.innodb_log_waits, SHOW_LONG}, {"log_write_requests", @@ -7696,6 +7700,7 @@ { char buf1[IO_SIZE], buf2[IO_SIZE]; mutex_t* mutex; + rw_lock_t* lock; #ifdef UNIV_DEBUG ulint rw_lock_count= 0; ulint rw_lock_count_spin_loop= 0; @@ -7766,6 +7771,31 @@ mutex_exit(&mutex_list_mutex); + mutex_enter(&rw_lock_list_mutex); + + lock = UT_LIST_GET_FIRST(rw_lock_list); + + while (lock != NULL) + { + if (lock->count_os_wait) + { + buf1len= my_snprintf(buf1, sizeof(buf1), "%s:%lu", + lock->cfile_name, (ulong) lock->cline); + buf2len= my_snprintf(buf2, sizeof(buf2), + "os_waits=%lu", lock->count_os_wait); + + if (stat_print(thd, innobase_hton_name, + hton_name_len, buf1, buf1len, + buf2, buf2len)) { + mutex_exit(&rw_lock_list_mutex); + DBUG_RETURN(1); + } + } + lock = UT_LIST_GET_NEXT(list, lock); + } + + mutex_exit(&rw_lock_list_mutex); + #ifdef UNIV_DEBUG buf2len= my_snprintf(buf2, sizeof(buf2), "count=%lu, spin_waits=%lu, spin_rounds=%lu, " @@ -7800,6 +7830,7 @@ return(FALSE); } } + rw_lock_t* lock; /**************************************************************************** diff --recursive -u innodb_plugin-1.0.1/include/buf0buf.ic innodb_plugin-1.0.1-new/include/buf0buf.ic --- innodb_plugin-1.0.1/include/buf0buf.ic 2008-03-04 00:57:07.000000000 -0800 +++ innodb_plugin-1.0.1-new/include/buf0buf.ic 2008-09-10 12:01:34.351166000 -0700 @@ -862,7 +862,7 @@ #ifdef UNIV_SYNC_DEBUG ibool ret; - ret = rw_lock_s_lock_func_nowait(&(block->debug_latch), file, line); + ret = rw_lock_s_lock_nowait(&(block->debug_latch), file, line); ut_a(ret); #endif /* UNIV_SYNC_DEBUG */ ut_ad(mutex_own(&block->mutex)); diff --recursive -u innodb_plugin-1.0.1/include/os0sync.h innodb_plugin-1.0.1-new/include/os0sync.h --- innodb_plugin-1.0.1/include/os0sync.h 2008-03-17 07:19:04.000000000 -0700 +++ innodb_plugin-1.0.1-new/include/os0sync.h 2008-09-16 23:00:58.207359000 -0700 @@ -261,6 +261,29 @@ /*===============*/ os_fast_mutex_t* fast_mutex); /* in: mutex to free */ +#ifdef HAVE_GCC_ATOMIC_BUILTINS +/************************************************************** +Atomic compare-and-swap for InnoDB. Currently requires GCC atomic builtins. */ +UNIV_INLINE +ibool +os_compare_and_swap( +/*================*/ + /* out: true if swapped */ + volatile lint* ptr, /* in: pointer to target */ + lint oldVal, /* in: value to compare to */ + lint newVal); /* in: value to swap in */ +/************************************************************** +Atomic increment for InnoDB. Currently requires GCC atomic builtins. */ +UNIV_INLINE +lint +os_atomic_increment( +/*================*/ + /* out: resulting value */ + volatile lint* ptr, /* in: pointer to target */ + lint amount); /* in: amount of increment */ + +#endif /* HAVE_GCC_ATOMIC_BUILTINS */ + #ifndef UNIV_NONINL #include "os0sync.ic" #endif diff --recursive -u innodb_plugin-1.0.1/include/os0sync.ic innodb_plugin-1.0.1-new/include/os0sync.ic --- innodb_plugin-1.0.1/include/os0sync.ic 2008-02-29 13:45:22.000000000 -0800 +++ innodb_plugin-1.0.1-new/include/os0sync.ic 2008-09-16 23:01:09.940447000 -0700 @@ -44,3 +44,38 @@ #endif #endif } + +#ifdef HAVE_GCC_ATOMIC_BUILTINS +/************************************************************** +Atomic compare-and-swap for InnoDB. Currently requires GCC atomic builtins. */ +UNIV_INLINE +ibool +os_compare_and_swap( +/*================*/ + /* out: true if swapped */ + volatile lint* ptr, /* in: pointer to target */ + lint oldVal, /* in: value to compare to */ + lint newVal) /* in: value to swap in */ +{ + if(__sync_bool_compare_and_swap(ptr, oldVal, newVal)) { + return(TRUE); + } + + return(FALSE); +} + +/************************************************************** +Atomic increment for InnoDB. Currently requires GCC atomic builtins. */ +UNIV_INLINE +lint +os_atomic_increment( +/*================*/ + /* out: resulting value */ + volatile lint* ptr, /* in: pointer to target */ + lint amount) /* in: amount of increment */ +{ + lint newVal = __sync_add_and_fetch(ptr, amount); + return newVal; +} + +#endif /* HAVE_GCC_ATOMIC_BUILTINS */ diff --recursive -u innodb_plugin-1.0.1/include/srv0srv.h innodb_plugin-1.0.1-new/include/srv0srv.h --- innodb_plugin-1.0.1/include/srv0srv.h 2008-05-06 09:00:25.000000000 -0700 +++ innodb_plugin-1.0.1-new/include/srv0srv.h 2008-06-04 16:10:30.224021000 -0700 @@ -527,6 +527,8 @@ ulint innodb_buffer_pool_read_ahead_rnd; ulint innodb_dblwr_pages_written; ulint innodb_dblwr_writes; + ibool innodb_have_atomic_builtins; + ibool innodb_heap_enabled; ulint innodb_log_waits; ulint innodb_log_write_requests; ulint innodb_log_writes; @@ -563,4 +565,3 @@ extern ulint srv_n_threads_active[]; #endif - diff --recursive -u innodb_plugin-1.0.1/include/sync0rw.h innodb_plugin-1.0.1-new/include/sync0rw.h --- innodb_plugin-1.0.1/include/sync0rw.h 2008-02-29 13:45:22.000000000 -0800 +++ innodb_plugin-1.0.1-new/include/sync0rw.h 2008-09-16 23:02:49.824098000 -0700 @@ -24,6 +24,12 @@ #define RW_X_LATCH 2 #define RW_NO_LATCH 3 +/* We decrement lock_word by this amount for each x_lock. It is also the +start value for the lock_word, meaning that it limits the maximum number +of concurrent read locks before the rw_lock breaks. The current value of +0x00100000 allows 1,048,575 concurrent readers and 2047 recursive writers.*/ +#define X_LOCK_DECR 0x00100000 + typedef struct rw_lock_struct rw_lock_t; #ifdef UNIV_SYNC_DEBUG typedef struct rw_lock_debug_struct rw_lock_debug_t; @@ -47,14 +53,14 @@ there may be waiters for the event */ #endif /* UNIV_SYNC_DEBUG */ -extern ulint rw_s_system_call_count; -extern ulint rw_s_spin_wait_count; -extern ulint rw_s_exit_count; -extern ulint rw_s_os_wait_count; -extern ulint rw_x_system_call_count; -extern ulint rw_x_spin_wait_count; -extern ulint rw_x_os_wait_count; -extern ulint rw_x_exit_count; +extern ib_int64_t rw_s_spin_wait_count; +extern ib_int64_t rw_s_spin_round_count; +extern ib_int64_t rw_s_exit_count; +extern ib_int64_t rw_s_os_wait_count; +extern ib_int64_t rw_x_spin_wait_count; +extern ib_int64_t rw_x_spin_round_count; +extern ib_int64_t rw_x_os_wait_count; +extern ib_int64_t rw_x_exit_count; /********************************************************************** Creates, or rather, initializes an rw-lock object in a specified memory @@ -127,8 +133,8 @@ NOTE! The following macros should be used in rw s-locking, not the corresponding function. */ -#define rw_lock_s_lock_nowait(M) rw_lock_s_lock_func_nowait(\ - (M), __FILE__, __LINE__) +#define rw_lock_s_lock_nowait(M, F, L) rw_lock_s_lock_low(\ + (M), 0, (F), (L)) /********************************************************************** NOTE! Use the corresponding macro, not directly this function, except if you supply the file name and line number. Lock an rw-lock in shared mode @@ -146,18 +152,6 @@ const char* file_name,/* in: file name where lock requested */ ulint line); /* in: line where requested */ /********************************************************************** -NOTE! Use the corresponding macro, not directly this function, except if -you supply the file name and line number. Lock an rw-lock in shared mode -for the current thread if the lock can be acquired immediately. */ -UNIV_INLINE -ibool -rw_lock_s_lock_func_nowait( -/*=======================*/ - /* out: TRUE if success */ - rw_lock_t* lock, /* in: pointer to rw-lock */ - const char* file_name,/* in: file name where lock requested */ - ulint line); /* in: line where requested */ -/********************************************************************** NOTE! Use the corresponding macro, not directly this function! Lock an rw-lock in exclusive mode for the current thread if the lock can be obtained immediately. */ @@ -341,6 +335,23 @@ rw_lock_get_reader_count( /*=====================*/ rw_lock_t* lock); +/********************************************************************** +Decrements lock_word the specified amount if it is greater than 0. +This is used by both s_lock and x_lock operations. */ +UNIV_INLINE +ibool +rw_lock_lock_word_decr( + /* out: TRUE if decr occurs */ + rw_lock_t* lock, /* in: rw-lock */ + ulint amount); /* in: amount to decrement */ +/********************************************************************** +Increments lock_word the specified amount and returns new value. */ +UNIV_INLINE +lint +rw_lock_lock_word_incr( + /* out: TRUE if decr occurs */ + rw_lock_t* lock, + ulint amount); /* in: rw-lock */ #ifdef UNIV_SYNC_DEBUG /********************************************************************** Checks if the thread has locked the rw-lock in the specified mode, with @@ -417,44 +428,28 @@ implementation of a read-write lock. Several threads may have a shared lock simultaneously in this lock, but only one writer may have an exclusive lock, in which case no shared locks are allowed. To prevent starving of a writer -blocked by readers, a writer may queue for the lock by setting the writer -field. Then no new readers are allowed in. */ +blocked by readers, a writer may queue for x-lock by decrementing lock_word: +no new readers will be let in while the thread waits for readers to exit. */ struct rw_lock_struct { - os_event_t event; /* Used by sync0arr.c for thread queueing */ - -#ifdef __WIN__ - os_event_t wait_ex_event; /* This windows specific event is - used by the thread which has set the - lock state to RW_LOCK_WAIT_EX. The - rw_lock design guarantees that this - thread will be the next one to proceed - once the current the event gets - signalled. See LEMMA 2 in sync0sync.c */ -#endif - - ulint reader_count; /* Number of readers who have locked this - lock in the shared mode */ - ulint writer; /* This field is set to RW_LOCK_EX if there - is a writer owning the lock (in exclusive - mode), RW_LOCK_WAIT_EX if a writer is - queueing for the lock, and - RW_LOCK_NOT_LOCKED, otherwise. */ - os_thread_id_t writer_thread; - /* Thread id of a possible writer thread */ - ulint writer_count; /* Number of times the same thread has - recursively locked the lock in the exclusive - mode */ - mutex_t mutex; /* The mutex protecting rw_lock_struct */ - ulint pass; /* Default value 0. This is set to some + volatile lint lock_word; + /* Holds the state of the lock. */ + volatile ulint waiters;/* 1: there are waiters */ + volatile ulint pass; /* Default value 0. This is set to some value != 0 given by the caller of an x-lock operation, if the x-lock is to be passed to another thread to unlock (which happens in asynchronous i/o). */ - ulint waiters; /* This ulint is set to 1 if there are - waiters (readers or writers) in the global - wait array, waiting for this rw_lock. - Otherwise, == 0. */ + volatile os_thread_id_t writer_thread; + /* Thread id of writer thread */ + os_event_t event; /* Used by sync0arr.c for thread queueing */ + os_event_t wait_ex_event; + /* Event for next-writer to wait on. A thread + must decrement lock_word before waiting. */ +#ifndef HAVE_GCC_ATOMIC_BUILTINS + mutex_t mutex; /* The mutex protecting rw_lock_struct */ +#endif /* HAVE_GCC_ATOMIC_BUILTINS */ + UT_LIST_NODE_T(rw_lock_t) list; /* All allocated rw locks are put into a list */ @@ -464,7 +459,9 @@ info list of the lock */ ulint level; /* Level in the global latching order. */ #endif /* UNIV_SYNC_DEBUG */ + ulint count_os_wait; /* Count of os_waits. May not be accurate */ const char* cfile_name;/* File name where lock created */ + /* last s-lock file/line is not guaranteed to be correct */ const char* last_s_file_name;/* File name where last s-locked */ const char* last_x_file_name;/* File name where last x-locked */ ibool writer_is_wait_ex; diff --recursive -u innodb_plugin-1.0.1/include/sync0rw.ic innodb_plugin-1.0.1-new/include/sync0rw.ic --- innodb_plugin-1.0.1/include/sync0rw.ic 2008-02-29 13:45:22.000000000 -0800 +++ innodb_plugin-1.0.1-new/include/sync0rw.ic 2008-09-16 23:02:39.399172000 -0700 @@ -62,40 +62,48 @@ { lock->waiters = flag; } + +/********************************************************************** +Returns the write-status of the lock - this function made more sense +with the old rw_lock implementation. + */ UNIV_INLINE ulint rw_lock_get_writer( /*===============*/ rw_lock_t* lock) { - return(lock->writer); -} -UNIV_INLINE -void -rw_lock_set_writer( -/*===============*/ - rw_lock_t* lock, - ulint flag) -{ - lock->writer = flag; + lint lock_word = lock->lock_word; + if(lock_word > 0) { + /* return NOT_LOCKED in s-lock state, like the writer + member of the old lock implementation. */ + return RW_LOCK_NOT_LOCKED; + } else if (((-lock_word) % X_LOCK_DECR) == 0) { + return RW_LOCK_EX; + } else { + ut_ad(lock_word > -X_LOCK_DECR); + return RW_LOCK_WAIT_EX; + } } + UNIV_INLINE ulint rw_lock_get_reader_count( /*=====================*/ rw_lock_t* lock) { - return(lock->reader_count); -} -UNIV_INLINE -void -rw_lock_set_reader_count( -/*=====================*/ - rw_lock_t* lock, - ulint count) -{ - lock->reader_count = count; + lint lock_word = lock->lock_word; + if(lock_word > 0) { + /* s-locked, no x-waiters */ + return(X_LOCK_DECR - lock_word); + } else if (lock_word < 0 && lock_word > -X_LOCK_DECR) { + /* s-locked, with x-waiters */ + return (ulint)(-lock_word); + } + return 0; } + +#ifndef HAVE_GCC_ATOMIC_BUILTINS UNIV_INLINE mutex_t* rw_lock_get_mutex( @@ -104,6 +112,7 @@ { return(&(lock->mutex)); } +#endif /********************************************************************** Returns the value of writer_count for the lock. Does not reserve the lock @@ -115,7 +124,87 @@ /* out: value of writer_count */ rw_lock_t* lock) /* in: rw-lock */ { - return(lock->writer_count); + lint lock_copy = lock->lock_word; + /* If there is a reader, lock_word is not divisible by X_LOCK_DECR */ + if(lock_copy > 0 || (-lock_copy) % X_LOCK_DECR != 0) { + return 0; + } + return ((-lock_copy) / X_LOCK_DECR) + 1; +} + +/********************************************************************** +Two different implementations for decrementing the lock_word of a rw_lock: +one for systems supporting atomic operations, one for others. This does +does not support recusive x-locks: they should be handled by the caller and +need not be atomic since they are performed by the current lock holder. +Returns true if the decrement was made, false if not. */ +UNIV_INLINE +ibool +rw_lock_lock_word_decr( + /* out: TRUE if decr occurs */ + rw_lock_t* lock, /* in: rw-lock */ + ulint amount) /* in: amount of decrement */ +{ + +#ifdef HAVE_GCC_ATOMIC_BUILTINS + + lint local_lock_word = lock->lock_word; + while (local_lock_word > 0) { + if(os_compare_and_swap(&(lock->lock_word), + local_lock_word, + local_lock_word - amount)) { + return TRUE; + } + local_lock_word = lock->lock_word; + } + return(FALSE); + +#else /* HAVE_GCC_ATOMIC_BUILTINS */ + + ibool success = FALSE; + mutex_enter(&(lock->mutex)); + if(lock->lock_word > 0) { + lock->lock_word -= amount; + success = TRUE; + } + mutex_exit(&(lock->mutex)); + return success; + +#endif /* HAVE_GCC_ATOMIC_BUILTINS */ + +} + +/********************************************************************** +Two different implementations for incrementing the lock_word of a rw_lock: +one for systems supporting atomic operations, one for others. +Returns the value of lock_word after increment. */ +UNIV_INLINE +lint +rw_lock_lock_word_incr( + /* out: lock->lock_word after increment */ + rw_lock_t* lock, /* in: rw-lock */ + ulint amount) /* in: amount of increment */ +{ + +#ifdef HAVE_GCC_ATOMIC_BUILTINS + + return(os_atomic_increment(&(lock->lock_word), amount)); + +#else /* HAVE_GCC_ATOMIC_BUILTINS */ + + lint local_lock_word; + + mutex_enter(&(lock->mutex)); + + lock->lock_word += amount; + local_lock_word = lock->lock_word; + + mutex_exit(&(lock->mutex)); + + return local_lock_word; + +#endif /* HAVE_GCC_ATOMIC_BUILTINS */ + } /********************************************************************** @@ -133,25 +222,21 @@ const char* file_name, /* in: file name where lock requested */ ulint line) /* in: line where requested */ { - ut_ad(mutex_own(rw_lock_get_mutex(lock))); - - /* Check if the writer field is free */ - - if (UNIV_LIKELY(lock->writer == RW_LOCK_NOT_LOCKED)) { - /* Set the shared lock by incrementing the reader count */ - lock->reader_count++; + /* TODO: study performance of UNIV_LIKELY branch prediction hints. */ + if (!rw_lock_lock_word_decr(lock, 1)) { + /* Locking did not succeed */ + return(FALSE); + } #ifdef UNIV_SYNC_DEBUG - rw_lock_add_debug_info(lock, pass, RW_LOCK_SHARED, file_name, - line); + rw_lock_add_debug_info(lock, pass, RW_LOCK_SHARED, file_name, line); #endif - lock->last_s_file_name = file_name; - lock->last_s_line = line; - - return(TRUE); /* locking succeeded */ - } + /* These debugging values are not set safely: they may be incorrect + or even refer to a line that is invalid for the file name. */ + lock->last_s_file_name = file_name; + lock->last_s_line = line; - return(FALSE); /* locking did not succeed */ + return(TRUE); /* locking succeeded */ } /********************************************************************** @@ -166,11 +251,10 @@ const char* file_name, /* in: file name where requested */ ulint line) /* in: line where lock requested */ { - ut_ad(lock->writer == RW_LOCK_NOT_LOCKED); - ut_ad(rw_lock_get_reader_count(lock) == 0); + ut_ad(lock->lock_word == X_LOCK_DECR); - /* Set the shared lock by incrementing the reader count */ - lock->reader_count++; + /* Indicate there is a new reader by decrementing lock_word */ + lock->lock_word--; lock->last_s_file_name = file_name; lock->last_s_line = line; @@ -193,12 +277,10 @@ ulint line) /* in: line where lock requested */ { ut_ad(rw_lock_validate(lock)); - ut_ad(rw_lock_get_reader_count(lock) == 0); - ut_ad(rw_lock_get_writer(lock) == RW_LOCK_NOT_LOCKED); + ut_ad(lock->lock_word == X_LOCK_DECR); - rw_lock_set_writer(lock, RW_LOCK_EX); + lock->lock_word -= X_LOCK_DECR; lock->writer_thread = os_thread_get_curr_id(); - lock->writer_count++; lock->pass = 0; lock->last_x_file_name = file_name; @@ -240,15 +322,12 @@ ut_ad(!rw_lock_own(lock, RW_LOCK_SHARED)); /* see NOTE above */ #endif /* UNIV_SYNC_DEBUG */ - mutex_enter(rw_lock_get_mutex(lock)); - - if (UNIV_LIKELY(rw_lock_s_lock_low(lock, pass, file_name, line))) { - mutex_exit(rw_lock_get_mutex(lock)); + /* TODO: study performance of UNIV_LIKELY branch prediction hints. */ + if (rw_lock_s_lock_low(lock, pass, file_name, line)) { return; /* Success */ } else { /* Did not succeed, try spin wait */ - mutex_exit(rw_lock_get_mutex(lock)); rw_lock_s_lock_spin(lock, pass, file_name, line); @@ -258,86 +337,66 @@ /********************************************************************** NOTE! Use the corresponding macro, not directly this function! Lock an -rw-lock in shared mode for the current thread if the lock can be acquired -immediately. */ +rw-lock in exclusive mode for the current thread if the lock can be +obtained immediately. */ UNIV_INLINE ibool -rw_lock_s_lock_func_nowait( +rw_lock_x_lock_func_nowait( /*=======================*/ /* out: TRUE if success */ rw_lock_t* lock, /* in: pointer to rw-lock */ const char* file_name,/* in: file name where lock requested */ ulint line) /* in: line where requested */ { - ibool success = FALSE; - - mutex_enter(rw_lock_get_mutex(lock)); - - if (lock->writer == RW_LOCK_NOT_LOCKED) { - /* Set the shared lock by incrementing the reader count */ - lock->reader_count++; + os_thread_id_t curr_thread = os_thread_get_curr_id(); -#ifdef UNIV_SYNC_DEBUG - rw_lock_add_debug_info(lock, 0, RW_LOCK_SHARED, file_name, - line); -#endif + ibool success; - lock->last_s_file_name = file_name; - lock->last_s_line = line; +#ifdef HAVE_GCC_ATOMIC_BUILTINS + success = os_compare_and_swap(&(lock->lock_word), X_LOCK_DECR, 0); +#else + success = FALSE; + mutex_enter(&(lock->mutex)); + if(lock->lock_word == X_LOCK_DECR) { + lock->lock_word = 0; success = TRUE; } + mutex_exit(&(lock->mutex)); - mutex_exit(rw_lock_get_mutex(lock)); - - return(success); -} - -/********************************************************************** -NOTE! Use the corresponding macro, not directly this function! Lock an -rw-lock in exclusive mode for the current thread if the lock can be -obtained immediately. */ -UNIV_INLINE -ibool -rw_lock_x_lock_func_nowait( -/*=======================*/ - /* out: TRUE if success */ - rw_lock_t* lock, /* in: pointer to rw-lock */ - const char* file_name,/* in: file name where lock requested */ - ulint line) /* in: line where requested */ -{ - ibool success = FALSE; - os_thread_id_t curr_thread = os_thread_get_curr_id(); - mutex_enter(rw_lock_get_mutex(lock)); - - if (UNIV_UNLIKELY(rw_lock_get_reader_count(lock) != 0)) { - } else if (UNIV_LIKELY(rw_lock_get_writer(lock) - == RW_LOCK_NOT_LOCKED)) { - rw_lock_set_writer(lock, RW_LOCK_EX); +#endif + if(success) { lock->writer_thread = curr_thread; lock->pass = 0; -relock: - lock->writer_count++; -#ifdef UNIV_SYNC_DEBUG - rw_lock_add_debug_info(lock, 0, RW_LOCK_EX, file_name, line); -#endif + } else if (!(lock->pass) && + os_thread_eq(lock->writer_thread, curr_thread)) { + /* Must verify pass first: otherwise another thread can + call move_ownership suddenly allowing recursive locks. + and after we have verified our thread_id matches + (though move_ownership has since changed it).*/ + + /* Relock: this lock_word modification is safe since no other + threads can modify (lock, unlock, or reserve) lock_word while + there is an exclusive writer and this is the writer thread. */ + lock->lock_word -= X_LOCK_DECR; - lock->last_x_file_name = file_name; - lock->last_x_line = line; + ut_ad(((-lock->lock_word) % X_LOCK_DECR) == 0); - success = TRUE; - } else if (rw_lock_get_writer(lock) == RW_LOCK_EX - && lock->pass == 0 - && os_thread_eq(lock->writer_thread, curr_thread)) { - goto relock; + } else { + /* Failure */ + return(FALSE); } +#ifdef UNIV_SYNC_DEBUG + rw_lock_add_debug_info(lock, 0, RW_LOCK_EX, file_name, line); +#endif - mutex_exit(rw_lock_get_mutex(lock)); + lock->last_x_file_name = file_name; + lock->last_x_line = line; ut_ad(rw_lock_validate(lock)); - return(success); + return(TRUE); } /********************************************************************** @@ -353,39 +412,21 @@ #endif ) { - mutex_t* mutex = &(lock->mutex); - ibool sg = FALSE; - - /* Acquire the mutex protecting the rw-lock fields */ - mutex_enter(mutex); - - /* Reset the shared lock by decrementing the reader count */ - - ut_a(lock->reader_count > 0); - lock->reader_count--; + ut_ad((lock->lock_word % X_LOCK_DECR) != 0); #ifdef UNIV_SYNC_DEBUG rw_lock_remove_debug_info(lock, pass, RW_LOCK_SHARED); #endif - /* If there may be waiters and this was the last s-lock, - signal the object */ - - if (UNIV_UNLIKELY(lock->waiters) - && lock->reader_count == 0) { - sg = TRUE; + /* Increment lock_word to indicate 1 less reader */ + if(rw_lock_lock_word_incr(lock, 1) == 0) { - rw_lock_set_waiters(lock, 0); - } - - mutex_exit(mutex); - - if (UNIV_UNLIKELY(sg)) { -#ifdef __WIN__ + /* wait_ex waiter exists. It may not be asleep, but we signal + anyway. We do not wake other waiters, because they can't + exist without wait_ex waiter and wait_ex waiter goes first.*/ os_event_set(lock->wait_ex_event); -#endif - os_event_set(lock->event); sync_array_object_signalled(sync_primary_wait_array); + } ut_ad(rw_lock_validate(lock)); @@ -404,16 +445,15 @@ /*====================*/ rw_lock_t* lock) /* in: rw-lock */ { - /* Reset the shared lock by decrementing the reader count */ - - ut_ad(lock->reader_count > 0); - - lock->reader_count--; + ut_ad(lock->lock_word < X_LOCK_DECR); #ifdef UNIV_SYNC_DEBUG rw_lock_remove_debug_info(lock, 0, RW_LOCK_SHARED); #endif + /* Decrease reader count by incrementing lock_word */ + lock->lock_word++; + ut_ad(!lock->waiters); ut_ad(rw_lock_validate(lock)); #ifdef UNIV_SYNC_PERF_STAT @@ -434,42 +474,32 @@ #endif ) { - ibool sg = FALSE; - - /* Acquire the mutex protecting the rw-lock fields */ - mutex_enter(&(lock->mutex)); - - /* Reset the exclusive lock if this thread no longer has an x-mode - lock */ - - ut_ad(lock->writer_count > 0); + ut_ad((lock->lock_word % X_LOCK_DECR) == 0); - lock->writer_count--; - - if (lock->writer_count == 0) { - rw_lock_set_writer(lock, RW_LOCK_NOT_LOCKED); - } + /* Must reset writer_thread while we still have the lock. + If we are not the last unlocker, we correct it later in the function, + which is harmless since we still hold the lock. */ + /* TODO: are there any risks of a thread id == -1 on any platform? */ + os_thread_id_t local_writer_thread = lock->writer_thread; + lock->writer_thread = -1; #ifdef UNIV_SYNC_DEBUG rw_lock_remove_debug_info(lock, pass, RW_LOCK_EX); #endif - /* If there may be waiters, signal the lock */ - if (UNIV_UNLIKELY(lock->waiters) - && lock->writer_count == 0) { - - sg = TRUE; - rw_lock_set_waiters(lock, 0); - } - - mutex_exit(&(lock->mutex)); + if(rw_lock_lock_word_incr(lock, X_LOCK_DECR) == X_LOCK_DECR) { + /* Lock is now free. May have to signal read/write waiters. + We do not need to signal wait_ex waiters, since they cannot + exist when there is a writer. */ + if(lock->waiters) { + rw_lock_set_waiters(lock, 0); + os_event_set(lock->event); + sync_array_object_signalled(sync_primary_wait_array); + } - if (UNIV_UNLIKELY(sg)) { -#ifdef __WIN__ - os_event_set(lock->wait_ex_event); -#endif - os_event_set(lock->event); - sync_array_object_signalled(sync_primary_wait_array); + } else { + /* We still hold x-lock, so we correct writer_thread. */ + lock->writer_thread = local_writer_thread; } ut_ad(rw_lock_validate(lock)); @@ -491,18 +521,14 @@ /* Reset the exclusive lock if this thread no longer has an x-mode lock */ - ut_ad(lock->writer_count > 0); - - lock->writer_count--; - - if (lock->writer_count == 0) { - rw_lock_set_writer(lock, RW_LOCK_NOT_LOCKED); - } + ut_ad((lock->lock_word % X_LOCK_DECR) == 0); #ifdef UNIV_SYNC_DEBUG rw_lock_remove_debug_info(lock, 0, RW_LOCK_EX); #endif + lock->lock_word += X_LOCK_DECR; + ut_ad(!lock->waiters); ut_ad(rw_lock_validate(lock)); diff --recursive -u innodb_plugin-1.0.1/include/sync0sync.h innodb_plugin-1.0.1-new/include/sync0sync.h --- innodb_plugin-1.0.1/include/sync0sync.h 2008-05-06 09:00:25.000000000 -0700 +++ innodb_plugin-1.0.1-new/include/sync0sync.h 2008-09-16 23:02:12.908604000 -0700 @@ -237,7 +237,7 @@ NOT to be used outside this module except in debugging! Gets the value of the lock word. */ UNIV_INLINE -ulint +byte mutex_get_lock_word( /*================*/ const mutex_t* mutex); /* in: mutex */ @@ -462,9 +462,11 @@ struct mutex_struct { os_event_t event; /* Used by sync0arr.c for the wait queue */ - ulint lock_word; /* This ulint is the target of the atomic - test-and-set instruction in Win32 */ -#if defined WIN32 && defined UNIV_CAN_USE_X86_ASSEMBLER + byte lock_word; /* This byte is the target of the atomic + test-and-set instruction in Win32 and + x86 32/64 with GCC 4.1.0 or later version */ +#if defined(_WIN32) && defined(UNIV_CAN_USE_X86_ASSEMBLER) +#elif defined(HAVE_GCC_ATOMIC_BUILTINS) #else os_fast_mutex_t os_fast_mutex; /* In other systems we use this OS mutex @@ -518,8 +520,7 @@ /* The number of system calls made in this module. Intended for performance monitoring. */ -extern ulint mutex_system_call_count; -extern ulint mutex_exit_count; +extern ib_int64_t mutex_exit_count; #ifdef UNIV_SYNC_DEBUG /* Latching order checks start when this is set TRUE */ diff --recursive -u innodb_plugin-1.0.1/include/sync0sync.ic innodb_plugin-1.0.1-new/include/sync0sync.ic --- innodb_plugin-1.0.1/include/sync0sync.ic 2008-02-29 13:45:22.000000000 -0800 +++ innodb_plugin-1.0.1-new/include/sync0sync.ic 2008-09-16 23:02:01.402423000 -0700 @@ -6,16 +6,6 @@ Created 9/5/1995 Heikki Tuuri *******************************************************/ -#if defined(not_defined) && defined(__GNUC__) && defined(UNIV_INTEL_X86) -/* %z0: Use the size of operand %0 which in our case is *m to determine -instruction size, it should end up as xchgl. "1" in the input constraint, -says that "in" has to go in the same place as "out".*/ -#define TAS(m, in, out) \ - asm volatile ("xchg%z0 %2, %0" \ - : "=g" (*(m)), "=r" (out) \ - : "1" (in)) /* Note: "1" here refers to "=r" (out) */ -#endif - /********************************************************************** Sets the waiters field in a mutex. */ UNIV_INTERN @@ -59,7 +49,7 @@ Performs an atomic test-and-set instruction to the lock_word field of a mutex. */ UNIV_INLINE -ulint +byte mutex_test_and_set( /*===============*/ /* out: the previous value of lock_word: 0 or @@ -67,18 +57,18 @@ mutex_t* mutex) /* in: mutex */ { #if defined(_WIN32) && defined(UNIV_CAN_USE_X86_ASSEMBLER) - ulint res; - ulint* lw; /* assembler code is used to ensure that + byte res; + byte* lw; /* assembler code is used to ensure that lock_word is loaded from memory */ ut_ad(mutex); - ut_ad(sizeof(ulint) == 4); + ut_ad(sizeof(byte) == 1); lw = &(mutex->lock_word); __asm MOV ECX, lw __asm MOV EDX, 1 - __asm XCHG EDX, DWORD PTR [ECX] - __asm MOV res, EDX + __asm XCHG DL, BYTE PTR [ECX] + __asm MOV res, DL /* The fence below would prevent this thread from reading the data structure protected by the mutex @@ -98,12 +88,8 @@ /* mutex_fence(); */ return(res); -#elif defined(not_defined) && defined(__GNUC__) && defined(UNIV_INTEL_X86) - ulint res; - - TAS(&mutex->lock_word, 1, res); - - return(res); +#elif defined(HAVE_GCC_ATOMIC_BUILTINS) + return __sync_lock_test_and_set(&(mutex->lock_word), 1); #else ibool ret; @@ -117,7 +103,7 @@ mutex->lock_word = 1; } - return(ret); + return((byte)ret); #endif } @@ -131,7 +117,7 @@ mutex_t* mutex) /* in: mutex */ { #if defined(_WIN32) && defined(UNIV_CAN_USE_X86_ASSEMBLER) - ulint* lw; /* assembler code is used to ensure that + byte* lw; /* assembler code is used to ensure that lock_word is loaded from memory */ ut_ad(mutex); @@ -139,11 +125,12 @@ __asm MOV EDX, 0 __asm MOV ECX, lw - __asm XCHG EDX, DWORD PTR [ECX] -#elif defined(not_defined) && defined(__GNUC__) && defined(UNIV_INTEL_X86) - ulint res; - - TAS(&mutex->lock_word, 0, res); + __asm XCHG DL, BYTE PTR [ECX] +#elif defined(HAVE_GCC_ATOMIC_BUILTINS) + /* In theory __sync_lock_release should be used to release the lock. + Unfortunately, it does not work properly alone. The workaround is + that more conservative __sync_lock_test_and_set is used instead. */ + __sync_lock_test_and_set(&(mutex->lock_word), 0); #else mutex->lock_word = 0; @@ -154,12 +141,12 @@ /********************************************************************** Gets the value of the lock word. */ UNIV_INLINE -ulint +byte mutex_get_lock_word( /*================*/ const mutex_t* mutex) /* in: mutex */ { - const volatile ulint* ptr; /* declared volatile to ensure that + const volatile byte* ptr; /* declared volatile to ensure that lock_word is loaded from memory */ ut_ad(mutex); diff --recursive -u innodb_plugin-1.0.1/include/univ.i innodb_plugin-1.0.1-new/include/univ.i --- innodb_plugin-1.0.1/include/univ.i 2008-05-06 05:10:09.000000000 -0700 +++ innodb_plugin-1.0.1-new/include/univ.i 2008-06-02 16:53:35.978821000 -0700 @@ -171,6 +171,9 @@ #define UNIV_SET_MEM_TO_ZERO #endif +/* Use malloc instead of innodb additional memory pool (great with tcmalloc) */ +#define UNIV_DISABLE_MEM_POOL + /* #define UNIV_SQL_DEBUG #define UNIV_LOG_DEBUG diff --recursive -u innodb_plugin-1.0.1/mem/mem0pool.c innodb_plugin-1.0.1-new/mem/mem0pool.c --- innodb_plugin-1.0.1/mem/mem0pool.c 2008-02-29 13:45:22.000000000 -0800 +++ innodb_plugin-1.0.1-new/mem/mem0pool.c 2008-09-09 18:41:44.973992000 -0700 @@ -331,6 +331,9 @@ or equal to the requested size) */ mem_pool_t* pool) /* in: memory pool */ { +#ifdef UNIV_DISABLE_MEM_POOL + return malloc(*psize); +#else /* UNIV_DISABLE_MEM_POOL */ mem_area_t* area; ulint size; ulint n; @@ -412,6 +415,7 @@ UNIV_MEM_ALLOC(MEM_AREA_EXTRA_SIZE + (byte*)area, *psize); return((void*)(MEM_AREA_EXTRA_SIZE + ((byte*)area))); +#endif /* UNIV_DISABLE_MEM_POOL */ } /************************************************************************ @@ -464,6 +468,9 @@ buffer */ mem_pool_t* pool) /* in: memory pool */ { +#ifdef UNIV_DISABLE_MEM_POOL + free(ptr); +#else /* UNIV_DISABLE_MEM_POOL */ mem_area_t* area; mem_area_t* buddy; void* new_ptr; @@ -575,6 +582,7 @@ mutex_exit(&(pool->mutex)); ut_ad(mem_pool_validate(pool)); +#endif /* UNIV_DISABLE_MEM_POOL */ } /************************************************************************ diff --recursive -u innodb_plugin-1.0.1/row/row0sel.c innodb_plugin-1.0.1-new/row/row0sel.c --- innodb_plugin-1.0.1/row/row0sel.c 2008-04-29 23:42:56.000000000 -0700 +++ innodb_plugin-1.0.1-new/row/row0sel.c 2008-06-02 16:53:35.992802000 -0700 @@ -1341,7 +1341,7 @@ rw_lock_s_lock(&btr_search_latch); search_latch_locked = TRUE; - } else if (btr_search_latch.writer_is_wait_ex) { + } else if (rw_lock_get_writer(&btr_search_latch) == RW_LOCK_WAIT_EX) { /* There is an x-latch request waiting: release the s-latch for a moment; as an s-latch here is often @@ -3448,7 +3448,7 @@ /* PHASE 0: Release a possible s-latch we are holding on the adaptive hash index latch if there is someone waiting behind */ - if (UNIV_UNLIKELY(btr_search_latch.writer != RW_LOCK_NOT_LOCKED) + if (UNIV_UNLIKELY(rw_lock_get_writer(&btr_search_latch) != RW_LOCK_NOT_LOCKED) && trx->has_search_latch) { /* There is an x-latch request on the adaptive hash index: diff --recursive -u innodb_plugin-1.0.1/srv/srv0srv.c innodb_plugin-1.0.1-new/srv/srv0srv.c --- innodb_plugin-1.0.1/srv/srv0srv.c 2008-05-06 09:00:25.000000000 -0700 +++ innodb_plugin-1.0.1-new/srv/srv0srv.c 2008-09-16 23:03:02.956330000 -0700 @@ -1799,6 +1799,16 @@ export_vars.innodb_buffer_pool_pages_misc = buf_pool->curr_size - UT_LIST_GET_LEN(buf_pool->LRU) - UT_LIST_GET_LEN(buf_pool->free); +#ifdef HAVE_GCC_ATOMIC_BUILTINS + export_vars.innodb_have_atomic_builtins = 1; +#else + export_vars.innodb_have_atomic_builtins = 0; +#endif +#ifdef UNIV_DISABLE_MEM_POOL + export_vars.innodb_heap_enabled = 0; +#else + export_vars.innodb_heap_enabled = 1; +#endif export_vars.innodb_page_size = UNIV_PAGE_SIZE; export_vars.innodb_log_waits = srv_log_waits; export_vars.innodb_os_log_written = srv_os_log_written; diff --recursive -u innodb_plugin-1.0.1/srv/srv0start.c innodb_plugin-1.0.1-new/srv/srv0start.c --- innodb_plugin-1.0.1/srv/srv0start.c 2008-05-06 09:00:25.000000000 -0700 +++ innodb_plugin-1.0.1-new/srv/srv0start.c 2008-09-16 23:03:13.246585000 -0700 @@ -1046,6 +1046,16 @@ "InnoDB: !!!!!!!! UNIV_MEM_DEBUG switched on !!!!!!!!!\n"); #endif +#ifdef UNIV_DISABLE_MEM_POOL + fprintf(stderr, + "InnoDB: The InnoDB memory heap has been disabled.\n"); +#endif + +#ifdef HAVE_GCC_ATOMIC_BUILTINS + fprintf(stderr, + "InnoDB: Mutex and rw_lock use GCC atomic builtins.\n"); +#endif + /* Since InnoDB does not currently clean up all its internal data structures in MySQL Embedded Server Library server_end(), we print an error message if someone tries to start up InnoDB a diff --recursive -u innodb_plugin-1.0.1/sync/sync0arr.c innodb_plugin-1.0.1-new/sync/sync0arr.c --- innodb_plugin-1.0.1/sync/sync0arr.c 2008-03-17 07:19:04.000000000 -0700 +++ innodb_plugin-1.0.1-new/sync/sync0arr.c 2008-09-16 23:04:43.809338000 -0700 @@ -295,25 +295,21 @@ } /*********************************************************************** -Puts the cell event in reset state. */ +Returns the event that the thread owning the cell waits for. */ static -ib_int64_t -sync_cell_event_reset( -/*==================*/ - /* out: value of signal_count - at the time of reset. */ - ulint type, /* in: lock type mutex/rw_lock */ - void* object) /* in: the rw_lock/mutex object */ +os_event_t +sync_cell_get_event( +/*================*/ + sync_cell_t* cell) /* in: non-empty sync array cell */ { + ulint type = cell->request_type; + if (type == SYNC_MUTEX) { - return(os_event_reset(((mutex_t *) object)->event)); -#ifdef __WIN__ + return(((mutex_t *) cell->wait_object)->event); } else if (type == RW_LOCK_WAIT_EX) { - return(os_event_reset( - ((rw_lock_t *) object)->wait_ex_event)); -#endif - } else { - return(os_event_reset(((rw_lock_t *) object)->event)); + return(((rw_lock_t *) cell->wait_object)->wait_ex_event); + } else { /* RW_LOCK_SHARED and RW_LOCK_EX wait on the same event */ + return(((rw_lock_t *) cell->wait_object)->event); } } @@ -332,6 +328,7 @@ ulint* index) /* out: index of the reserved cell */ { sync_cell_t* cell; + os_event_t event; ulint i; ut_a(object); @@ -370,8 +367,8 @@ /* Make sure the event is reset and also store the value of signal_count at which the event was reset. */ - cell->signal_count = sync_cell_event_reset(type, - object); + event = sync_cell_get_event(cell); + cell->signal_count = os_event_reset(event); cell->reservation_time = time(NULL); @@ -411,19 +408,7 @@ ut_a(!cell->waiting); ut_ad(os_thread_get_curr_id() == cell->thread); - if (cell->request_type == SYNC_MUTEX) { - event = ((mutex_t*) cell->wait_object)->event; -#ifdef __WIN__ - /* On windows if the thread about to wait is the one which - has set the state of the rw_lock to RW_LOCK_WAIT_EX, then - it waits on a special event i.e.: wait_ex_event. */ - } else if (cell->request_type == RW_LOCK_WAIT_EX) { - event = ((rw_lock_t*) cell->wait_object)->wait_ex_event; -#endif - } else { - event = ((rw_lock_t*) cell->wait_object)->event; - } - + event = sync_cell_get_event(cell); cell->waiting = TRUE; #ifdef UNIV_SYNC_DEBUG @@ -462,6 +447,7 @@ mutex_t* mutex; rw_lock_t* rwlock; ulint type; + ulint writer; type = cell->request_type; @@ -491,9 +477,7 @@ (ulong) mutex->waiters); } else if (type == RW_LOCK_EX -#ifdef __WIN__ || type == RW_LOCK_WAIT_EX -#endif || type == RW_LOCK_SHARED) { fputs(type == RW_LOCK_EX ? "X-lock on" : "S-lock on", file); @@ -504,22 +488,25 @@ " RW-latch at %p created in file %s line %lu\n", (void*) rwlock, rwlock->cfile_name, (ulong) rwlock->cline); - if (rwlock->writer != RW_LOCK_NOT_LOCKED) { + writer = rw_lock_get_writer(rwlock); + if (writer != RW_LOCK_NOT_LOCKED) { fprintf(file, "a writer (thread id %lu) has" " reserved it in mode %s", (ulong) os_thread_pf(rwlock->writer_thread), - rwlock->writer == RW_LOCK_EX + writer == RW_LOCK_EX ? " exclusive\n" : " wait exclusive\n"); } fprintf(file, - "number of readers %lu, waiters flag %lu\n" + "number of readers %lu, waiters flag %lu, " + "lock_word: %ld\n" "Last time read locked in file %s line %lu\n" "Last time write locked in file %s line %lu\n", - (ulong) rwlock->reader_count, + (ulong) rw_lock_get_reader_count(rwlock), (ulong) rwlock->waiters, + rwlock->lock_word, rwlock->last_s_file_name, (ulong) rwlock->last_s_line, rwlock->last_x_file_name, @@ -778,28 +765,30 @@ return(TRUE); } - } else if (cell->request_type == RW_LOCK_EX - || cell->request_type == RW_LOCK_WAIT_EX) { + } else if (cell->request_type == RW_LOCK_EX) { lock = cell->wait_object; - if (rw_lock_get_reader_count(lock) == 0 - && rw_lock_get_writer(lock) == RW_LOCK_NOT_LOCKED) { + /* X_LOCK_DECR is the unlocked state */ + if (lock->lock_word == X_LOCK_DECR) { return(TRUE); } - if (rw_lock_get_reader_count(lock) == 0 - && rw_lock_get_writer(lock) == RW_LOCK_WAIT_EX - && os_thread_eq(lock->writer_thread, cell->thread)) { + } else if (cell->request_type == RW_LOCK_WAIT_EX) { + + lock = cell->wait_object; + + /* lock_word == 0 means all readers have left */ + if (lock->lock_word == 0) { return(TRUE); } - } else if (cell->request_type == RW_LOCK_SHARED) { lock = cell->wait_object; - if (rw_lock_get_writer(lock) == RW_LOCK_NOT_LOCKED) { + /* lock_word > 0 means no writer or reserved writer */ + if (lock->lock_word > 0) { return(TRUE); } @@ -844,11 +833,15 @@ /*========================*/ sync_array_t* arr) /* in: wait array */ { +#ifdef HAVE_GCC_ATOMIC_BUILTINS + __sync_fetch_and_add(&(arr->sg_count),1); +#else sync_array_enter(arr); arr->sg_count++; sync_array_exit(arr); +#endif } /************************************************************************** @@ -868,6 +861,7 @@ sync_cell_t* cell; ulint count; ulint i; + os_event_t event; sync_array_enter(arr); @@ -877,36 +871,20 @@ while (count < arr->n_reserved) { cell = sync_array_get_nth_cell(arr, i); + i++; - if (cell->wait_object != NULL) { - + if (cell->wait_object == NULL) { + continue; + } count++; if (sync_arr_cell_can_wake_up(cell)) { - if (cell->request_type == SYNC_MUTEX) { - mutex_t* mutex; - - mutex = cell->wait_object; - os_event_set(mutex->event); -#ifdef __WIN__ - } else if (cell->request_type - == RW_LOCK_WAIT_EX) { - rw_lock_t* lock; + event = sync_cell_get_event(cell); - lock = cell->wait_object; - os_event_set(lock->wait_ex_event); -#endif - } else { - rw_lock_t* lock; - - lock = cell->wait_object; - os_event_set(lock->event); - } - } + os_event_set(event); } - i++; } sync_array_exit(arr); @@ -1026,4 +1004,3 @@ sync_array_exit(arr); } - diff --recursive -u innodb_plugin-1.0.1/sync/sync0rw.c innodb_plugin-1.0.1-new/sync/sync0rw.c --- innodb_plugin-1.0.1/sync/sync0rw.c 2008-02-29 13:45:22.000000000 -0800 +++ innodb_plugin-1.0.1-new/sync/sync0rw.c 2008-09-16 23:04:35.127050000 -0700 @@ -15,35 +15,111 @@ #include "mem0mem.h" #include "srv0srv.h" -/* number of system calls made during shared latching */ -UNIV_INTERN ulint rw_s_system_call_count = 0; +/* + 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: + +pass: This is only set to 1 to prevent recursive x-locks. It must + be set as specified by x_lock caller after the lock_word + indicates that the thread holds the lock, but before that + thread resumes execution. It must be reset to 0 during the + final x_unlock, but before the lock_word status is updated. + When an x_lock or move_ownership call wishes to change + pass, it must first update the writer_thread appropriately. +writer_thread: Must be set to the writers thread_id after the lock_word + indicates that the thread holds the lock, but before that + thread resumes execution. It must be reset to -1 during the + final x_unlock, but before the lock_word status is updated. + This ensures that when the lock_word indicates that an x_lock + is held, the only legitimate values for writer_thread are -1 + (x_lock function hasn't completed) or the writer's thread_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. +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 ulint rw_s_spin_wait_count = 0; +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 ulint rw_s_os_wait_count = 0; +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 ulint rw_s_exit_count = 0; - -/* number of system calls made during exclusive latching */ -UNIV_INTERN ulint rw_x_system_call_count = 0; +UNIV_INTERN ib_int64_t rw_s_exit_count = 0; /* number of spin waits on rw-latches, resulted during exclusive (write) locks */ -UNIV_INTERN ulint rw_x_spin_wait_count = 0; +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 ulint rw_x_os_wait_count = 0; +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 ulint rw_x_exit_count = 0; +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; @@ -119,6 +195,7 @@ /* If this is the very first time a synchronization object is created, then the following call initializes the sync system. */ +#ifndef HAVE_GCC_ATOMIC_BUILTINS mutex_create(rw_lock_get_mutex(lock), SYNC_NO_ORDER_CHECK); lock->mutex.cfile_name = cfile_name; @@ -129,12 +206,12 @@ lock->mutex.mutex_type = 1; #endif /* UNIV_DEBUG && !UNIV_HOTBACKUP */ - rw_lock_set_waiters(lock, 0); - rw_lock_set_writer(lock, RW_LOCK_NOT_LOCKED); - lock->writer_count = 0; - rw_lock_set_reader_count(lock, 0); +#endif /* HAVE_GCC_ATOMIC_BUILTINS */ - lock->writer_is_wait_ex = FALSE; + lock->lock_word = X_LOCK_DECR; + rw_lock_set_waiters(lock, 0); + lock->writer_thread = -1; + lock->pass = 0; #ifdef UNIV_SYNC_DEBUG UT_LIST_INIT(lock->debug_list); @@ -147,15 +224,13 @@ 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(NULL); - -#ifdef __WIN__ lock->wait_ex_event = os_event_create(NULL); -#endif mutex_enter(&rw_lock_list_mutex); @@ -180,20 +255,19 @@ rw_lock_t* lock) /* in: rw-lock */ { ut_ad(rw_lock_validate(lock)); - ut_a(rw_lock_get_writer(lock) == RW_LOCK_NOT_LOCKED); + ut_a(lock->lock_word == X_LOCK_DECR); ut_a(rw_lock_get_waiters(lock) == 0); - ut_a(rw_lock_get_reader_count(lock) == 0); lock->magic_n = 0; +#ifndef HAVE_GCC_ATOMIC_BUILTINS mutex_free(rw_lock_get_mutex(lock)); +#endif /* HAVE_GCC_ATOMIC_BUILTINS */ mutex_enter(&rw_lock_list_mutex); os_event_free(lock->event); -#ifdef __WIN__ os_event_free(lock->wait_ex_event); -#endif if (UT_LIST_GET_PREV(list, lock)) { ut_a(UT_LIST_GET_PREV(list, lock)->magic_n == RW_LOCK_MAGIC_N); @@ -219,19 +293,12 @@ { ut_a(lock); - mutex_enter(rw_lock_get_mutex(lock)); + ulint waiters = rw_lock_get_waiters(lock); + lint lock_word = lock->lock_word; ut_a(lock->magic_n == RW_LOCK_MAGIC_N); - ut_a((rw_lock_get_reader_count(lock) == 0) - || (rw_lock_get_writer(lock) != RW_LOCK_EX)); - ut_a((rw_lock_get_writer(lock) == RW_LOCK_EX) - || (rw_lock_get_writer(lock) == RW_LOCK_WAIT_EX) - || (rw_lock_get_writer(lock) == RW_LOCK_NOT_LOCKED)); - ut_a((rw_lock_get_waiters(lock) == 0) - || (rw_lock_get_waiters(lock) == 1)); - ut_a((lock->writer != RW_LOCK_EX) || (lock->writer_count > 0)); - - mutex_exit(rw_lock_get_mutex(lock)); + ut_a(waiters == 0 || waiters == 1); + ut_a(lock_word > -X_LOCK_DECR ||(-lock_word) % X_LOCK_DECR == 0); return(TRUE); } @@ -253,18 +320,15 @@ ulint line) /* in: line where requested */ { ulint index; /* index of the reserved wait cell */ - ulint i; /* spin round count */ + ulint i = 0; /* spin round count */ ut_ad(rw_lock_validate(lock)); + rw_s_spin_wait_count++; /* Count calls to this function */ lock_loop: - rw_s_spin_wait_count++; /* Spin waiting for the writer field to become free */ - i = 0; - - while (rw_lock_get_writer(lock) != RW_LOCK_NOT_LOCKED - && i < SYNC_SPIN_ROUNDS) { + while (i < SYNC_SPIN_ROUNDS && lock->lock_word <= 0) { if (srv_spin_wait_delay) { ut_delay(ut_rnd_interval(0, srv_spin_wait_delay)); } @@ -285,28 +349,32 @@ lock->cfile_name, (ulong) lock->cline, (ulong) i); } - mutex_enter(rw_lock_get_mutex(lock)); - /* We try once again to obtain the lock */ - if (TRUE == rw_lock_s_lock_low(lock, pass, file_name, line)) { - mutex_exit(rw_lock_get_mutex(lock)); + rw_s_spin_round_count += i; return; /* Success */ } else { - /* If we get here, locking did not succeed, we may - suspend the thread to wait in the wait array */ - rw_s_system_call_count++; + 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. */ rw_lock_set_waiters(lock, 1); - mutex_exit(rw_lock_get_mutex(lock)); + if (TRUE == rw_lock_s_lock_low(lock, pass, file_name, line)) { + sync_array_free_cell(sync_primary_wait_array, index); + return; /* Success */ + } if (srv_print_latch_waits) { fprintf(stderr, @@ -317,11 +385,13 @@ (ulong) lock->cline); } - rw_s_system_call_count++; + /* 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; } } @@ -343,113 +413,149 @@ { ut_ad(rw_lock_is_locked(lock, RW_LOCK_EX)); +#ifdef HAVE_GCC_ATOMIC_BUILTINS + os_thread_id_t local_writer_thread = lock->writer_thread; + os_thread_id_t new_writer_thread = os_thread_get_curr_id(); + while (TRUE) { + if (local_writer_thread != -1) { + if(os_compare_and_swap( + &(lock->writer_thread), + local_writer_thread, + new_writer_thread)) { + break; + } + } + local_writer_thread = lock->writer_thread; + } + lock->pass = 0; +#else /* HAVE_GCC_ATOMIC_BUILTINS */ mutex_enter(&(lock->mutex)); - lock->writer_thread = os_thread_get_curr_id(); - lock->pass = 0; - mutex_exit(&(lock->mutex)); +#endif /* HAVE_GCC_ATOMIC_BUILTINS */ } /********************************************************************** -Low-level function for acquiring an exclusive lock. */ +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 -ulint -rw_lock_x_lock_low( -/*===============*/ - /* out: RW_LOCK_NOT_LOCKED if did - not succeed, RW_LOCK_EX if success, - RW_LOCK_WAIT_EX, if got wait reservation */ +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 */ { - ut_ad(mutex_own(rw_lock_get_mutex(lock))); + ulint index; + ulint i = 0; - if (rw_lock_get_writer(lock) == RW_LOCK_NOT_LOCKED) { + ut_ad(lock->lock_word <= 0); - if (rw_lock_get_reader_count(lock) == 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; + } - rw_lock_set_writer(lock, RW_LOCK_EX); - lock->writer_thread = os_thread_get_curr_id(); - lock->writer_count++; - lock->pass = pass; + /* 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_EX, + rw_lock_add_debug_info(lock, pass, RW_LOCK_WAIT_EX, file_name, line); #endif - lock->last_x_file_name = file_name; - lock->last_x_line = (unsigned int) line; - - /* Locking succeeded, we may return */ - return(RW_LOCK_EX); - } else { - /* There are readers, we have to wait */ - rw_lock_set_writer(lock, RW_LOCK_WAIT_EX); - lock->writer_thread = os_thread_get_curr_id(); - lock->pass = pass; - lock->writer_is_wait_ex = TRUE; + sync_array_wait_event(sync_primary_wait_array, + index); #ifdef UNIV_SYNC_DEBUG - rw_lock_add_debug_info(lock, pass, RW_LOCK_WAIT_EX, - file_name, line); + rw_lock_remove_debug_info(lock, pass, + RW_LOCK_WAIT_EX); #endif - - return(RW_LOCK_WAIT_EX); + /* 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; +} - } else if ((rw_lock_get_writer(lock) == RW_LOCK_WAIT_EX) - && os_thread_eq(lock->writer_thread, - os_thread_get_curr_id())) { - - if (rw_lock_get_reader_count(lock) == 0) { +/********************************************************************** +Low-level function for acquiring an exclusive lock. */ +UNIV_INLINE +ibool +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(); + ut_ad(curr_thread != -1); /* We use -1 as the unlocked value. */ - rw_lock_set_writer(lock, RW_LOCK_EX); - lock->writer_count++; - lock->pass = pass; - lock->writer_is_wait_ex = FALSE; + if(rw_lock_lock_word_decr(lock, X_LOCK_DECR)) { + ut_ad(lock->writer_thread == -1); + /* Decrement occurred: we are writer or next-writer. */ + lock->writer_thread = curr_thread; + lock->pass = pass; + rw_lock_x_lock_wait(lock, #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); + pass, #endif + file_name, line); - lock->last_x_file_name = file_name; - lock->last_x_line = (unsigned int) line; - - /* Locking succeeded, we may return */ - return(RW_LOCK_EX); + } else { + /* Decrement failed: relock or failed lock */ + /* Must verify pass first: otherwise another thread can + call move_ownership suddenly allowing recursive locks. + and after we have verified our thread_id matches + (though move_ownership has since changed it).*/ + if(!pass && !(lock->pass) && + os_thread_eq(lock->writer_thread, curr_thread)) { + /* Relock */ + lock->lock_word -= X_LOCK_DECR; + } else { + /* Another thread locked before us */ + return(FALSE); } - - return(RW_LOCK_WAIT_EX); - - } else if ((rw_lock_get_writer(lock) == RW_LOCK_EX) - && os_thread_eq(lock->writer_thread, - os_thread_get_curr_id()) - && (lock->pass == 0) - && (pass == 0)) { - - lock->writer_count++; - + } #ifdef UNIV_SYNC_DEBUG - rw_lock_add_debug_info(lock, pass, RW_LOCK_EX, file_name, - line); + 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; - lock->last_x_file_name = file_name; - lock->last_x_line = (unsigned int) line; - - /* Locking succeeded, we may return */ - return(RW_LOCK_EX); - } - - /* Locking did not succeed */ - return(RW_LOCK_NOT_LOCKED); + return(TRUE); } /********************************************************************** @@ -472,47 +578,30 @@ ulint line) /* in: line where requested */ { ulint index; /* index of the reserved wait cell */ - ulint state; /* lock state acquired */ ulint i; /* spin round count */ + ibool spinning = FALSE; ut_ad(rw_lock_validate(lock)); -lock_loop: - /* Acquire the mutex protecting the rw-lock fields */ - mutex_enter_fast(&(lock->mutex)); - - state = rw_lock_x_lock_low(lock, pass, file_name, line); + i = 0; - mutex_exit(&(lock->mutex)); +lock_loop: - if (state == RW_LOCK_EX) { + if (rw_lock_x_lock_low(lock, pass, file_name, line)) { + rw_x_spin_round_count += i; return; /* Locking succeeded */ - } else if (state == RW_LOCK_NOT_LOCKED) { - - /* Spin waiting for the writer field to become free */ - i = 0; - - while (rw_lock_get_writer(lock) != RW_LOCK_NOT_LOCKED - && i < SYNC_SPIN_ROUNDS) { - if (srv_spin_wait_delay) { - ut_delay(ut_rnd_interval(0, - srv_spin_wait_delay)); - } + } else { - i++; - } - if (i == SYNC_SPIN_ROUNDS) { - os_thread_yield(); + if (!spinning) { + spinning = TRUE; + rw_x_spin_wait_count++; } - } else if (state == RW_LOCK_WAIT_EX) { - /* Spin waiting for the reader count field to become zero */ - i = 0; - - while (rw_lock_get_reader_count(lock) != 0 - && i < SYNC_SPIN_ROUNDS) { + /* 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)); @@ -522,12 +611,13 @@ } if (i == SYNC_SPIN_ROUNDS) { os_thread_yield(); + } else { + goto lock_loop; } - } else { - i = 0; /* Eliminate a compiler warning */ - ut_error; } + rw_x_spin_round_count += i; + if (srv_print_latch_waits) { fprintf(stderr, "Thread %lu spin wait rw-x-lock at %p" @@ -536,39 +626,20 @@ lock->cfile_name, (ulong) lock->cline, (ulong) i); } - rw_x_spin_wait_count++; - - /* We try once again to obtain the lock. Acquire the mutex protecting - the rw-lock fields */ - - mutex_enter(rw_lock_get_mutex(lock)); - - state = rw_lock_x_lock_low(lock, pass, file_name, line); - - if (state == RW_LOCK_EX) { - mutex_exit(rw_lock_get_mutex(lock)); - - return; /* Locking succeeded */ - } - - rw_x_system_call_count++; - sync_array_reserve_cell(sync_primary_wait_array, lock, -#ifdef __WIN__ - /* On windows RW_LOCK_WAIT_EX signifies - that this thread should wait on the - special wait_ex_event. */ - (state == RW_LOCK_WAIT_EX) - ? RW_LOCK_WAIT_EX : -#endif 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_waiters(lock, 1); - mutex_exit(rw_lock_get_mutex(lock)); + if (rw_lock_x_lock_low(lock, pass, file_name, line)) { + sync_array_free_cell(sync_primary_wait_array, index); + return; /* Locking succeeded */ + } if (srv_print_latch_waits) { fprintf(stderr, @@ -578,11 +649,13 @@ lock->cfile_name, (ulong) lock->cline); } - rw_x_system_call_count++; + /* 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; } @@ -730,7 +803,7 @@ ut_ad(lock); ut_ad(rw_lock_validate(lock)); - mutex_enter(&(lock->mutex)); + rw_lock_debug_mutex_enter(); info = UT_LIST_GET_FIRST(lock->debug_list); @@ -740,7 +813,7 @@ && (info->pass == 0) && (info->lock_type == lock_type)) { - mutex_exit(&(lock->mutex)); + rw_lock_debug_mutex_exit(); /* Found! */ return(TRUE); @@ -748,7 +821,7 @@ info = UT_LIST_GET_NEXT(list, info); } - mutex_exit(&(lock->mutex)); + rw_lock_debug_mutex_exit(); return(FALSE); } @@ -770,22 +843,18 @@ ut_ad(lock); ut_ad(rw_lock_validate(lock)); - mutex_enter(&(lock->mutex)); - if (lock_type == RW_LOCK_SHARED) { - if (lock->reader_count > 0) { + if (rw_lock_get_reader_count(lock) > 0) { ret = TRUE; } } else if (lock_type == RW_LOCK_EX) { - if (lock->writer == RW_LOCK_EX) { + if (rw_lock_get_writer(lock) == RW_LOCK_EX) { ret = TRUE; } } else { ut_error; } - mutex_exit(&(lock->mutex)); - return(ret); } @@ -814,11 +883,10 @@ count++; +#ifndef HAVE_GCC_ATOMIC_BUILTINS mutex_enter(&(lock->mutex)); - - if ((rw_lock_get_writer(lock) != RW_LOCK_NOT_LOCKED) - || (rw_lock_get_reader_count(lock) != 0) - || (rw_lock_get_waiters(lock) != 0)) { +#endif + if (lock->lock_word != X_LOCK_DECR) { fprintf(file, "RW-LOCK: %p ", (void*) lock); @@ -834,8 +902,10 @@ info = UT_LIST_GET_NEXT(list, info); } } - +#ifndef HAVE_GCC_ATOMIC_BUILTINS mutex_exit(&(lock->mutex)); +#endif + lock = UT_LIST_GET_NEXT(list, lock); } @@ -858,9 +928,10 @@ "RW-LATCH INFO\n" "RW-LATCH: %p ", (void*) lock); - if ((rw_lock_get_writer(lock) != RW_LOCK_NOT_LOCKED) - || (rw_lock_get_reader_count(lock) != 0) - || (rw_lock_get_waiters(lock) != 0)) { +#ifndef HAVE_GCC_ATOMIC_BUILTINS + mutex_enter(&(lock->mutex)); +#endif + if (lock->lock_word != X_LOCK_DECR) { if (rw_lock_get_waiters(lock)) { fputs(" Waiters for the lock exist\n", stderr); @@ -874,6 +945,9 @@ info = UT_LIST_GET_NEXT(list, info); } } +#ifndef HAVE_GCC_ATOMIC_BUILTINS + mutex_exit(&(lock->mutex)); +#endif } /************************************************************************* @@ -922,14 +996,11 @@ lock = UT_LIST_GET_FIRST(rw_lock_list); while (lock != NULL) { - mutex_enter(rw_lock_get_mutex(lock)); - if ((rw_lock_get_writer(lock) != RW_LOCK_NOT_LOCKED) - || (rw_lock_get_reader_count(lock) != 0)) { + if (lock->lock_word != X_LOCK_DECR) { count++; } - mutex_exit(rw_lock_get_mutex(lock)); lock = UT_LIST_GET_NEXT(list, lock); } diff --recursive -u innodb_plugin-1.0.1/sync/sync0sync.c innodb_plugin-1.0.1-new/sync/sync0sync.c --- innodb_plugin-1.0.1/sync/sync0sync.c 2008-05-06 09:00:25.000000000 -0700 +++ innodb_plugin-1.0.1-new/sync/sync0sync.c 2008-09-16 23:04:57.041470000 -0700 @@ -138,18 +138,13 @@ signalled unconditionally at the release of the lock. Q.E.D. */ -/* The number of system calls made in this module. Intended for performance -monitoring. */ - -UNIV_INTERN ulint mutex_system_call_count = 0; - /* Number of spin waits on mutexes: for performance monitoring */ /* round=one iteration of a spin loop */ -UNIV_INTERN ulint mutex_spin_round_count = 0; -UNIV_INTERN ulint mutex_spin_wait_count = 0; -UNIV_INTERN ulint mutex_os_wait_count = 0; -UNIV_INTERN ulint mutex_exit_count = 0; +UNIV_INTERN ib_int64_t mutex_spin_round_count = 0; +UNIV_INTERN ib_int64_t mutex_spin_wait_count = 0; +UNIV_INTERN ib_int64_t mutex_os_wait_count = 0; +UNIV_INTERN ib_int64_t mutex_exit_count = 0; /* The global array of wait cells for implementation of the database's own mutexes and read-write locks */ @@ -219,6 +214,8 @@ { #if defined(_WIN32) && defined(UNIV_CAN_USE_X86_ASSEMBLER) mutex_reset_lock_word(mutex); +#elif defined(HAVE_GCC_ATOMIC_BUILTINS) + mutex_reset_lock_word(mutex); #else os_fast_mutex_init(&(mutex->os_fast_mutex)); mutex->lock_word = 0; @@ -309,7 +306,9 @@ os_event_free(mutex->event); -#if !defined(_WIN32) || !defined(UNIV_CAN_USE_X86_ASSEMBLER) +#if defined(_WIN32) && defined(UNIV_CAN_USE_X86_ASSEMBLER) +#elif defined(HAVE_GCC_ATOMIC_BUILTINS) +#else os_fast_mutex_free(&(mutex->os_fast_mutex)); #endif /* If we free the mutex protecting the mutex list (freeing is @@ -426,6 +425,12 @@ #endif /* UNIV_DEBUG && !UNIV_HOTBACKUP */ ut_ad(mutex); + /* This update is not thread safe, but we don't mind if the count + isn't exact. Moved out of ifdef that follows because we are willing + to sacrifice the cost of counting this as the data is valuable. + Count the number of calls to mutex_spin_wait. */ + mutex_spin_wait_count++; + mutex_loop: i = 0; @@ -438,7 +443,6 @@ spin_loop: #if defined UNIV_DEBUG && !defined UNIV_HOTBACKUP - mutex_spin_wait_count++; mutex->count_spin_loop++; #endif /* UNIV_DEBUG && !UNIV_HOTBACKUP */ @@ -503,8 +507,6 @@ sync_array_reserve_cell(sync_primary_wait_array, mutex, SYNC_MUTEX, file_name, line, &index); - mutex_system_call_count++; - /* The memory order of the array reservation and the change in the waiters field is important: when we suspend a thread, we first reserve the cell and then set waiters field to 1. When threads are @@ -551,7 +553,6 @@ mutex->cfile_name, (ulong) mutex->cline, (ulong) i); #endif - mutex_system_call_count++; mutex_os_wait_count++; #ifndef UNIV_HOTBACKUP @@ -1333,21 +1334,31 @@ FILE* file) /* in: file where to print */ { #ifdef UNIV_SYNC_DEBUG - fprintf(file, "Mutex exits %lu, rws exits %lu, rwx exits %lu\n", + fprintf(file, "Mutex exits %llu, rws exits %llu, rwx exits %llu\n", mutex_exit_count, rw_s_exit_count, rw_x_exit_count); #endif fprintf(file, - "Mutex spin waits %lu, rounds %lu, OS waits %lu\n" - "RW-shared spins %lu, OS waits %lu;" - " RW-excl spins %lu, OS waits %lu\n", - (ulong) mutex_spin_wait_count, - (ulong) mutex_spin_round_count, - (ulong) mutex_os_wait_count, - (ulong) rw_s_spin_wait_count, - (ulong) rw_s_os_wait_count, - (ulong) rw_x_spin_wait_count, - (ulong) rw_x_os_wait_count); + "Mutex spin waits %llu, rounds %llu, OS waits %llu\n" + "RW-shared spins %llu, OS waits %llu;" + " RW-excl spins %llu, OS waits %llu\n", + mutex_spin_wait_count, + mutex_spin_round_count, + mutex_os_wait_count, + rw_s_spin_wait_count, + rw_s_os_wait_count, + rw_x_spin_wait_count, + rw_x_os_wait_count); + + fprintf(file, + "Spin rounds per wait: %.2f mutex, %.2f RW-shared, " + "%.2f RW-excl\n", + (double) mutex_spin_round_count / + (mutex_spin_wait_count ? mutex_spin_wait_count : 1), + (double) rw_s_spin_round_count / + (rw_s_spin_wait_count ? rw_s_spin_wait_count : 1), + (double) rw_x_spin_round_count / + (rw_x_spin_wait_count ? rw_x_spin_wait_count : 1)); } /***********************************************************************