diff options
Diffstat (limited to 'storage/maria/ma_pagecache.c')
| -rw-r--r-- | storage/maria/ma_pagecache.c | 5719 |
1 files changed, 5719 insertions, 0 deletions
diff --git a/storage/maria/ma_pagecache.c b/storage/maria/ma_pagecache.c new file mode 100644 index 00000000..144b10a8 --- /dev/null +++ b/storage/maria/ma_pagecache.c @@ -0,0 +1,5719 @@ +/* Copyright (C) 2000-2008 MySQL AB, 2008-2011 Monty Program Ab + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; version 2 of the License. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA */ + +/* + These functions handle page caching for Maria tables. + + One cache can handle many files. + It must contain buffers of the same blocksize. + init_pagecache() should be used to init cache handler. + + The free list (free_block_list) is a stack like structure. + When a block is freed by free_block(), it is pushed onto the stack. + When a new block is required it is first tried to pop one from the stack. + If the stack is empty, it is tried to get a never-used block from the pool. + If this is empty too, then a block is taken from the LRU ring, flushing it + to disk, if necessary. This is handled in find_block(). + With the new free list, the blocks can have three temperatures: + hot, warm and cold (which is free). This is remembered in the block header + by the enum PCBLOCK_TEMPERATURE temperature variable. Remembering the + temperature is necessary to correctly count the number of warm blocks, + which is required to decide when blocks are allowed to become hot. Whenever + a block is inserted to another (sub-)chain, we take the old and new + temperature into account to decide if we got one more or less warm block. + blocks_unused is the sum of never used blocks in the pool and of currently + free blocks. blocks_used is the number of blocks fetched from the pool and + as such gives the maximum number of in-use blocks at any time. + + TODO: Write operation locks whole cache till the end of the operation. + Should be fixed. +*/ + +#include "maria_def.h" +#include <m_string.h> +#include "ma_pagecache.h" +#include "ma_blockrec.h" +#include <my_bit.h> +#include <errno.h> + +/* + Some compilation flags have been added specifically for this module + to control the following: + - not to let a thread to yield the control when reading directly + from page cache, which might improve performance in many cases; + to enable this add: + #define SERIALIZED_READ_FROM_CACHE + - to set an upper bound for number of threads simultaneously + using the page cache; this setting helps to determine an optimal + size for hash table and improve performance when the number of + blocks in the page cache much less than the number of threads + accessing it; + to set this number equal to <N> add + #define MAX_THREADS <N> + - to substitute calls of mysql_cond_wait for calls of + mysql_cond_timedwait (wait with timeout set up); + this setting should be used only when you want to trap a deadlock + situation, which theoretically should not happen; + to set timeout equal to <T> seconds add + #define PAGECACHE_TIMEOUT <T> + - to enable the module traps and to send debug information from + page cache module to a special debug log add: + #define PAGECACHE_DEBUG + the name of this debug log file <LOG NAME> can be set through: + #define PAGECACHE_DEBUG_LOG <LOG NAME> + if the name is not defined, it's set by default; + if the PAGECACHE_DEBUG flag is not set up and we are in a debug + mode, i.e. when ! defined(DBUG_OFF), the debug information from the + module is sent to the regular debug log. + + Example of the settings: + #define SERIALIZED_READ_FROM_CACHE + #define MAX_THREADS 100 + #define PAGECACHE_TIMEOUT 1 + #define PAGECACHE_DEBUG + #define PAGECACHE_DEBUG_LOG "my_pagecache_debug.log" +*/ +#undef PAGECACHE_DEBUG +#define PAGECACHE_DEBUG_LOG "my_pagecache_debug.log" +#define _VARARGS(X) X + +/* + In key cache we have external raw locking here we use + SERIALIZED_READ_FROM_CACHE to avoid problem of reading + not consistent data from the page. + (keycache functions (key_cache_read(), key_cache_insert() and + key_cache_write()) rely on external MyISAM lock, we don't) +*/ +#define SERIALIZED_READ_FROM_CACHE yes + +#define PCBLOCK_INFO(B) \ + DBUG_PRINT("info", \ + ("block: %p fd: %lu page: %lu status: 0x%x " \ + "hshL: %p requests: %u/%u wrlocks: %u rdlocks: %u " \ + "rdlocks_q: %u pins: %u type: %s", \ + (B), \ + (ulong)((B)->hash_link ? \ + (B)->hash_link->file.file : \ + 0), \ + (ulong)((B)->hash_link ? \ + (B)->hash_link->pageno : \ + 0), \ + (uint) (B)->status, \ + (B)->hash_link, \ + (uint) (B)->requests, \ + (uint)((B)->hash_link ? \ + (B)->hash_link->requests : \ + 0), \ + (B)->wlocks, (B)->rlocks, (B)->rlocks_queue, \ + (uint)(B)->pins, \ + page_cache_page_type_str[(B)->type])) + +/* TODO: put it to my_static.c */ +my_bool my_disable_flush_pagecache_blocks= 0; + +#define STRUCT_PTR(TYPE, MEMBER, a) \ + (TYPE *) ((char *) (a) - offsetof(TYPE, MEMBER)) + +/* types of condition variables */ +#define COND_FOR_REQUESTED 0 /* queue of thread waiting for read operation */ +#define COND_FOR_SAVED 1 /* queue of thread waiting for flush */ +#define COND_FOR_WRLOCK 2 /* queue of write lock */ +#define COND_SIZE 3 /* number of COND_* queues */ + +typedef mysql_cond_t KEYCACHE_CONDVAR; + +/* descriptor of the page in the page cache block buffer */ +struct st_pagecache_page +{ + PAGECACHE_FILE file; /* file to which the page belongs to */ + pgcache_page_no_t pageno; /* number of the page in the file */ +}; + +/* element in the chain of a hash table bucket */ +struct st_pagecache_hash_link +{ + struct st_pagecache_hash_link + *next, **prev; /* to connect links in the same bucket */ + struct st_pagecache_block_link + *block; /* reference to the block for the page: */ + PAGECACHE_FILE file; /* from such a file */ + pgcache_page_no_t pageno; /* this page */ + uint requests; /* number of requests for the page */ +}; + +/* simple states of a block */ +#define PCBLOCK_ERROR 1 /* an error occurred when performing disk i/o */ +#define PCBLOCK_READ 2 /* there is an active page in the block buffer */ + +/* + A tread is reading the data to the page. + If the page contained old changed data, it will be written out with + this state set on the block. + The page is not yet ready to be used for reading. +*/ +#define PCBLOCK_IN_SWITCH 4 +/* + Block does not accept new requests for old page that would cause + the page to be pinned or written to. + (Reads that copies the block can still continue). + This state happens when another thread is waiting for readers to finish + to read data to the block (after the block, if it was changed, has been + flushed out to disk). +*/ +#define PCBLOCK_REASSIGNED 8 +#define PCBLOCK_IN_FLUSH 16 /* block is in flush operation */ +#define PCBLOCK_CHANGED 32 /* block buffer contains a dirty page */ +#define PCBLOCK_DIRECT_W 64 /* possible direct write to the block */ +#define PCBLOCK_DEL_WRITE 128 /* should be written on delete */ +#define PCBLOCK_BIG_READ 256 /* the first block of the big read in progress + or not first block which other thread wait + to be read in big read operation */ + +/* page status, returned by find_block */ +#define PAGE_READ 0 +#define PAGE_TO_BE_READ 1 +#define PAGE_WAIT_TO_BE_READ 2 + +/* block temperature determines in which (sub-)chain the block currently is */ +enum PCBLOCK_TEMPERATURE { PCBLOCK_COLD /*free*/ , PCBLOCK_WARM , PCBLOCK_HOT }; + +/* debug info */ +#ifdef DBUG_TRACE +static const char *page_cache_page_type_str[]= +{ + /* used only for control page type changing during debugging */ + "EMPTY", + "PLAIN", + "LSN", + "READ_UNKNOWN" +}; + +static const char *page_cache_page_write_mode_str[]= +{ + "DELAY", + "DONE" +}; + +static const char *page_cache_page_lock_str[]= +{ + "free -> free", + "read -> read", + "write -> write", + "free -> read", + "free -> write", + "read -> free", + "write -> free", + "write -> read" +}; + +static const char *page_cache_page_pin_str[]= +{ + "pinned -> pinned", + "unpinned -> unpinned", + "unpinned -> pinned", + "pinned -> unpinned" +}; +#endif /* DBUG_TRACE */ + +#ifndef DBUG_OFF +typedef struct st_pagecache_pin_info +{ + struct st_pagecache_pin_info *next, **prev; + struct st_my_thread_var *thread; +} PAGECACHE_PIN_INFO; + +/* + st_pagecache_lock_info structure should be kept in next, prev, thread part + compatible with st_pagecache_pin_info to be compatible in functions. +*/ + +typedef struct st_pagecache_lock_info +{ + struct st_pagecache_lock_info *next, **prev; + struct st_my_thread_var *thread; + my_bool write_lock; +} PAGECACHE_LOCK_INFO; + + +/* service functions maintain debugging info about pin & lock */ + + +/* + Links information about thread pinned/locked the block to the list + + SYNOPSIS + info_link() + list the list to link in + node the node which should be linked +*/ + +static void info_link(PAGECACHE_PIN_INFO **list, PAGECACHE_PIN_INFO *node) +{ + if ((node->next= *list)) + node->next->prev= &(node->next); + *list= node; + node->prev= list; +} + + +/* + Unlinks information about thread pinned/locked the block from the list + + SYNOPSIS + info_unlink() + node the node which should be unlinked +*/ + +static void info_unlink(PAGECACHE_PIN_INFO *node) +{ + if ((*node->prev= node->next)) + node->next->prev= node->prev; +} + + +/* + Finds information about given thread in the list of threads which + pinned/locked this block. + + SYNOPSIS + info_find() + list the list where to find the thread + thread thread ID (reference to the st_my_thread_var + of the thread) + any return any thread of the list + + RETURN + 0 - the thread was not found + pointer to the information node of the thread in the list, or, if 'any', + to any thread of the list. +*/ + +static PAGECACHE_PIN_INFO *info_find(PAGECACHE_PIN_INFO *list, + struct st_my_thread_var *thread, + my_bool any) +{ + register PAGECACHE_PIN_INFO *i= list; + if (any) + return i; + for(; i != 0; i= i->next) + if (i->thread == thread) + return i; + return 0; +} + +#endif /* !DBUG_OFF */ + +/* page cache block */ +struct st_pagecache_block_link +{ + struct st_pagecache_block_link + *next_used, **prev_used; /* to connect links in the LRU chain (ring) */ + struct st_pagecache_block_link + *next_changed, **prev_changed; /* for lists of file dirty/clean blocks */ + struct st_pagecache_hash_link + *hash_link; /* backward ptr to referring hash_link */ +#ifndef DBUG_OFF + PAGECACHE_PIN_INFO *pin_list; + PAGECACHE_LOCK_INFO *lock_list; +#endif + KEYCACHE_CONDVAR *condvar; /* condition variable for 'no readers' event */ + uchar *buffer; /* buffer for the block page */ + pthread_t write_locker; + + ulonglong last_hit_time; /* timestamp of the last hit */ + WQUEUE + wqueue[COND_SIZE]; /* queues on waiting requests for new/old pages */ + uint32 requests; /* number of requests for the block */ + uint32 pins; /* pin counter */ + uint32 wlocks; /* write locks counter */ + uint32 rlocks; /* read locks counter */ + uint32 rlocks_queue; /* rd. locks waiting wr. lock of this thread */ + uint16 status; /* state of the block */ + int16 error; /* error code for block in case of error */ + enum PCBLOCK_TEMPERATURE temperature; /* block temperature: cold, warm, hot*/ + enum pagecache_page_type type; /* type of the block */ + uint hits_left; /* number of hits left until promotion */ + /** @brief LSN when first became dirty; LSN_MAX means "not yet set" */ + LSN rec_lsn; +}; + +/** @brief information describing a run of flush_pagecache_blocks_int() */ +struct st_file_in_flush +{ + File file; + /** + @brief threads waiting for the thread currently flushing this file to be + done + */ + WQUEUE flush_queue; + /** + @brief if the thread currently flushing the file has a non-empty + first_in_switch list. + */ + my_bool first_in_switch; +}; + +#ifndef DBUG_OFF +/* debug checks */ + +#ifdef NOT_USED +static my_bool info_check_pin(PAGECACHE_BLOCK_LINK *block, + enum pagecache_page_pin mode + __attribute__((unused))) +{ + struct st_my_thread_var *thread= my_thread_var; + PAGECACHE_PIN_INFO *info= info_find(block->pin_list, thread); + DBUG_ENTER("info_check_pin"); + DBUG_PRINT("enter", ("thread: 0x%lx pin: %s", + (ulong) thread, page_cache_page_pin_str[mode])); + if (info) + { + if (mode == PAGECACHE_PIN_LEFT_UNPINNED) + { + DBUG_PRINT("info", + ("info_check_pin: thread: 0x%lx block: 0x%lx ; LEFT_UNPINNED!!!", + (ulong)thread, (ulong)block)); + DBUG_RETURN(1); + } + else if (mode == PAGECACHE_PIN) + { + DBUG_PRINT("info", + ("info_check_pin: thread: 0x%lx block: 0x%lx ; PIN!!!", + (ulong)thread, (ulong)block)); + DBUG_RETURN(1); + } + } + else + { + if (mode == PAGECACHE_PIN_LEFT_PINNED) + { + DBUG_PRINT("info", + ("info_check_pin: thread: 0x%lx block: 0x%lx ; LEFT_PINNED!!!", + (ulong)thread, (ulong)block)); + DBUG_RETURN(1); + } + else if (mode == PAGECACHE_UNPIN) + { + DBUG_PRINT("info", + ("info_check_pin: thread: 0x%lx block: 0x%lx ; UNPIN!!!", + (ulong)thread, (ulong)block)); + DBUG_RETURN(1); + } + } + DBUG_RETURN(0); +} + + +/* + Debug function which checks current lock/pin state and requested changes + + SYNOPSIS + info_check_lock() + lock requested lock changes + pin requested pin changes + + RETURN + 0 - OK + 1 - Error +*/ + +static my_bool info_check_lock(PAGECACHE_BLOCK_LINK *block, + enum pagecache_page_lock lock, + enum pagecache_page_pin pin) +{ + struct st_my_thread_var *thread= my_thread_var; + PAGECACHE_LOCK_INFO *info= + (PAGECACHE_LOCK_INFO *) info_find((PAGECACHE_PIN_INFO *) block->lock_list, + thread); + DBUG_ENTER("info_check_lock"); + switch(lock) { + case PAGECACHE_LOCK_LEFT_UNLOCKED: + if (pin != PAGECACHE_PIN_LEFT_UNPINNED || + info) + goto error; + break; + case PAGECACHE_LOCK_LEFT_READLOCKED: + if ((pin != PAGECACHE_PIN_LEFT_UNPINNED && + pin != PAGECACHE_PIN_LEFT_PINNED) || + info == 0 || info->write_lock) + goto error; + break; + case PAGECACHE_LOCK_LEFT_WRITELOCKED: + if (pin != PAGECACHE_PIN_LEFT_PINNED || + info == 0 || !info->write_lock) + goto error; + break; + case PAGECACHE_LOCK_READ: + if ((pin != PAGECACHE_PIN_LEFT_UNPINNED && + pin != PAGECACHE_PIN) || + info != 0) + goto error; + break; + case PAGECACHE_LOCK_WRITE: + if (pin != PAGECACHE_PIN || + info != 0) + goto error; + break; + case PAGECACHE_LOCK_READ_UNLOCK: + if ((pin != PAGECACHE_PIN_LEFT_UNPINNED && + pin != PAGECACHE_UNPIN) || + info == 0 || info->write_lock) + goto error; + break; + case PAGECACHE_LOCK_WRITE_UNLOCK: + if (pin != PAGECACHE_UNPIN || + info == 0 || !info->write_lock) + goto error; + break; + case PAGECACHE_LOCK_WRITE_TO_READ: + if ((pin != PAGECACHE_PIN_LEFT_PINNED && + pin != PAGECACHE_UNPIN) || + info == 0 || !info->write_lock) + goto error; + break; + } + DBUG_RETURN(0); +error: + DBUG_PRINT("info", + ("info_check_lock: thread: 0x%lx block 0x%lx: info: %d wrt: %d," + "to lock: %s, to pin: %s", + (ulong) thread, (ulong) block, MY_TEST(info), + (info ? info->write_lock : 0), + page_cache_page_lock_str[lock], + page_cache_page_pin_str[pin])); + DBUG_RETURN(1); +} +#endif /* NOT_USED */ +#endif /* !DBUG_OFF */ + +#define FLUSH_CACHE 2000 /* sort this many blocks at once */ + +static my_bool free_block(PAGECACHE *pagecache, PAGECACHE_BLOCK_LINK *block, + my_bool abort_if_pinned); +static void unlink_hash(PAGECACHE *pagecache, PAGECACHE_HASH_LINK *hash_link); +#ifndef DBUG_OFF +static void test_key_cache(PAGECACHE *pagecache, + const char *where, my_bool lock); +#endif + +#define PAGECACHE_HASH(p, f, pos) (((size_t) (pos) + \ + (size_t) (f).file) & (p->hash_entries-1)) +#define FILE_HASH(f,cache) ((uint) (f).file & (cache->changed_blocks_hash_size-1)) + +#define DEFAULT_PAGECACHE_DEBUG_LOG "pagecache_debug.log" + +#if defined(PAGECACHE_DEBUG) +static FILE *pagecache_debug_log= NULL; +static void pagecache_debug_print _VARARGS((const char *fmt, ...)); +#define PAGECACHE_DEBUG_OPEN \ + if (!pagecache_debug_log) \ + { \ + if ((pagecache_debug_log= fopen(PAGECACHE_DEBUG_LOG, "w"))) \ + (void) setvbuf(pagecache_debug_log, NULL, _IOLBF, BUFSIZ); \ + } + +#define PAGECACHE_DEBUG_CLOSE \ + if (pagecache_debug_log) \ + { \ + fclose(pagecache_debug_log); \ + pagecache_debug_log= 0; \ + } +#else +#define PAGECACHE_DEBUG_OPEN +#define PAGECACHE_DEBUG_CLOSE +#endif /* defined(PAGECACHE_DEBUG_LOG) */ + +#if defined(PAGECACHE_DEBUG) +#define KEYCACHE_PRINT(l, m) KEYCACHE_DBUG_PRINT(l,m) + +#ifdef PAGECACHE_DEBUG_DLOG +#define KEYCACHE_DBUG_PRINT(l, m) \ + { if (pagecache_debug_log) \ + { \ + fprintf(pagecache_debug_log, "%s: ", l); \ + DBUG_PRINT("PCDEBUG", ("%s: ", l)); \ + } \ + pagecache_debug_print m; } +#else +#define KEYCACHE_DBUG_PRINT(l, m) \ + { if (pagecache_debug_log) \ + fprintf(pagecache_debug_log, "%s: ", l); \ + pagecache_debug_print m; } +#endif + + +#define KEYCACHE_DBUG_ASSERT(a) \ + { if (! (a) && pagecache_debug_log) \ + fclose(pagecache_debug_log); \ + DBUG_ASSERT(a); } +#else +#define KEYCACHE_PRINT(l, m) +#define KEYCACHE_DBUG_PRINT(l, m) DBUG_PRINT(l, m) +#define KEYCACHE_DBUG_ASSERT(a) DBUG_ASSERT(a) +#endif /* defined(PAGECACHE_DEBUG) */ + +#if defined(PAGECACHE_DEBUG) || defined(DBUG_TRACE) +static my_thread_id pagecache_thread_id; +#define KEYCACHE_THREAD_TRACE(l) \ + KEYCACHE_DBUG_PRINT(l,("|thread %lld",pagecache_thread_id)) + +#define KEYCACHE_THREAD_TRACE_BEGIN(l) \ + { struct st_my_thread_var *thread_var= my_thread_var; \ + pagecache_thread_id= thread_var->id; \ + KEYCACHE_DBUG_PRINT(l,("[thread %lld",pagecache_thread_id)); \ + } + +#define KEYCACHE_THREAD_TRACE_END(l) \ + KEYCACHE_DBUG_PRINT(l,("]thread %lld",pagecache_thread_id)) +#else +#define KEYCACHE_PRINT(l,m) +#define KEYCACHE_THREAD_TRACE_BEGIN(l) +#define KEYCACHE_THREAD_TRACE_END(l) +#define KEYCACHE_THREAD_TRACE(l) +#endif /* defined(PAGECACHE_DEBUG) || defined(DBUG_TRACE) */ + +#define PCBLOCK_NUMBER(p, b) \ + ((uint) (((char*)(b)-(char *) p->block_root)/sizeof(PAGECACHE_BLOCK_LINK))) +#define PAGECACHE_HASH_LINK_NUMBER(p, h) \ + ((uint) (((char*)(h)-(char *) p->hash_link_root)/ \ + sizeof(PAGECACHE_HASH_LINK))) + +#if (defined(PAGECACHE_TIMEOUT) && !defined(_WIN32)) || defined(PAGECACHE_DEBUG) +static int pagecache_pthread_cond_wait(mysql_cond_t *cond, + mysql_mutex_t *mutex); +#else +#define pagecache_pthread_cond_wait mysql_cond_wait +#endif + +#if defined(PAGECACHE_DEBUG) +static int ___pagecache_pthread_mutex_lock(mysql_mutex_t *mutex); +static void ___pagecache_pthread_mutex_unlock(mysql_mutex_t *mutex); +static int ___pagecache_pthread_cond_signal(mysql_cond_t *cond); +#define pagecache_pthread_mutex_lock(M) \ +{ DBUG_PRINT("lock", ("mutex lock %p %u", (M), __LINE__)); \ + ___pagecache_pthread_mutex_lock(M);} +#define pagecache_pthread_mutex_unlock(M) \ +{ DBUG_PRINT("lock", ("mutex unlock %p %u", (M), __LINE__)); \ + ___pagecache_pthread_mutex_unlock(M);} +#define pagecache_pthread_cond_signal(M) \ +{ DBUG_PRINT("lock", ("signal %p %u", (M), __LINE__)); \ + ___pagecache_pthread_cond_signal(M);} +#else +#define pagecache_pthread_mutex_lock mysql_mutex_lock +#define pagecache_pthread_mutex_unlock mysql_mutex_unlock +#define pagecache_pthread_cond_signal mysql_cond_signal +#endif /* defined(PAGECACHE_DEBUG) */ + +extern my_bool translog_flush(TRANSLOG_ADDRESS lsn); + +/* + Write page to the disk + + SYNOPSIS + pagecache_fwrite() + pagecache - page cache pointer + filedesc - pagecache file descriptor structure + buffer - buffer which we will write + type - page type (plain or with LSN) + flags - MYF() flags + + RETURN + 0 - OK + 1 - Error +*/ + +static my_bool pagecache_fwrite(PAGECACHE *pagecache, + PAGECACHE_FILE *filedesc, + uchar *buffer, + pgcache_page_no_t pageno, + enum pagecache_page_type type + __attribute__((unused)), + myf flags) +{ + int res; + PAGECACHE_IO_HOOK_ARGS args; + DBUG_ENTER("pagecache_fwrite"); + DBUG_ASSERT(type != PAGECACHE_READ_UNKNOWN_PAGE); + +#ifdef EXTRA_DEBUG_BITMAP + /* + This code is very good when debugging changes in bitmaps or dirty lists + The above define should be defined for all Aria files if you want to + debug either of the above issues. + */ + + if (pagecache->extra_debug) + { + char buff[80]; + uint len= my_sprintf(buff, + (buff, "fwrite: fd: %d id: %u page: %llu", + filedesc->file, + _ma_file_callback_to_id(filedesc->callback_data), + pageno)); + (void) translog_log_debug_info(0, LOGREC_DEBUG_INFO_QUERY, + (uchar*) buff, len); + } +#endif + + /* initialize hooks args */ + args.page= buffer; + args.pageno= pageno; + args.data= filedesc->callback_data; + + /* Todo: Integrate this with write_callback so we have only one callback */ + if ((*filedesc->flush_log_callback)(&args)) + DBUG_RETURN(1); + DBUG_PRINT("info", ("pre_write_hook:%p data: %p", + filedesc->pre_write_hook, + filedesc->callback_data)); + if ((*filedesc->pre_write_hook)(&args)) + { + DBUG_PRINT("error", ("write callback problem")); + DBUG_RETURN(1); + } +#if __has_feature(memory_sanitizer) /* FIXME: encryption.aria_tiny etc. fail */ + /* FIXME: ENGINE=Aria occasionally writes uninitialized data */ + __msan_unpoison(args.page, pagecache->block_size); +#endif + res= (int)my_pwrite(filedesc->file, args.page, pagecache->block_size, + ((my_off_t) pageno << pagecache->shift), flags); + (*filedesc->post_write_hook)(res, &args); + DBUG_RETURN(res); +} + + +/* + Read page from the disk + + SYNOPSIS + pagecache_fread() + pagecache - page cache pointer + filedesc - pagecache file descriptor structure + buffer - buffer in which we will read + pageno - page number + flags - MYF() flags +*/ +#define pagecache_fread(pagecache, filedesc, buffer, pageno, flags) \ + mysql_file_pread((filedesc)->file, buffer, pagecache->block_size, \ + ((my_off_t) pageno << pagecache->shift), flags) + + +/** + @brief set rec_lsn of pagecache block (if it is needed) + + @param block block where to set rec_lsn + @param first_REDO_LSN_for_page the LSN to set +*/ + +static inline void pagecache_set_block_rec_lsn(PAGECACHE_BLOCK_LINK *block, + LSN first_REDO_LSN_for_page) +{ + if (block->rec_lsn == LSN_MAX) + block->rec_lsn= first_REDO_LSN_for_page; + else + DBUG_ASSERT(cmp_translog_addr(block->rec_lsn, + first_REDO_LSN_for_page) <= 0); +} + + +/* + next_power(value) is 2 at the power of (1+floor(log2(value))); + e.g. next_power(2)=4, next_power(3)=4. +*/ +static inline uint next_power(uint value) +{ + return (uint) my_round_up_to_next_power((uint32) value) << 1; +} + + +/* + Initialize a page cache + + SYNOPSIS + init_pagecache() + pagecache pointer to a page cache data structure + key_cache_block_size size of blocks to keep cached data + use_mem total memory to use for the key cache + division_limit division limit (may be zero) + age_threshold age threshold (may be zero) + block_size size of block (should be power of 2) + my_read_flags Flags used for all pread/pwrite calls + Usually MY_WME in case of recovery + + RETURN VALUE + number of blocks in the key cache, if successful, + 0 - otherwise. + + NOTES. + if pagecache->inited != 0 we assume that the key cache + is already initialized. This is for now used by myisamchk, but shouldn't + be something that a program should rely on! + + It's assumed that no two threads call this function simultaneously + referring to the same key cache handle. + +*/ + +size_t init_pagecache(PAGECACHE *pagecache, size_t use_mem, + uint division_limit, uint age_threshold, + uint block_size, + uint changed_blocks_hash_size, + myf my_readwrite_flags) +{ + size_t blocks, hash_links, length; + int error; + DBUG_ENTER("init_pagecache"); + DBUG_ASSERT(block_size >= 512); + + // By default we init usual cache (variables will be assigned to switch to s3) + pagecache->big_block_read= NULL; + pagecache->big_block_free= NULL; + + PAGECACHE_DEBUG_OPEN; + if (pagecache->inited && pagecache->disk_blocks > 0) + { + DBUG_PRINT("warning",("key cache already in use")); + DBUG_RETURN(0); + } + + pagecache->global_cache_w_requests= pagecache->global_cache_r_requests= 0; + pagecache->global_cache_read= pagecache->global_cache_write= 0; + pagecache->disk_blocks= -1; + if (! pagecache->inited) + { + if (mysql_mutex_init(key_PAGECACHE_cache_lock, + &pagecache->cache_lock, MY_MUTEX_INIT_FAST) || + my_hash_init(PSI_INSTRUMENT_ME, &pagecache->files_in_flush, + &my_charset_bin, 32, offsetof(struct st_file_in_flush, file), + sizeof(((struct st_file_in_flush *)NULL)->file), + NULL, NULL, 0)) + goto err; + pagecache->inited= 1; + pagecache->in_init= 0; + pagecache->resize_queue.last_thread= NULL; + } + + pagecache->mem_size= use_mem; + pagecache->block_size= block_size; + pagecache->shift= my_bit_log2_uint64(block_size); + pagecache->readwrite_flags= my_readwrite_flags | MY_NABP | MY_WAIT_IF_FULL; + pagecache->org_readwrite_flags= pagecache->readwrite_flags; + DBUG_PRINT("info", ("block_size: %u", block_size)); + DBUG_ASSERT(((uint)(1 << pagecache->shift)) == block_size); + + blocks= use_mem / (sizeof(PAGECACHE_BLOCK_LINK) + + 2 * sizeof(PAGECACHE_HASH_LINK) + + sizeof(PAGECACHE_HASH_LINK*) * + 5/4 + block_size); + /* Changed blocks hash needs to be a power of 2 */ + changed_blocks_hash_size= my_round_up_to_next_power(MY_MAX(changed_blocks_hash_size, + MIN_PAGECACHE_CHANGED_BLOCKS_HASH_SIZE)); + + /* + We need to support page cache with just one block to be able to do + scanning of rows-in-block files + */ + for ( ; ; ) + { + if (blocks < 8) + { + my_message(ENOMEM, "Not enough memory to allocate 8 pagecache pages", + MYF(0)); + my_errno= ENOMEM; + goto err; + } + /* Set my_hash_entries to the next bigger 2 power */ + if ((pagecache->hash_entries= next_power((uint)blocks)) < + (blocks) * 5/4) + pagecache->hash_entries<<= 1; + hash_links= 2 * blocks; +#if defined(MAX_THREADS) + if (hash_links < MAX_THREADS + blocks - 1) + hash_links= MAX_THREADS + blocks - 1; +#endif + while ((length= (ALIGN_SIZE(blocks * sizeof(PAGECACHE_BLOCK_LINK)) + + ALIGN_SIZE(sizeof(PAGECACHE_HASH_LINK*) * + pagecache->hash_entries) + + ALIGN_SIZE(hash_links * sizeof(PAGECACHE_HASH_LINK)) + + sizeof(PAGECACHE_BLOCK_LINK*)* (changed_blocks_hash_size*2))) + + (blocks << pagecache->shift) > use_mem && blocks > 8) + blocks--; + /* Allocate memory for cache page buffers */ + pagecache->mem_size= blocks * pagecache->block_size; + if ((pagecache->block_mem= + my_large_malloc(&pagecache->mem_size, MYF(MY_WME)))) + { + /* + Allocate memory for blocks, hash_links and hash entries; + For each block 2 hash links are allocated + */ + if (my_multi_malloc_large(PSI_INSTRUMENT_ME, MYF(MY_ZEROFILL), + &pagecache->block_root, + (ulonglong) (blocks * + sizeof(PAGECACHE_BLOCK_LINK)), + &pagecache->hash_root, + (ulonglong) (sizeof(PAGECACHE_HASH_LINK*) * + pagecache->hash_entries), + &pagecache->hash_link_root, + (ulonglong) (hash_links * + sizeof(PAGECACHE_HASH_LINK)), + &pagecache->changed_blocks, + (ulonglong) (sizeof(PAGECACHE_BLOCK_LINK*) * + changed_blocks_hash_size), + &pagecache->file_blocks, + (ulonglong) (sizeof(PAGECACHE_BLOCK_LINK*) * + changed_blocks_hash_size), + NullS)) + break; + my_large_free(pagecache->block_mem, pagecache->mem_size); + pagecache->block_mem= 0; + } + blocks= blocks / 4*3; + } + pagecache->blocks_unused= blocks; + pagecache->disk_blocks= blocks; + pagecache->hash_links= hash_links; + pagecache->hash_links_used= 0; + pagecache->free_hash_list= NULL; + pagecache->blocks_used= pagecache->blocks_changed= 0; + + pagecache->global_blocks_changed= 0; + pagecache->blocks_available=0; /* For debugging */ + + /* The LRU chain is empty after initialization */ + pagecache->used_last= NULL; + pagecache->used_ins= NULL; + pagecache->free_block_list= NULL; + pagecache->time= 0; + pagecache->warm_blocks= 0; + pagecache->min_warm_blocks= (division_limit ? + blocks * division_limit / 100 + 1 : + blocks); + pagecache->age_threshold= (age_threshold ? + blocks * age_threshold / 100 : + blocks); + pagecache->changed_blocks_hash_size= changed_blocks_hash_size; + + pagecache->cnt_for_resize_op= 0; + pagecache->resize_in_flush= 0; + pagecache->can_be_used= 1; + + pagecache->waiting_for_hash_link.last_thread= NULL; + pagecache->waiting_for_block.last_thread= NULL; + DBUG_PRINT("exit", + ("disk_blocks: %zu block_root: %p hash_entries: %zu\ + hash_root: %p hash_links: %zu hash_link_root: %p", + (size_t)pagecache->disk_blocks, pagecache->block_root, + pagecache->hash_entries, pagecache->hash_root, + (size_t)pagecache->hash_links, pagecache->hash_link_root)); + + pagecache->blocks= pagecache->disk_blocks > 0 ? pagecache->disk_blocks : 0; + DBUG_RETURN((size_t)pagecache->disk_blocks); + +err: + error= my_errno; + pagecache->disk_blocks= 0; + pagecache->blocks= 0; + if (pagecache->block_mem) + { + my_large_free(pagecache->block_mem, pagecache->mem_size); + pagecache->block_mem= NULL; + } + if (pagecache->block_root) + { + my_free(pagecache->block_root); + pagecache->block_root= NULL; + } + my_errno= error; + pagecache->can_be_used= 0; + DBUG_RETURN(0); +} + + +/* + Flush all blocks in the key cache to disk +*/ + +#ifdef NOT_USED +static int flush_all_key_blocks(PAGECACHE *pagecache) +{ +#if defined(PAGECACHE_DEBUG) + uint cnt=0; +#endif + while (pagecache->blocks_changed > 0) + { + PAGECACHE_BLOCK_LINK *block; + for (block= pagecache->used_last->next_used ; ; block=block->next_used) + { + if (block->hash_link) + { +#if defined(PAGECACHE_DEBUG) + cnt++; + KEYCACHE_DBUG_ASSERT(cnt <= pagecache->blocks_used); +#endif + if (flush_pagecache_blocks_int(pagecache, &block->hash_link->file, + FLUSH_RELEASE, NULL, NULL)) + return 1; + break; + } + if (block == pagecache->used_last) + break; + } + } + return 0; +} +#endif /* NOT_USED */ + +/* + Resize a key cache + + SYNOPSIS + resize_pagecache() + pagecache pointer to a page cache data structure + use_mem total memory to use for the new key cache + division_limit new division limit (if not zero) + age_threshold new age threshold (if not zero) + + RETURN VALUE + number of blocks in the key cache, if successful, + 0 - otherwise. + + NOTES. + The function first compares the memory size parameter + with the key cache value. + + If they differ the function free the the memory allocated for the + old key cache blocks by calling the end_pagecache function and + then rebuilds the key cache with new blocks by calling + init_key_cache. + + The function starts the operation only when all other threads + performing operations with the key cache let her to proceed + (when cnt_for_resize=0). + + Before being usable, this function needs: + - to receive fixes for BUG#17332 "changing key_buffer_size on a running + server can crash under load" similar to those done to the key cache + - to have us (Sanja) look at the additional constraints placed on + resizing, due to the page locking specific to this page cache. + So we disable it for now. +*/ +#ifdef NOT_USED /* keep disabled until code is fixed see above !! */ +size_t resize_pagecache(PAGECACHE *pagecache, + size_t use_mem, uint division_limit, + uint age_threshold, uint changed_blocks_hash_size) +{ + size_t blocks; + struct st_my_thread_var *thread; + WQUEUE *wqueue; + DBUG_ENTER("resize_pagecache"); + + if (!pagecache->inited) + DBUG_RETURN(pagecache->disk_blocks); + + if(use_mem == pagecache->mem_size) + { + change_pagecache_param(pagecache, division_limit, age_threshold); + DBUG_RETURN(pagecache->disk_blocks); + } + + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + + wqueue= &pagecache->resize_queue; + thread= my_thread_var; + wqueue_link_into_queue(wqueue, thread); + + while (wqueue->last_thread->next != thread) + { + pagecache_pthread_cond_wait(&thread->suspend, &pagecache->cache_lock); + } + + pagecache->resize_in_flush= 1; + if (flush_all_key_blocks(pagecache)) + { + /* TODO: if this happens, we should write a warning in the log file ! */ + pagecache->resize_in_flush= 0; + blocks= 0; + pagecache->can_be_used= 0; + goto finish; + } + pagecache->resize_in_flush= 0; + pagecache->can_be_used= 0; + while (pagecache->cnt_for_resize_op) + { + DBUG_PRINT("wait", ("suspend thread %s %ld", thread->name, thread->id)); + pagecache_pthread_cond_wait(&thread->suspend, &pagecache->cache_lock); + } + + end_pagecache(pagecache, 0); /* Don't free mutex */ + /* The following will work even if use_mem is 0 */ + blocks= init_pagecache(pagecache, pagecache->block_size, use_mem, + division_limit, age_threshold, changed_blocks_hash_size, + pagecache->readwrite_flags); + +finish: + wqueue_unlink_from_queue(wqueue, thread); + /* Signal for the next resize request to proceeed if any */ + if (wqueue->last_thread) + { + DBUG_PRINT("signal", + ("thread %s %ld", wqueue->last_thread->next->name, + wqueue->last_thread->next->id)); + pagecache_pthread_cond_signal(&wqueue->last_thread->next->suspend); + } + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + DBUG_RETURN(blocks); +} +#endif /* 0 */ + + +/* + Increment counter blocking resize key cache operation +*/ +static inline void inc_counter_for_resize_op(PAGECACHE *pagecache) +{ + mysql_mutex_assert_owner(&pagecache->cache_lock); + pagecache->cnt_for_resize_op++; +} + + +/* + Decrement counter blocking resize key cache operation; + Signal the operation to proceed when counter becomes equal zero +*/ + +static inline void dec_counter_for_resize_op(PAGECACHE *pagecache) +{ + struct st_my_thread_var *last_thread; + mysql_mutex_assert_owner(&pagecache->cache_lock); + if (!--pagecache->cnt_for_resize_op && + (last_thread= pagecache->resize_queue.last_thread)) + { + DBUG_PRINT("signal", + ("thread %s %ld", last_thread->next->name, + (ulong) last_thread->next->id)); + pagecache_pthread_cond_signal(&last_thread->next->suspend); + } +} + +/* + Change the page cache parameters + + SYNOPSIS + change_pagecache_param() + pagecache pointer to a page cache data structure + division_limit new division limit (if not zero) + age_threshold new age threshold (if not zero) + + RETURN VALUE + none + + NOTES. + Presently the function resets the key cache parameters + concerning midpoint insertion strategy - division_limit and + age_threshold. +*/ + +void change_pagecache_param(PAGECACHE *pagecache, uint division_limit, + uint age_threshold) +{ + DBUG_ENTER("change_pagecache_param"); + + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + if (division_limit) + pagecache->min_warm_blocks= (pagecache->disk_blocks * + division_limit / 100 + 1); + if (age_threshold) + pagecache->age_threshold= (pagecache->disk_blocks * + age_threshold / 100); + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + DBUG_VOID_RETURN; +} + + +/* + Check that pagecache was used and cleaned up properly. +*/ + +#ifndef DBUG_OFF +void check_pagecache_is_cleaned_up(PAGECACHE *pagecache) +{ + DBUG_ENTER("check_pagecache_is_cleaned_up"); + /* + Ensure we called inc_counter_for_resize_op and dec_counter_for_resize_op + the same number of times. (If not, a resize() could never happen. + */ + DBUG_ASSERT(pagecache->cnt_for_resize_op == 0); + + if (pagecache->disk_blocks > 0) + { + if (pagecache->block_mem) + { + uint i; + for (i=0 ; i < pagecache->blocks_used ; i++) + { + DBUG_ASSERT(pagecache->block_root[i].status == 0); + DBUG_ASSERT(pagecache->block_root[i].type == PAGECACHE_EMPTY_PAGE); + } + } + } + DBUG_VOID_RETURN; +} +#endif + + +/* + Removes page cache from memory. Does NOT flush pages to disk. + + SYNOPSIS + end_pagecache() + pagecache page cache handle + cleanup Complete free (Free also mutex for key cache) + + RETURN VALUE + none +*/ + +void end_pagecache(PAGECACHE *pagecache, my_bool cleanup) +{ + DBUG_ENTER("end_pagecache"); + DBUG_PRINT("enter", ("key_cache: %p", pagecache)); + + if (!pagecache->inited) + DBUG_VOID_RETURN; + + if (pagecache->disk_blocks > 0) + { +#ifndef DBUG_OFF + check_pagecache_is_cleaned_up(pagecache); +#endif + + if (pagecache->block_mem) + { + my_large_free(pagecache->block_mem, pagecache->mem_size); + pagecache->block_mem= NULL; + my_free(pagecache->block_root); + pagecache->block_root= NULL; + } + pagecache->disk_blocks= -1; + /* Reset blocks_changed to be safe if flush_all_key_blocks is called */ + pagecache->blocks_changed= 0; + } + + DBUG_PRINT("status", ("used: %zu changed: %zu w_requests: %llu " + "writes: %llu r_requests: %llu reads: %llu", + pagecache->blocks_used, + pagecache->global_blocks_changed, + pagecache->global_cache_w_requests, + pagecache->global_cache_write, + pagecache->global_cache_r_requests, + pagecache->global_cache_read)); + + if (cleanup) + { + my_hash_free(&pagecache->files_in_flush); + mysql_mutex_destroy(&pagecache->cache_lock); + pagecache->inited= pagecache->can_be_used= 0; + PAGECACHE_DEBUG_CLOSE; + } + DBUG_VOID_RETURN; +} /* end_pagecache */ + + +/* + Unlink a block from the chain of dirty/clean blocks +*/ + +static inline void unlink_changed(PAGECACHE_BLOCK_LINK *block) +{ + if (block->next_changed) + block->next_changed->prev_changed= block->prev_changed; + *block->prev_changed= block->next_changed; +} + + +/* + Link a block into the chain of dirty/clean blocks +*/ + +static inline void link_changed(PAGECACHE_BLOCK_LINK *block, + PAGECACHE_BLOCK_LINK **phead) +{ + block->prev_changed= phead; + if ((block->next_changed= *phead)) + (*phead)->prev_changed= &block->next_changed; + *phead= block; +} + + +/* + Unlink a block from the chain of dirty/clean blocks, if it's asked for, + and link it to the chain of clean blocks for the specified file +*/ + +static void link_to_file_list(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block, + PAGECACHE_FILE *file, my_bool unlink_flag) +{ + if (unlink_flag) + unlink_changed(block); + link_changed(block, &pagecache->file_blocks[FILE_HASH(*file, pagecache)]); + if (block->status & PCBLOCK_CHANGED) + { + block->status&= ~(PCBLOCK_CHANGED | PCBLOCK_DEL_WRITE); + block->rec_lsn= LSN_MAX; + pagecache->blocks_changed--; + pagecache->global_blocks_changed--; + } +} + + +/* + Unlink a block from the chain of clean blocks for the specified + file and link it to the chain of dirty blocks for this file +*/ + +static inline void link_to_changed_list(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block) +{ + unlink_changed(block); + link_changed(block, + &pagecache->changed_blocks[FILE_HASH(block->hash_link->file, pagecache)]); + block->status|=PCBLOCK_CHANGED; + pagecache->blocks_changed++; + pagecache->global_blocks_changed++; +} + + +/* + Link a block to the LRU chain at the beginning or at the end of + one of two parts. + + SYNOPSIS + link_block() + pagecache pointer to a page cache data structure + block pointer to the block to link to the LRU chain + hot <-> to link the block into the hot subchain + at_end <-> to link the block at the end of the subchain + + RETURN VALUE + none + + NOTES. + The LRU chain is represented by a circular list of block structures. + The list is double-linked of the type (**prev,*next) type. + The LRU chain is divided into two parts - hot and warm. + There are two pointers to access the last blocks of these two + parts. The beginning of the warm part follows right after the + end of the hot part. + Only blocks of the warm part can be used for replacement. + The first block from the beginning of this subchain is always + taken for eviction (pagecache->last_used->next) + + LRU chain: +------+ H O T +------+ + +----| end |----...<----| beg |----+ + | +------+last +------+ | + v<-link in latest hot (new end) | + | link in latest warm (new end)->^ + | +------+ W A R M +------+ | + +----| beg |---->...----| end |----+ + +------+ +------+ins + first for eviction +*/ + +static void link_block(PAGECACHE *pagecache, PAGECACHE_BLOCK_LINK *block, + my_bool hot, my_bool at_end) +{ + PAGECACHE_BLOCK_LINK *ins; + PAGECACHE_BLOCK_LINK **ptr_ins; + DBUG_ENTER("link_block"); + + PCBLOCK_INFO(block); + KEYCACHE_DBUG_ASSERT(! (block->hash_link && block->hash_link->requests)); + if (!hot && pagecache->waiting_for_block.last_thread) + { + /* Signal that in the LRU warm sub-chain an available block has appeared */ + struct st_my_thread_var *last_thread= + pagecache->waiting_for_block.last_thread; + struct st_my_thread_var *first_thread= last_thread->next; + struct st_my_thread_var *next_thread= first_thread; + PAGECACHE_HASH_LINK *hash_link= + (PAGECACHE_HASH_LINK *) first_thread->keycache_link; + struct st_my_thread_var *thread; + + DBUG_ASSERT(block->requests + block->wlocks + block->rlocks + + block->pins == 0); + DBUG_ASSERT(block->next_used == NULL); + + do + { + thread= next_thread; + next_thread= thread->next; + /* + We notify about the event all threads that ask + for the same page as the first thread in the queue + */ + if ((PAGECACHE_HASH_LINK *) thread->keycache_link == hash_link) + { + DBUG_PRINT("signal", ("thread: %s %ld", thread->name, + (ulong) thread->id)); + pagecache_pthread_cond_signal(&thread->suspend); + wqueue_unlink_from_queue(&pagecache->waiting_for_block, thread); + block->requests++; + } + } + while (thread != last_thread); + DBUG_PRINT("hash", ("hash_link (link block): %p, hash_link: %p -> %p", + hash_link, hash_link->block, block)); + hash_link->block= block; + /* Ensure that no other thread tries to use this block */ + block->status|= PCBLOCK_REASSIGNED; + + DBUG_PRINT("signal", ("after signal")); +#if defined(PAGECACHE_DEBUG) + KEYCACHE_DBUG_PRINT("link_block", + ("linked,unlinked block: %u status: %x #requests: %u #available: %u", + PCBLOCK_NUMBER(pagecache, block), block->status, + block->requests, pagecache->blocks_available)); +#endif + DBUG_VOID_RETURN; + } + ptr_ins= hot ? &pagecache->used_ins : &pagecache->used_last; + ins= *ptr_ins; + if (ins) + { + ins->next_used->prev_used= &block->next_used; + block->next_used= ins->next_used; + block->prev_used= &ins->next_used; + ins->next_used= block; + if (at_end) + *ptr_ins= block; + } + else + { + /* The LRU chain is empty */ + pagecache->used_last= pagecache->used_ins= block->next_used= block; + block->prev_used= &block->next_used; + } + KEYCACHE_THREAD_TRACE("link_block"); +#if defined(PAGECACHE_DEBUG) + pagecache->blocks_available++; + KEYCACHE_DBUG_PRINT("link_block", + ("linked block: %u:%1u status: %x #requests: %u #available: %u", + PCBLOCK_NUMBER(pagecache, block), at_end, block->status, + block->requests, pagecache->blocks_available)); + KEYCACHE_DBUG_ASSERT(pagecache->blocks_available <= + pagecache->blocks_used); +#endif + DBUG_VOID_RETURN; +} + + +/* + Unlink a block from the LRU chain + + SYNOPSIS + unlink_block() + pagecache pointer to a page cache data structure + block pointer to the block to unlink from the LRU chain + + RETURN VALUE + none + + NOTES. + See NOTES for link_block +*/ + +static void unlink_block(PAGECACHE *pagecache, PAGECACHE_BLOCK_LINK *block) +{ + DBUG_ENTER("unlink_block"); + DBUG_PRINT("pagecache", ("unlink %p", block)); + DBUG_ASSERT(block->next_used != NULL); + if (block->next_used == block) + { + /* The list contains only one member */ + pagecache->used_last= pagecache->used_ins= NULL; + } + else + { + block->next_used->prev_used= block->prev_used; + *block->prev_used= block->next_used; + if (pagecache->used_last == block) + pagecache->used_last= STRUCT_PTR(PAGECACHE_BLOCK_LINK, + next_used, block->prev_used); + if (pagecache->used_ins == block) + pagecache->used_ins= STRUCT_PTR(PAGECACHE_BLOCK_LINK, + next_used, block->prev_used); + } + block->next_used= NULL; + + KEYCACHE_THREAD_TRACE("unlink_block"); +#if defined(PAGECACHE_DEBUG) + KEYCACHE_DBUG_ASSERT(pagecache->blocks_available != 0); + pagecache->blocks_available--; + KEYCACHE_DBUG_PRINT("pagecache", + ("unlinked block: %p (%u) status: %x #requests: %u #available: %u", + block, PCBLOCK_NUMBER(pagecache, block), + block->status, + block->requests, pagecache->blocks_available)); + PCBLOCK_INFO(block); +#endif + DBUG_VOID_RETURN; +} + + +/* + Register requests for a block + + SYNOPSIS + reg_requests() + pagecache this page cache reference + block the block we request reference + count how many requests we register (it is 1 everywhere) + + NOTE + Registration of request means we are going to use this block so we exclude + it from the LRU if it is first request +*/ +static void reg_requests(PAGECACHE *pagecache, PAGECACHE_BLOCK_LINK *block, + int count) +{ + DBUG_ENTER("reg_requests"); + PCBLOCK_INFO(block); + if (! block->requests) + /* First request for the block unlinks it */ + unlink_block(pagecache, block); + block->requests+= count; + DBUG_VOID_RETURN; +} + + +/* + Unregister request for a block + linking it to the LRU chain if it's the last request + + SYNOPSIS + unreg_request() + pagecache pointer to a page cache data structure + block pointer to the block to link to the LRU chain + at_end <-> to link the block at the end of the LRU chain + + RETURN VALUE + none + + NOTES. + Every linking to the LRU chain decrements by one a special block + counter (if it's positive). If the at_end parameter is TRUE the block is + added either at the end of warm sub-chain or at the end of hot sub-chain. + It is added to the hot subchain if its counter is zero and number of + blocks in warm sub-chain is not less than some low limit (determined by + the division_limit parameter). Otherwise the block is added to the warm + sub-chain. If the at_end parameter is FALSE the block is always added + at beginning of the warm sub-chain. + Thus a warm block can be promoted to the hot sub-chain when its counter + becomes zero for the first time. + At the same time the block at the very beginning of the hot subchain + might be moved to the beginning of the warm subchain if it stays untouched + for a too long time (this time is determined by parameter age_threshold). +*/ + +static void unreg_request(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block, int at_end) +{ + DBUG_ENTER("unreg_request"); + DBUG_PRINT("enter", ("block %p (%u) status: %x requests: %u", + block, PCBLOCK_NUMBER(pagecache, block), + block->status, block->requests)); + PCBLOCK_INFO(block); + DBUG_ASSERT(block->requests > 0); + if (! --block->requests) + { + my_bool hot; + if (block->hits_left) + block->hits_left--; + hot= !block->hits_left && at_end && + pagecache->warm_blocks > pagecache->min_warm_blocks; + if (hot) + { + if (block->temperature == PCBLOCK_WARM) + pagecache->warm_blocks--; + block->temperature= PCBLOCK_HOT; + KEYCACHE_DBUG_PRINT("unreg_request", ("#warm_blocks: %zu", + pagecache->warm_blocks)); + } + link_block(pagecache, block, hot, (my_bool)at_end); + block->last_hit_time= pagecache->time; + pagecache->time++; + + block= pagecache->used_ins; + /* Check if we should link a hot block to the warm block */ + if (block && pagecache->time - block->last_hit_time > + pagecache->age_threshold) + { + unlink_block(pagecache, block); + link_block(pagecache, block, 0, 0); + if (block->temperature != PCBLOCK_WARM) + { + pagecache->warm_blocks++; + block->temperature= PCBLOCK_WARM; + } + KEYCACHE_DBUG_PRINT("unreg_request", ("#warm_blocks: %zu", + pagecache->warm_blocks)); + } + } + DBUG_VOID_RETURN; +} + +/* + Remove a reader of the page in block +*/ + +static inline void remove_reader(PAGECACHE_BLOCK_LINK *block) +{ + DBUG_ENTER("remove_reader"); + PCBLOCK_INFO(block); + DBUG_ASSERT(block->hash_link->requests > 0); + if (! --block->hash_link->requests && block->condvar) + pagecache_pthread_cond_signal(block->condvar); + DBUG_VOID_RETURN; +} + + +/* + Wait until the last reader of the page in block + signals on its termination +*/ + +static inline void wait_for_readers(PAGECACHE *pagecache + __attribute__((unused)), + PAGECACHE_BLOCK_LINK *block + __attribute__((unused))) +{ + struct st_my_thread_var *thread= my_thread_var; + DBUG_ASSERT(block->condvar == NULL); + while (block->hash_link->requests) + { + DBUG_ENTER("wait_for_readers"); + DBUG_PRINT("wait", + ("suspend thread: %s %ld block: %u", + thread->name, (ulong) thread->id, + PCBLOCK_NUMBER(pagecache, block))); + block->condvar= &thread->suspend; + pagecache_pthread_cond_wait(&thread->suspend, &pagecache->cache_lock); + block->condvar= NULL; + DBUG_VOID_RETURN; + } +} + + +/* + Wait until the flush of the page is done. +*/ + +static void wait_for_flush(PAGECACHE *pagecache + __attribute__((unused)), + PAGECACHE_BLOCK_LINK *block + __attribute__((unused))) +{ + struct st_my_thread_var *thread= my_thread_var; + DBUG_ENTER("wait_for_flush"); + wqueue_add_to_queue(&block->wqueue[COND_FOR_SAVED], thread); + do + { + DBUG_PRINT("wait", + ("suspend thread %s %ld", thread->name, (ulong) thread->id)); + pagecache_pthread_cond_wait(&thread->suspend, + &pagecache->cache_lock); + } + while(thread->next); + DBUG_VOID_RETURN; +} + + +/* + Add a hash link to a bucket in the hash_table +*/ + +static inline void link_hash(PAGECACHE_HASH_LINK **start, + PAGECACHE_HASH_LINK *hash_link) +{ + if (*start) + (*start)->prev= &hash_link->next; + hash_link->next= *start; + hash_link->prev= start; + *start= hash_link; +} + + +/* + Remove a hash link from the hash table +*/ + +static void unlink_hash(PAGECACHE *pagecache, PAGECACHE_HASH_LINK *hash_link) +{ + DBUG_ENTER("unlink_hash"); + DBUG_PRINT("enter", ("hash_link: %p block: %p fd: %u pos: %lu requests: %u", + hash_link, hash_link->block, (uint) hash_link->file.file, + (ulong) hash_link->pageno, + hash_link->requests)); + DBUG_ASSERT(hash_link->requests == 0); + DBUG_ASSERT(!hash_link->block || hash_link->block->pins == 0); + + if ((*hash_link->prev= hash_link->next)) + hash_link->next->prev= hash_link->prev; + + hash_link->block= NULL; + if (pagecache->waiting_for_hash_link.last_thread) + { + /* Signal that a free hash link has appeared */ + struct st_my_thread_var *last_thread= + pagecache->waiting_for_hash_link.last_thread; + struct st_my_thread_var *first_thread= last_thread->next; + struct st_my_thread_var *next_thread= first_thread; + PAGECACHE_PAGE *first_page= (PAGECACHE_PAGE *) (first_thread->keycache_link); + struct st_my_thread_var *thread; + + hash_link->file= first_page->file; + DBUG_ASSERT(first_page->pageno < ((1ULL) << 40)); + hash_link->pageno= first_page->pageno; + do + { + PAGECACHE_PAGE *page; + thread= next_thread; + page= (PAGECACHE_PAGE *) thread->keycache_link; + next_thread= thread->next; + /* + We notify about the event all threads that ask + for the same page as the first thread in the queue + */ + if (page->file.file == hash_link->file.file && + page->pageno == hash_link->pageno) + { + DBUG_PRINT("signal", ("thread %s %ld", thread->name, + (ulong) thread->id)); + pagecache_pthread_cond_signal(&thread->suspend); + wqueue_unlink_from_queue(&pagecache->waiting_for_hash_link, thread); + } + } + while (thread != last_thread); + + /* + Add this to the hash, so that the waiting threads can find it + when they retry the call to get_hash_link(). This entry is special + in that it has no associated block. + */ + link_hash(&pagecache->hash_root[PAGECACHE_HASH(pagecache, + hash_link->file, + hash_link->pageno)], + hash_link); + DBUG_VOID_RETURN; + } + + /* Add hash to free hash list */ + hash_link->next= pagecache->free_hash_list; + pagecache->free_hash_list= hash_link; + DBUG_VOID_RETURN; +} + + +/* + Get the hash link for the page if it is in the cache (do not put the + page in the cache if it is absent there) + + SYNOPSIS + get_present_hash_link() + pagecache Pagecache reference + file file ID + pageno page number in the file + start where to put pointer to found hash bucket (for + direct referring it) + + RETURN + found hashlink pointer +*/ + +static PAGECACHE_HASH_LINK *get_present_hash_link(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + pgcache_page_no_t pageno, + PAGECACHE_HASH_LINK ***start) +{ + reg1 PAGECACHE_HASH_LINK *hash_link; +#if defined(PAGECACHE_DEBUG) + int cnt; +#endif + DBUG_ENTER("get_present_hash_link"); + DBUG_PRINT("enter", ("fd: %u pos: %lu", (uint) file->file, (ulong) pageno)); + + /* + Find the bucket in the hash table for the pair (file, pageno); + start contains the head of the bucket list, + hash_link points to the first member of the list + */ + hash_link= *(*start= &pagecache->hash_root[PAGECACHE_HASH(pagecache, + *file, pageno)]); +#if defined(PAGECACHE_DEBUG) + cnt= 0; +#endif + /* Look for an element for the pair (file, pageno) in the bucket chain */ + while (hash_link && + (hash_link->pageno != pageno || + hash_link->file.file != file->file)) + { + hash_link= hash_link->next; +#if defined(PAGECACHE_DEBUG) + cnt++; + if (! (cnt <= pagecache->hash_links_used)) + { + int i; + for (i=0, hash_link= **start ; + i < cnt ; i++, hash_link= hash_link->next) + { + KEYCACHE_DBUG_PRINT("get_present_hash_link", ("fd: %u pos: %lu", + (uint) hash_link->file.file, (ulong) hash_link->pageno)); + } + } + KEYCACHE_DBUG_ASSERT(cnt <= pagecache->hash_links_used); +#endif + } + if (hash_link) + { + DBUG_PRINT("exit", ("hash_link: %p", hash_link)); + /* Register the request for the page */ + hash_link->requests++; + } + /* + As soon as the caller will release the page cache's lock, "hash_link" + will be potentially obsolete (unusable) information. + */ + DBUG_RETURN(hash_link); +} + + +/* + Get the hash link for a page +*/ + +static PAGECACHE_HASH_LINK *get_hash_link(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + pgcache_page_no_t pageno) +{ + reg1 PAGECACHE_HASH_LINK *hash_link; + PAGECACHE_HASH_LINK **start; + DBUG_ENTER("get_hash_link"); + +restart: + /* try to find the page in the cache */ + hash_link= get_present_hash_link(pagecache, file, pageno, + &start); + if (!hash_link) + { + /* There is no hash link in the hash table for the pair (file, pageno) */ + if (pagecache->free_hash_list) + { + DBUG_PRINT("info", ("free_hash_list: %p free_hash_list->next: %p", + pagecache->free_hash_list, + pagecache->free_hash_list->next)); + hash_link= pagecache->free_hash_list; + pagecache->free_hash_list= hash_link->next; + } + else if (pagecache->hash_links_used < pagecache->hash_links) + { + hash_link= &pagecache->hash_link_root[pagecache->hash_links_used++]; + } + else + { + /* Wait for a free hash link */ + struct st_my_thread_var *thread= my_thread_var; + PAGECACHE_PAGE page; + page.file= *file; + page.pageno= pageno; + thread->keycache_link= (void *) &page; + wqueue_link_into_queue(&pagecache->waiting_for_hash_link, thread); + DBUG_PRINT("wait", + ("suspend thread %s %ld", thread->name, (ulong) thread->id)); + pagecache_pthread_cond_wait(&thread->suspend, + &pagecache->cache_lock); + thread->keycache_link= NULL; + DBUG_PRINT("thread", ("restarting...")); + goto restart; + } + hash_link->file= *file; + DBUG_ASSERT(pageno < ((1ULL) << 40)); + hash_link->pageno= pageno; + link_hash(start, hash_link); + /* Register the request for the page */ + hash_link->requests++; + DBUG_ASSERT(hash_link->block == 0); + DBUG_ASSERT(hash_link->requests == 1); + } + else + { + /* + We have to copy the flush_log callback, as it may change if the table + goes from non_transactional to transactional during recovery + */ + hash_link->file.flush_log_callback= file->flush_log_callback; + } + DBUG_PRINT("exit", ("hash_link: %p block: %p", hash_link, + hash_link->block)); + DBUG_RETURN(hash_link); +} + + +/* + Get a block for the file page requested by a pagecache read/write operation; + If the page is not in the cache return a free block, if there is none + return the lru block after saving its buffer if the page is dirty. + + SYNOPSIS + + find_block() + pagecache pointer to a page cache data structure + file handler for the file to read page from + pageno number of the page in the file + init_hits_left how initialize the block counter for the page + wrmode <-> get for writing + block_is_copied 1 if block will be copied from page cache under + the pagelock mutex. + reg_req Register request to the page. Normally all pages + should be registered; The only time it's ok to + not register a page is when the page is already + pinned (and thus registered) by the same thread. + page_st out {PAGE_READ,PAGE_TO_BE_READ,PAGE_WAIT_TO_BE_READ} + + RETURN VALUE + Pointer to the found block if successful, 0 - otherwise + + NOTES. + For the page from file positioned at pageno the function checks whether + the page is in the key cache specified by the first parameter. + If this is the case it immediately returns the block. + If not, the function first chooses a block for this page. If there is + no not used blocks in the key cache yet, the function takes the block + at the very beginning of the warm sub-chain. It saves the page in that + block if it's dirty before returning the pointer to it. + The function returns in the page_st parameter the following values: + PAGE_READ - if page already in the block, + PAGE_TO_BE_READ - if it is to be read yet by the current thread + WAIT_TO_BE_READ - if it is to be read by another thread + If an error occurs THE PCBLOCK_ERROR bit is set in the block status. + It might happen that there are no blocks in LRU chain (in warm part) - + all blocks are unlinked for some read/write operations. Then the function + waits until first of this operations links any block back. +*/ + +static PAGECACHE_BLOCK_LINK *find_block(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + pgcache_page_no_t pageno, + int init_hits_left, + my_bool wrmode, + my_bool block_is_copied, + my_bool reg_req, + my_bool fast, + int *page_st) +{ + PAGECACHE_HASH_LINK *hash_link; + PAGECACHE_BLOCK_LINK *block; + int error= 0; + int page_status; + DBUG_ENTER("find_block"); + DBUG_PRINT("enter", ("fd: %d pos: %lu wrmode: %d block_is_copied: %d", + file->file, (ulong) pageno, wrmode, block_is_copied)); + KEYCACHE_PRINT("find_block", ("fd: %d pos: %lu wrmode: %d", + file->file, (ulong) pageno, + wrmode)); +#if !defined(DBUG_OFF) && defined(EXTRA_DEBUG) + DBUG_EXECUTE("check_pagecache", + test_key_cache(pagecache, "start of find_block", 0);); +#endif + DBUG_ASSERT(!fast || !wrmode); + +restart: + /* Find the hash link for the requested page (file, pageno) */ + hash_link= get_hash_link(pagecache, file, pageno); + + page_status= -1; + if ((block= hash_link->block) && + block->hash_link == hash_link && (block->status & PCBLOCK_READ)) + page_status= PAGE_READ; + + if (wrmode && pagecache->resize_in_flush) + { + /* This is a write request during the flush phase of a resize operation */ + + if (page_status != PAGE_READ) + { + /* We don't need the page in the cache: we are going to write on disk */ + DBUG_ASSERT(hash_link->requests > 0); + hash_link->requests--; + unlink_hash(pagecache, hash_link); + return 0; + } + if (!(block->status & PCBLOCK_IN_FLUSH)) + { + DBUG_ASSERT(hash_link->requests > 0); + hash_link->requests--; + /* + Remove block to invalidate the page in the block buffer + as we are going to write directly on disk. + Although we have an exclusive lock for the updated key part + the control can be yielded by the current thread as we might + have unfinished readers of other key parts in the block + buffer. Still we are guaranteed not to have any readers + of the key part we are writing into until the block is + removed from the cache as we set the PCBLOCK_REASSIGNED + flag (see the code below that handles reading requests). + */ + free_block(pagecache, block, 0); + return 0; + } + /* Wait until the page is flushed on disk */ + DBUG_ASSERT(hash_link->requests > 0); + hash_link->requests--; + wait_for_flush(pagecache, block); + + /* Invalidate page in the block if it has not been done yet */ + DBUG_ASSERT(block->status); /* Should always be true */ + if (block->status) + free_block(pagecache, block, 0); + return 0; + } + + if (page_status == PAGE_READ && + (block->status & (PCBLOCK_IN_SWITCH | PCBLOCK_REASSIGNED))) + { + /* This is a request for a page to be removed from cache */ + + KEYCACHE_DBUG_PRINT("find_block", + ("request for old page in block: %u " + "wrmode: %d block->status: %d", + PCBLOCK_NUMBER(pagecache, block), wrmode, + block->status)); + /* + Only reading requests can proceed until the old dirty page is flushed, + all others are to be suspended, then resubmitted + */ + if (!wrmode && block_is_copied && !(block->status & PCBLOCK_REASSIGNED)) + { + if (reg_req) + reg_requests(pagecache, block, 1); + } + else + { + /* + When we come here either PCBLOCK_REASSIGNED or PCBLOCK_IN_SWITCH are + active. In both cases wqueue_release_queue() is called when the + state changes. + */ + DBUG_ASSERT(block->hash_link == hash_link); + remove_reader(block); + KEYCACHE_DBUG_PRINT("find_block", + ("request waiting for old page to be saved")); + { + struct st_my_thread_var *thread= my_thread_var; + /* Put the request into the queue of those waiting for the old page */ + wqueue_add_to_queue(&block->wqueue[COND_FOR_SAVED], thread); + /* Wait until the request can be resubmitted */ + do + { + DBUG_PRINT("wait", + ("suspend thread %s %ld", thread->name, + (ulong) thread->id)); + pagecache_pthread_cond_wait(&thread->suspend, + &pagecache->cache_lock); + } + while(thread->next); + } + KEYCACHE_DBUG_PRINT("find_block", + ("request for old page resubmitted")); + DBUG_PRINT("info", ("restarting...")); + /* Resubmit the request */ + goto restart; + } + } + else + { + /* This is a request for a new page or for a page not to be removed */ + if (! block) + { + DBUG_PRINT("info", ("request for a new page")); + /* No block is assigned for the page yet */ + if (pagecache->blocks_unused) + { + DBUG_PRINT("info", ("there is never used blocks")); + if (pagecache->free_block_list) + { + /* There is a block in the free list. */ + block= pagecache->free_block_list; + pagecache->free_block_list= block->next_used; + block->next_used= NULL; + } + else + { + /* There are some never used blocks, take first of them */ + block= &pagecache->block_root[pagecache->blocks_used]; + block->buffer= ADD_TO_PTR(pagecache->block_mem, + (pagecache->blocks_used* + pagecache->block_size), + uchar*); + pagecache->blocks_used++; + } + pagecache->blocks_unused--; + DBUG_ASSERT(block->wlocks == 0); + DBUG_ASSERT(block->rlocks == 0); + DBUG_ASSERT(block->rlocks_queue == 0); + DBUG_ASSERT(block->pins == 0); + block->status= 0; +#ifdef DBUG_ASSERT_EXISTS + block->type= PAGECACHE_EMPTY_PAGE; +#endif + DBUG_ASSERT(reg_req); + block->requests= 1; + block->temperature= PCBLOCK_COLD; + block->hits_left= init_hits_left; + block->last_hit_time= 0; + block->rec_lsn= LSN_MAX; + link_to_file_list(pagecache, block, file, 0); + DBUG_PRINT("hash", + ("block (no block assigned): %p hash_link: %p -> %p", + block, block->hash_link, hash_link)); + block->hash_link= hash_link; + DBUG_PRINT("hash", + ("hash_link (no block assignment): %p hash_link: %p -> %p", + hash_link, hash_link->block, block)); + hash_link->block= block; + page_status= PAGE_TO_BE_READ; + DBUG_PRINT("info", ("page to be read set for page %p (%u)", + block, PCBLOCK_NUMBER(pagecache, block))); + KEYCACHE_PRINT("find_block", + ("got free or never used block %u", + PCBLOCK_NUMBER(pagecache, block))); + } + else + { + DBUG_PRINT("info", ("there is NOT never used blocks")); + /* There are no never used blocks, use a block from the LRU chain */ + + /* + Ensure that we are going to register the block. + (This should be true as a new block could not have been + pinned by caller). + */ + DBUG_ASSERT(reg_req); + + if (! pagecache->used_last) + { + struct st_my_thread_var *thread; + DBUG_PRINT("info", ("there is NOT UNUSED blocks")); + /* + Wait until a new block is added to the LRU chain; + several threads might wait here for the same page, + all of them must get the same block. + + The block is given to us by the next thread executing + link_block(). + */ + if (fast) + { + DBUG_ASSERT(hash_link->requests == 0); + unlink_hash(pagecache, hash_link); + DBUG_PRINT("info", ("fast and no blocks in LRU")); + + KEYCACHE_DBUG_PRINT("find_block", + ("fast and no blocks in LRU")); + DBUG_RETURN(0); + } + + thread= my_thread_var; + thread->keycache_link= (void *) hash_link; + wqueue_link_into_queue(&pagecache->waiting_for_block, thread); + do + { + DBUG_PRINT("wait", + ("suspend thread %s %ld", thread->name, + (ulong) thread->id)); + pagecache_pthread_cond_wait(&thread->suspend, + &pagecache->cache_lock); + } + while (thread->next); + thread->keycache_link= NULL; + block= hash_link->block; + /* Ensure that the block is registered */ + DBUG_ASSERT(block->requests >= 1); + } + else + { + DBUG_PRINT("info", ("take a block from LRU")); + /* + Take the first block from the LRU chain + unlinking it from the chain + */ + block= pagecache->used_last->next_used; + if (fast && + ((block->status & (PCBLOCK_IN_FLUSH | PCBLOCK_CHANGED)) || + (block->hash_link && block->hash_link != hash_link && + block->hash_link->requests))) + { + DBUG_ASSERT(hash_link->requests == 0); + unlink_hash(pagecache, hash_link); + DBUG_PRINT("info", ("fast and LRU block is in switch or has " + "readers")); + KEYCACHE_DBUG_PRINT("find_block", + ("fast and LRU block is in switch or has " + "readers")); + DBUG_RETURN (0); + } + if (reg_req) + reg_requests(pagecache, block, 1); + DBUG_PRINT("hash", ("hash_link (LRU): %p, hash_link: %p -> %p", + hash_link, hash_link->block, block)); + hash_link->block= block; + DBUG_ASSERT(block->requests == 1); + } + + PCBLOCK_INFO(block); + + DBUG_ASSERT(block->hash_link == hash_link || + !(block->status & PCBLOCK_IN_SWITCH)); + + if (block->hash_link != hash_link && + ! (block->status & PCBLOCK_IN_SWITCH) ) + { + /* If another thread is flushing the block, wait for it. */ + if (block->status & PCBLOCK_IN_FLUSH) + wait_for_flush(pagecache, block); + + /* this is a primary request for a new page */ + DBUG_ASSERT(block->wlocks == 0); + DBUG_ASSERT(block->rlocks == 0); + DBUG_ASSERT(block->rlocks_queue == 0); + DBUG_ASSERT(block->pins == 0); + block->status|= PCBLOCK_IN_SWITCH; + + KEYCACHE_DBUG_PRINT("find_block", + ("got block %u for new page", + PCBLOCK_NUMBER(pagecache, block))); + + if (block->status & PCBLOCK_CHANGED) + { + /* The block contains a dirty page - push it out of the cache */ + + KEYCACHE_DBUG_PRINT("find_block", ("block is dirty")); + + /* + The call is thread safe because only the current + thread might change the block->hash_link value + */ + DBUG_ASSERT(block->pins == 0); + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + error= pagecache_fwrite(pagecache, + &block->hash_link->file, + block->buffer, + block->hash_link->pageno, + block->type, + pagecache->readwrite_flags); + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + pagecache->global_cache_write++; + } + + block->status|= PCBLOCK_REASSIGNED; + if (block->hash_link) + { + /* + Wait until all pending read requests + for this page are executed + (we could have avoided this waiting, if we had read + a page in the cache in a sweep, without yielding control) + */ + wait_for_readers(pagecache, block); + + /* Remove the hash link for this page from the hash table */ + unlink_hash(pagecache, block->hash_link); + + /* All pending requests for this page must be resubmitted */ + if (block->wqueue[COND_FOR_SAVED].last_thread) + wqueue_release_queue(&block->wqueue[COND_FOR_SAVED]); + } + link_to_file_list(pagecache, block, file, + (my_bool)(block->hash_link ? 1 : 0)); + + DBUG_PRINT("hash", ("block (LRU): %p, hash_link: %p -> %p", + block, block->hash_link, hash_link)); + block->hash_link= hash_link; + PCBLOCK_INFO(block); + block->hits_left= init_hits_left; + block->last_hit_time= 0; + block->status= error ? PCBLOCK_ERROR : 0; + block->error= error ? (int16) my_errno : 0; +#ifdef DBUG_ASSERT_EXISTS + block->type= PAGECACHE_EMPTY_PAGE; + if (error) + my_debug_put_break_here(); +#endif + page_status= PAGE_TO_BE_READ; + DBUG_PRINT("info", ("page to be read set for page %p", block)); + + KEYCACHE_DBUG_ASSERT(block->hash_link->block == block); + KEYCACHE_DBUG_ASSERT(hash_link->block->hash_link == hash_link); + } + else + { + /* This is for secondary requests for a new page only */ + KEYCACHE_DBUG_PRINT("find_block", + ("block->hash_link: %p hash_link: %p " + "block->status: %u", block->hash_link, + hash_link, block->status )); + page_status= (((block->hash_link == hash_link) && + (block->status & PCBLOCK_READ)) ? + PAGE_READ : PAGE_WAIT_TO_BE_READ); + } + } + } + else + { + /* + The block was found in the cache. It's either a already read + block or a block waiting to be read by another thread. + */ + if (reg_req) + reg_requests(pagecache, block, 1); + KEYCACHE_DBUG_PRINT("find_block", + ("block->hash_link: %p hash_link: %p " + "block->status: %u", block->hash_link, + hash_link, block->status )); + /* + block->hash_link != hash_link can only happen when + the block is in PCBLOCK_IN_SWITCH above (is flushed out + to be replaced by another block). The SWITCH code will change + block->hash_link to point to hash_link. + */ + KEYCACHE_DBUG_ASSERT(block->hash_link == hash_link || + block->status & PCBLOCK_IN_SWITCH); + page_status= (((block->hash_link == hash_link) && + (block->status & PCBLOCK_READ)) ? + PAGE_READ : PAGE_WAIT_TO_BE_READ); + } + } + + KEYCACHE_DBUG_ASSERT(page_status != -1); + *page_st= page_status; + DBUG_PRINT("info", + ("block: %p fd: %u pos: %lu block->status: %u page_status: %u", + block, (uint) file->file, + (ulong) pageno, block->status, (uint) page_status)); + KEYCACHE_PRINT("find_block", + ("block: %p fd: %d pos: %lu block->status: %u page_status: %d", + block, file->file, (ulong) pageno, block->status, + page_status)); + +#if !defined(DBUG_OFF) && defined(EXTRA_DEBUG) + DBUG_EXECUTE("check_pagecache", + test_key_cache(pagecache, "end of find_block",0);); +#endif + KEYCACHE_THREAD_TRACE("find_block:end"); + DBUG_RETURN(block); +} + + +static void add_pin(PAGECACHE_BLOCK_LINK *block) +{ + DBUG_ENTER("add_pin"); + DBUG_PRINT("enter", ("block: %p pins: %u", block, block->pins)); + PCBLOCK_INFO(block); + block->pins++; +#ifndef DBUG_OFF + { + PAGECACHE_PIN_INFO *info= + (PAGECACHE_PIN_INFO *)my_malloc(PSI_INSTRUMENT_ME, sizeof(PAGECACHE_PIN_INFO), MYF(0)); + info->thread= my_thread_var; + info_link(&block->pin_list, info); + } +#endif + DBUG_VOID_RETURN; +} + +static void remove_pin(PAGECACHE_BLOCK_LINK *block, my_bool any +#ifdef DBUG_OFF + __attribute__((unused)) +#endif + ) +{ + DBUG_ENTER("remove_pin"); + DBUG_PRINT("enter", ("block: %p pins: %u any: %d", block, block->pins, + (int)any)); + PCBLOCK_INFO(block); + DBUG_ASSERT(block->pins > 0); + block->pins--; +#ifndef DBUG_OFF + { + PAGECACHE_PIN_INFO *info= info_find(block->pin_list, my_thread_var, any); + DBUG_ASSERT(info != 0); + info_unlink(info); + my_free(info); + } +#endif + DBUG_VOID_RETURN; +} +#ifndef DBUG_OFF +static void info_add_lock(PAGECACHE_BLOCK_LINK *block, my_bool wl) +{ + PAGECACHE_LOCK_INFO *info= + (PAGECACHE_LOCK_INFO *)my_malloc(PSI_INSTRUMENT_ME, sizeof(PAGECACHE_LOCK_INFO), MYF(0)); + info->thread= my_thread_var; + info->write_lock= wl; + info_link((PAGECACHE_PIN_INFO **)&block->lock_list, + (PAGECACHE_PIN_INFO *)info); +} +static void info_remove_lock(PAGECACHE_BLOCK_LINK *block) +{ + PAGECACHE_LOCK_INFO *info= + (PAGECACHE_LOCK_INFO *)info_find((PAGECACHE_PIN_INFO *)block->lock_list, + my_thread_var, FALSE); + DBUG_ASSERT(info != 0); + info_unlink((PAGECACHE_PIN_INFO *)info); + my_free(info); +} +static void info_change_lock(PAGECACHE_BLOCK_LINK *block, my_bool wl) +{ + PAGECACHE_LOCK_INFO *info= + (PAGECACHE_LOCK_INFO *)info_find((PAGECACHE_PIN_INFO *)block->lock_list, + my_thread_var, FALSE); + DBUG_ASSERT(info != 0); + DBUG_ASSERT(info->write_lock != wl); + info->write_lock= wl; +} +#else +#define info_add_lock(B,W) +#define info_remove_lock(B) +#define info_change_lock(B,W) +#endif + + +/** + @brief waiting for lock for read and write lock + + @parem pagecache pointer to a page cache data structure + @parem block the block to work with + @param file file of the block when it was locked + @param pageno page number of the block when it was locked + @param lock_type MY_PTHREAD_LOCK_READ or MY_PTHREAD_LOCK_WRITE + + @retval 0 OK + @retval 1 Can't lock this block, need retry +*/ + +static my_bool pagecache_wait_lock(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block, + PAGECACHE_FILE file, + pgcache_page_no_t pageno, + uint lock_type) +{ + /* Lock failed we will wait */ + struct st_my_thread_var *thread= my_thread_var; + DBUG_ENTER("pagecache_wait_lock"); + DBUG_PRINT("info", ("fail to lock, waiting... %p", block)); + thread->lock_type= lock_type; + wqueue_add_to_queue(&block->wqueue[COND_FOR_WRLOCK], thread); + dec_counter_for_resize_op(pagecache); + do + { + DBUG_PRINT("wait", + ("suspend thread %s %ld", thread->name, (ulong) thread->id)); + pagecache_pthread_cond_wait(&thread->suspend, + &pagecache->cache_lock); + } + while(thread->next); + inc_counter_for_resize_op(pagecache); + PCBLOCK_INFO(block); + if ((block->status & (PCBLOCK_REASSIGNED | PCBLOCK_IN_SWITCH)) || + !block->hash_link || + file.file != block->hash_link->file.file || + pageno != block->hash_link->pageno) + { + DBUG_PRINT("info", ("the block %p changed => need retry " + "status: %x files %d != %d or pages %lu != %lu", + block, block->status, file.file, + block->hash_link ? block->hash_link->file.file : -1, + (ulong) pageno, + (ulong) (block->hash_link ? block->hash_link->pageno : 0))); + DBUG_RETURN(1); + } + DBUG_RETURN(0); +} + +/** + @brief Put on the block write lock + + @parem pagecache pointer to a page cache data structure + @parem block the block to work with + + @note We have loose scheme for locking by the same thread: + * Downgrade to read lock if no other locks are taken + * Our scheme of locking allow for the same thread + - the same kind of lock + - taking read lock if write lock present + - downgrading to read lock if still other place the same + thread keep write lock + * But unlock operation number should be the same to lock operation. + * If we try to get read lock having active write locks we put read + locks to queue, and as soon as write lock(s) gone the read locks + from queue came in force. + * If read lock is unlocked earlier then it came to force it + just removed from the queue + + @retval 0 OK + @retval 1 Can't lock this block, need retry +*/ + +static my_bool get_wrlock(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block) +{ + PAGECACHE_FILE file= block->hash_link->file; + pgcache_page_no_t pageno= block->hash_link->pageno; + pthread_t locker= pthread_self(); + DBUG_ENTER("get_wrlock"); + DBUG_PRINT("info", ("the block %p " + "files %d(%d) pages %lu(%lu)", + block, file.file, block->hash_link->file.file, + (ulong) pageno, (ulong) block->hash_link->pageno)); + PCBLOCK_INFO(block); + /* + We assume that the same thread will try write lock on block on which it + has already read lock. + */ + while ((block->wlocks && !pthread_equal(block->write_locker, locker)) || + block->rlocks) + { + /* Lock failed we will wait */ + if (pagecache_wait_lock(pagecache, block, file, pageno, + MY_PTHREAD_LOCK_WRITE)) + DBUG_RETURN(1); + } + /* we are doing it by global cache mutex protection, so it is OK */ + block->wlocks++; + block->write_locker= locker; + DBUG_PRINT("info", ("WR lock set, block %p", block)); + DBUG_RETURN(0); +} + + +/* + @brief Put on the block read lock + + @param pagecache pointer to a page cache data structure + @param block the block to work with + @param user_file Unique handler per handler file. Used to check if + we request many write locks withing the same + statement + + @note see note for get_wrlock(). + + @retvalue 0 OK + @retvalue 1 Can't lock this block, need retry +*/ + +static my_bool get_rdlock(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block) +{ + PAGECACHE_FILE file= block->hash_link->file; + pgcache_page_no_t pageno= block->hash_link->pageno; + pthread_t locker= pthread_self(); + DBUG_ENTER("get_rdlock"); + DBUG_PRINT("info", ("the block %p " + "files %d(%d) pages %lu(%lu)", + block, file.file, block->hash_link->file.file, + (ulong) pageno, (ulong) block->hash_link->pageno)); + PCBLOCK_INFO(block); + while (block->wlocks && !pthread_equal(block->write_locker, locker)) + { + /* Lock failed we will wait */ + if (pagecache_wait_lock(pagecache, block, file, pageno, + MY_PTHREAD_LOCK_READ)) + DBUG_RETURN(1); + } + /* we are doing it by global cache mutex protection, so it is OK */ + if (block->wlocks) + { + DBUG_ASSERT(pthread_equal(block->write_locker, locker)); + block->rlocks_queue++; + DBUG_PRINT("info", ("RD lock put into queue, block %p", block)); + } + else + { + block->rlocks++; + DBUG_PRINT("info", ("RD lock set, block %p", block)); + } + DBUG_RETURN(0); +} + + +/* + @brief Remove write lock from the block + + @param pagecache pointer to a page cache data structure + @param block the block to work with + @param read_lock downgrade to read lock + + @note see note for get_wrlock(). +*/ + +static void release_wrlock(PAGECACHE_BLOCK_LINK *block, my_bool read_lock) +{ + DBUG_ENTER("release_wrlock"); + PCBLOCK_INFO(block); + DBUG_ASSERT(block->wlocks > 0); + DBUG_ASSERT(block->rlocks == 0); + DBUG_ASSERT(block->pins > 0); + if (read_lock) + block->rlocks_queue++; + if (block->wlocks == 1) + { + block->rlocks= block->rlocks_queue; + block->rlocks_queue= 0; + } + block->wlocks--; + if (block->wlocks > 0) + DBUG_VOID_RETURN; /* Multiple write locked */ + DBUG_PRINT("info", ("WR lock reset, block %p", block)); + /* release all threads waiting for read lock or one waiting for write */ + if (block->wqueue[COND_FOR_WRLOCK].last_thread) + wqueue_release_one_locktype_from_queue(&block->wqueue[COND_FOR_WRLOCK]); + PCBLOCK_INFO(block); + DBUG_VOID_RETURN; +} + +/* + @brief Remove read lock from the block + + @param pagecache pointer to a page cache data structure + @param block the block to work with + + @note see note for get_wrlock(). +*/ + +static void release_rdlock(PAGECACHE_BLOCK_LINK *block) +{ + DBUG_ENTER("release_wrlock"); + PCBLOCK_INFO(block); + if (block->wlocks) + { + DBUG_ASSERT(pthread_equal(block->write_locker, pthread_self())); + DBUG_ASSERT(block->rlocks == 0); + DBUG_ASSERT(block->rlocks_queue > 0); + block->rlocks_queue--; + DBUG_PRINT("info", ("RD lock queue decreased, block %p", block)); + DBUG_VOID_RETURN; + } + DBUG_ASSERT(block->rlocks > 0); + DBUG_ASSERT(block->rlocks_queue == 0); + block->rlocks--; + DBUG_PRINT("info", ("RD lock decreased, block %p", block)); + if (block->rlocks > 0) + DBUG_VOID_RETURN; /* Multiple write locked */ + DBUG_PRINT("info", ("RD lock reset, block %p", block)); + /* release all threads waiting for read lock or one waiting for write */ + if (block->wqueue[COND_FOR_WRLOCK].last_thread) + wqueue_release_one_locktype_from_queue(&block->wqueue[COND_FOR_WRLOCK]); + PCBLOCK_INFO(block); + DBUG_VOID_RETURN; +} + +/** + @brief Try to lock/unlock and pin/unpin the block + + @param pagecache pointer to a page cache data structure + @param block the block to work with + @param lock lock change mode + @param pin pinchange mode + @param file File handler requesting pin + @param any allow unpinning block pinned by any thread; possible + only if not locked, see pagecache_unlock_by_link() + + @retval 0 OK + @retval 1 Try to lock the block failed +*/ + +static my_bool make_lock_and_pin(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block, + enum pagecache_page_lock lock, + enum pagecache_page_pin pin, + my_bool any) +{ + DBUG_ENTER("make_lock_and_pin"); + DBUG_PRINT("enter", ("block: %p (%u) lock: %s pin: %s any %d", + block, PCBLOCK_NUMBER(pagecache, block), + page_cache_page_lock_str[lock], + page_cache_page_pin_str[pin], (int)any)); + PCBLOCK_INFO(block); + + DBUG_ASSERT(block); + DBUG_ASSERT(!any || + ((lock == PAGECACHE_LOCK_LEFT_UNLOCKED) && + (pin == PAGECACHE_UNPIN))); + DBUG_ASSERT(block->hash_link->block == block); + + switch (lock) { + case PAGECACHE_LOCK_WRITE: /* free -> write */ + /* Writelock and pin the buffer */ + if (get_wrlock(pagecache, block)) + { + /* Couldn't lock because block changed status => need retry */ + goto retry; + } + + /* The cache is locked so nothing afraid of */ + add_pin(block); + info_add_lock(block, 1); + break; + case PAGECACHE_LOCK_WRITE_TO_READ: /* write -> read */ + case PAGECACHE_LOCK_WRITE_UNLOCK: /* write -> free */ + /* Removes write lock and puts read lock */ + release_wrlock(block, lock == PAGECACHE_LOCK_WRITE_TO_READ); + /* fall through */ + case PAGECACHE_LOCK_READ_UNLOCK: /* read -> free */ + if (lock == PAGECACHE_LOCK_READ_UNLOCK) + release_rdlock(block); + /* fall through */ + case PAGECACHE_LOCK_LEFT_READLOCKED: /* read -> read */ + if (pin == PAGECACHE_UNPIN) + { + remove_pin(block, FALSE); + } + if (lock == PAGECACHE_LOCK_WRITE_TO_READ) + { + info_change_lock(block, 0); + } + else if (lock == PAGECACHE_LOCK_WRITE_UNLOCK || + lock == PAGECACHE_LOCK_READ_UNLOCK) + { + info_remove_lock(block); + } + break; + case PAGECACHE_LOCK_READ: /* free -> read */ + if (get_rdlock(pagecache, block)) + { + /* Couldn't lock because block changed status => need retry */ + goto retry; + } + + if (pin == PAGECACHE_PIN) + { + /* The cache is locked so nothing afraid off */ + add_pin(block); + } + info_add_lock(block, 0); + break; + case PAGECACHE_LOCK_LEFT_UNLOCKED: /* free -> free */ + if (pin == PAGECACHE_UNPIN) + { + remove_pin(block, any); + } + /* fall through */ + case PAGECACHE_LOCK_LEFT_WRITELOCKED: /* write -> write */ + break; /* do nothing */ + default: + DBUG_ASSERT(0); /* Never should happened */ + } + + PCBLOCK_INFO(block); + DBUG_RETURN(0); +retry: + DBUG_PRINT("INFO", ("Retry block %p", block)); + PCBLOCK_INFO(block); + DBUG_ASSERT(block->hash_link->requests > 0); + block->hash_link->requests--; + DBUG_RETURN(1); +} + + +/** + @brief Reading of a big block in the S3 storage engine. + + @param pagecache Page cache + @param block Block to read + + @note + + Page cache is segmented in logical blocks of size 'block_size'. All + read request are for blocks of 'block_size'. + + When using a file with 'big blocks', the file is split into a + header, header size (for index information) and then blocks of + big_block_size. he last block may be smaller than big_block_size. + All 'big blocks' are a multiple of block_size. + The header is never read into the page cache. It's used to store + the table definition and status and is only read by open(). + + When wanting to read a block, we register a read request for that + block and for the first block that is part of the big block read. We + also put a special flag on the first block so that if another thread + would want to do a big block read, it will wait on signal, and then + check if the block it requested is now in the page cache. If it's + not in the cache it will retry. + + After the big block is read, we will put all read block that was not in the + page cache. Blocks that where already in page cache will not be touched + and will not be added first in the FIFO. + + The block for which we had a read request is added first in FIFO and + returned. +*/ + +#ifdef WITH_S3_STORAGE_ENGINE +static void read_big_block(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block) +{ + int page_st; + size_t big_block_size_in_pages; + size_t offset; + pgcache_page_no_t page, our_page; + pgcache_page_no_t page_to_read; + PAGECACHE_BLOCK_LINK *block_to_read= NULL; + PAGECACHE_IO_HOOK_ARGS args; + S3_BLOCK data; + DBUG_ENTER("read_big_block"); + DBUG_PRINT("enter", ("read BIG block: %p", block)); + bzero((void*) &data, sizeof(data)); + + DBUG_ASSERT(block->hash_link->file.big_block_size % + pagecache->block_size == 0); + big_block_size_in_pages= + block->hash_link->file.big_block_size / pagecache->block_size; + + our_page= block->hash_link->pageno; + + /* find first page of the big block (page_to_read) */ + page_to_read= ((block->hash_link->pageno - + block->hash_link->file.head_blocks) / + big_block_size_in_pages); + page_to_read= (page_to_read * big_block_size_in_pages + + block->hash_link->file.head_blocks); + if (page_to_read != our_page) + { + block_to_read= find_block(pagecache, &block->hash_link->file, + page_to_read, 1, + FALSE, TRUE /* copy under protection (?)*/, + TRUE /*register*/, FALSE, &page_st); + DBUG_ASSERT(block_to_read == block_to_read->hash_link->block); + + if (block_to_read->status & PCBLOCK_ERROR) + { + /* We get first block with an error so all operation failed */ + DBUG_PRINT("error", ("Got error when reading first page")); + block->status|= PCBLOCK_ERROR; + block->error= block_to_read->error; + remove_reader(block_to_read); + unreg_request(pagecache, block_to_read, 1); + DBUG_VOID_RETURN; + } + if (block_to_read->status & PCBLOCK_BIG_READ) + { + /* + Other thread is reading the big block so we will wait when it will + have read our block for us + */ + struct st_my_thread_var *thread; + /* + Either the page was not yet read and there is another thread + doing the read (page_st == PAGE_WAIT_TO_BE_READ) or the page + was just read and there are other threads waiting for the page + but they have not yet unmarked the PCLBOCK_BIG_READ flag + (page_st == PAGE_READ) + */ + DBUG_ASSERT(page_st == PAGE_READ || page_st == PAGE_WAIT_TO_BE_READ); + block->status|= PCBLOCK_BIG_READ; // will be read by other thread + /* + Block read failed because somebody else is reading the first block + (and all other blocks part of this one). + Wait until block is available. + */ + thread= my_thread_var; + /* Put the request into a queue and wait until it can be processed */ + wqueue_add_to_queue(&block_to_read->wqueue[COND_FOR_REQUESTED], thread); + do + { + DBUG_PRINT("wait", + ("suspend thread %s %ld", thread->name, + (ulong) thread->id)); + pagecache_pthread_cond_wait(&thread->suspend, + &pagecache->cache_lock); + } + while (thread->next); + // page should be read by other thread + DBUG_ASSERT(block->status & PCBLOCK_READ || + block->status & PCBLOCK_ERROR); + /* + It is possible that other thread already removed the flag (in + case of two threads waiting) but it will not harm to try to + remove it even in that case. + */ + block->status&= ~PCBLOCK_BIG_READ; + // all is read => lets finish nice + DBUG_ASSERT(block_to_read != block); + remove_reader(block_to_read); + unreg_request(pagecache, block_to_read, 1); + DBUG_VOID_RETURN; + } + else + { + // only primary request here, PAGE_WAIT_TO_BE_READ is impossible + DBUG_ASSERT(page_st != PAGE_WAIT_TO_BE_READ); + } + } + else + { + block_to_read= block; + page_st= PAGE_TO_BE_READ; + } + + DBUG_ASSERT(!(block_to_read->status & PCBLOCK_BIG_READ)); + // Mark the first page of a big block + block_to_read->status|= PCBLOCK_BIG_READ; + + // Don't keep cache locked during the possible slow read from s3 + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + + // perform read of big block + args.page= NULL; + args.pageno= page_to_read; + args.data= block->hash_link->file.callback_data; + + pagecache->global_cache_read++; + if (pagecache->big_block_read(pagecache, &args, &block->hash_link->file, + &data)) + { + pagecache->big_block_free(&data); + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + block_to_read->status|= PCBLOCK_ERROR; + block_to_read->error= (int16) my_errno; + + /* Handle the block that we originally wanted with read */ + block->status|= PCBLOCK_ERROR; + block->error= block_to_read->error; + goto error; + } + + /* + We need to keep the mutex locked while filling pages. + As there is no changed blocks to flush, this operation should + be reasonable fast + */ + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + + /* Copy the first page to the cache */ + if (page_st != PAGE_READ) + { + DBUG_ASSERT(page_st != PAGE_WAIT_TO_BE_READ); + memcpy(block_to_read->buffer, data.str, pagecache->block_size); + block_to_read->status|= PCBLOCK_READ; + } + else + { + DBUG_ASSERT(block_to_read->status & PCBLOCK_READ); + } + /* Signal that all pending requests for this page now can be processed */ + if (block_to_read->wqueue[COND_FOR_REQUESTED].last_thread) + wqueue_release_queue(&block_to_read->wqueue[COND_FOR_REQUESTED]); + + /* Copy the rest of the pages */ + for (offset= pagecache->block_size, page= page_to_read + 1; + offset < data.length; + offset+= pagecache->block_size, page++) + { + DBUG_ASSERT(offset + pagecache->block_size <= data.length); + if (page == our_page) + { + DBUG_ASSERT(!(block->status & PCBLOCK_READ)); + memcpy(block->buffer, data.str + offset, pagecache->block_size); + block->status|= PCBLOCK_READ; + } + else + { + PAGECACHE_BLOCK_LINK *bl; + bl= find_block(pagecache, &block->hash_link->file, page, 1, + FALSE, TRUE /* copy under protection (?)*/, + TRUE /*register*/, TRUE /*fast*/, &page_st); + if (!bl) + { + /* + We can not get this page easy. + Maybe we will be lucky with other pages, + also among other pages can be page which waited by other thread + */ + continue; + } + DBUG_ASSERT(bl == bl->hash_link->block); + if ((bl->status & PCBLOCK_ERROR) == 0 && + (page_st == PAGE_TO_BE_READ || // page should be read + (page_st == PAGE_WAIT_TO_BE_READ && + (bl->status & PCBLOCK_BIG_READ)))) // or page waited by other thread + { + memcpy(bl->buffer, data.str + offset, pagecache->block_size); + bl->status|= PCBLOCK_READ; + } + remove_reader(bl); + unreg_request(pagecache, bl, 1); + /* Signal that all pending requests for this page now can be processed */ + if (bl->wqueue[COND_FOR_REQUESTED].last_thread) + wqueue_release_queue(&bl->wqueue[COND_FOR_REQUESTED]); + } + } + if (page < our_page) + { + /* we break earlier, but still have to fill page what was requested */ + DBUG_ASSERT(!(block->status & PCBLOCK_READ)); + memcpy(block->buffer, + data.str + ((our_page - page_to_read) * pagecache->block_size), + pagecache->block_size); + block->status|= PCBLOCK_READ; + } + pagecache->big_block_free(&data); + +end: + block_to_read->status&= ~PCBLOCK_BIG_READ; + if (block_to_read != block) + { + /* Unlock the 'first block' in the big read */ + remove_reader(block_to_read); + unreg_request(pagecache, block_to_read, 1); + } + /* Signal that all pending requests for this page now can be processed */ + if (block->wqueue[COND_FOR_REQUESTED].last_thread) + wqueue_release_queue(&block->wqueue[COND_FOR_REQUESTED]); + DBUG_VOID_RETURN; + +error: + /* + Read failed. Mark all readers waiting for the a block covered by the + big block that the read failed + */ + for (offset= 0, page= page_to_read + 1; + offset < big_block_size_in_pages; + offset++) + { + if (page != our_page) + { + PAGECACHE_BLOCK_LINK *bl; + bl= find_block(pagecache, &block->hash_link->file, page, 1, + FALSE, TRUE /* copy under protection (?)*/, + TRUE /*register*/, TRUE /*fast*/, &page_st); + if (!bl) + { + /* + We can not get this page easy. + Maybe we will be lucky with other pages, + also among other pages can be page which waited by other thread + */ + continue; + } + DBUG_ASSERT(bl == bl->hash_link->block); + if ((bl->status & PCBLOCK_ERROR) == 0 && + (page_st == PAGE_TO_BE_READ || // page should be read + (page_st == PAGE_WAIT_TO_BE_READ && + (bl->status & PCBLOCK_BIG_READ)))) // or page waited by other thread + { + bl->status|= PCBLOCK_ERROR; + bl->error= (int16) my_errno; + } + remove_reader(bl); + unreg_request(pagecache, bl, 1); + /* Signal that all pending requests for this page now can be processed */ + if (bl->wqueue[COND_FOR_REQUESTED].last_thread) + wqueue_release_queue(&bl->wqueue[COND_FOR_REQUESTED]); + } + } + goto end; +} +#endif /* WITH_S3_STORAGE_ENGINE */ + + +/* + Read into a key cache block buffer from disk. + + SYNOPSIS + + read_block() + pagecache pointer to a page cache data structure + block block to which buffer the data is to be read + primary <-> the current thread will read the data + + RETURN VALUE + None + + NOTES. + The function either reads a page data from file to the block buffer, + or waits until another thread reads it. What page to read is determined + by a block parameter - reference to a hash link for this page. + If an error occurs THE PCBLOCK_ERROR bit is set in the block status. + + On entry cache_lock is locked +*/ + +static void read_block(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block, + my_bool primary) +{ + DBUG_ENTER("read_block"); + DBUG_PRINT("enter", ("read block: %p primary: %d", block, primary)); + if (primary) + { + size_t error; + PAGECACHE_IO_HOOK_ARGS args; + /* + This code is executed only by threads + that submitted primary requests + */ + + pagecache->global_cache_read++; + /* + Page is not in buffer yet, is to be read from disk + Here other threads may step in and register as secondary readers. + They will register in block->wqueue[COND_FOR_REQUESTED]. + */ + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + args.page= block->buffer; + args.pageno= block->hash_link->pageno; + args.data= block->hash_link->file.callback_data; + error= (*block->hash_link->file.pre_read_hook)(&args); + if (!error) + { + error= pagecache_fread(pagecache, &block->hash_link->file, + args.page, + block->hash_link->pageno, + pagecache->readwrite_flags); + } + error= (*block->hash_link->file.post_read_hook)(error != 0, &args); + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + if (error) + { + DBUG_ASSERT(maria_in_recovery || !maria_assert_if_crashed_table); + block->status|= PCBLOCK_ERROR; + block->error= (int16) my_errno; + my_debug_put_break_here(); + } + else + { + block->status|= PCBLOCK_READ; + } + DBUG_PRINT("read_block", + ("primary request: new page in cache")); + /* Signal that all pending requests for this page now can be processed */ + if (block->wqueue[COND_FOR_REQUESTED].last_thread) + wqueue_release_queue(&block->wqueue[COND_FOR_REQUESTED]); + } + else + { + /* + This code is executed only by threads + that submitted secondary requests + */ + + struct st_my_thread_var *thread= my_thread_var; + /* Put the request into a queue and wait until it can be processed */ + wqueue_add_to_queue(&block->wqueue[COND_FOR_REQUESTED], thread); + do + { + DBUG_PRINT("wait", + ("suspend thread %s %ld", thread->name, + (ulong) thread->id)); + pagecache_pthread_cond_wait(&thread->suspend, + &pagecache->cache_lock); + } + while (thread->next); + DBUG_PRINT("read_block", + ("secondary request: new page in cache")); + } + DBUG_VOID_RETURN; +} + + +/** + @brief Set LSN on the page to the given one if the given LSN is bigger + + @param pagecache pointer to a page cache data structure + @param lsn LSN to set + @param block block to check and set +*/ + +static void check_and_set_lsn(PAGECACHE *pagecache, + LSN lsn, PAGECACHE_BLOCK_LINK *block) +{ + LSN old; + DBUG_ENTER("check_and_set_lsn"); + /* + In recovery, we can _ma_unpin_all_pages() to put a LSN on page, though + page would be PAGECACHE_PLAIN_PAGE (transactionality temporarily disabled + to not log REDOs). + */ + DBUG_ASSERT((block->type == PAGECACHE_LSN_PAGE) || maria_in_recovery); + old= lsn_korr(block->buffer); + DBUG_PRINT("info", ("old lsn: " LSN_FMT " new lsn: " LSN_FMT, + LSN_IN_PARTS(old), LSN_IN_PARTS(lsn))); + if (cmp_translog_addr(lsn, old) > 0) + { + + DBUG_ASSERT(block->type != PAGECACHE_READ_UNKNOWN_PAGE); + lsn_store(block->buffer, lsn); + /* we stored LSN in page so we dirtied it */ + if (!(block->status & PCBLOCK_CHANGED)) + link_to_changed_list(pagecache, block); + } + DBUG_VOID_RETURN; +} + + +/** + @brief Unlock/unpin page and put LSN stamp if it need + + @param pagecache pointer to a page cache data structure + @pagam file handler for the file for the block of data to be read + @param pageno number of the block of data in the file + @param lock lock change + @param pin pin page + @param first_REDO_LSN_for_page do not set it if it is zero + @param lsn if it is not LSN_IMPOSSIBLE (0) and it + is bigger then LSN on the page it will be written on + the page + @param was_changed should be true if the page was write locked with + direct link giving and the page was changed + + @note + Pininig uses requests registration mechanism it works following way: + | beginnig | ending | + | of func. | of func. | + ----------------------------+-------------+---------------+ + PAGECACHE_PIN_LEFT_PINNED | - | - | + PAGECACHE_PIN_LEFT_UNPINNED | reg request | unreg request | + PAGECACHE_PIN | reg request | - | + PAGECACHE_UNPIN | - | unreg request | + + +*/ + +void pagecache_unlock(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + pgcache_page_no_t pageno, + enum pagecache_page_lock lock, + enum pagecache_page_pin pin, + LSN first_REDO_LSN_for_page, + LSN lsn, my_bool was_changed) +{ + PAGECACHE_BLOCK_LINK *block; + int page_st; + DBUG_ENTER("pagecache_unlock"); + DBUG_PRINT("enter", ("fd: %u page: %lu %s %s", + (uint) file->file, (ulong) pageno, + page_cache_page_lock_str[lock], + page_cache_page_pin_str[pin])); + /* we do not allow any lock/pin increasing here */ + DBUG_ASSERT(pin != PAGECACHE_PIN); + DBUG_ASSERT(lock != PAGECACHE_LOCK_READ && lock != PAGECACHE_LOCK_WRITE); + + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + /* + As soon as we keep lock cache can be used, and we have lock because want + to unlock. + */ + DBUG_ASSERT(pagecache->can_be_used); + + inc_counter_for_resize_op(pagecache); + /* See NOTE for pagecache_unlock about registering requests */ + block= find_block(pagecache, file, pageno, 0, 0, 0, + pin == PAGECACHE_PIN_LEFT_UNPINNED, FALSE, &page_st); + PCBLOCK_INFO(block); + DBUG_ASSERT(block != 0 && page_st == PAGE_READ); + if (first_REDO_LSN_for_page) + { + DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE_UNLOCK); + DBUG_ASSERT(pin == PAGECACHE_UNPIN); + pagecache_set_block_rec_lsn(block, first_REDO_LSN_for_page); + } + if (lsn != LSN_IMPOSSIBLE) + check_and_set_lsn(pagecache, lsn, block); + + /* if we lock for write we must link the block to changed blocks */ + DBUG_ASSERT((block->status & PCBLOCK_DIRECT_W) == 0 || + (lock == PAGECACHE_LOCK_WRITE_UNLOCK || + lock == PAGECACHE_LOCK_WRITE_TO_READ || + lock == PAGECACHE_LOCK_LEFT_WRITELOCKED)); + /* + if was_changed then status should be PCBLOCK_DIRECT_W or marked + as dirty + */ + DBUG_ASSERT(!was_changed || (block->status & PCBLOCK_DIRECT_W) || + (block->status & PCBLOCK_CHANGED)); + if ((block->status & PCBLOCK_DIRECT_W) && + (lock == PAGECACHE_LOCK_WRITE_UNLOCK || + lock == PAGECACHE_LOCK_WRITE_TO_READ)) + { + if (!(block->status & PCBLOCK_CHANGED) && was_changed) + link_to_changed_list(pagecache, block); + block->status&= ~PCBLOCK_DIRECT_W; + DBUG_PRINT("info", ("Drop PCBLOCK_DIRECT_W for block: %p", block)); + } + + if (make_lock_and_pin(pagecache, block, lock, pin, FALSE)) + { + DBUG_ASSERT(0); /* should not happend */ + } + + remove_reader(block); + /* + Link the block into the LRU chain if it's the last submitted request + for the block and block will not be pinned. + See NOTE for pagecache_unlock about registering requests. + */ + if (pin != PAGECACHE_PIN_LEFT_PINNED) + unreg_request(pagecache, block, 1); + + dec_counter_for_resize_op(pagecache); + + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + + DBUG_VOID_RETURN; +} + + +/* + Unpin page + + SYNOPSIS + pagecache_unpin() + pagecache pointer to a page cache data structure + file handler for the file for the block of data to be read + pageno number of the block of data in the file + lsn if it is not LSN_IMPOSSIBLE (0) and it + is bigger then LSN on the page it will be written on + the page +*/ + +void pagecache_unpin(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + pgcache_page_no_t pageno, + LSN lsn) +{ + PAGECACHE_BLOCK_LINK *block; + int page_st; + DBUG_ENTER("pagecache_unpin"); + DBUG_PRINT("enter", ("fd: %u page: %lu", + (uint) file->file, (ulong) pageno)); + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + /* + As soon as we keep lock cache can be used, and we have lock bacause want + aunlock. + */ + DBUG_ASSERT(pagecache->can_be_used); + + inc_counter_for_resize_op(pagecache); + /* See NOTE for pagecache_unlock about registering requests */ + block= find_block(pagecache, file, pageno, 0, 0, 0, 0, FALSE, &page_st); + DBUG_ASSERT(block != 0); + DBUG_ASSERT(page_st == PAGE_READ); + /* we can't unpin such page without unlock */ + DBUG_ASSERT((block->status & PCBLOCK_DIRECT_W) == 0); + + if (lsn != LSN_IMPOSSIBLE) + check_and_set_lsn(pagecache, lsn, block); + + /* + we can just unpin only with keeping read lock because: + a) we can't pin without any lock + b) we can't unpin keeping write lock + */ + if (make_lock_and_pin(pagecache, block, + PAGECACHE_LOCK_LEFT_READLOCKED, + PAGECACHE_UNPIN, FALSE)) + DBUG_ASSERT(0); /* should not happend */ + + remove_reader(block); + /* + Link the block into the LRU chain if it's the last submitted request + for the block and block will not be pinned. + See NOTE for pagecache_unlock about registering requests + */ + unreg_request(pagecache, block, 1); + + dec_counter_for_resize_op(pagecache); + + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + + DBUG_VOID_RETURN; +} + + +/** + @brief Unlock/unpin page and put LSN stamp if it need + (uses direct block/page pointer) + + @param pagecache pointer to a page cache data structure + @param link direct link to page (returned by read or write) + @param lock lock change + @param pin pin page + @param first_REDO_LSN_for_page do not set it if it is LSN_IMPOSSIBLE (0) + @param lsn if it is not LSN_IMPOSSIBLE and it is bigger then + LSN on the page it will be written on the page + @param was_changed should be true if the page was write locked with + direct link giving and the page was changed + @param any allow unpinning block pinned by any thread; possible + only if not locked + + @note 'any' is a hack so that _ma_bitmap_unpin_all() is allowed to unpin + non-locked bitmap pages pinned by other threads. Because it always uses + PAGECACHE_LOCK_LEFT_UNLOCKED and PAGECACHE_UNPIN + (see write_changed_bitmap()), the hack is limited to these conditions. +*/ + +void pagecache_unlock_by_link(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block, + enum pagecache_page_lock lock, + enum pagecache_page_pin pin, + LSN first_REDO_LSN_for_page, + LSN lsn, my_bool was_changed, + my_bool any) +{ + DBUG_ENTER("pagecache_unlock_by_link"); + DBUG_PRINT("enter", ("block: %p fd: %u page: %lu changed: %d %s %s", + block, (uint) block->hash_link->file.file, + (ulong) block->hash_link->pageno, was_changed, + page_cache_page_lock_str[lock], + page_cache_page_pin_str[pin])); + /* + We do not allow any lock/pin increasing here and page can't be + unpinned because we use direct link. + */ + DBUG_ASSERT(pin != PAGECACHE_PIN); + DBUG_ASSERT(pin != PAGECACHE_PIN_LEFT_UNPINNED); + DBUG_ASSERT(lock != PAGECACHE_LOCK_READ); + DBUG_ASSERT(lock != PAGECACHE_LOCK_WRITE); + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + if (pin == PAGECACHE_PIN_LEFT_UNPINNED && + lock == PAGECACHE_LOCK_READ_UNLOCK) + { + if (make_lock_and_pin(pagecache, block, lock, pin, FALSE)) + DBUG_ASSERT(0); /* should not happend */ + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + DBUG_VOID_RETURN; + } + + /* + As soon as we keep lock cache can be used, and we have lock because want + unlock. + */ + DBUG_ASSERT(pagecache->can_be_used); + + inc_counter_for_resize_op(pagecache); + if (was_changed) + { + if (first_REDO_LSN_for_page != LSN_IMPOSSIBLE) + { + /* + LOCK_READ_UNLOCK is ok here as the page may have first locked + with WRITE lock that was temporarly converted to READ lock before + it's unpinned + */ + DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE_UNLOCK || + lock == PAGECACHE_LOCK_READ_UNLOCK); + DBUG_ASSERT(pin == PAGECACHE_UNPIN); + pagecache_set_block_rec_lsn(block, first_REDO_LSN_for_page); + } + if (lsn != LSN_IMPOSSIBLE) + check_and_set_lsn(pagecache, lsn, block); + /* + Reset error flag. Mark also that page is active; This may not have + been the case if there was an error reading the page + */ + block->status= (block->status & ~PCBLOCK_ERROR) | PCBLOCK_READ; + } + + /* if we lock for write we must link the block to changed blocks */ + DBUG_ASSERT((block->status & PCBLOCK_DIRECT_W) == 0 || + (lock == PAGECACHE_LOCK_WRITE_UNLOCK || + lock == PAGECACHE_LOCK_WRITE_TO_READ || + lock == PAGECACHE_LOCK_LEFT_WRITELOCKED)); + /* + If was_changed then status should be PCBLOCK_DIRECT_W or marked + as dirty + */ + DBUG_ASSERT(!was_changed || (block->status & PCBLOCK_DIRECT_W) || + (block->status & PCBLOCK_CHANGED)); + if ((block->status & PCBLOCK_DIRECT_W) && + (lock == PAGECACHE_LOCK_WRITE_UNLOCK || + lock == PAGECACHE_LOCK_WRITE_TO_READ)) + { + if (!(block->status & PCBLOCK_CHANGED) && was_changed) + link_to_changed_list(pagecache, block); + block->status&= ~PCBLOCK_DIRECT_W; + DBUG_PRINT("info", ("Drop PCBLOCK_DIRECT_W for block: %p", block)); + } + + if (make_lock_and_pin(pagecache, block, lock, pin, any)) + DBUG_ASSERT(0); /* should not happend */ + + /* + Link the block into the LRU chain if it's the last submitted request + for the block and block will not be pinned. + See NOTE for pagecache_unlock about registering requests. + */ + if (pin != PAGECACHE_PIN_LEFT_PINNED) + unreg_request(pagecache, block, 1); + + dec_counter_for_resize_op(pagecache); + + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + + DBUG_VOID_RETURN; +} + + +/* + Unpin page + (uses direct block/page pointer) + + SYNOPSIS + pagecache_unpin_by_link() + pagecache pointer to a page cache data structure + link direct link to page (returned by read or write) + lsn if it is not LSN_IMPOSSIBLE (0) and it + is bigger then LSN on the page it will be written on + the page +*/ + +void pagecache_unpin_by_link(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block, + LSN lsn) +{ + DBUG_ENTER("pagecache_unpin_by_link"); + DBUG_PRINT("enter", ("block: %p fd: %u page: %lu", + block, (uint) block->hash_link->file.file, + (ulong) block->hash_link->pageno)); + + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + /* + As soon as we keep lock cache can be used, and we have lock because want + unlock. + */ + DBUG_ASSERT(pagecache->can_be_used); + /* we can't unpin such page without unlock */ + DBUG_ASSERT((block->status & PCBLOCK_DIRECT_W) == 0); + + inc_counter_for_resize_op(pagecache); + + if (lsn != LSN_IMPOSSIBLE) + check_and_set_lsn(pagecache, lsn, block); + + /* + We can just unpin only with keeping read lock because: + a) we can't pin without any lock + b) we can't unpin keeping write lock + */ + if (make_lock_and_pin(pagecache, block, + PAGECACHE_LOCK_LEFT_READLOCKED, + PAGECACHE_UNPIN, FALSE)) + DBUG_ASSERT(0); /* should not happend */ + + /* + Link the block into the LRU chain if it's the last submitted request + for the block and block will not be pinned. + See NOTE for pagecache_unlock about registering requests. + */ + unreg_request(pagecache, block, 1); + + dec_counter_for_resize_op(pagecache); + + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + + DBUG_VOID_RETURN; +} + +/* description of how to change lock before and after read/write */ +struct rw_lock_change +{ + my_bool need_lock_change; /* need changing of lock at the end */ + enum pagecache_page_lock new_lock; /* lock at the beginning */ + enum pagecache_page_lock unlock_lock; /* lock at the end */ +}; + +/* description of how to change pin before and after read/write */ +struct rw_pin_change +{ + enum pagecache_page_pin new_pin; /* pin status at the beginning */ + enum pagecache_page_pin unlock_pin; /* pin status at the end */ +}; + +/** + Depending on the lock which the user wants in pagecache_read(), we + need to acquire a first type of lock at start of pagecache_read(), and + downgrade it to a second type of lock at end. For example, if user + asked for no lock (PAGECACHE_LOCK_LEFT_UNLOCKED) this translates into + taking first a read lock PAGECACHE_LOCK_READ (to rightfully block on + existing write locks) then read then unlock the lock i.e. change lock + to PAGECACHE_LOCK_READ_UNLOCK (the "1" below tells that a change is + needed). +*/ + +static struct rw_lock_change lock_to_read[8]= +{ + { /*PAGECACHE_LOCK_LEFT_UNLOCKED*/ + 1, + PAGECACHE_LOCK_READ, PAGECACHE_LOCK_READ_UNLOCK + }, + { /*PAGECACHE_LOCK_LEFT_READLOCKED*/ + 0, + PAGECACHE_LOCK_LEFT_READLOCKED, PAGECACHE_LOCK_LEFT_READLOCKED + }, + { /*PAGECACHE_LOCK_LEFT_WRITELOCKED*/ + 0, + PAGECACHE_LOCK_LEFT_WRITELOCKED, PAGECACHE_LOCK_LEFT_WRITELOCKED + }, + { /*PAGECACHE_LOCK_READ*/ + 1, + PAGECACHE_LOCK_READ, PAGECACHE_LOCK_LEFT_READLOCKED + }, + { /*PAGECACHE_LOCK_WRITE*/ + 1, + PAGECACHE_LOCK_WRITE, PAGECACHE_LOCK_LEFT_WRITELOCKED + }, + { /*PAGECACHE_LOCK_READ_UNLOCK*/ + 1, + PAGECACHE_LOCK_LEFT_READLOCKED, PAGECACHE_LOCK_READ_UNLOCK + }, + { /*PAGECACHE_LOCK_WRITE_UNLOCK*/ + 1, + PAGECACHE_LOCK_LEFT_WRITELOCKED, PAGECACHE_LOCK_WRITE_UNLOCK + }, + { /*PAGECACHE_LOCK_WRITE_TO_READ*/ + 1, + PAGECACHE_LOCK_LEFT_WRITELOCKED, PAGECACHE_LOCK_WRITE_TO_READ + } +}; + +/** + Two sets of pin modes (every as for lock upper but for pinning). The + difference between sets if whether we are going to provide caller with + reference on the block or not +*/ + +static struct rw_pin_change lock_to_pin[2][8]= +{ + { + { /*PAGECACHE_LOCK_LEFT_UNLOCKED*/ + PAGECACHE_PIN_LEFT_UNPINNED, + PAGECACHE_PIN_LEFT_UNPINNED + }, + { /*PAGECACHE_LOCK_LEFT_READLOCKED*/ + PAGECACHE_PIN_LEFT_UNPINNED, + PAGECACHE_PIN_LEFT_UNPINNED, + }, + { /*PAGECACHE_LOCK_LEFT_WRITELOCKED*/ + PAGECACHE_PIN_LEFT_PINNED, + PAGECACHE_PIN_LEFT_PINNED + }, + { /*PAGECACHE_LOCK_READ*/ + PAGECACHE_PIN_LEFT_UNPINNED, + PAGECACHE_PIN_LEFT_UNPINNED + }, + { /*PAGECACHE_LOCK_WRITE*/ + PAGECACHE_PIN, + PAGECACHE_PIN_LEFT_PINNED + }, + { /*PAGECACHE_LOCK_READ_UNLOCK*/ + PAGECACHE_PIN_LEFT_UNPINNED, + PAGECACHE_PIN_LEFT_UNPINNED + }, + { /*PAGECACHE_LOCK_WRITE_UNLOCK*/ + PAGECACHE_PIN_LEFT_PINNED, + PAGECACHE_UNPIN + }, + { /*PAGECACHE_LOCK_WRITE_TO_READ*/ + PAGECACHE_PIN_LEFT_PINNED, + PAGECACHE_UNPIN + } + }, + { + { /*PAGECACHE_LOCK_LEFT_UNLOCKED*/ + PAGECACHE_PIN_LEFT_UNPINNED, + PAGECACHE_PIN_LEFT_UNPINNED + }, + { /*PAGECACHE_LOCK_LEFT_READLOCKED*/ + PAGECACHE_PIN_LEFT_UNPINNED, + PAGECACHE_PIN_LEFT_UNPINNED, + }, + { /*PAGECACHE_LOCK_LEFT_WRITELOCKED*/ + PAGECACHE_PIN_LEFT_PINNED, + PAGECACHE_PIN_LEFT_PINNED + }, + { /*PAGECACHE_LOCK_READ*/ + PAGECACHE_PIN, + PAGECACHE_PIN_LEFT_PINNED + }, + { /*PAGECACHE_LOCK_WRITE*/ + PAGECACHE_PIN, + PAGECACHE_PIN_LEFT_PINNED + }, + { /*PAGECACHE_LOCK_READ_UNLOCK*/ + PAGECACHE_PIN_LEFT_UNPINNED, + PAGECACHE_PIN_LEFT_UNPINNED + }, + { /*PAGECACHE_LOCK_WRITE_UNLOCK*/ + PAGECACHE_PIN_LEFT_PINNED, + PAGECACHE_UNPIN + }, + { /*PAGECACHE_LOCK_WRITE_TO_READ*/ + PAGECACHE_PIN_LEFT_PINNED, + PAGECACHE_PIN_LEFT_PINNED, + } + } +}; + + +/* + @brief Read a block of data from a cached file into a buffer; + + @param pagecache pointer to a page cache data structure + @param file handler for the file for the block of data to be read + @param pageno number of the block of data in the file + @param level determines the weight of the data + @param buff buffer to where the data must be placed + @param type type of the page + @param lock lock change + @param link link to the page if we pin it + + @return address from where the data is placed if successful, 0 - otherwise. + + @note Pin will be chosen according to lock parameter (see lock_to_pin) + + @note 'buff', if not NULL, must be long-aligned. + + @note If buff==0 then we provide reference on the page so should keep the + page pinned. +*/ + +uchar *pagecache_read(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + pgcache_page_no_t pageno, + uint level, + uchar *buff, + enum pagecache_page_type type, + enum pagecache_page_lock lock, + PAGECACHE_BLOCK_LINK **page_link) +{ + my_bool error= 0; + enum pagecache_page_pin + new_pin= lock_to_pin[buff==0][lock].new_pin, + unlock_pin= lock_to_pin[buff==0][lock].unlock_pin; + PAGECACHE_BLOCK_LINK *fake_link; + my_bool reg_request; +#ifdef DBUG_TRACE + char llbuf[22]; +#endif + DBUG_ENTER("pagecache_read"); + DBUG_PRINT("enter", ("fd: %u page: %s buffer: %p level: %u " + "t:%s (%d)%s->%s %s->%s big block: %d", + (uint) file->file, ullstr(pageno, llbuf), + buff, level, + page_cache_page_type_str[type], + lock_to_read[lock].need_lock_change, + page_cache_page_lock_str[lock_to_read[lock].new_lock], + page_cache_page_lock_str[lock_to_read[lock].unlock_lock], + page_cache_page_pin_str[new_pin], + page_cache_page_pin_str[unlock_pin], + MY_TEST(pagecache->big_block_read))); + DBUG_ASSERT(buff != 0 || (buff == 0 && (unlock_pin == PAGECACHE_PIN || + unlock_pin == PAGECACHE_PIN_LEFT_PINNED))); + DBUG_ASSERT(pageno < ((1ULL) << 40)); + + if (!page_link) + page_link= &fake_link; + *page_link= 0; /* Catch errors */ + +restart: + + /* + If we use big block than the big block is multiple of blocks and we + have enouch blocks in cache + */ + DBUG_ASSERT(!pagecache->big_block_read || + (file->big_block_size != 0 && + file->big_block_size % pagecache->block_size == 0)); + + if (pagecache->can_be_used) + { + /* Key cache is used */ + PAGECACHE_BLOCK_LINK *block; + uint status; + int UNINIT_VAR(page_st); + + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + if (!pagecache->can_be_used) + { + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + goto no_key_cache; + } + + inc_counter_for_resize_op(pagecache); + pagecache->global_cache_r_requests++; + /* See NOTE for pagecache_unlock about registering requests. */ + reg_request= ((new_pin == PAGECACHE_PIN_LEFT_UNPINNED) || + (new_pin == PAGECACHE_PIN) || + pagecache->big_block_read); + block= find_block(pagecache, file, pageno, level, + lock == PAGECACHE_LOCK_WRITE, buff != 0, + reg_request, FALSE, &page_st); + DBUG_PRINT("info", ("Block type: %s current type %s", + page_cache_page_type_str[block->type], + page_cache_page_type_str[type])); + if (((block->status & PCBLOCK_ERROR) == 0) && (page_st != PAGE_READ)) + { +#ifdef WITH_S3_STORAGE_ENGINE + if (!pagecache->big_block_read || page_st == PAGE_WAIT_TO_BE_READ) +#endif /* WITH_S3_STORAGE_ENGINE */ + { + /* The requested page is to be read into the block buffer */ + read_block(pagecache, block, page_st == PAGE_TO_BE_READ); + DBUG_PRINT("info", ("read is done")); + } +#ifdef WITH_S3_STORAGE_ENGINE + else + { + /* It is big read and this thread should read */ + DBUG_ASSERT(page_st == PAGE_TO_BE_READ); + + read_big_block(pagecache, block); + + if (!((new_pin == PAGECACHE_PIN_LEFT_UNPINNED) || + (new_pin == PAGECACHE_PIN))) + { + /* we registered request only for big_block_read */ + unreg_request(pagecache, block, 1); + } + } +#endif /* WITH_S3_STORAGE_ENGINE */ + } + /* + Assert after block is read. Imagine two concurrent SELECTs on same + table (thread1 and 2), which want to pagecache_read() the same + pageno/fileno. Thread1 calls find_block(), decides to evict a dirty + page from LRU; while it's writing this dirty page to disk, it is + pre-empted and thread2 runs its find_block(), gets the block (in + PAGE_TO_BE_READ state). This block is still containing the in-eviction + dirty page so has an its type, which cannot be tested. + So thread2 has to wait for read_block() to finish (when it wakes up in + read_block(), it's woken up by read_block() of thread1, which implies + that block's type was set to EMPTY by thread1 as part of find_block()). + */ + DBUG_ASSERT(block->type == PAGECACHE_EMPTY_PAGE || + block->type == type || + type == PAGECACHE_LSN_PAGE || + type == PAGECACHE_READ_UNKNOWN_PAGE || + block->type == PAGECACHE_READ_UNKNOWN_PAGE); + if (type != PAGECACHE_READ_UNKNOWN_PAGE || + block->type == PAGECACHE_EMPTY_PAGE) + block->type= type; + + if (make_lock_and_pin(pagecache, block, lock_to_read[lock].new_lock, + new_pin, FALSE)) + { + /* + We failed to write lock the block, cache is unlocked, + we will try to get the block again. + */ + if (reg_request) + unreg_request(pagecache, block, 1); + dec_counter_for_resize_op(pagecache); + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + DBUG_PRINT("info", ("restarting...")); + goto restart; + } + + status= block->status; + if (!buff) + { + buff= block->buffer; + /* possibly we will write here (resolved on unlock) */ + if ((lock == PAGECACHE_LOCK_WRITE || + lock == PAGECACHE_LOCK_LEFT_WRITELOCKED)) + { + block->status|= PCBLOCK_DIRECT_W; + DBUG_PRINT("info", ("Set PCBLOCK_DIRECT_W for block: %p", block)); + } + } + else + { + if (status & PCBLOCK_READ) + { +#if !defined(SERIALIZED_READ_FROM_CACHE) + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); +#endif + + DBUG_ASSERT((pagecache->block_size & 511) == 0); + /* Copy data from the cache buffer */ + memcpy(buff, block->buffer, pagecache->block_size); + +#if !defined(SERIALIZED_READ_FROM_CACHE) + pagecache_pthread_mutex_lock(&pagecache->cache_lock); +#endif + } + } + + remove_reader(block); + if (lock_to_read[lock].need_lock_change) + { + if (make_lock_and_pin(pagecache, block, + lock_to_read[lock].unlock_lock, + unlock_pin, FALSE)) + { + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + DBUG_ASSERT(0); + return (uchar*) 0; + } + } + /* + Link the block into the LRU chain if it's the last submitted request + for the block and block will not be pinned. + See NOTE for pagecache_unlock about registering requests. + */ + if (unlock_pin == PAGECACHE_PIN_LEFT_UNPINNED || + unlock_pin == PAGECACHE_UNPIN) + unreg_request(pagecache, block, 1); + else + *page_link= block; + + dec_counter_for_resize_op(pagecache); + + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + + if (status & PCBLOCK_ERROR) + { + my_errno= block->error; + DBUG_ASSERT(my_errno != 0); + DBUG_PRINT("error", ("Got error %d when doing page read", my_errno)); + DBUG_RETURN((uchar *) 0); + } + + DBUG_RETURN(buff); + } + +no_key_cache: /* Key cache is not used */ + + /* We can't use mutex here as the key cache may not be initialized */ + pagecache->global_cache_r_requests++; + pagecache->global_cache_read++; + + { + PAGECACHE_IO_HOOK_ARGS args; + args.page= buff; + args.pageno= pageno; + args.data= file->callback_data; + error= (* file->pre_read_hook)(&args); + if (!error) + { + error= pagecache_fread(pagecache, file, args.page, pageno, + pagecache->readwrite_flags) != 0; + } + error= (* file->post_read_hook)(error, &args); + } + + DBUG_RETURN(error ? (uchar*) 0 : buff); +} + + +/* + @brief Set/reset flag that page always should be flushed on delete + + @param pagecache pointer to a page cache data structure + @param link direct link to page (returned by read or write) + @param write write on delete flag value + +*/ + +void pagecache_set_write_on_delete_by_link(PAGECACHE_BLOCK_LINK *block) +{ + DBUG_ENTER("pagecache_set_write_on_delete_by_link"); + DBUG_PRINT("enter", ("fd: %d block %p %d -> TRUE", + block->hash_link->file.file, + block, (int) block->status & PCBLOCK_DEL_WRITE)); + DBUG_ASSERT(block->pins); /* should be pinned */ + DBUG_ASSERT(block->wlocks); /* should be write locked */ + + block->status|= PCBLOCK_DEL_WRITE; + + DBUG_VOID_RETURN; +} + + +/* + @brief Delete page from the buffer (common part for link and file/page) + + @param pagecache pointer to a page cache data structure + @param block direct link to page (returned by read or write) + @param page_link hash link of the block + @param flush flush page if it is dirty + + @retval 0 deleted or was not present at all + @retval 1 error +*/ + +static my_bool pagecache_delete_internal(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block, + PAGECACHE_HASH_LINK *page_link, + my_bool flush) +{ + my_bool error= 0; + if (block->status & PCBLOCK_IN_FLUSH) + { + /* + this call is just 'hint' for the cache to free the page so we will + not interferes with flushing process but must return success + */ + goto out; + } + if (block->status & PCBLOCK_CHANGED) + { + flush= (flush || (block->status & PCBLOCK_DEL_WRITE)); + if (flush) + { + /* The block contains a dirty page - push it out of the cache */ + + KEYCACHE_DBUG_PRINT("find_block", ("block is dirty")); + + /* + The call is thread safe because only the current + thread might change the block->hash_link value + */ + DBUG_ASSERT(block->pins == 1); + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + error= pagecache_fwrite(pagecache, + &block->hash_link->file, + block->buffer, + block->hash_link->pageno, + block->type, + pagecache->readwrite_flags); + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + pagecache->global_cache_write++; + + if (error) + { + block->status|= PCBLOCK_ERROR; + block->error= (int16) my_errno; + my_debug_put_break_here(); + goto out; + } + } + else + { + PAGECACHE_IO_HOOK_ARGS args; + PAGECACHE_FILE *filedesc= &block->hash_link->file; + args.page= block->buffer; + args.pageno= block->hash_link->pageno; + args.data= filedesc->callback_data; + /* We are not going to write the page but have to call callbacks */ + DBUG_PRINT("info", ("flush_callback: %p data: %p", + filedesc->flush_log_callback, + filedesc->callback_data)); + if ((*filedesc->flush_log_callback)(&args)) + { + DBUG_PRINT("error", ("flush or write callback problem")); + error= 1; + goto out; + } + } + pagecache->blocks_changed--; + pagecache->global_blocks_changed--; + /* + free_block() will change the status and rec_lsn of the block so no + need to change them here. + */ + } + /* Cache is locked, so we can relese page before freeing it */ + if (make_lock_and_pin(pagecache, block, + PAGECACHE_LOCK_WRITE_UNLOCK, + PAGECACHE_UNPIN, FALSE)) + DBUG_ASSERT(0); + DBUG_ASSERT(block->hash_link->requests > 0); + page_link->requests--; + /* See NOTE for pagecache_unlock() about registering requests. */ + free_block(pagecache, block, 0); + dec_counter_for_resize_op(pagecache); + return 0; + +out: + /* Cache is locked, so we can relese page before freeing it */ + if (make_lock_and_pin(pagecache, block, + PAGECACHE_LOCK_WRITE_UNLOCK, + PAGECACHE_UNPIN, FALSE)) + DBUG_ASSERT(0); + page_link->requests--; + unreg_request(pagecache, block, 1); + dec_counter_for_resize_op(pagecache); + return error; +} + + +/* + @brief Delete page from the buffer by link + + @param pagecache pointer to a page cache data structure + @param link direct link to page (returned by read or write) + @param lock lock change + @param flush flush page if it is dirty + + @retval 0 deleted or was not present at all + @retval 1 error + + @note lock can be only PAGECACHE_LOCK_LEFT_WRITELOCKED (page was + write locked before) or PAGECACHE_LOCK_WRITE (delete will write + lock page before delete) +*/ + +my_bool pagecache_delete_by_link(PAGECACHE *pagecache, + PAGECACHE_BLOCK_LINK *block, + enum pagecache_page_lock lock, + my_bool flush) +{ + my_bool error= 0; + enum pagecache_page_pin pin= PAGECACHE_PIN_LEFT_PINNED; + DBUG_ENTER("pagecache_delete_by_link"); + DBUG_PRINT("enter", ("fd: %d block %p %s %s", + block->hash_link->file.file, + block, + page_cache_page_lock_str[lock], + page_cache_page_pin_str[pin])); + DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE || + lock == PAGECACHE_LOCK_LEFT_WRITELOCKED); + DBUG_ASSERT(block->pins != 0); /* should be pinned */ + + if (pagecache->can_be_used) + { + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + if (!pagecache->can_be_used) + goto end; + + /* + This block should be pinned (i.e. has not zero request counter) => + Such block can't be chosen for eviction. + */ + DBUG_ASSERT((block->status & + (PCBLOCK_IN_SWITCH | PCBLOCK_REASSIGNED)) == 0); + + /* This lock is deleted in pagecache_delete_internal() called below */ + inc_counter_for_resize_op(pagecache); + /* + make_lock_and_pin() can't fail here, because we are keeping pin on the + block and it can't be evicted (which is cause of lock fail and retry) + */ + if (make_lock_and_pin(pagecache, block, lock, pin, FALSE)) + DBUG_ASSERT(0); + + /* + get_present_hash_link() side effect emulation before call + pagecache_delete_internal() + */ + block->hash_link->requests++; + + error= pagecache_delete_internal(pagecache, block, block->hash_link, + flush); +end: + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + } + + DBUG_RETURN(error); +} + + +/** + @brief Returns "hits" for promotion + + @return "hits" for promotion +*/ + +uint pagecache_pagelevel(PAGECACHE_BLOCK_LINK *block) +{ + return block->hits_left; +} + +/* + @brief Adds "hits" to the page + + @param link direct link to page (returned by read or write) + @param level number of "hits" which we add to the page +*/ + +void pagecache_add_level_by_link(PAGECACHE_BLOCK_LINK *block, + uint level) +{ + DBUG_ASSERT(block->pins != 0); /* should be pinned */ + /* + Operation is just for statistics so it is not really important + if it interfere with other hit increasing => we are doing it without + locking the pagecache. + */ + block->hits_left+= level; +} + +/* + @brief Delete page from the buffer + + @param pagecache pointer to a page cache data structure + @param file handler for the file for the block of data to be read + @param pageno number of the block of data in the file + @param lock lock change + @param flush flush page if it is dirty + + @retval 0 deleted or was not present at all + @retval 1 error + + @note lock can be only PAGECACHE_LOCK_LEFT_WRITELOCKED (page was + write locked before) or PAGECACHE_LOCK_WRITE (delete will write + lock page before delete) +*/ +static enum pagecache_page_pin lock_to_pin_one_phase[8]= +{ + PAGECACHE_PIN_LEFT_UNPINNED /*PAGECACHE_LOCK_LEFT_UNLOCKED*/, + PAGECACHE_PIN_LEFT_UNPINNED /*PAGECACHE_LOCK_LEFT_READLOCKED*/, + PAGECACHE_PIN_LEFT_PINNED /*PAGECACHE_LOCK_LEFT_WRITELOCKED*/, + PAGECACHE_PIN_LEFT_UNPINNED /*PAGECACHE_LOCK_READ*/, + PAGECACHE_PIN /*PAGECACHE_LOCK_WRITE*/, + PAGECACHE_PIN_LEFT_UNPINNED /*PAGECACHE_LOCK_READ_UNLOCK*/, + PAGECACHE_UNPIN /*PAGECACHE_LOCK_WRITE_UNLOCK*/, + PAGECACHE_UNPIN /*PAGECACHE_LOCK_WRITE_TO_READ*/ +}; + +my_bool pagecache_delete(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + pgcache_page_no_t pageno, + enum pagecache_page_lock lock, + my_bool flush) +{ + my_bool error= 0; + enum pagecache_page_pin pin= lock_to_pin_one_phase[lock]; + DBUG_ENTER("pagecache_delete"); + DBUG_PRINT("enter", ("fd: %u page: %lu %s %s", + (uint) file->file, (ulong) pageno, + page_cache_page_lock_str[lock], + page_cache_page_pin_str[pin])); + DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE || + lock == PAGECACHE_LOCK_LEFT_WRITELOCKED); + DBUG_ASSERT(pin == PAGECACHE_PIN || + pin == PAGECACHE_PIN_LEFT_PINNED); +restart: + + DBUG_ASSERT(pageno < ((1ULL) << 40)); + if (pagecache->can_be_used) + { + /* Key cache is used */ + reg1 PAGECACHE_BLOCK_LINK *block; + PAGECACHE_HASH_LINK **unused_start, *page_link; + + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + if (!pagecache->can_be_used) + goto end; + + inc_counter_for_resize_op(pagecache); + page_link= get_present_hash_link(pagecache, file, pageno, &unused_start); + if (!page_link) + { + DBUG_PRINT("info", ("There is no such page in the cache")); + dec_counter_for_resize_op(pagecache); + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + DBUG_RETURN(0); + } + block= page_link->block; + if (block->status & (PCBLOCK_REASSIGNED | PCBLOCK_IN_SWITCH)) + { + DBUG_PRINT("info", ("Block %p already is %s", + block, + ((block->status & PCBLOCK_REASSIGNED) ? + "reassigned" : "in switch"))); + PCBLOCK_INFO(block); + page_link->requests--; + dec_counter_for_resize_op(pagecache); + goto end; + } + /* See NOTE for pagecache_unlock about registering requests. */ + if (pin == PAGECACHE_PIN) + reg_requests(pagecache, block, 1); + if (make_lock_and_pin(pagecache, block, lock, pin, FALSE)) + { + /* + We failed to writelock the block, cache is unlocked, and last write + lock is released, we will try to get the block again. + */ + if (pin == PAGECACHE_PIN) + unreg_request(pagecache, block, 1); + dec_counter_for_resize_op(pagecache); + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + dec_counter_for_resize_op(pagecache); + DBUG_PRINT("info", ("restarting...")); + goto restart; + } + + /* we can't delete with opened direct link for write */ + DBUG_ASSERT((block->status & PCBLOCK_DIRECT_W) == 0); + + error= pagecache_delete_internal(pagecache, block, page_link, flush); +end: + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + } + + DBUG_RETURN(error); +} + + +my_bool pagecache_delete_pages(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + pgcache_page_no_t pageno, + uint page_count, + enum pagecache_page_lock lock, + my_bool flush) +{ + pgcache_page_no_t page_end; + DBUG_ENTER("pagecache_delete_pages"); + DBUG_ASSERT(page_count > 0); + + page_end= pageno + page_count; + do + { + if (pagecache_delete(pagecache, file, pageno, + lock, flush)) + DBUG_RETURN(1); + } while (++pageno != page_end); + DBUG_RETURN(0); +} + + +/** + @brief Writes a buffer into a cached file. + + @param pagecache pointer to a page cache data structure + @param file handler for the file to write data to + @param pageno number of the block of data in the file + @param level determines the weight of the data + @param buff buffer with the data + @param type type of the page + @param lock lock change + @param pin pin page + @param write_mode how to write page + @param link link to the page if we pin it + @param first_REDO_LSN_for_page the lsn to set rec_lsn + @param offset offset in the page + @param size size of data + @param validator read page validator + @param validator_data the validator data + + @retval 0 if a success. + @retval 1 Error. +*/ + +static struct rw_lock_change write_lock_change_table[]= +{ + {1, + PAGECACHE_LOCK_WRITE, + PAGECACHE_LOCK_WRITE_UNLOCK} /*PAGECACHE_LOCK_LEFT_UNLOCKED*/, + {0, /*unsupported (we can't write having the block read locked) */ + PAGECACHE_LOCK_LEFT_UNLOCKED, + PAGECACHE_LOCK_LEFT_UNLOCKED} /*PAGECACHE_LOCK_LEFT_READLOCKED*/, + {0, PAGECACHE_LOCK_LEFT_WRITELOCKED, 0} /*PAGECACHE_LOCK_LEFT_WRITELOCKED*/, + {1, + PAGECACHE_LOCK_WRITE, + PAGECACHE_LOCK_WRITE_TO_READ} /*PAGECACHE_LOCK_READ*/, + {0, PAGECACHE_LOCK_WRITE, 0} /*PAGECACHE_LOCK_WRITE*/, + {0, /*unsupported (we can't write having the block read locked) */ + PAGECACHE_LOCK_LEFT_UNLOCKED, + PAGECACHE_LOCK_LEFT_UNLOCKED} /*PAGECACHE_LOCK_READ_UNLOCK*/, + {1, + PAGECACHE_LOCK_LEFT_WRITELOCKED, + PAGECACHE_LOCK_WRITE_UNLOCK } /*PAGECACHE_LOCK_WRITE_UNLOCK*/, + {1, + PAGECACHE_LOCK_LEFT_WRITELOCKED, + PAGECACHE_LOCK_WRITE_TO_READ} /*PAGECACHE_LOCK_WRITE_TO_READ*/ +}; + + +static struct rw_pin_change write_pin_change_table[]= +{ + {PAGECACHE_PIN_LEFT_PINNED, + PAGECACHE_PIN_LEFT_PINNED} /*PAGECACHE_PIN_LEFT_PINNED*/, + {PAGECACHE_PIN, + PAGECACHE_UNPIN} /*PAGECACHE_PIN_LEFT_UNPINNED*/, + {PAGECACHE_PIN, + PAGECACHE_PIN_LEFT_PINNED} /*PAGECACHE_PIN*/, + {PAGECACHE_PIN_LEFT_PINNED, + PAGECACHE_UNPIN} /*PAGECACHE_UNPIN*/ +}; + + +/** + @note 'buff', if not NULL, must be long-aligned. +*/ + +my_bool pagecache_write_part(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + pgcache_page_no_t pageno, + uint level, + uchar *buff, + enum pagecache_page_type type, + enum pagecache_page_lock lock, + enum pagecache_page_pin pin, + enum pagecache_write_mode write_mode, + PAGECACHE_BLOCK_LINK **page_link, + LSN first_REDO_LSN_for_page, + uint offset, uint size) +{ + PAGECACHE_BLOCK_LINK *block= NULL; + PAGECACHE_BLOCK_LINK *fake_link; + my_bool error= 0; + int need_lock_change= write_lock_change_table[lock].need_lock_change; + my_bool reg_request; +#ifdef DBUG_TRACE + char llbuf[22]; +#endif + DBUG_ENTER("pagecache_write_part"); + DBUG_PRINT("enter", ("fd: %u page: %s level: %u type: %s lock: %s " + "pin: %s mode: %s offset: %u size %u", + (uint) file->file, ullstr(pageno, llbuf), level, + page_cache_page_type_str[type], + page_cache_page_lock_str[lock], + page_cache_page_pin_str[pin], + page_cache_page_write_mode_str[write_mode], + offset, size)); + DBUG_ASSERT(type != PAGECACHE_READ_UNKNOWN_PAGE); + DBUG_ASSERT(lock != PAGECACHE_LOCK_LEFT_READLOCKED); + DBUG_ASSERT(lock != PAGECACHE_LOCK_READ_UNLOCK); + DBUG_ASSERT(offset + size <= pagecache->block_size); + DBUG_ASSERT(pageno < ((1ULL) << 40)); + DBUG_ASSERT(pagecache->big_block_read == 0); + + if (!page_link) + page_link= &fake_link; + *page_link= 0; + +restart: + +#if !defined(DBUG_OFF) && defined(EXTRA_DEBUG) + DBUG_EXECUTE("check_pagecache", + test_key_cache(pagecache, "start of key_cache_write", 1);); +#endif + + if (pagecache->can_be_used) + { + /* Key cache is used */ + int page_st; + my_bool need_page_ready_signal= FALSE; + + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + if (!pagecache->can_be_used) + { + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + goto no_key_cache; + } + + inc_counter_for_resize_op(pagecache); + pagecache->global_cache_w_requests++; + /* + Here we register a request if the page was not already pinned. + See NOTE for pagecache_unlock about registering requests. + */ + reg_request= ((pin == PAGECACHE_PIN_LEFT_UNPINNED) || + (pin == PAGECACHE_PIN)); + block= find_block(pagecache, file, pageno, level, + TRUE, FALSE, + reg_request, FALSE, &page_st); + if (!block) + { + DBUG_ASSERT(write_mode != PAGECACHE_WRITE_DONE); + /* It happens only for requests submitted during resize operation */ + dec_counter_for_resize_op(pagecache); + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + /* Write to the disk key cache is in resize at the moment*/ + goto no_key_cache; + } + DBUG_PRINT("info", ("page status: %d", page_st)); + if (!(block->status & PCBLOCK_ERROR) && + ((page_st == PAGE_TO_BE_READ && + (offset || size < pagecache->block_size)) || + (page_st == PAGE_WAIT_TO_BE_READ))) + { + /* The requested page is to be read into the block buffer */ + read_block(pagecache, block, + (my_bool)(page_st == PAGE_TO_BE_READ)); + DBUG_PRINT("info", ("read is done")); + } + else if (page_st == PAGE_TO_BE_READ) + { + need_page_ready_signal= TRUE; + } + + DBUG_ASSERT(block->type == PAGECACHE_EMPTY_PAGE || + block->type == PAGECACHE_READ_UNKNOWN_PAGE || + block->type == type || + /* this is for when going to non-trans to trans */ + (block->type == PAGECACHE_PLAIN_PAGE && + type == PAGECACHE_LSN_PAGE)); + block->type= type; + /* we write to the page so it has no sense to keep the flag */ + block->status&= ~PCBLOCK_DIRECT_W; + DBUG_PRINT("info", ("Drop PCBLOCK_DIRECT_W for block: %p", block)); + + if (make_lock_and_pin(pagecache, block, + write_lock_change_table[lock].new_lock, + (need_lock_change ? + write_pin_change_table[pin].new_pin : + pin), FALSE)) + { + /* + We failed to writelock the block, cache is unlocked, and last write + lock is released, we will try to get the block again. + */ + if (reg_request) + unreg_request(pagecache, block, 1); + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + DBUG_PRINT("info", ("restarting...")); + goto restart; + } + + if (write_mode == PAGECACHE_WRITE_DONE) + { + if (block->status & PCBLOCK_ERROR) + { + my_debug_put_break_here(); + DBUG_PRINT("warning", ("Writing on page with error")); + } + else + { + /* Copy data from buff */ + memcpy(block->buffer + offset, buff, size); + block->status= PCBLOCK_READ; + KEYCACHE_DBUG_PRINT("key_cache_insert", + ("Page injection")); + /* Signal that all pending requests for this now can be processed. */ + if (block->wqueue[COND_FOR_REQUESTED].last_thread) + wqueue_release_queue(&block->wqueue[COND_FOR_REQUESTED]); + } + } + else + { + if (! (block->status & PCBLOCK_CHANGED)) + link_to_changed_list(pagecache, block); + + memcpy(block->buffer + offset, buff, size); + block->status|= PCBLOCK_READ; + /* Page is correct again if we made a full write in it */ + if (size == pagecache->block_size) + block->status&= ~PCBLOCK_ERROR; + } + + if (need_page_ready_signal && + block->wqueue[COND_FOR_REQUESTED].last_thread) + wqueue_release_queue(&block->wqueue[COND_FOR_REQUESTED]); + + if (first_REDO_LSN_for_page) + { + /* single write action of the last write action */ + DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE_UNLOCK || + lock == PAGECACHE_LOCK_LEFT_UNLOCKED); + DBUG_ASSERT(pin == PAGECACHE_UNPIN || + pin == PAGECACHE_PIN_LEFT_UNPINNED); + pagecache_set_block_rec_lsn(block, first_REDO_LSN_for_page); + } + + if (need_lock_change) + { + /* + We don't set rec_lsn of the block; this is ok as for the + Maria-block-record's pages, we always keep pages pinned here. + */ + if (make_lock_and_pin(pagecache, block, + write_lock_change_table[lock].unlock_lock, + write_pin_change_table[pin].unlock_pin, FALSE)) + DBUG_ASSERT(0); + } + + /* Unregister the request */ + DBUG_ASSERT(block->hash_link->requests > 0); + block->hash_link->requests--; + /* See NOTE for pagecache_unlock about registering requests. */ + if (pin == PAGECACHE_PIN_LEFT_UNPINNED || pin == PAGECACHE_UNPIN) + { + unreg_request(pagecache, block, 1); + DBUG_ASSERT(page_link == &fake_link); + } + else + *page_link= block; + + if (block->status & PCBLOCK_ERROR) + { + error= 1; + my_debug_put_break_here(); + } + + dec_counter_for_resize_op(pagecache); + + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + + goto end; + } + +no_key_cache: + /* + We can't by pass the normal page cache operations because need + whole page for calling callbacks & so on. + This branch should not be used for now (but it is fixed as it + should be just to avoid confusing) + */ + DBUG_ASSERT(0); + /* Key cache is not used */ + if (write_mode == PAGECACHE_WRITE_DELAY) + { + /* We can't use mutex here as the key cache may not be initialized */ + pagecache->global_cache_w_requests++; + pagecache->global_cache_write++; + if (offset != 0 || size != pagecache->block_size) + { + uchar *page_buffer= (uchar *) alloca(pagecache->block_size); + PAGECACHE_IO_HOOK_ARGS args; + args.page= page_buffer; + args.pageno= pageno; + args.data= file->callback_data; + + pagecache->global_cache_read++; + error= (*file->pre_read_hook)(&args); + if (!error) + { + error= pagecache_fread(pagecache, file, + page_buffer, + pageno, + pagecache->readwrite_flags) != 0; + } + if ((*file->post_read_hook)(error, &args)) + { + DBUG_PRINT("error", ("read callback problem")); + error= 1; + goto end; + } + memcpy((char *)page_buffer + offset, buff, size); + buff= page_buffer; + } + if (pagecache_fwrite(pagecache, file, buff, pageno, type, + pagecache->readwrite_flags)) + error= 1; + } + +end: +#if !defined(DBUG_OFF) && defined(EXTRA_DEBUG) + DBUG_EXECUTE("exec", + test_key_cache(pagecache, "end of key_cache_write", 1);); +#endif + if (block) + PCBLOCK_INFO(block); + else + DBUG_PRINT("info", ("No block")); + DBUG_RETURN(error); +} + + +/* + Free block: remove reference to it from hash table, + remove it from the chain file of dirty/clean blocks + and add it to the free list. +*/ + +static my_bool free_block(PAGECACHE *pagecache, PAGECACHE_BLOCK_LINK *block, + my_bool abort_if_pinned) +{ + uint status= block->status; + KEYCACHE_THREAD_TRACE("free block"); + KEYCACHE_DBUG_PRINT("free_block", + ("block: %u hash_link %p", + PCBLOCK_NUMBER(pagecache, block), + block->hash_link)); + mysql_mutex_assert_owner(&pagecache->cache_lock); + if (block->hash_link) + { + /* + While waiting for readers to finish, new readers might request the + block. But since we set block->status|= PCBLOCK_REASSIGNED, they + will wait on block->wqueue[COND_FOR_SAVED]. They must be signaled + later. + */ + block->status|= PCBLOCK_REASSIGNED; + wait_for_readers(pagecache, block); + if (unlikely(abort_if_pinned) && unlikely(block->pins)) + { + /* + Block got pinned while waiting for readers. + This can only happens when called from flush_pagecache_blocks_int() + when flushing blocks as part of prepare for maria_close() or from + flush_cached_blocks() + */ + block->status&= ~PCBLOCK_REASSIGNED; + unreg_request(pagecache, block, 0); + + /* All pending requests for this page must be resubmitted. */ + if (block->wqueue[COND_FOR_SAVED].last_thread) + wqueue_release_queue(&block->wqueue[COND_FOR_SAVED]); + return 1; + } + unlink_hash(pagecache, block->hash_link); + } + + unlink_changed(block); + DBUG_ASSERT(block->wlocks == 0); + DBUG_ASSERT(block->rlocks == 0); + DBUG_ASSERT(block->rlocks_queue == 0); + DBUG_ASSERT(block->pins == 0); + DBUG_ASSERT((block->status & ~(PCBLOCK_ERROR | PCBLOCK_READ | PCBLOCK_IN_FLUSH | PCBLOCK_CHANGED | PCBLOCK_REASSIGNED | PCBLOCK_DEL_WRITE)) == 0); + DBUG_ASSERT(block->requests >= 1); + DBUG_ASSERT(block->next_used == NULL); + block->status= 0; +#ifdef DBUG_ASSERT_EXISTS + block->type= PAGECACHE_EMPTY_PAGE; +#endif + block->rec_lsn= LSN_MAX; + DBUG_PRINT("hash", ("block (Free): %p, hash_link: %p -> NULL", + block, block->hash_link)); + block->hash_link= NULL; + if (block->temperature == PCBLOCK_WARM) + pagecache->warm_blocks--; + block->temperature= PCBLOCK_COLD; + KEYCACHE_THREAD_TRACE("free block"); + KEYCACHE_DBUG_PRINT("free_block", + ("block is freed")); + unreg_request(pagecache, block, 0); + + /* + Block->requests is != 0 if unreg_requests()/link_block() gave the block + to a waiting thread + */ + if (!block->requests) + { + DBUG_ASSERT(block->next_used != 0); + + /* Remove the free block from the LRU ring. */ + unlink_block(pagecache, block); + /* Insert the free block in the free list. */ + block->next_used= pagecache->free_block_list; + pagecache->free_block_list= block; + /* Keep track of the number of currently unused blocks. */ + pagecache->blocks_unused++; + } + else + { + /* keep flag set by link_block() */ + block->status= status & PCBLOCK_REASSIGNED; + } + + /* All pending requests for this page must be resubmitted. */ + if (block->wqueue[COND_FOR_SAVED].last_thread) + wqueue_release_queue(&block->wqueue[COND_FOR_SAVED]); + + return 0; +} + + +static int cmp_sec_link(PAGECACHE_BLOCK_LINK **a, PAGECACHE_BLOCK_LINK **b) +{ + return (((*a)->hash_link->pageno < (*b)->hash_link->pageno) ? -1 : + ((*a)->hash_link->pageno > (*b)->hash_link->pageno) ? 1 : 0); +} + + +/** + @brief Flush a portion of changed blocks to disk, free used blocks + if requested + + @param pagecache This page cache reference. + @param file File which should be flushed + @param cache Beginning of array of the block. + @param end Reference to the block after last in the array. + @param flush_type Type of the flush. + @param first_errno Where to store first errno of the flush. + + + @return Operation status + @retval PCFLUSH_OK OK + @retval PCFLUSH_ERROR There was errors during the flush process. + @retval PCFLUSH_PINNED Pinned blocks was met and skipped. + @retval PCFLUSH_PINNED_AND_ERROR PCFLUSH_ERROR and PCFLUSH_PINNED. +*/ + +static int flush_cached_blocks(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + PAGECACHE_BLOCK_LINK **cache, + PAGECACHE_BLOCK_LINK **end, + enum flush_type type, + int *first_errno) +{ + int rc= PCFLUSH_OK; + my_bool error; + uint count= (uint) (end-cache); + DBUG_ENTER("flush_cached_blocks"); + *first_errno= 0; + + /* Don't lock the cache during the flush */ + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + /* + As all blocks referred in 'cache' are marked by PCBLOCK_IN_FLUSH + we are guaranteed that no thread will change them + */ + qsort((uchar*) cache, count, sizeof(*cache), (qsort_cmp) cmp_sec_link); + + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + for (; cache != end; cache++) + { + PAGECACHE_BLOCK_LINK *block= *cache; + + /* + In the case of non_transactional tables we want to flush also + block pinned with reads. This is becasue we may have other + threads reading the block during flush, as non transactional + tables can have many readers while the one writer is doing the + flush. + We don't want to do flush pinned blocks during checkpoint. + We detect the checkpoint case by checking if type is LAZY. + */ + if ((type == FLUSH_KEEP_LAZY && block->pins) || block->wlocks) + { + KEYCACHE_DBUG_PRINT("flush_cached_blocks", + ("block: %u (%p) pinned", + PCBLOCK_NUMBER(pagecache, block), block)); + DBUG_PRINT("info", ("block: %u (%p) pinned", + PCBLOCK_NUMBER(pagecache, block), block)); + PCBLOCK_INFO(block); + /* undo the mark put by flush_pagecache_blocks_int(): */ + block->status&= ~PCBLOCK_IN_FLUSH; + rc|= PCFLUSH_PINNED; + DBUG_PRINT("warning", ("Page pinned")); + unreg_request(pagecache, block, 1); + if (!*first_errno) + *first_errno= HA_ERR_INTERNAL_ERROR; + continue; + } + if (make_lock_and_pin(pagecache, block, + PAGECACHE_LOCK_READ, PAGECACHE_PIN, FALSE)) + DBUG_ASSERT(0); + + KEYCACHE_PRINT("flush_cached_blocks", + ("block: %u (%p) to be flushed", + PCBLOCK_NUMBER(pagecache, block), block)); + DBUG_PRINT("info", ("block: %u (%p) to be flushed", + PCBLOCK_NUMBER(pagecache, block), block)); + PCBLOCK_INFO(block); + + /** + @todo IO If page is contiguous with next page to flush, group flushes + in one single my_pwrite(). + */ + /** + It is important to use block->hash_link->file below and not 'file', as + the first one is right and the second may have different out-of-date + content (see StaleFilePointersInFlush in ma_checkpoint.c). + @todo change argument of functions to be File. + */ + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + error= pagecache_fwrite(pagecache, &block->hash_link->file, + block->buffer, + block->hash_link->pageno, + block->type, + pagecache->readwrite_flags); + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + + if (make_lock_and_pin(pagecache, block, + PAGECACHE_LOCK_READ_UNLOCK, + PAGECACHE_UNPIN, FALSE)) + DBUG_ASSERT(0); + + pagecache->global_cache_write++; + if (error) + { + block->status|= PCBLOCK_ERROR; + block->error= (int16) my_errno; + my_debug_put_break_here(); + if (!*first_errno) + *first_errno= my_errno ? my_errno : -1; + rc|= PCFLUSH_ERROR; + } + /* + Let to proceed for possible waiting requests to write to the block page. + It might happen only during an operation to resize the key cache. + */ + if (block->wqueue[COND_FOR_SAVED].last_thread) + wqueue_release_queue(&block->wqueue[COND_FOR_SAVED]); + /* type will never be FLUSH_IGNORE_CHANGED here */ + if (! (type == FLUSH_KEEP || type == FLUSH_KEEP_LAZY || + type == FLUSH_FORCE_WRITE)) + { + if (!free_block(pagecache, block, 1)) + { + pagecache->blocks_changed--; + pagecache->global_blocks_changed--; + } + else + { + block->status&= ~PCBLOCK_IN_FLUSH; + link_to_file_list(pagecache, block, file, 1); + } + } + else + { + block->status&= ~PCBLOCK_IN_FLUSH; + link_to_file_list(pagecache, block, file, 1); + unreg_request(pagecache, block, 1); + } + } + DBUG_RETURN(rc); +} + + +/** + @brief flush all blocks for a file to disk but don't do any mutex locks + + @param pagecache pointer to a pagecache data structure + @param file handler for the file to flush to + @param flush_type type of the flush + @param filter optional function which tells what blocks to flush; + can be non-NULL only if FLUSH_KEEP, FLUSH_KEEP_LAZY + or FLUSH_FORCE_WRITE. + @param filter_arg an argument to pass to 'filter'. Information about + the block will be passed too. + + @note + Flushes all blocks having the same OS file descriptor as 'file->file', so + can flush blocks having '*block->hash_link->file' != '*file'. + + @note + This function doesn't do any mutex locks because it needs to be called + both from flush_pagecache_blocks and flush_all_key_blocks (the later one + does the mutex lock in the resize_pagecache() function). + + @note + This function can cause problems if two threads call it + concurrently on the same file (look for "PageCacheFlushConcurrencyBugs" + in ma_checkpoint.c); to avoid them, it has internal logic to serialize in + this situation. + + @return Operation status + @retval PCFLUSH_OK OK + @retval PCFLUSH_ERROR There was errors during the flush process. + @retval PCFLUSH_PINNED Pinned blocks was met and skipped. + @retval PCFLUSH_PINNED_AND_ERROR PCFLUSH_ERROR and PCFLUSH_PINNED. +*/ + +static int flush_pagecache_blocks_int(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + enum flush_type type, + PAGECACHE_FLUSH_FILTER filter, + void *filter_arg) +{ + PAGECACHE_BLOCK_LINK *cache_buff[FLUSH_CACHE],**cache; + int last_errno= 0; + int rc= PCFLUSH_OK; + DBUG_ENTER("flush_pagecache_blocks_int"); + DBUG_PRINT("enter", + ("fd: %d blocks_used: %zu blocks_changed: %zu type: %d", + file->file, pagecache->blocks_used, pagecache->blocks_changed, + type)); + +#if !defined(DBUG_OFF) && defined(EXTRA_DEBUG) + DBUG_EXECUTE("check_pagecache", + test_key_cache(pagecache, + "start of flush_pagecache_blocks", 0);); +#endif + + cache= cache_buff; + if (pagecache->disk_blocks > 0 && + (!my_disable_flush_pagecache_blocks || + (type != FLUSH_KEEP && type != FLUSH_KEEP_LAZY))) + { + /* + Key cache exists. If my_disable_flush_pagecache_blocks is true it + disables the operation but only FLUSH_KEEP[_LAZY]: other flushes still + need to be allowed: FLUSH_RELEASE has to free blocks, and + FLUSH_FORCE_WRITE is to overrule my_disable_flush_pagecache_blocks. + */ + int error= 0; + uint count= 0; + PAGECACHE_BLOCK_LINK **pos, **end; + PAGECACHE_BLOCK_LINK *first_in_switch= NULL; + PAGECACHE_BLOCK_LINK *block, *next; +#if defined(PAGECACHE_DEBUG) + uint cnt= 0; +#endif + + struct st_file_in_flush us_flusher, *other_flusher; + us_flusher.file= file->file; + us_flusher.flush_queue.last_thread= NULL; + us_flusher.first_in_switch= FALSE; + while ((other_flusher= (struct st_file_in_flush *) + my_hash_search(&pagecache->files_in_flush, (uchar *)&file->file, + sizeof(file->file)))) + { + /* + File is in flush already: wait, unless FLUSH_KEEP_LAZY. "Flusher" + means "who can mark PCBLOCK_IN_FLUSH", i.e. caller of + flush_pagecache_blocks_int(). + */ + struct st_my_thread_var *thread; + if (type == FLUSH_KEEP_LAZY) + { + DBUG_PRINT("info",("FLUSH_KEEP_LAZY skips")); + DBUG_RETURN(0); + } + thread= my_thread_var; + wqueue_add_to_queue(&other_flusher->flush_queue, thread); + do + { + DBUG_PRINT("wait", + ("(1) suspend thread %s %ld", + thread->name, (ulong) thread->id)); + pagecache_pthread_cond_wait(&thread->suspend, + &pagecache->cache_lock); + } + while (thread->next); + } + /* we are the only flusher of this file now */ + while (my_hash_insert(&pagecache->files_in_flush, (uchar *)&us_flusher)) + { + /* + Out of memory, wait for flushers to empty the hash and retry; should + rarely happen. Other threads are flushing the file; when done, they + are going to remove themselves from the hash, and thus memory will + appear again. However, this memory may be stolen by yet another thread + (for a purpose unrelated to page cache), before we retry + my_hash_insert(). So the loop may run for long. Only if the thread was + killed do we abort the loop, returning 1 (error) which can cause the + table to be marked as corrupted (cf maria_chk_size(), maria_close()) + and thus require a table check. + */ + DBUG_ASSERT(0); + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + if (my_thread_var->abort) + DBUG_RETURN(1); /* End if aborted by user */ + sleep(10); + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + } + + if (type != FLUSH_IGNORE_CHANGED) + { + /* + Count how many key blocks we have to cache to be able + to flush all dirty pages with minimum seek moves. + */ + for (block= pagecache->changed_blocks[FILE_HASH(*file, pagecache)] ; + block; + block= block->next_changed) + { + if (block->hash_link->file.file == file->file) + { + count++; + KEYCACHE_DBUG_ASSERT(count<= pagecache->blocks_used); + } + } + count++; /* Allocate one extra for easy end-of-buffer test */ + /* Allocate a new buffer only if its bigger than the one we have */ + if (count > FLUSH_CACHE && + !(cache= + (PAGECACHE_BLOCK_LINK**) + my_malloc(PSI_INSTRUMENT_ME, sizeof(PAGECACHE_BLOCK_LINK*)*count, MYF(0)))) + { + cache= cache_buff; + count= FLUSH_CACHE; + } + } + + /* Retrieve the blocks and write them to a buffer to be flushed */ +restart: + end= (pos= cache)+count; + for (block= pagecache->changed_blocks[FILE_HASH(*file, pagecache)] ; + block; + block= next) + { +#if defined(PAGECACHE_DEBUG) + cnt++; + KEYCACHE_DBUG_ASSERT(cnt <= pagecache->blocks_used); +#endif + next= block->next_changed; + if (block->hash_link->file.file != file->file) + continue; + if (filter != NULL) + { + int filter_res= (*filter)(block->type, block->hash_link->pageno, + block->rec_lsn, filter_arg); + DBUG_PRINT("info",("filter returned %d", filter_res)); + if (filter_res == FLUSH_FILTER_SKIP_TRY_NEXT) + continue; + if (filter_res == FLUSH_FILTER_SKIP_ALL) + break; + DBUG_ASSERT(filter_res == FLUSH_FILTER_OK); + } + { + DBUG_ASSERT(!(block->status & PCBLOCK_IN_FLUSH)); + /* + We care only for the blocks for which flushing was not + initiated by other threads as a result of page swapping + */ + if (! (block->status & PCBLOCK_IN_SWITCH)) + { + /* + Mark the block with BLOCK_IN_FLUSH in order not to let + other threads to use it for new pages and interfere with + our sequence of flushing dirty file pages + */ + block->status|= PCBLOCK_IN_FLUSH; + + reg_requests(pagecache, block, 1); + if (type != FLUSH_IGNORE_CHANGED) + { + *pos++= block; + /* It's not a temporary file */ + if (pos == end) + { + /* + This happens only if there is not enough + memory for the big block + */ + if ((rc|= flush_cached_blocks(pagecache, file, cache, + end, type, &error)) & + (PCFLUSH_ERROR | PCFLUSH_PINNED)) + last_errno=error; + DBUG_PRINT("info", ("restarting...")); + /* + Restart the scan as some other thread might have changed + the changed blocks chain: the blocks that were in switch + state before the flush started have to be excluded + */ + goto restart; + } + } + else + { + /* It's a temporary file */ + pagecache->blocks_changed--; + pagecache->global_blocks_changed--; + free_block(pagecache, block, 0); + } + } + else if (type != FLUSH_KEEP_LAZY) + { + /* + Link the block into a list of blocks 'in switch', and then we will + wait for this list to be empty, which means they have been flushed + */ + unlink_changed(block); + link_changed(block, &first_in_switch); + us_flusher.first_in_switch= TRUE; + } + } + } + if (pos != cache) + { + if ((rc|= flush_cached_blocks(pagecache, file, cache, pos, type, + &error)) & + (PCFLUSH_ERROR | PCFLUSH_PINNED)) + last_errno= error; + } + /* Wait until list of blocks in switch is empty */ + while (first_in_switch) + { +#if defined(PAGECACHE_DEBUG) + cnt= 0; +#endif + block= first_in_switch; + { + struct st_my_thread_var *thread= my_thread_var; + wqueue_add_to_queue(&block->wqueue[COND_FOR_SAVED], thread); + do + { + DBUG_PRINT("wait", + ("(2) suspend thread %s %ld", + thread->name, (ulong) thread->id)); + pagecache_pthread_cond_wait(&thread->suspend, + &pagecache->cache_lock); + } + while (thread->next); + } +#if defined(PAGECACHE_DEBUG) + cnt++; + KEYCACHE_DBUG_ASSERT(cnt <= pagecache->blocks_used); +#endif + } + us_flusher.first_in_switch= FALSE; + /* The following happens very seldom */ + if (! (type == FLUSH_KEEP || type == FLUSH_KEEP_LAZY || + type == FLUSH_FORCE_WRITE)) + { + /* + this code would free all blocks while filter maybe handled only a + few, that is not possible. + */ + DBUG_ASSERT(filter == NULL); +#if defined(PAGECACHE_DEBUG) + cnt=0; +#endif + for (block= pagecache->file_blocks[FILE_HASH(*file, pagecache)] ; + block; + block= next) + { +#if defined(PAGECACHE_DEBUG) + cnt++; + KEYCACHE_DBUG_ASSERT(cnt <= pagecache->blocks_used); +#endif + next= block->next_changed; + if (block->hash_link->file.file == file->file && + !block->pins && + (! (block->status & PCBLOCK_CHANGED) + || type == FLUSH_IGNORE_CHANGED)) + { + reg_requests(pagecache, block, 1); + free_block(pagecache, block, 1); + } + } + } + /* wake up others waiting to flush this file */ + my_hash_delete(&pagecache->files_in_flush, (uchar *)&us_flusher); + if (us_flusher.flush_queue.last_thread) + wqueue_release_queue(&us_flusher.flush_queue); + } + + DBUG_EXECUTE("check_pagecache", + test_key_cache(pagecache, "end of flush_pagecache_blocks", 0);); + if (cache != cache_buff) + my_free(cache); + if (rc != 0) + { + if (last_errno) + my_errno= last_errno; /* Return first error */ + DBUG_PRINT("error", ("Got error: %d", my_errno)); + } + DBUG_RETURN(rc); +} + + +/** + @brief flush all blocks for a file to disk + + @param pagecache pointer to a pagecache data structure + @param file handler for the file to flush to + @param flush_type type of the flush + @param filter optional function which tells what blocks to flush; + can be non-NULL only if FLUSH_KEEP, FLUSH_KEEP_LAZY + or FLUSH_FORCE_WRITE. + @param filter_arg an argument to pass to 'filter'. Information about + the block will be passed too. + + @return Operation status + @retval PCFLUSH_OK OK + @retval PCFLUSH_ERROR There was errors during the flush process. + @retval PCFLUSH_PINNED Pinned blocks was met and skipped. + @retval PCFLUSH_PINNED_AND_ERROR PCFLUSH_ERROR and PCFLUSH_PINNED. +*/ + +int flush_pagecache_blocks_with_filter(PAGECACHE *pagecache, + PAGECACHE_FILE *file, + enum flush_type type, + PAGECACHE_FLUSH_FILTER filter, + void *filter_arg) +{ + int res; + DBUG_ENTER("flush_pagecache_blocks_with_filter"); + DBUG_PRINT("enter", ("pagecache: %p", pagecache)); + + if (pagecache->disk_blocks <= 0) + DBUG_RETURN(0); + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + inc_counter_for_resize_op(pagecache); + res= flush_pagecache_blocks_int(pagecache, file, type, filter, filter_arg); + dec_counter_for_resize_op(pagecache); + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + DBUG_RETURN(res); +} + + +/* + Reset the counters of a key cache. + + SYNOPSIS + reset_pagecache_counters() + name the name of a key cache + pagecache pointer to the pagecache to be reset + + DESCRIPTION + This procedure is used to reset the counters of all currently used key + caches, both the default one and the named ones. + + RETURN + 0 on success (always because it can't fail) +*/ + +int reset_pagecache_counters(const char *name __attribute__((unused)), + PAGECACHE *pagecache) +{ + DBUG_ENTER("reset_pagecache_counters"); + if (!pagecache->inited) + { + DBUG_PRINT("info", ("Key cache %s not initialized.", name)); + DBUG_RETURN(0); + } + DBUG_PRINT("info", ("Resetting counters for key cache %s.", name)); + + pagecache->global_blocks_changed= 0; /* Key_blocks_not_flushed */ + pagecache->global_cache_r_requests= 0; /* Key_read_requests */ + pagecache->global_cache_read= 0; /* Key_reads */ + pagecache->global_cache_w_requests= 0; /* Key_write_requests */ + pagecache->global_cache_write= 0; /* Key_writes */ + DBUG_RETURN(0); +} + + +/** + @brief Allocates a buffer and stores in it some info about all dirty pages + + Does the allocation because the caller cannot know the size itself. + Memory freeing is to be done by the caller (if the "str" member of the + LEX_STRING is not NULL). + Ignores all pages of another type than PAGECACHE_LSN_PAGE, because they + are not interesting for a checkpoint record. + The caller has the intention of doing checkpoints. + + @param pagecache pointer to the page cache + @param[out] str pointer to where the allocated buffer, and + its size, will be put + @param[out] min_rec_lsn pointer to where the minimum rec_lsn of all + relevant dirty pages will be put + @return Operation status + @retval 0 OK + @retval 1 Error +*/ + +my_bool pagecache_collect_changed_blocks_with_lsn(PAGECACHE *pagecache, + LEX_STRING *str, + LSN *min_rec_lsn) +{ + my_bool error= 0; + size_t stored_list_size= 0; + uint file_hash; + char *ptr; + LSN minimum_rec_lsn= LSN_MAX; + DBUG_ENTER("pagecache_collect_changed_blocks_with_LSN"); + + DBUG_ASSERT(NULL == str->str); + /* + We lock the entire cache but will be quick, just reading/writing a few MBs + of memory at most. + */ + pagecache_pthread_mutex_lock(&pagecache->cache_lock); + for (;;) + { + struct st_file_in_flush *other_flusher; + for (file_hash= 0; + (other_flusher= (struct st_file_in_flush *) + my_hash_element(&pagecache->files_in_flush, file_hash)) != NULL && + !other_flusher->first_in_switch; + file_hash++) + {} + if (other_flusher == NULL) + break; + /* + other_flusher.first_in_switch is true: some thread is flushing a file + and has removed dirty blocks from changed_blocks[] while they were still + dirty (they were being evicted (=>flushed) by yet another thread, which + may not have flushed the block yet so it may still be dirty). + If Checkpoint proceeds now, it will not see the page. If there is a + crash right after writing the checkpoint record, before the page is + flushed, at recovery the page will be wrongly ignored because it won't + be in the dirty pages list in the checkpoint record. So wait. + */ + { + struct st_my_thread_var *thread= my_thread_var; + wqueue_add_to_queue(&other_flusher->flush_queue, thread); + do + { + DBUG_PRINT("wait", + ("suspend thread %s %ld", thread->name, + (ulong) thread->id)); + pagecache_pthread_cond_wait(&thread->suspend, + &pagecache->cache_lock); + } + while (thread->next); + } + } + + /* Count how many dirty pages are interesting */ + for (file_hash= 0; file_hash < pagecache->changed_blocks_hash_size; file_hash++) + { + PAGECACHE_BLOCK_LINK *block; + for (block= pagecache->changed_blocks[file_hash] ; + block; + block= block->next_changed) + { + /* + Q: is there something subtle with block->hash_link: can it be NULL? + does it have to be == hash_link->block... ? + */ + DBUG_ASSERT(block->hash_link != NULL); + DBUG_ASSERT(block->status & PCBLOCK_CHANGED); + /* + Note that we don't store bitmap pages, or pages from non-transactional + (like temporary) tables. Don't checkpoint during Recovery which uses + PAGECACHE_PLAIN_PAGE. + */ + if (block->type != PAGECACHE_LSN_PAGE) + continue; /* no need to store it */ + stored_list_size++; + } + } + + compile_time_assert(sizeof(pagecache->blocks) <= 8); + str->length= 8 + /* number of dirty pages */ + (2 + /* table id */ + 1 + /* data or index file */ + 5 + /* pageno */ + LSN_STORE_SIZE /* rec_lsn */ + ) * stored_list_size; + if (NULL == (str->str= my_malloc(PSI_INSTRUMENT_ME, str->length, MYF(MY_WME)))) + goto err; + ptr= str->str; + int8store(ptr, (ulonglong)stored_list_size); + ptr+= 8; + DBUG_PRINT("info", ("found %zu dirty pages", stored_list_size)); + if (stored_list_size == 0) + goto end; + for (file_hash= 0; file_hash < pagecache->changed_blocks_hash_size; file_hash++) + { + PAGECACHE_BLOCK_LINK *block; + for (block= pagecache->changed_blocks[file_hash] ; + block; + block= block->next_changed) + { + uint16 table_id; + MARIA_SHARE *share; + if (block->type != PAGECACHE_LSN_PAGE) + continue; /* no need to store it in the checkpoint record */ + share= (MARIA_SHARE *)(block->hash_link->file.callback_data); + table_id= share->id; + int2store(ptr, table_id); + ptr+= 2; + ptr[0]= (share->kfile.file == block->hash_link->file.file); + ptr++; + DBUG_ASSERT(block->hash_link->pageno < ((1ULL) << 40)); + page_store(ptr, block->hash_link->pageno); + ptr+= PAGE_STORE_SIZE; + lsn_store(ptr, block->rec_lsn); + ptr+= LSN_STORE_SIZE; + if (block->rec_lsn != LSN_MAX) + { + DBUG_ASSERT(LSN_VALID(block->rec_lsn)); + if (cmp_translog_addr(block->rec_lsn, minimum_rec_lsn) < 0) + minimum_rec_lsn= block->rec_lsn; + } /* otherwise, some trn->rec_lsn should hold the correct info */ + } + } +end: + pagecache_pthread_mutex_unlock(&pagecache->cache_lock); + *min_rec_lsn= minimum_rec_lsn; + DBUG_RETURN(error); + +err: + error= 1; + goto end; +} + + +#ifndef DBUG_OFF + +/** + Verifies that a file has no dirty pages. +*/ + +void pagecache_file_no_dirty_page(PAGECACHE *pagecache, PAGECACHE_FILE *file) +{ + File fd= file->file; + PAGECACHE_BLOCK_LINK *block; + for (block= pagecache->changed_blocks[FILE_HASH(*file, pagecache)]; + block != NULL; + block= block->next_changed) + if (block->hash_link->file.file == fd) + { + DBUG_PRINT("info", ("pagecache_file_not_in error")); + PCBLOCK_INFO(block); + DBUG_ASSERT(0); + } +} + + +/* + Test if disk-cache is ok +*/ +static void test_key_cache(PAGECACHE *pagecache __attribute__((unused)), + const char *where __attribute__((unused)), + my_bool lock __attribute__((unused))) +{ + /* TODO */ +} +#endif + +uchar *pagecache_block_link_to_buffer(PAGECACHE_BLOCK_LINK *block) +{ + return block->buffer; +} + +#if defined(PAGECACHE_TIMEOUT) + +#define KEYCACHE_DUMP_FILE "pagecache_dump.txt" +#define MAX_QUEUE_LEN 100 + + +static void pagecache_dump(PAGECACHE *pagecache) +{ + FILE *pagecache_dump_file=fopen(KEYCACHE_DUMP_FILE, "w"); + struct st_my_thread_var *last; + struct st_my_thread_var *thread; + PAGECACHE_BLOCK_LINK *block; + PAGECACHE_HASH_LINK *hash_link; + PAGECACHE_PAGE *page; + uint i; + + fprintf(pagecache_dump_file, "thread: %s %ld\n", thread->name, + (ulong) thread->id); + + i=0; + thread=last=waiting_for_hash_link.last_thread; + fprintf(pagecache_dump_file, "queue of threads waiting for hash link\n"); + if (thread) + do + { + thread= thread->next; + page= (PAGECACHE_PAGE *) thread->keycache_link; + fprintf(pagecache_dump_file, + "thread: %s %ld, (file,pageno)=(%u,%lu)\n", + thread->name, (ulong) thread->id, + (uint) page->file.file,(ulong) page->pageno); + if (++i == MAX_QUEUE_LEN) + break; + } + while (thread != last); + + i=0; + thread=last=waiting_for_block.last_thread; + fprintf(pagecache_dump_file, "queue of threads waiting for block\n"); + if (thread) + do + { + thread=thread->next; + hash_link= (PAGECACHE_HASH_LINK *) thread->keycache_link; + fprintf(pagecache_dump_file, + "thread: %s %u hash_link:%u (file,pageno)=(%u,%lu)\n", + thread->name, (ulong) thread->id, + (uint) PAGECACHE_HASH_LINK_NUMBER(pagecache, hash_link), + (uint) hash_link->file.file,(ulong) hash_link->pageno); + if (++i == MAX_QUEUE_LEN) + break; + } + while (thread != last); + + for (i=0 ; i < pagecache->blocks_used ; i++) + { + int j; + block= &pagecache->block_root[i]; + hash_link= block->hash_link; + fprintf(pagecache_dump_file, + "block:%u hash_link:%d status:%x #requests=%u waiting_for_readers:%d\n", + i, (int) (hash_link ? + PAGECACHE_HASH_LINK_NUMBER(pagecache, hash_link) : + -1), + block->status, block->requests, block->condvar ? 1 : 0); + for (j=0 ; j < COND_SIZE; j++) + { + PAGECACHE_WQUEUE *wqueue=&block->wqueue[j]; + thread= last= wqueue->last_thread; + fprintf(pagecache_dump_file, "queue #%d\n", j); + if (thread) + { + do + { + thread=thread->next; + fprintf(pagecache_dump_file, + "thread: %s %ld\n", thread->name, (ulong) thread->id); + if (++i == MAX_QUEUE_LEN) + break; + } + while (thread != last); + } + } + } + fprintf(pagecache_dump_file, "LRU chain:"); + block= pagecache= used_last; + if (block) + { + do + { + block= block->next_used; + fprintf(pagecache_dump_file, + "block:%u, ", PCBLOCK_NUMBER(pagecache, block)); + } + while (block != pagecache->used_last); + } + fprintf(pagecache_dump_file, "\n"); + + fclose(pagecache_dump_file); +} + +#endif /* defined(PAGECACHE_TIMEOUT) */ + +#if defined(PAGECACHE_TIMEOUT) && !defined(_WIN32) + + +static int pagecache_pthread_cond_wait(mysql_cond_t *cond, + mysql_mutex_t *mutex) +{ + int rc; + struct timeval now; /* time when we started waiting */ + struct timespec timeout; /* timeout value for the wait function */ + struct timezone tz; +#if defined(PAGECACHE_DEBUG) + int cnt=0; +#endif + + /* Get current time */ + gettimeofday(&now, &tz); + /* Prepare timeout value */ + timeout.tv_sec= now.tv_sec + PAGECACHE_TIMEOUT; + /* + timeval uses microseconds. + timespec uses nanoseconds. + 1 nanosecond = 1000 micro seconds + */ + timeout.tv_nsec= now.tv_usec * 1000; + KEYCACHE_THREAD_TRACE_END("started waiting"); +#if defined(PAGECACHE_DEBUG) + cnt++; + if (cnt % 100 == 0) + fprintf(pagecache_debug_log, "waiting...\n"); + fflush(pagecache_debug_log); +#endif + rc= mysql_cond_timedwait(cond, mutex, &timeout); + KEYCACHE_THREAD_TRACE_BEGIN("finished waiting"); + if (rc == ETIMEDOUT || rc == ETIME) + { +#if defined(PAGECACHE_DEBUG) + fprintf(pagecache_debug_log,"aborted by pagecache timeout\n"); + fclose(pagecache_debug_log); + abort(); +#endif + pagecache_dump(); + } + +#if defined(PAGECACHE_DEBUG) + KEYCACHE_DBUG_ASSERT(rc != ETIMEDOUT); +#else + assert(rc != ETIMEDOUT); +#endif + return rc; +} +#else +#if defined(PAGECACHE_DEBUG) +static int pagecache_pthread_cond_wait(mysql_cond_t *cond, + mysql_mutex_t *mutex) +{ + int rc; + KEYCACHE_THREAD_TRACE_END("started waiting"); + rc= mysql_cond_wait(cond, mutex); + KEYCACHE_THREAD_TRACE_BEGIN("finished waiting"); + return rc; +} +#endif +#endif /* defined(PAGECACHE_TIMEOUT) && !defined(_WIN32) */ + + +#if defined(PAGECACHE_DEBUG) +static int ___pagecache_pthread_mutex_lock(mysql_mutex_t *mutex) +{ + int rc; + rc= mysql_mutex_lock(mutex); + KEYCACHE_THREAD_TRACE_BEGIN(""); + return rc; +} + + +static void ___pagecache_pthread_mutex_unlock(mysql_mutex_t *mutex) +{ + KEYCACHE_THREAD_TRACE_END(""); + mysql_mutex_unlock(mutex); +} + + +static int ___pagecache_pthread_cond_signal(mysql_cond_t *cond) +{ + int rc; + KEYCACHE_THREAD_TRACE("signal"); + rc= mysql_cond_signal(cond); + return rc; +} + + +static void pagecache_debug_print(const char * fmt, ...) +{ + va_list args; + va_start(args,fmt); + if (pagecache_debug_log) + { + vfprintf(pagecache_debug_log, fmt, args); + fputc('\n',pagecache_debug_log); +#ifdef PAGECACHE_DEBUG_DLOG + _db_doprnt_(fmt, args); +#endif + } + va_end(args); +} + +void pagecache_debug_log_close(void) +{ + if (pagecache_debug_log) + fclose(pagecache_debug_log); +} +#endif /* defined(PAGECACHE_DEBUG) */ + +/** + @brief null hooks +*/ + +static my_bool null_pre_hook(PAGECACHE_IO_HOOK_ARGS *args + __attribute__((unused))) +{ + return 0; +} + +static my_bool null_post_read_hook(int res, PAGECACHE_IO_HOOK_ARGS *args + __attribute__((unused))) +{ + return res != 0; +} + +static void null_post_write_hook(int res __attribute__((unused)), + PAGECACHE_IO_HOOK_ARGS *args + __attribute__((unused))) +{ + return; +} + +void pagecache_file_set_null_hooks(PAGECACHE_FILE *file) +{ + file->pre_read_hook= null_pre_hook; + file->post_read_hook= null_post_read_hook; + file->pre_write_hook= null_pre_hook; + file->post_write_hook= null_post_write_hook; + file->flush_log_callback= null_pre_hook; + file->callback_data= NULL; + file->head_blocks= file->big_block_size= 0; +} |