mirror/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-01-16 03:52:35 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge 88-214-96-46.dialup.umc.net.ua:/Users/bell/mysql/bk/mysql-maria

Browse source 
into  88-214-96-46.dialup.umc.net.ua:/Users/bell/mysql/bk/work-maria-test


storage/maria/ma_pagecache.c:
  Auto merged
storage/maria/unittest/Makefile.am:
  merge
storage/maria/unittest/ma_pagecache_consist.c:
  merge
This commit is contained in:
unknown 2008-02-22 23:21:27 +02:00
commit daacf696e5
7 changed files with 652 additions and 93 deletions
Show stats 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
#define MY_PTHREAD_LOCK_READ 0
#define MY_PTHREAD_LOCK_WRITE 1
struct st_my_thread_var
{
int thr_errno;
@ -688,6 +691,7 @@ struct st_my_thread_var
my_bool init;
struct st_my_thread_var *next,**prev;
void *opt_info;
uint lock_type; /* used by conditional release the queue */
#ifndef DBUG_OFF
void *dbug;
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,
pthread_mutex_t *lock);
void wqueue_release_queue(WQUEUE *wqueue);
void wqueue_release_one_locktype_from_queue(WQUEUE *wqueue);
#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

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
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"
@ -96,7 +99,7 @@
#define PCBLOCK_INFO(B) \
DBUG_PRINT("info", \
("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)->hash_link ? \
(B)->hash_link->file.file : \
@ -110,7 +113,7 @@
(uint)((B)->hash_link ? \
(B)->hash_link->requests : \
0), \
block->wlocks, \
block->wlocks, block->rlocks, block->rlocks_queue, \
(uint)(B)->pins))
/* TODO: put it to my_static.c */
@ -298,11 +301,7 @@ struct st_pagecache_block_link
#endif
KEYCACHE_CONDVAR *condvar; /* condition variable for 'no readers' event */
uchar *buffer; /* buffer for the block page */
#ifdef THREAD
pthread_t write_locker;
#else
int write_locker;
#endif
ulonglong last_hit_time; /* timestamp of the last hit */
WQUEUE
@ -311,7 +310,9 @@ struct st_pagecache_block_link
uint status; /* state of the block */
uint pins; /* pin 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 */
uint hits_left; /* number of hits left until promotion */
/** @brief LSN when first became dirty; LSN_MAX means "not yet set" */
@ -1938,6 +1939,8 @@ restart:
}
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;
#ifndef DBUG_OFF
@ -2003,6 +2006,8 @@ restart:
}
PCBLOCK_INFO(block);
DBUG_ASSERT(block->wlocks == 0);
DBUG_ASSERT(block->rlocks == 0);
DBUG_ASSERT(block->rlocks_queue == 0);
DBUG_ASSERT(block->pins == 0);
if (block->hash_link != hash_link &&
@ -2010,6 +2015,8 @@ restart:
{
/* 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;
@ -2196,17 +2203,82 @@ static void info_change_lock(PAGECACHE_BLOCK_LINK *block, my_bool wl)
#define info_change_lock(B,W)
#endif
/*
Put on the block write lock
SYNOPSIS
get_wrlock()
pagecache pointer to a page cache data structure
block the block to work with
/**
@brief waiting for lock for read and write lock
RETURN
0 - OK
1 - Can't lock this block, need retry
@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 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,
@ -2214,11 +2286,7 @@ static my_bool get_wrlock(PAGECACHE *pagecache,
{
PAGECACHE_FILE file= block->hash_link->file;
pgcache_page_no_t pageno= block->hash_link->pageno;
#ifdef THREAD
pthread_t locker= pthread_self();
#else
int locker= 0;
#endif
DBUG_ENTER("get_wrlock");
DBUG_PRINT("info", ("the block 0x%lx "
"files %d(%d) pages %lu(%lu)",
@ -2226,37 +2294,17 @@ static my_bool get_wrlock(PAGECACHE *pagecache,
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))
/*
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 */
#ifdef THREAD
struct st_my_thread_var *thread= my_thread_var;
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));
if (translog_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++;
@ -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
release_wrlock()
pagecache pointer to a page cache data structure
block the block to work with
@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
RETURN
0 - OK
@note see note for get_wrlock().
@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");
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 0x%lx", (ulong)block));
#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)
wqueue_release_queue(&block->wqueue[COND_FOR_WRLOCK]);
wqueue_release_one_locktype_from_queue(&block->wqueue[COND_FOR_WRLOCK]);
#endif
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 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
@ -2325,9 +2467,10 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
#ifndef DBUG_OFF
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),
block->wlocks,
block->wlocks, block->rlocks, block->rlocks_queue,
block->pins,
page_cache_page_lock_str[lock],
page_cache_page_pin_str[pin]));
@ -2340,7 +2483,7 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
/* Writelock and pin the buffer */
if (get_wrlock(pagecache, block))
{
/* can't lock => need retry */
/* Couldn't lock because block changed status => need retry */
goto retry;
}
@ -2350,13 +2493,13 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
break;
case PAGECACHE_LOCK_WRITE_TO_READ: /* write -> read */
case PAGECACHE_LOCK_WRITE_UNLOCK: /* write -> free */
/*
Removes write lock and puts read lock (which is nothing in our
implementation)
*/
release_wrlock(block);
/* 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)
{
@ -2373,6 +2516,12 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
}
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 */
@ -2385,6 +2534,7 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
{
remove_pin(block);
}
/* fall through */
case PAGECACHE_LOCK_LEFT_WRITELOCKED: /* write -> write */
break; /* do nothing */
default:
@ -3009,6 +3159,9 @@ restart:
test((pin == PAGECACHE_PIN_LEFT_UNPINNED) ||
(pin == PAGECACHE_PIN)),
&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))
{
DBUG_PRINT("info", ("read block 0x%lx", (ulong)block));
@ -3751,6 +3904,8 @@ static void free_block(PAGECACHE *pagecache, PAGECACHE_BLOCK_LINK *block)
unlink_changed(block);
DBUG_ASSERT(block->wlocks == 0);
DBUG_ASSERT(block->rlocks == 0);
DBUG_ASSERT(block->rlocks_queue == 0);
DBUG_ASSERT(block->pins == 0);
block->status= 0;
#ifndef DBUG_OFF
@ -3857,6 +4012,7 @@ static int flush_cached_blocks(PAGECACHE *pagecache,
if (make_lock_and_pin(pagecache, block,
PAGECACHE_LOCK_WRITE, PAGECACHE_PIN))
DBUG_ASSERT(0);
DBUG_ASSERT(block->pins == 1);
KEYCACHE_DBUG_PRINT("flush_cached_blocks",
("block: %u (0x%lx) to be flushed",

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

@ -39,6 +39,7 @@ noinst_PROGRAMS = ma_control_file-t trnman-t lockman2-t \
ma_pagecache_consist_64kRD-t \
ma_pagecache_consist_1kWR-t \
ma_pagecache_consist_64kWR-t \
ma_pagecache_rwconsist_1k-t \
ma_test_loghandler-t \
ma_test_loghandler_multigroup-t \
ma_test_loghandler_multithread-t \
@ -96,6 +97,9 @@ ma_pagecache_consist_1kWR_t_CPPFLAGS = $(ma_pagecache_common_cppflags) -DTEST_P
ma_pagecache_consist_64kWR_t_SOURCES = $(ma_pagecache_consist_src)
ma_pagecache_consist_64kWR_t_CPPFLAGS = $(ma_pagecache_common_cppflags) -DTEST_PAGE_SIZE=65536 -DTEST_WRITERS
ma_pagecache_rwconsist_1k_t_SOURCES = ma_pagecache_rwconsist.c
ma_pagecache_rwconsist_1k_t_CPPFLAGS = -DTEST_PAGE_SIZE=1024
# 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
CLEANFILES = maria_log_control page_cache_test_file_1 \

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

@ -273,46 +273,49 @@ void writer(int num)
static void *test_thread_reader(void *arg)
{
int param=*((int*) arg);
DBUG_ENTER("test_reader");
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);
ok(1, "reader%d: done", 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();
DBUG_RETURN(0);
DBUG_PRINT("info", ("Thread %s ended\n", my_thread_name()));
pthread_mutex_lock(&LOCK_thread_count);
ok(1, "reader%d: done", 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);
DBUG_ENTER("test_writer");
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()));
pthread_mutex_lock(&LOCK_thread_count);
ok(1, "writer%d: done", 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();
DBUG_RETURN(0);
DBUG_PRINT("info", ("Thread %s ended\n", my_thread_name()));
pthread_mutex_lock(&LOCK_thread_count);
ok(1, "writer%d: done", 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)))
{

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();
}
}