database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-08-08 11:22:35 +03:00
Code Activity

Locking for read and write made waiting for each other (with loose

Browse Source 
scheme for the same thread locking).


include/my_pthread.h:
  Added variable for lock diffirentiation.
include/wqueue.h:
  New release call.
mysys/wqueue.c:
  New release call in case of read/write lock.
storage/maria/ma_pagecache.c:
  Locking for read and write made waitimg for each other.
storage/maria/unittest/Makefile.am:
  New test added.
storage/maria/unittest/ma_pagecache_consist.c:
  Fixed thread initialization in the test.
storage/maria/unittest/ma_pagecache_rwconsist.c:
  New BitKeeper file ``storage/maria/unittest/ma_pagecache_rwconsist.c''
This commit is contained in:
unknown
2008-02-22 22:32:34 +02:00
parent 3aaa0f60cd
commit 190de95f6f
7 changed files with 652 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
include/my_pthread.h
View File

@@ -674,6 +674,9 @@ extern int pthread_dummy(int);
#endif #endif
#endif #endif
#define MY_PTHREAD_LOCK_READ 0
#define MY_PTHREAD_LOCK_WRITE 1
struct st_my_thread_var struct st_my_thread_var
{ {
int thr_errno; int thr_errno;
@@ -688,6 +691,7 @@ struct st_my_thread_var
my_bool init; my_bool init;
struct st_my_thread_var *next,**prev; struct st_my_thread_var *next,**prev;
void *opt_info; void *opt_info;
uint lock_type; /* used by conditional release the queue */
#ifndef DBUG_OFF #ifndef DBUG_OFF
void *dbug; void *dbug;
char name[THREAD_NAME_SIZE+1]; char name[THREAD_NAME_SIZE+1];

1
include/wqueue.h
View File

@@ -20,6 +20,7 @@ void wqueue_add_and_wait(WQUEUE *wqueue,
struct st_my_thread_var *thread, struct st_my_thread_var *thread,
pthread_mutex_t *lock); pthread_mutex_t *lock);
void wqueue_release_queue(WQUEUE *wqueue); void wqueue_release_queue(WQUEUE *wqueue);
void wqueue_release_one_locktype_from_queue(WQUEUE *wqueue);
#endif #endif

43
mysys/wqueue.c
View File

@@ -136,6 +136,49 @@ void wqueue_release_queue(WQUEUE *wqueue)
} }
/**
@brief Removes all threads waiting for read or first one waiting for write.
@param wqueue pointer to the queue structure
@apram thread pointer to the thread to be added to the queue
*/
void wqueue_release_one_locktype_from_queue(WQUEUE *wqueue)
{
struct st_my_thread_var *last= wqueue->last_thread;
struct st_my_thread_var *next= last->next;
struct st_my_thread_var **prev= &wqueue->last_thread;
struct st_my_thread_var *thread;
uint first_type= next->lock_type;
if (first_type == MY_PTHREAD_LOCK_WRITE)
{
/* release first waiting for write lock */
thread= next;
pthread_cond_signal(&thread->suspend);
wqueue->last_thread= next;
thread->next= NULL;
return;
}
do
{
thread= next;
next= thread->next;
if (thread->lock_type == MY_PTHREAD_LOCK_WRITE)
{
/* skip waiting for write lock */
*prev= thread;
prev= &thread->next;
}
else
{
/* release waiting for read lock */
pthread_cond_signal(&thread->suspend);
thread->next= NULL;
}
} while (thread != last);
*prev= NULL;
}
/* /*
Add thread and wait Add thread and wait

288
storage/maria/ma_pagecache.c
View File

@@ -37,6 +37,9 @@
blocks_unused is the sum of never used blocks in the pool and of currently 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 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. 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 "maria_def.h"
@@ -96,7 +99,7 @@
#define PCBLOCK_INFO(B) \ #define PCBLOCK_INFO(B) \
DBUG_PRINT("info", \ DBUG_PRINT("info", \
("block: 0x%lx fd: %lu page: %lu s: %0x hshL: 0x%lx req: %u/%u " \ ("block: 0x%lx fd: %lu page: %lu s: %0x hshL: 0x%lx req: %u/%u " \
"wrlocks: %u pins: %u", \ "wrlocks: %u rdlocks %u rdlocks_q: %u pins: %u", \
(ulong)(B), \ (ulong)(B), \
(ulong)((B)->hash_link ? \ (ulong)((B)->hash_link ? \
(B)->hash_link->file.file : \ (B)->hash_link->file.file : \
@@ -110,7 +113,7 @@
(uint)((B)->hash_link ? \ (uint)((B)->hash_link ? \
(B)->hash_link->requests : \ (B)->hash_link->requests : \
0), \ 0), \
block->wlocks, \ block->wlocks, block->rlocks, block->rlocks_queue, \
(uint)(B)->pins)) (uint)(B)->pins))
/* TODO: put it to my_static.c */ /* TODO: put it to my_static.c */
@@ -298,11 +301,7 @@ struct st_pagecache_block_link
#endif #endif
KEYCACHE_CONDVAR *condvar; /* condition variable for 'no readers' event */ KEYCACHE_CONDVAR *condvar; /* condition variable for 'no readers' event */
uchar *buffer; /* buffer for the block page */ uchar *buffer; /* buffer for the block page */
#ifdef THREAD
pthread_t write_locker; pthread_t write_locker;
#else
int write_locker;
#endif
ulonglong last_hit_time; /* timestamp of the last hit */ ulonglong last_hit_time; /* timestamp of the last hit */
WQUEUE WQUEUE
@@ -311,7 +310,9 @@ struct st_pagecache_block_link
uint status; /* state of the block */ uint status; /* state of the block */
uint pins; /* pin counter */ uint pins; /* pin counter */
uint wlocks; /* write locks counter */ uint wlocks; /* write locks counter */
enum PCBLOCK_TEMPERATURE temperature; /* block temperature: cold, warm, hot */ uint rlocks; /* read locks counter */
uint rlocks_queue; /* rd. locks waiting wr. lock of this thread */
enum PCBLOCK_TEMPERATURE temperature; /* block temperature: cold, warm, hot*/
enum pagecache_page_type type; /* type of the block */ enum pagecache_page_type type; /* type of the block */
uint hits_left; /* number of hits left until promotion */ uint hits_left; /* number of hits left until promotion */
/** @brief LSN when first became dirty; LSN_MAX means "not yet set" */ /** @brief LSN when first became dirty; LSN_MAX means "not yet set" */
@@ -1938,6 +1939,8 @@ restart:
} }
pagecache->blocks_unused--; pagecache->blocks_unused--;
DBUG_ASSERT(block->wlocks == 0); DBUG_ASSERT(block->wlocks == 0);
DBUG_ASSERT(block->rlocks == 0);
DBUG_ASSERT(block->rlocks_queue == 0);
DBUG_ASSERT(block->pins == 0); DBUG_ASSERT(block->pins == 0);
block->status= 0; block->status= 0;
#ifndef DBUG_OFF #ifndef DBUG_OFF
@@ -2003,6 +2006,8 @@ restart:
} }
PCBLOCK_INFO(block); PCBLOCK_INFO(block);
DBUG_ASSERT(block->wlocks == 0); DBUG_ASSERT(block->wlocks == 0);
DBUG_ASSERT(block->rlocks == 0);
DBUG_ASSERT(block->rlocks_queue == 0);
DBUG_ASSERT(block->pins == 0); DBUG_ASSERT(block->pins == 0);
if (block->hash_link != hash_link && if (block->hash_link != hash_link &&
@@ -2010,6 +2015,8 @@ restart:
{ {
/* this is a primary request for a new page */ /* this is a primary request for a new page */
DBUG_ASSERT(block->wlocks == 0); DBUG_ASSERT(block->wlocks == 0);
DBUG_ASSERT(block->rlocks == 0);
DBUG_ASSERT(block->rlocks_queue == 0);
DBUG_ASSERT(block->pins == 0); DBUG_ASSERT(block->pins == 0);
block->status|= PCBLOCK_IN_SWITCH; block->status|= PCBLOCK_IN_SWITCH;
@@ -2196,17 +2203,82 @@ static void info_change_lock(PAGECACHE_BLOCK_LINK *block, my_bool wl)
#define info_change_lock(B,W) #define info_change_lock(B,W)
#endif #endif
/*
Put on the block write lock
SYNOPSIS /**
get_wrlock() @brief waiting for lock for read and write lock
pagecache pointer to a page cache data structure
block the block to work with
RETURN @parem pagecache pointer to a page cache data structure
0 - OK @parem block the block to work with
1 - Can't lock this block, need retry @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 translog_wait_lock(PAGECACHE *pagecache,
PAGECACHE_BLOCK_LINK *block,
PAGECACHE_FILE file,
pgcache_page_no_t pageno,
uint lock_type)
{
/* Lock failed we will wait */
#ifdef THREAD
struct st_my_thread_var *thread= my_thread_var;
DBUG_ENTER("translog_wait_lock");
DBUG_PRINT("info", ("fail to lock, waiting... 0x%lx", (ulong)block));
thread->lock_type= lock_type;
wqueue_add_to_queue(&block->wqueue[COND_FOR_WRLOCK], thread);
dec_counter_for_resize_op(pagecache);
do
{
KEYCACHE_DBUG_PRINT("get_wrlock: wait",
("suspend thread %ld", thread->id));
pagecache_pthread_cond_wait(&thread->suspend,
&pagecache->cache_lock);
}
while(thread->next);
#else
DBUG_ASSERT(0);
#endif
PCBLOCK_INFO(block);
if ((block->status & (PCBLOCK_REASSIGNED | PCBLOCK_IN_SWITCH)) ||
file.file != block->hash_link->file.file ||
pageno != block->hash_link->pageno)
{
DBUG_PRINT("info", ("the block 0x%lx changed => need retry "
"status: %x files %d != %d or pages %lu != %lu",
(ulong)block, block->status,
file.file, block->hash_link->file.file,
(ulong) pageno, (ulong) block->hash_link->pageno));
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, static my_bool get_wrlock(PAGECACHE *pagecache,
@@ -2214,11 +2286,7 @@ static my_bool get_wrlock(PAGECACHE *pagecache,
{ {
PAGECACHE_FILE file= block->hash_link->file; PAGECACHE_FILE file= block->hash_link->file;
pgcache_page_no_t pageno= block->hash_link->pageno; pgcache_page_no_t pageno= block->hash_link->pageno;
#ifdef THREAD
pthread_t locker= pthread_self(); pthread_t locker= pthread_self();
#else
int locker= 0;
#endif
DBUG_ENTER("get_wrlock"); DBUG_ENTER("get_wrlock");
DBUG_PRINT("info", ("the block 0x%lx " DBUG_PRINT("info", ("the block 0x%lx "
"files %d(%d) pages %lu(%lu)", "files %d(%d) pages %lu(%lu)",
@@ -2226,37 +2294,17 @@ static my_bool get_wrlock(PAGECACHE *pagecache,
file.file, block->hash_link->file.file, file.file, block->hash_link->file.file,
(ulong) pageno, (ulong) block->hash_link->pageno)); (ulong) pageno, (ulong) block->hash_link->pageno));
PCBLOCK_INFO(block); PCBLOCK_INFO(block);
while (block->wlocks && !pthread_equal(block->write_locker, locker)) /*
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 */ /* Lock failed we will wait */
#ifdef THREAD if (translog_wait_lock(pagecache, block, file, pageno,
struct st_my_thread_var *thread= my_thread_var; MY_PTHREAD_LOCK_WRITE))
DBUG_PRINT("info", ("fail to lock, waiting... 0x%lx", (ulong)block));
wqueue_add_to_queue(&block->wqueue[COND_FOR_WRLOCK], thread);
dec_counter_for_resize_op(pagecache);
do
{
KEYCACHE_DBUG_PRINT("get_wrlock: wait",
("suspend thread %ld", thread->id));
pagecache_pthread_cond_wait(&thread->suspend,
&pagecache->cache_lock);
}
while(thread->next);
#else
DBUG_ASSERT(0);
#endif
PCBLOCK_INFO(block);
if ((block->status & (PCBLOCK_REASSIGNED | PCBLOCK_IN_SWITCH)) ||
file.file != block->hash_link->file.file ||
pageno != block->hash_link->pageno)
{
DBUG_PRINT("info", ("the block 0x%lx changed => need retry "
"status: %x files %d != %d or pages %lu != %lu",
(ulong)block, block->status,
file.file, block->hash_link->file.file,
(ulong) pageno, (ulong) block->hash_link->pageno));
DBUG_RETURN(1); DBUG_RETURN(1);
}
} }
/* we are doing it by global cache mutex protection, so it is OK */ /* we are doing it by global cache mutex protection, so it is OK */
block->wlocks++; block->wlocks++;
@@ -2267,36 +2315,130 @@ static my_bool get_wrlock(PAGECACHE *pagecache,
/* /*
Remove write lock from the block @brief Put on the block read lock
SYNOPSIS @param pagecache pointer to a page cache data structure
release_wrlock() @param block the block to work with
pagecache pointer to a page cache data structure @param user_file Unique handler per handler file. Used to check if
block the block to work with we request many write locks withing the same
statement
RETURN @note see note for get_wrlock().
0 - OK
@retvalue 0 OK
@retvalue 1 Can't lock this block, need retry
*/ */
static void release_wrlock(PAGECACHE_BLOCK_LINK *block) 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 0x%lx "
"files %d(%d) pages %lu(%lu)",
(ulong) 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 (translog_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 0x%lx", (ulong)block));
}
else
{
block->rlocks++;
DBUG_PRINT("info", ("RD lock set, block 0x%lx", (ulong)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"); DBUG_ENTER("release_wrlock");
PCBLOCK_INFO(block); PCBLOCK_INFO(block);
DBUG_ASSERT(block->wlocks > 0); DBUG_ASSERT(block->wlocks > 0);
DBUG_ASSERT(block->rlocks == 0);
DBUG_ASSERT(block->pins > 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--; block->wlocks--;
if (block->wlocks > 0) if (block->wlocks > 0)
DBUG_VOID_RETURN; /* Multiple write locked */ DBUG_VOID_RETURN; /* Multiple write locked */
DBUG_PRINT("info", ("WR lock reset, block 0x%lx", (ulong)block)); DBUG_PRINT("info", ("WR lock reset, block 0x%lx", (ulong)block));
#ifdef THREAD #ifdef THREAD
/* release all threads waiting for write lock */ /* release all threads waiting for read lock or one waiting for write */
if (block->wqueue[COND_FOR_WRLOCK].last_thread) if (block->wqueue[COND_FOR_WRLOCK].last_thread)
wqueue_release_queue(&block->wqueue[COND_FOR_WRLOCK]); wqueue_release_one_locktype_from_queue(&block->wqueue[COND_FOR_WRLOCK]);
#endif #endif
PCBLOCK_INFO(block); PCBLOCK_INFO(block);
DBUG_VOID_RETURN; 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 0x%lx", (ulong)block));
DBUG_VOID_RETURN;
}
DBUG_ASSERT(block->rlocks > 0);
DBUG_ASSERT(block->rlocks_queue == 0);
block->rlocks--;
DBUG_PRINT("info", ("RD lock decreased, block 0x%lx", (ulong)block));
if (block->rlocks > 0)
DBUG_VOID_RETURN; /* Multiple write locked */
DBUG_PRINT("info", ("RD lock reset, block 0x%lx", (ulong)block));
#ifdef THREAD
/* 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]);
#endif
PCBLOCK_INFO(block);
DBUG_VOID_RETURN;
}
/* /*
Try to lock/unlock and pin/unpin the block Try to lock/unlock and pin/unpin the block
@@ -2325,9 +2467,10 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
#ifndef DBUG_OFF #ifndef DBUG_OFF
if (block) if (block)
{ {
DBUG_PRINT("enter", ("block: 0x%lx (%u) wrlocks: %u pins: %u lock: %s pin: %s", DBUG_PRINT("enter", ("block: 0x%lx (%u) wrlocks: %u rdlocks: %u "
"rdlocks_q: %u pins: %u lock: %s pin: %s",
(ulong)block, PCBLOCK_NUMBER(pagecache, block), (ulong)block, PCBLOCK_NUMBER(pagecache, block),
block->wlocks, block->wlocks, block->rlocks, block->rlocks_queue,
block->pins, block->pins,
page_cache_page_lock_str[lock], page_cache_page_lock_str[lock],
page_cache_page_pin_str[pin])); page_cache_page_pin_str[pin]));
@@ -2340,7 +2483,7 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
/* Writelock and pin the buffer */ /* Writelock and pin the buffer */
if (get_wrlock(pagecache, block)) if (get_wrlock(pagecache, block))
{ {
/* can't lock => need retry */ /* Couldn't lock because block changed status => need retry */
goto retry; goto retry;
} }
@@ -2350,13 +2493,13 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
break; break;
case PAGECACHE_LOCK_WRITE_TO_READ: /* write -> read */ case PAGECACHE_LOCK_WRITE_TO_READ: /* write -> read */
case PAGECACHE_LOCK_WRITE_UNLOCK: /* write -> free */ case PAGECACHE_LOCK_WRITE_UNLOCK: /* write -> free */
/* /* Removes write lock and puts read lock */
Removes write lock and puts read lock (which is nothing in our release_wrlock(block, lock == PAGECACHE_LOCK_WRITE_TO_READ);
implementation)
*/
release_wrlock(block);
/* fall through */ /* fall through */
case PAGECACHE_LOCK_READ_UNLOCK: /* read -> free */ 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 */ case PAGECACHE_LOCK_LEFT_READLOCKED: /* read -> read */
if (pin == PAGECACHE_UNPIN) if (pin == PAGECACHE_UNPIN)
{ {
@@ -2373,6 +2516,12 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
} }
break; break;
case PAGECACHE_LOCK_READ: /* free -> read */ 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) if (pin == PAGECACHE_PIN)
{ {
/* The cache is locked so nothing afraid off */ /* The cache is locked so nothing afraid off */
@@ -2385,6 +2534,7 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
{ {
remove_pin(block); remove_pin(block);
} }
/* fall through */
case PAGECACHE_LOCK_LEFT_WRITELOCKED: /* write -> write */ case PAGECACHE_LOCK_LEFT_WRITELOCKED: /* write -> write */
break; /* do nothing */ break; /* do nothing */
default: default:
@@ -3004,6 +3154,9 @@ restart:
test((pin == PAGECACHE_PIN_LEFT_UNPINNED) || test((pin == PAGECACHE_PIN_LEFT_UNPINNED) ||
(pin == PAGECACHE_PIN)), (pin == PAGECACHE_PIN)),
&page_st); &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)) if (((block->status & PCBLOCK_ERROR) == 0) && (page_st != PAGE_READ))
{ {
DBUG_PRINT("info", ("read block 0x%lx", (ulong)block)); DBUG_PRINT("info", ("read block 0x%lx", (ulong)block));
@@ -3741,6 +3894,8 @@ static void free_block(PAGECACHE *pagecache, PAGECACHE_BLOCK_LINK *block)
unlink_changed(block); unlink_changed(block);
DBUG_ASSERT(block->wlocks == 0); DBUG_ASSERT(block->wlocks == 0);
DBUG_ASSERT(block->rlocks == 0);
DBUG_ASSERT(block->rlocks_queue == 0);
DBUG_ASSERT(block->pins == 0); DBUG_ASSERT(block->pins == 0);
block->status= 0; block->status= 0;
#ifndef DBUG_OFF #ifndef DBUG_OFF
@@ -3847,6 +4002,7 @@ static int flush_cached_blocks(PAGECACHE *pagecache,
if (make_lock_and_pin(pagecache, block, if (make_lock_and_pin(pagecache, block,
PAGECACHE_LOCK_WRITE, PAGECACHE_PIN)) PAGECACHE_LOCK_WRITE, PAGECACHE_PIN))
DBUG_ASSERT(0); DBUG_ASSERT(0);
DBUG_ASSERT(block->pins == 1);
KEYCACHE_DBUG_PRINT("flush_cached_blocks", KEYCACHE_DBUG_PRINT("flush_cached_blocks",
("block: %u (0x%lx) to be flushed", ("block: %u (0x%lx) to be flushed",

4
storage/maria/unittest/Makefile.am
View File

@@ -38,6 +38,7 @@ noinst_PROGRAMS = ma_control_file-t trnman-t lockman2-t \
ma_pagecache_consist_64kRD-t \ ma_pagecache_consist_64kRD-t \
ma_pagecache_consist_1kWR-t \ ma_pagecache_consist_1kWR-t \
ma_pagecache_consist_64kWR-t \ ma_pagecache_consist_64kWR-t \
ma_pagecache_rwconsist_1k-t \
ma_test_loghandler-t \ ma_test_loghandler-t \
ma_test_loghandler_multigroup-t \ ma_test_loghandler_multigroup-t \
ma_test_loghandler_multithread-t \ ma_test_loghandler_multithread-t \
@@ -95,6 +96,9 @@ ma_pagecache_consist_1kWR_t_CPPFLAGS = $(ma_pagecache_common_cppflags) -DPAGE_S
ma_pagecache_consist_64kWR_t_SOURCES = $(ma_pagecache_consist_src) ma_pagecache_consist_64kWR_t_SOURCES = $(ma_pagecache_consist_src)
ma_pagecache_consist_64kWR_t_CPPFLAGS = $(ma_pagecache_common_cppflags) -DPAGE_SIZE=65536 -DTEST_WRITERS ma_pagecache_consist_64kWR_t_CPPFLAGS = $(ma_pagecache_common_cppflags) -DPAGE_SIZE=65536 -DTEST_WRITERS
ma_pagecache_rwconsist_1k_t_SOURCES = ma_pagecache_rwconsist.c
ma_pagecache_rwconsist_1k_t_CPPFLAGS = -DPAGE_SIZE=1024
# the generic lock manager may not be used in the end and lockman1-t crashes, # the generic lock manager may not be used in the end and lockman1-t crashes,
# so we don't build lockman-t and lockman1-t # so we don't build lockman-t and lockman1-t
CLEANFILES = maria_log_control page_cache_test_file_1 \ CLEANFILES = maria_log_control page_cache_test_file_1 \

57
storage/maria/unittest/ma_pagecache_consist.c
View File

@@ -266,46 +266,49 @@ void writer(int num)
static void *test_thread_reader(void *arg) static void *test_thread_reader(void *arg)
{ {
int param=*((int*) arg); int param=*((int*) arg);
DBUG_ENTER("test_reader");
my_thread_init(); my_thread_init();
{
DBUG_ENTER("test_reader");
DBUG_PRINT("enter", ("param: %d", param));
DBUG_PRINT("enter", ("param: %d", param)); reader(param);
reader(param); DBUG_PRINT("info", ("Thread %s ended\n", my_thread_name()));
pthread_mutex_lock(&LOCK_thread_count);
DBUG_PRINT("info", ("Thread %s ended\n", my_thread_name())); ok(1, "reader%d: done\n", param);
pthread_mutex_lock(&LOCK_thread_count); thread_count--;
ok(1, "reader%d: done\n", param); VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
thread_count--; pthread_mutex_unlock(&LOCK_thread_count);
VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */ free((uchar*) arg);
pthread_mutex_unlock(&LOCK_thread_count); my_thread_end();
free((uchar*) arg); }
my_thread_end(); return 0;
DBUG_RETURN(0);
} }
static void *test_thread_writer(void *arg) static void *test_thread_writer(void *arg)
{ {
int param=*((int*) arg); int param=*((int*) arg);
DBUG_ENTER("test_writer");
my_thread_init(); my_thread_init();
DBUG_PRINT("enter", ("param: %d", param)); {
DBUG_ENTER("test_writer");
DBUG_PRINT("enter", ("param: %d", param));
writer(param); writer(param);
DBUG_PRINT("info", ("Thread %s ended\n", my_thread_name())); DBUG_PRINT("info", ("Thread %s ended\n", my_thread_name()));
pthread_mutex_lock(&LOCK_thread_count); pthread_mutex_lock(&LOCK_thread_count);
ok(1, "writer%d: done\n", param); ok(1, "writer%d: done\n", param);
thread_count--; thread_count--;
VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */ VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
pthread_mutex_unlock(&LOCK_thread_count); pthread_mutex_unlock(&LOCK_thread_count);
free((uchar*) arg); free((uchar*) arg);
my_thread_end(); my_thread_end();
DBUG_RETURN(0); }
return 0;
} }
int main(int argc __attribute__((unused)), int main(int argc __attribute__((unused)),
char **argv __attribute__((unused))) char **argv __attribute__((unused)))
{ {

348
storage/maria/unittest/ma_pagecache_rwconsist.c Normal file
View File

@@ -0,0 +1,348 @@
/*
TODO: use pthread_join instead of wait_for_thread_count_to_be_zero, like in
my_atomic-t.c (see BUG#22320).
*/
#include <tap.h>
#include <my_sys.h>
#include <m_string.h>
#include "test_file.h"
#include <tap.h>
#define PCACHE_SIZE (PAGE_SIZE*1024*8)
#ifndef DBUG_OFF
static const char* default_dbug_option;
#endif
#define SLEEP usleep(5)
static char *file1_name= (char*)"page_cache_test_file_1";
static PAGECACHE_FILE file1;
static pthread_cond_t COND_thread_count;
static pthread_mutex_t LOCK_thread_count;
static uint thread_count;
static PAGECACHE pagecache;
static uint number_of_readers= 5;
static uint number_of_writers= 5;
static uint number_of_read_tests= 2000;
static uint number_of_write_tests= 1000;
static uint read_sleep_limit= 3;
static uint report_divisor= 50;
/**
@brief Dummy pagecache callback.
*/
static my_bool
dummy_callback(uchar *page __attribute__((unused)),
pgcache_page_no_t page_no __attribute__((unused)),
uchar* data_ptr __attribute__((unused)))
{
return 0;
}
/**
@brief Dummy pagecache callback.
*/
static void
dummy_fail_callback(uchar* data_ptr __attribute__((unused)))
{
return;
}
/**
@brief Checks page consistency
@param buff pointer to the page content
@param task task ID
*/
void check_page(uchar *buff, int task)
{
uint i;
DBUG_ENTER("check_page");
for (i= 1; i < PAGE_SIZE; i++)
{
if (buff[0] != buff[i])
goto err;
}
DBUG_VOID_RETURN;
err:
diag("Task %d char #%u '%u' != '%u'", task, i, (uint) buff[0],
(uint) buff[i]);
DBUG_PRINT("err", ("try to flush"));
exit(1);
}
void reader(int num)
{
unsigned char *buff;
uint i;
PAGECACHE_BLOCK_LINK *link;
for (i= 0; i < number_of_read_tests; i++)
{
if (i % report_divisor == 0)
diag("Reader %d - %u", num, i);
buff= pagecache_read(&pagecache, &file1, 0, 3, NULL,
PAGECACHE_PLAIN_PAGE,
PAGECACHE_LOCK_READ,
&link);
check_page(buff, num);
pagecache_unlock_by_link(&pagecache, link,
PAGECACHE_LOCK_READ_UNLOCK,
PAGECACHE_UNPIN, 0, 0, 0);
{
int lim= rand() % read_sleep_limit;
int j;
for (j= 0; j < lim; j++)
SLEEP;
}
}
}
void writer(int num)
{
uint i;
uchar *buff;
PAGECACHE_BLOCK_LINK *link;
for (i= 0; i < number_of_write_tests; i++)
{
uchar c= (uchar) rand() % 256;
if (i % report_divisor == 0)
diag("Writer %d - %u", num, i);
buff= pagecache_read(&pagecache, &file1, 0, 3, NULL,
PAGECACHE_PLAIN_PAGE,
PAGECACHE_LOCK_WRITE,
&link);
check_page(buff, num);
bfill(buff, PAGE_SIZE / 2, c);
SLEEP;
bfill(buff + PAGE_SIZE/2, PAGE_SIZE / 2, c);
check_page(buff, num);
pagecache_unlock_by_link(&pagecache, link,
PAGECACHE_LOCK_WRITE_UNLOCK,
PAGECACHE_UNPIN, 0, 0, 1);
SLEEP;
}
}
static void *test_thread_reader(void *arg)
{
int param=*((int*) arg);
my_thread_init();
{
DBUG_ENTER("test_reader");
DBUG_PRINT("enter", ("param: %d", param));
reader(param);
DBUG_PRINT("info", ("Thread %s ended\n", my_thread_name()));
pthread_mutex_lock(&LOCK_thread_count);
ok(1, "reader%d: done\n", param);
thread_count--;
VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
pthread_mutex_unlock(&LOCK_thread_count);
free((uchar*) arg);
my_thread_end();
}
return 0;
}
static void *test_thread_writer(void *arg)
{
int param=*((int*) arg);
my_thread_init();
{
DBUG_ENTER("test_writer");
writer(param);
DBUG_PRINT("info", ("Thread %s ended\n", my_thread_name()));
pthread_mutex_lock(&LOCK_thread_count);
ok(1, "writer%d: done\n", param);
thread_count--;
VOID(pthread_cond_signal(&COND_thread_count)); /* Tell main we are ready */
pthread_mutex_unlock(&LOCK_thread_count);
free((uchar*) arg);
my_thread_end();
}
return 0;
}
int main(int argc __attribute__((unused)),
char **argv __attribute__((unused)))
{
pthread_t tid;
pthread_attr_t thr_attr;
int *param, error, pagen;
MY_INIT(argv[0]);
#ifndef DBUG_OFF
#if defined(__WIN__)
default_dbug_option= "d:t:i:O,\\test_pagecache_consist.trace";
#else
default_dbug_option= "d:t:i:o,/tmp/test_pagecache_consist.trace";
#endif
if (argc > 1)
{
DBUG_SET(default_dbug_option);
DBUG_SET_INITIAL(default_dbug_option);
}
#endif
{
DBUG_ENTER("main");
DBUG_PRINT("info", ("Main thread: %s\n", my_thread_name()));
plan(number_of_writers + number_of_readers);
SKIP_BIG_TESTS(number_of_writers + number_of_readers)
{
if ((file1.file= my_open(file1_name,
O_CREAT | O_TRUNC | O_RDWR, MYF(0))) == -1)
{
diag( "Got error during file1 creation from open() (errno: %d)\n",
errno);
exit(1);
}
pagecache_file_init(file1, &dummy_callback, &dummy_callback,
&dummy_fail_callback, &dummy_callback, NULL);
DBUG_PRINT("info", ("file1: %d", file1.file));
if (my_chmod(file1_name, S_IRWXU | S_IRWXG | S_IRWXO, MYF(MY_WME)))
exit(1);
my_pwrite(file1.file, (const uchar*) "test file", 9, 0, MYF(0));
if ((error= pthread_cond_init(&COND_thread_count, NULL)))
{
diag( "COND_thread_count: %d from pthread_cond_init (errno: %d)\n",
error, errno);
exit(1);
}
if ((error= pthread_mutex_init(&LOCK_thread_count, MY_MUTEX_INIT_FAST)))
{
diag( "LOCK_thread_count: %d from pthread_cond_init (errno: %d)\n",
error, errno);
exit(1);
}
if ((error= pthread_attr_init(&thr_attr)))
{
diag("Got error: %d from pthread_attr_init (errno: %d)\n",
error,errno);
exit(1);
}
if ((error= pthread_attr_setdetachstate(&thr_attr, PTHREAD_CREATE_DETACHED)))
{
diag(
"Got error: %d from pthread_attr_setdetachstate (errno: %d)\n",
error,errno);
exit(1);
}
#ifdef HAVE_THR_SETCONCURRENCY
VOID(thr_setconcurrency(2));
#endif
if ((pagen= init_pagecache(&pagecache, PCACHE_SIZE, 0, 0,
PAGE_SIZE, 0)) == 0)
{
diag("Got error: init_pagecache() (errno: %d)\n",
errno);
exit(1);
}
DBUG_PRINT("info", ("Page cache %d pages", pagen));
{
unsigned char *buffr= malloc(PAGE_SIZE);
memset(buffr, '\0', PAGE_SIZE);
pagecache_write(&pagecache, &file1, 0, 3, buffr,
PAGECACHE_PLAIN_PAGE,
PAGECACHE_LOCK_LEFT_UNLOCKED,
PAGECACHE_PIN_LEFT_UNPINNED,
PAGECACHE_WRITE_DELAY,
0, LSN_IMPOSSIBLE);
}
pthread_mutex_lock(&LOCK_thread_count);
while (number_of_readers != 0 || number_of_writers != 0)
{
if (number_of_readers != 0)
{
param=(int*) malloc(sizeof(int));
*param= number_of_readers + number_of_writers;
if ((error= pthread_create(&tid, &thr_attr, test_thread_reader,
(void*) param)))
{
diag("Got error: %d from pthread_create (errno: %d)\n",
error,errno);
exit(1);
}
thread_count++;
number_of_readers--;
}
if (number_of_writers != 0)
{
param=(int*) malloc(sizeof(int));
*param= number_of_writers + number_of_readers;
if ((error= pthread_create(&tid, &thr_attr, test_thread_writer,
(void*) param)))
{
diag("Got error: %d from pthread_create (errno: %d)\n",
error,errno);
exit(1);
}
thread_count++;
number_of_writers--;
}
}
DBUG_PRINT("info", ("Thread started"));
pthread_mutex_unlock(&LOCK_thread_count);
pthread_attr_destroy(&thr_attr);
/* wait finishing */
pthread_mutex_lock(&LOCK_thread_count);
while (thread_count)
{
if ((error= pthread_cond_wait(&COND_thread_count,&LOCK_thread_count)))
diag("COND_thread_count: %d from pthread_cond_wait\n",error);
}
pthread_mutex_unlock(&LOCK_thread_count);
DBUG_PRINT("info", ("thread ended"));
end_pagecache(&pagecache, 1);
DBUG_PRINT("info", ("Page cache ended"));
if (my_close(file1.file, MYF(0)) != 0)
{
diag( "Got error during file1 closing from close() (errno: %d)\n",
errno);
exit(1);
}
my_delete(file1_name, MYF(0));
DBUG_PRINT("info", ("file1 (%d) closed", file1.file));
DBUG_PRINT("info", ("Program end"));
} /* SKIP_BIG_TESTS */
my_end(0);
return exit_status();
}
}