database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-07-27 18:02:13 +03:00
Code Activity

multithreaded repair-by-sort code

Browse Source 
parallel read access to IO_CACHE
This commit is contained in:
serg@serg.mysql.com
2002-06-19 23:54:45 +00:00
parent 2874917198
commit 5c83ae3fca
8 changed files with 1255 additions and 497 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
include/my_sys.h
View File

@ -36,7 +36,7 @@ extern int NEAR my_errno; /* Last error in mysys */
#include <m_ctype.h> /* for CHARSET_INFO */ #include <m_ctype.h> /* for CHARSET_INFO */
#endif #endif
#include <stdarg.h> #include <stdarg.h>
#define MYSYS_PROGRAM_USES_CURSES() { error_handler_hook = my_message_curses; mysys_uses_curses=1; } #define MYSYS_PROGRAM_USES_CURSES() { error_handler_hook = my_message_curses; mysys_uses_curses=1; }
#define MYSYS_PROGRAM_DONT_USE_CURSES() { error_handler_hook = my_message_no_curses; mysys_uses_curses=0;} #define MYSYS_PROGRAM_DONT_USE_CURSES() { error_handler_hook = my_message_no_curses; mysys_uses_curses=0;}
@ -278,7 +278,7 @@ extern struct my_file_info
{ {
my_string name; my_string name;
enum file_type type; enum file_type type;
#if defined(THREAD) && !defined(HAVE_PREAD) #if defined(THREAD) && !defined(HAVE_PREAD)
pthread_mutex_t mutex; pthread_mutex_t mutex;
#endif #endif
} my_file_info[MY_NFILE]; } my_file_info[MY_NFILE];
@ -299,6 +299,45 @@ typedef struct st_dynamic_string {
struct st_io_cache; struct st_io_cache;
typedef int (*IO_CACHE_CALLBACK)(struct st_io_cache*); typedef int (*IO_CACHE_CALLBACK)(struct st_io_cache*);
#ifdef THREAD
typedef struct st_io_cache_share
{
/* to sync on reads into buffer */
pthread_mutex_t mutex;
pthread_cond_t cond;
int count;
/* actual IO_CACHE that filled the buffer */
struct st_io_cache *active;
/* the following will go implemented whenever the need arises */
#ifdef NOT_IMPLEMENTED
/* whether the structure should be free'd */
my_bool alloced;
#endif
} IO_CACHE_SHARE;
#define lock_io_cache(info) \
( \
(errno=pthread_mutex_lock(&((info)->share->mutex))) ? -1 : ( \
(info)->share->count ? ( \
--((info)->share->count), \
pthread_cond_wait(&((info)->share->cond), \
&((info)->share->mutex)), \
(++((info)->share->count) ? \
pthread_mutex_unlock(&((info)->share->mutex)) : 0)) \
: 1 ) \
)
#define unlock_io_cache(info) \
( \
pthread_cond_broadcast(&((info)->share->cond)), \
pthread_mutex_unlock (&((info)->share->mutex)) \
)
/* -- to catch errors
#else
#define lock_io_cache(info)
#define unlock_io_cache(info)
*/
#endif
typedef struct st_io_cache /* Used when cacheing files */ typedef struct st_io_cache /* Used when cacheing files */
{ {
@ -331,10 +370,16 @@ typedef struct st_io_cache /* Used when cacheing files */
WRITE_CACHE, and &read_pos and &read_end respectively otherwise WRITE_CACHE, and &read_pos and &read_end respectively otherwise
*/ */
byte **current_pos, **current_end; byte **current_pos, **current_end;
/* The lock is for append buffer used in SEQ_READ_APPEND cache */
#ifdef THREAD #ifdef THREAD
/* The lock is for append buffer used in SEQ_READ_APPEND cache
need mutex copying from append buffer to read buffer. */
pthread_mutex_t append_buffer_lock; pthread_mutex_t append_buffer_lock;
/* need mutex copying from append buffer to read buffer */ /* The following is used when several threads are reading the
same file in parallel. They are synchronized on disk
accesses reading the cached part of the file asynchronously.
It should be set to NULL to disable the feature. Only
READ_CACHE mode is supported. */
IO_CACHE_SHARE *share;
#endif #endif
/* a caller will use my_b_read() macro to read from the cache /* a caller will use my_b_read() macro to read from the cache
if the data is already in cache, it will be simply copied with if the data is already in cache, it will be simply copied with
@ -626,6 +671,12 @@ extern my_bool reinit_io_cache(IO_CACHE *info,enum cache_type type,
my_off_t seek_offset,pbool use_async_io, my_off_t seek_offset,pbool use_async_io,
pbool clear_cache); pbool clear_cache);
extern int _my_b_read(IO_CACHE *info,byte *Buffer,uint Count); extern int _my_b_read(IO_CACHE *info,byte *Buffer,uint Count);
#ifdef THREAD
extern int _my_b_read_r(IO_CACHE *info,byte *Buffer,uint Count);
extern int init_io_cache_share(IO_CACHE *info,
IO_CACHE_SHARE *s, uint num_threads);
extern int remove_io_thread(IO_CACHE *info);
#endif
extern int _my_b_seq_read(IO_CACHE *info,byte *Buffer,uint Count); extern int _my_b_seq_read(IO_CACHE *info,byte *Buffer,uint Count);
extern int _my_b_net_read(IO_CACHE *info,byte *Buffer,uint Count); extern int _my_b_net_read(IO_CACHE *info,byte *Buffer,uint Count);
extern int _my_b_get(IO_CACHE *info); extern int _my_b_get(IO_CACHE *info);

73
include/myisam.h
View File

@ -314,24 +314,6 @@ typedef struct st_sort_key_blocks { /* Used when sorting */
int inited; int inited;
} SORT_KEY_BLOCKS; } SORT_KEY_BLOCKS;
struct st_mi_check_param;
typedef struct st_sort_info {
MI_INFO *info;
struct st_mi_check_param *param;
enum data_file_type new_data_file_type;
SORT_KEY_BLOCKS *key_block,*key_block_end;
uint key,find_length,real_key_length;
my_off_t pos,max_pos,filepos,start_recpos,filelength,dupp,buff_length;
ha_rows max_records;
ulonglong unique[MI_MAX_KEY_SEG+1];
my_bool fix_datafile;
char *record,*buff;
void *wordlist, *wordptr;
MI_KEYDEF *keyinfo;
MI_KEYSEG *keyseg;
} SORT_INFO;
typedef struct st_mi_check_param typedef struct st_mi_check_param
{ {
ulonglong auto_increment_value; ulonglong auto_increment_value;
@ -354,7 +336,6 @@ typedef struct st_mi_check_param
int tmpfile_createflag; int tmpfile_createflag;
myf myf_rw; myf myf_rw;
IO_CACHE read_cache; IO_CACHE read_cache;
SORT_INFO sort_info;
ulonglong unique_count[MI_MAX_KEY_SEG+1]; ulonglong unique_count[MI_MAX_KEY_SEG+1];
ha_checksum key_crc[MI_MAX_POSSIBLE_KEY]; ha_checksum key_crc[MI_MAX_POSSIBLE_KEY];
ulong rec_per_key_part[MI_MAX_KEY_SEG*MI_MAX_POSSIBLE_KEY]; ulong rec_per_key_part[MI_MAX_KEY_SEG*MI_MAX_POSSIBLE_KEY];
@ -363,17 +344,42 @@ typedef struct st_mi_check_param
char *op_name; char *op_name;
} MI_CHECK; } MI_CHECK;
typedef struct st_sort_info {
typedef struct st_mi_sortinfo { MI_INFO *info;
MI_CHECK *param;
enum data_file_type new_data_file_type;
SORT_KEY_BLOCKS *key_block,*key_block_end;
uint kei, total_keys;
my_off_t filelength,dupp,buff_length;
ha_rows max_records; ha_rows max_records;
SORT_INFO *sort_info; char *buff;
char *tmpdir;
int (*key_cmp)(SORT_INFO *info, const void *, const void *);
int (*key_read)(SORT_INFO *info,void *buff);
int (*key_write)(SORT_INFO *info, const void *buff);
void (*lock_in_memory)(MI_CHECK *info);
uint key_length;
myf myf_rw; myf myf_rw;
/* sync things*/
uint got_error, threads_running;
pthread_mutex_t mutex;
pthread_cond_t cond;
} SORT_INFO;
typedef struct st_mi_sort_param {
pthread_t thr;
IO_CACHE read_cache;
ulonglong unique[MI_MAX_KEY_SEG+1];
uint key, key_length,real_key_length,sortbuff_size;
uint maxbuffers, keys, find_length, sort_keys_length;
uchar **sort_keys;
void *wordlist, *wordptr;
MI_KEYDEF *keyinfo;
SORT_INFO *sort_info;
IO_CACHE tempfile, tempfile_for_exceptions;
DYNAMIC_ARRAY buffpek;
my_off_t pos,max_pos,filepos,start_recpos;
my_bool fix_datafile;
char *record;
char *tmpdir;
int (*key_cmp)(struct st_mi_sort_param *, const void *, const void *);
int (*key_read)(struct st_mi_sort_param *,void *);
int (*key_write)(struct st_mi_sort_param *, const void *);
void (*lock_in_memory)(MI_CHECK *);
} MI_SORT_PARAM; } MI_SORT_PARAM;
/* functions in mi_check */ /* functions in mi_check */
@ -398,14 +404,17 @@ int flush_blocks(MI_CHECK *param, File file);
void update_auto_increment_key(MI_CHECK *param, MI_INFO *info, void update_auto_increment_key(MI_CHECK *param, MI_INFO *info,
my_bool repair); my_bool repair);
int update_state_info(MI_CHECK *param, MI_INFO *info,uint update); int update_state_info(MI_CHECK *param, MI_INFO *info,uint update);
void update_key_parts(MI_KEYDEF *keyinfo, ulong *rec_per_key_part,
ulonglong *unique, ulonglong records);
int filecopy(MI_CHECK *param, File to,File from,my_off_t start, int filecopy(MI_CHECK *param, File to,File from,my_off_t start,
my_off_t length, const char *type); my_off_t length, const char *type);
int movepoint(MI_INFO *info,byte *record,my_off_t oldpos, int movepoint(MI_INFO *info,byte *record,my_off_t oldpos,
my_off_t newpos, uint prot_key); my_off_t newpos, uint prot_key);
int sort_write_record(SORT_INFO *sort_info); int sort_write_record(MI_SORT_PARAM *sort_param);
int write_data_suffix(MI_CHECK *param, MI_INFO *info); int write_data_suffix(SORT_INFO *sort_info, my_bool fix_datafile);
int _create_index_by_sort(MI_SORT_PARAM *info,my_bool no_messages, int _create_index_by_sort(MI_SORT_PARAM *info,my_bool no_messages, ulong);
ulong); void *_thr_find_all_keys(MI_SORT_PARAM *info);
int _thr_write_keys(MI_SORT_PARAM *sort_param);
int test_if_almost_full(MI_INFO *info); int test_if_almost_full(MI_INFO *info);
int recreate_table(MI_CHECK *param, MI_INFO **org_info, char *filename); int recreate_table(MI_CHECK *param, MI_INFO **org_info, char *filename);
void mi_disable_non_unique_index(MI_INFO *info, ha_rows rows); void mi_disable_non_unique_index(MI_INFO *info, ha_rows rows);

64
myisam/mi_cache.c
View File

@ -14,15 +14,27 @@
along with this program; if not, write to the Free Software along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
/* Functions for read record cacheing with myisam */ /*
/* Used instead of my_b_read() to allow for no-cacheed seeks */ Functions for read record cacheing with myisam
Used for reading dynamic/compressed records from datafile.
Can fetch data directly from file (outside cache),
if reading a small chunk straight before the cached part (with possible
overlap).
Can be explicitly asked not to use cache (by not setting READING_NEXT in
flag) - useful for occasional out-of-cache reads, when the next read is
expected to hit the cache again.
Allows "partial read" errors in the record header (when READING_HEADER flag
is set) - unread part is bzero'ed
Note: out-of-cache reads are disabled for shared IO_CACHE's
*/
#include "myisamdef.h" #include "myisamdef.h"
/* Copy block from cache if it`s in it. If re_read_if_possibly is */
/* set read to cache (if after current file-position) else read to */
/* buff */
int _mi_read_cache(IO_CACHE *info, byte *buff, my_off_t pos, uint length, int _mi_read_cache(IO_CACHE *info, byte *buff, my_off_t pos, uint length,
int flag) int flag)
{ {
@ -31,7 +43,7 @@ int _mi_read_cache(IO_CACHE *info, byte *buff, my_off_t pos, uint length,
char *in_buff_pos; char *in_buff_pos;
DBUG_ENTER("_mi_read_cache"); DBUG_ENTER("_mi_read_cache");
if (pos < info->pos_in_file) if (pos < info->pos_in_file && ! info->share)
{ {
read_length=length; read_length=length;
if ((my_off_t) read_length > (my_off_t) (info->pos_in_file-pos)) if ((my_off_t) read_length > (my_off_t) (info->pos_in_file-pos))
@ -44,7 +56,8 @@ int _mi_read_cache(IO_CACHE *info, byte *buff, my_off_t pos, uint length,
pos+=read_length; pos+=read_length;
buff+=read_length; buff+=read_length;
} }
if ((offset= (my_off_t) (pos - info->pos_in_file)) < if (pos >= info->pos_in_file &&
(offset= (my_off_t) (pos - info->pos_in_file)) <
(my_off_t) (info->read_end - info->request_pos)) (my_off_t) (info->read_end - info->request_pos))
{ {
in_buff_pos=info->request_pos+(uint) offset; in_buff_pos=info->request_pos+(uint) offset;
@ -57,10 +70,10 @@ int _mi_read_cache(IO_CACHE *info, byte *buff, my_off_t pos, uint length,
} }
else else
in_buff_length=0; in_buff_length=0;
if (flag & READING_NEXT) if (flag & READING_NEXT || info->share)
{ {
if (pos != ((info)->pos_in_file + if (pos !=
(uint) ((info)->read_end - (info)->request_pos))) (info->pos_in_file + (uint) (info->read_end - info->request_pos)))
{ {
info->pos_in_file=pos; /* Force start here */ info->pos_in_file=pos; /* Force start here */
info->read_pos=info->read_end=info->request_pos; /* Everything used */ info->read_pos=info->read_end=info->request_pos; /* Everything used */
@ -70,34 +83,25 @@ int _mi_read_cache(IO_CACHE *info, byte *buff, my_off_t pos, uint length,
info->read_pos=info->read_end; /* All block used */ info->read_pos=info->read_end; /* All block used */
if (!(*info->read_function)(info,buff,length)) if (!(*info->read_function)(info,buff,length))
DBUG_RETURN(0); DBUG_RETURN(0);
if (!(flag & READING_HEADER) || info->error == -1 || read_length=info->error;
(uint) info->error+in_buff_length < 3) }
{ else
DBUG_PRINT("error", {
("Error %d reading next-multi-part block (Got %d bytes)", info->seek_not_done=1;
my_errno, info->error)); if ((read_length=my_pread(info->file,buff,length,pos,MYF(0))) == length)
if (!my_errno || my_errno == -1) DBUG_RETURN(0);
my_errno=HA_ERR_WRONG_IN_RECORD;
DBUG_RETURN(1);
}
bzero(buff+info->error,MI_BLOCK_INFO_HEADER_LENGTH - in_buff_length -
(uint) info->error);
DBUG_RETURN(0);
} }
info->seek_not_done=1;
if ((read_length=my_pread(info->file,buff,length,pos,MYF(0))) == length)
DBUG_RETURN(0);
if (!(flag & READING_HEADER) || (int) read_length == -1 || if (!(flag & READING_HEADER) || (int) read_length == -1 ||
read_length+in_buff_length < 3) read_length+in_buff_length < 3)
{ {
DBUG_PRINT("error", DBUG_PRINT("error",
("Error %d reading new block (Got %d bytes)", ("Error %d reading next-multi-part block (Got %d bytes)",
my_errno, (int) read_length)); my_errno, (int) read_length));
if (!my_errno || my_errno == -1) if (!my_errno || my_errno == -1)
my_errno=HA_ERR_WRONG_IN_RECORD; my_errno=HA_ERR_WRONG_IN_RECORD;
DBUG_RETURN(1); DBUG_RETURN(1);
} }
bzero(buff+read_length,MI_BLOCK_INFO_HEADER_LENGTH - in_buff_length - bzero(buff+read_length,MI_BLOCK_INFO_HEADER_LENGTH - in_buff_length -
read_length); read_length);
DBUG_RETURN(0); DBUG_RETURN(0);
} /* _mi_read_cache */ } /* _mi_read_cache */

914
myisam/mi_check.c
View File

File diff suppressed because it is too large Load Diff

56
myisam/myisamchk.c
View File

@ -75,7 +75,8 @@ static int mi_sort_records(MI_CHECK *param,
uint sort_key, uint sort_key,
my_bool write_info, my_bool write_info,
my_bool update_index); my_bool update_index);
static int sort_record_index(MI_CHECK *param,MI_INFO *info,MI_KEYDEF *keyinfo, static int sort_record_index(MI_SORT_PARAM *sort_param,MI_INFO *info,
MI_KEYDEF *keyinfo,
my_off_t page,uchar *buff,uint sortkey, my_off_t page,uchar *buff,uint sortkey,
File new_file, my_bool update_index); File new_file, my_bool update_index);
@ -1327,7 +1328,7 @@ static int mi_sort_records(MI_CHECK *param,
register MI_INFO *info, my_string name, register MI_INFO *info, my_string name,
uint sort_key, uint sort_key,
my_bool write_info, my_bool write_info,
my_bool update_index) my_bool update_index)
{ {
int got_error; int got_error;
uint key; uint key;
@ -1337,11 +1338,14 @@ static int mi_sort_records(MI_CHECK *param,
ha_rows old_record_count; ha_rows old_record_count;
MYISAM_SHARE *share=info->s; MYISAM_SHARE *share=info->s;
char llbuff[22],llbuff2[22]; char llbuff[22],llbuff2[22];
SORT_INFO *sort_info= &param->sort_info; SORT_INFO sort_info;
MI_SORT_PARAM sort_param;
DBUG_ENTER("sort_records"); DBUG_ENTER("sort_records");
bzero((char*) sort_info,sizeof(*sort_info)); bzero((char*)&sort_info,sizeof(sort_info));
sort_info->param=param; bzero((char*)&sort_param,sizeof(sort_param));
sort_param.sort_info=&sort_info;
sort_info.param=param;
keyinfo= &share->keyinfo[sort_key]; keyinfo= &share->keyinfo[sort_key];
got_error=1; got_error=1;
temp_buff=0; temp_buff=0;
@ -1377,7 +1381,7 @@ static int mi_sort_records(MI_CHECK *param,
mi_check_print_error(param,"Not enough memory for key block"); mi_check_print_error(param,"Not enough memory for key block");
goto err; goto err;
} }
if (!(sort_info->record=(byte*) my_malloc((uint) share->base.pack_reclength, if (!(sort_param.record=(byte*) my_malloc((uint) share->base.pack_reclength,
MYF(0)))) MYF(0))))
{ {
mi_check_print_error(param,"Not enough memory for record"); mi_check_print_error(param,"Not enough memory for record");
@ -1419,18 +1423,18 @@ static int mi_sort_records(MI_CHECK *param,
} }
/* Setup param for sort_write_record */ /* Setup param for sort_write_record */
sort_info->info=info; sort_info.info=info;
sort_info->new_data_file_type=share->data_file_type; sort_info.new_data_file_type=share->data_file_type;
sort_info->fix_datafile=1; sort_param.fix_datafile=1;
sort_info->filepos=share->pack.header_length; sort_param.filepos=share->pack.header_length;
old_record_count=info->state->records; old_record_count=info->state->records;
info->state->records=0; info->state->records=0;
if (sort_info->new_data_file_type != COMPRESSED_RECORD) if (sort_info.new_data_file_type != COMPRESSED_RECORD)
share->state.checksum=0; share->state.checksum=0;
if (sort_record_index(param, info,keyinfo,share->state.key_root[sort_key], if (sort_record_index(&sort_param,info,keyinfo,share->state.key_root[sort_key],
temp_buff, sort_key,new_file,update_index) || temp_buff, sort_key,new_file,update_index) ||
write_data_suffix(param, info) || write_data_suffix(&sort_info,1) ||
flush_io_cache(&info->rec_cache)) flush_io_cache(&info->rec_cache))
goto err; goto err;
@ -1448,7 +1452,7 @@ static int mi_sort_records(MI_CHECK *param,
info->state->del=0; info->state->del=0;
info->state->empty=0; info->state->empty=0;
share->state.dellink= HA_OFFSET_ERROR; share->state.dellink= HA_OFFSET_ERROR;
info->state->data_file_length=sort_info->filepos; info->state->data_file_length=sort_param.filepos;
share->state.split=info->state->records; /* Only hole records */ share->state.split=info->state->records; /* Only hole records */
share->state.version=(ulong) time((time_t*) 0); share->state.version=(ulong) time((time_t*) 0);
@ -1472,11 +1476,11 @@ err:
{ {
my_afree((gptr) temp_buff); my_afree((gptr) temp_buff);
} }
my_free(sort_info->record,MYF(MY_ALLOW_ZERO_PTR)); my_free(sort_param.record,MYF(MY_ALLOW_ZERO_PTR));
info->opt_flag&= ~(READ_CACHE_USED | WRITE_CACHE_USED); info->opt_flag&= ~(READ_CACHE_USED | WRITE_CACHE_USED);
VOID(end_io_cache(&info->rec_cache)); VOID(end_io_cache(&info->rec_cache));
my_free(sort_info->buff,MYF(MY_ALLOW_ZERO_PTR)); my_free(sort_info.buff,MYF(MY_ALLOW_ZERO_PTR));
sort_info->buff=0; sort_info.buff=0;
share->state.sortkey=sort_key; share->state.sortkey=sort_key;
DBUG_RETURN(flush_blocks(param, share->kfile) | got_error); DBUG_RETURN(flush_blocks(param, share->kfile) | got_error);
} /* sort_records */ } /* sort_records */
@ -1484,7 +1488,8 @@ err:
/* Sort records recursive using one index */ /* Sort records recursive using one index */
static int sort_record_index(MI_CHECK *param,MI_INFO *info, MI_KEYDEF *keyinfo, static int sort_record_index(MI_SORT_PARAM *sort_param,MI_INFO *info,
MI_KEYDEF *keyinfo,
my_off_t page, uchar *buff, uint sort_key, my_off_t page, uchar *buff, uint sort_key,
File new_file,my_bool update_index) File new_file,my_bool update_index)
{ {
@ -1493,7 +1498,8 @@ static int sort_record_index(MI_CHECK *param,MI_INFO *info, MI_KEYDEF *keyinfo,
my_off_t next_page,rec_pos; my_off_t next_page,rec_pos;
uchar lastkey[MI_MAX_KEY_BUFF]; uchar lastkey[MI_MAX_KEY_BUFF];
char llbuff[22]; char llbuff[22];
SORT_INFO *sort_info= &param->sort_info; SORT_INFO *sort_info= sort_param->sort_info;
MI_CHECK *param=sort_info->param;
DBUG_ENTER("sort_record_index"); DBUG_ENTER("sort_record_index");
nod_flag=mi_test_if_nod(buff); nod_flag=mi_test_if_nod(buff);
@ -1524,7 +1530,7 @@ static int sort_record_index(MI_CHECK *param,MI_INFO *info, MI_KEYDEF *keyinfo,
llstr(next_page,llbuff)); llstr(next_page,llbuff));
goto err; goto err;
} }
if (sort_record_index(param, info,keyinfo,next_page,temp_buff,sort_key, if (sort_record_index(sort_param, info,keyinfo,next_page,temp_buff,sort_key,
new_file, update_index)) new_file, update_index))
goto err; goto err;
} }
@ -1535,23 +1541,23 @@ static int sort_record_index(MI_CHECK *param,MI_INFO *info, MI_KEYDEF *keyinfo,
break; break;
rec_pos= _mi_dpos(info,0,lastkey+key_length); rec_pos= _mi_dpos(info,0,lastkey+key_length);
if ((*info->s->read_rnd)(info,sort_info->record,rec_pos,0)) if ((*info->s->read_rnd)(info,sort_param->record,rec_pos,0))
{ {
mi_check_print_error(param,"%d when reading datafile",my_errno); mi_check_print_error(param,"%d when reading datafile",my_errno);
goto err; goto err;
} }
if (rec_pos != sort_info->filepos && update_index) if (rec_pos != sort_param->filepos && update_index)
{ {
_mi_dpointer(info,keypos-nod_flag-info->s->rec_reflength, _mi_dpointer(info,keypos-nod_flag-info->s->rec_reflength,
sort_info->filepos); sort_param->filepos);
if (movepoint(info,sort_info->record,rec_pos,sort_info->filepos, if (movepoint(info,sort_param->record,rec_pos,sort_param->filepos,
sort_key)) sort_key))
{ {
mi_check_print_error(param,"%d when updating key-pointers",my_errno); mi_check_print_error(param,"%d when updating key-pointers",my_errno);
goto err; goto err;
} }
} }
if (sort_write_record(sort_info)) if (sort_write_record(sort_param))
goto err; goto err;
} }
/* Clear end of block to get better compression if the table is backuped */ /* Clear end of block to get better compression if the table is backuped */

2
myisam/myisamdef.h
View File

@ -658,7 +658,7 @@ int _mi_init_bulk_insert(MI_INFO *info);
void mi_check_print_error _VARARGS((MI_CHECK *param, const char *fmt,...)); void mi_check_print_error _VARARGS((MI_CHECK *param, const char *fmt,...));
void mi_check_print_warning _VARARGS((MI_CHECK *param, const char *fmt,...)); void mi_check_print_warning _VARARGS((MI_CHECK *param, const char *fmt,...));
void mi_check_print_info _VARARGS((MI_CHECK *param, const char *fmt,...)); void mi_check_print_info _VARARGS((MI_CHECK *param, const char *fmt,...));
int flush_pending_blocks(MI_CHECK *param); int flush_pending_blocks(MI_SORT_PARAM *param);
#ifdef __cplusplus #ifdef __cplusplus
} }

466
myisam/sort.c
View File

@ -27,7 +27,7 @@
#endif #endif
#include <queues.h> #include <queues.h>
/* static variabels */ /* static variables */
#undef MIN_SORT_MEMORY #undef MIN_SORT_MEMORY
#undef MYF_RW #undef MYF_RW
#undef DISK_BUFFER_SIZE #undef DISK_BUFFER_SIZE
@ -35,44 +35,43 @@
#define MERGEBUFF 15 #define MERGEBUFF 15
#define MERGEBUFF2 31 #define MERGEBUFF2 31
#define MIN_SORT_MEMORY (4096-MALLOC_OVERHEAD) #define MIN_SORT_MEMORY (4096-MALLOC_OVERHEAD)
#define MYF_RW MYF(MY_NABP | MY_WME | MY_WAIT_IF_FULL) #define MYF_RW MYF(MY_NABP | MY_WME | MY_WAIT_IF_FULL)
#define DISK_BUFFER_SIZE (IO_SIZE*16) #define DISK_BUFFER_SIZE (IO_SIZE*16)
typedef struct st_buffpek { typedef struct st_buffpek {
my_off_t file_pos; /* Where we are in the sort file */ my_off_t file_pos; /* Where we are in the sort file */
uchar *base,*key; /* Key pointers */ uchar *base,*key; /* Key pointers */
ha_rows count; /* Number of rows in table */ ha_rows count; /* Number of rows in table */
ulong mem_count; /* numbers of keys in memory */ ulong mem_count; /* numbers of keys in memory */
ulong max_keys; /* Max keys in buffert */ ulong max_keys; /* Max keys in buffert */
} BUFFPEK; } BUFFPEK;
extern void print_error _VARARGS((const char *fmt,...)); extern void print_error _VARARGS((const char *fmt,...));
/* functions defined in this file */ /* functions defined in this file */
static ha_rows NEAR_F find_all_keys(MI_SORT_PARAM *info,uint keys, static ha_rows NEAR_F find_all_keys(MI_SORT_PARAM *info,uint keys,
uchar **sort_keys, uchar **sort_keys,
DYNAMIC_ARRAY *buffpek,int *maxbuffer, DYNAMIC_ARRAY *buffpek,int *maxbuffer,
IO_CACHE *tempfile, IO_CACHE *tempfile,
IO_CACHE *tempfile_for_exceptions); IO_CACHE *tempfile_for_exceptions);
static int NEAR_F write_keys(MI_SORT_PARAM *info,uchar * *sort_keys, static int NEAR_F write_keys(MI_SORT_PARAM *info,uchar * *sort_keys,
uint count, BUFFPEK *buffpek,IO_CACHE *tempfile); uint count, BUFFPEK *buffpek,IO_CACHE *tempfile);
static int NEAR_F write_key(MI_SORT_PARAM *info, uchar *key, IO_CACHE *tempfile); static int NEAR_F write_key(MI_SORT_PARAM *info, uchar *key, IO_CACHE *tempfile);
static int NEAR_F write_index(MI_SORT_PARAM *info,uchar * *sort_keys, static int NEAR_F write_index(MI_SORT_PARAM *info,uchar * *sort_keys,
uint count); uint count);
static int NEAR_F merge_many_buff(MI_SORT_PARAM *info,uint keys, static int NEAR_F merge_many_buff(MI_SORT_PARAM *info,uint keys,
uchar * *sort_keys, uchar * *sort_keys,
BUFFPEK *buffpek,int *maxbuffer, BUFFPEK *buffpek,int *maxbuffer,
IO_CACHE *t_file); IO_CACHE *t_file);
static uint NEAR_F read_to_buffer(IO_CACHE *fromfile,BUFFPEK *buffpek, static uint NEAR_F read_to_buffer(IO_CACHE *fromfile,BUFFPEK *buffpek,
uint sort_length); uint sort_length);
static int NEAR_F merge_buffers(MI_SORT_PARAM *info,uint keys, static int NEAR_F merge_buffers(MI_SORT_PARAM *info,uint keys,
IO_CACHE *from_file, IO_CACHE *to_file, IO_CACHE *from_file, IO_CACHE *to_file,
uchar * *sort_keys, BUFFPEK *lastbuff, uchar * *sort_keys, BUFFPEK *lastbuff,
BUFFPEK *Fb, BUFFPEK *Tb); BUFFPEK *Fb, BUFFPEK *Tb);
static int NEAR_F merge_index(MI_SORT_PARAM *,uint,uchar **,BUFFPEK *, int, static int NEAR_F merge_index(MI_SORT_PARAM *,uint,uchar **,BUFFPEK *, int,
IO_CACHE *); IO_CACHE *);
/* Creates a index of sorted keys */ /* Creates a index of sorted keys */
/* Returns 0 if everything went ok */ /* Returns 0 if everything went ok */
@ -95,7 +94,7 @@ int _create_index_by_sort(MI_SORT_PARAM *info,my_bool no_messages,
maxbuffer=1; maxbuffer=1;
memavl=max(sortbuff_size,MIN_SORT_MEMORY); memavl=max(sortbuff_size,MIN_SORT_MEMORY);
records= info->max_records; records= info->sort_info->max_records;
sort_length= info->key_length; sort_length= info->key_length;
LINT_INIT(keys); LINT_INIT(keys);
@ -174,13 +173,13 @@ int _create_index_by_sort(MI_SORT_PARAM *info,my_bool no_messages,
goto err; /* purecov: inspected */ goto err; /* purecov: inspected */
} }
if (flush_pending_blocks(info->sort_info->param)) if (flush_pending_blocks(info))
goto err; goto err;
if (my_b_inited(&tempfile_for_exceptions)) if (my_b_inited(&tempfile_for_exceptions))
{ {
MI_INFO *index=info->sort_info->info; MI_INFO *index=info->sort_info->info;
uint keyno=info->sort_info->key; uint keyno=info->key;
uint key_length, ref_length=index->s->rec_reflength; uint key_length, ref_length=index->s->rec_reflength;
if (flush_io_cache(&tempfile_for_exceptions) || if (flush_io_cache(&tempfile_for_exceptions) ||
@ -224,9 +223,9 @@ static ha_rows NEAR_F find_all_keys(MI_SORT_PARAM *info, uint keys,
idx=error=0; idx=error=0;
sort_keys[0]=(uchar*) (sort_keys+keys); sort_keys[0]=(uchar*) (sort_keys+keys);
while(!(error=(*info->key_read)(info->sort_info,sort_keys[idx]))) while(!(error=(*info->key_read)(info,sort_keys[idx])))
{ {
if (info->sort_info->real_key_length > info->key_length) if (info->real_key_length > info->key_length)
{ {
if (write_key(info,sort_keys[idx],tempfile_for_exceptions)) if (write_key(info,sort_keys[idx],tempfile_for_exceptions))
DBUG_RETURN(HA_POS_ERROR); /* purecov: inspected */ DBUG_RETURN(HA_POS_ERROR); /* purecov: inspected */
@ -259,21 +258,262 @@ static ha_rows NEAR_F find_all_keys(MI_SORT_PARAM *info, uint keys,
DBUG_RETURN((*maxbuffer)*(keys-1)+idx); DBUG_RETURN((*maxbuffer)*(keys-1)+idx);
} /* find_all_keys */ } /* find_all_keys */
/* Write all keys in memory to file for later merge */ /* Search after all keys and place them in a temp. file */
void *_thr_find_all_keys(MI_SORT_PARAM *info)
{
int error,skr;
uint memavl,old_memavl,keys,sort_length;
uint idx, maxbuffer;
uchar **sort_keys;
MI_KEYSEG *keyseg;
my_b_clear(&info->tempfile);
my_b_clear(&info->tempfile_for_exceptions);
bzero((char*) &info->buffpek,sizeof(info->buffpek));
bzero((char*) &info->unique, sizeof(info->unique));
sort_keys= (uchar **) NULL; error= 1;
if (info->sort_info->got_error)
goto err;
memavl=max(info->sortbuff_size, MIN_SORT_MEMORY);
idx= info->sort_info->max_records;
sort_length= info->key_length;
while (memavl >= MIN_SORT_MEMORY)
{
if ((my_off_t) (idx+1)*(sort_length+sizeof(char*)) <=
(my_off_t) memavl)
keys= idx+1;
else
do
{
skr=maxbuffer;
if (memavl < sizeof(BUFFPEK)*maxbuffer ||
(keys=(memavl-sizeof(BUFFPEK)*maxbuffer)/
(sort_length+sizeof(char*))) <= 1)
{
mi_check_print_error(info->sort_info->param,
"sort_buffer_size is to small");
goto err;
}
}
while ((maxbuffer= (int) (idx/(keys-1)+1)) != skr);
if ((sort_keys=(uchar **)my_malloc(keys*(sort_length+sizeof(char*))+
((info->keyinfo->flag & HA_FULLTEXT) ? HA_FT_MAXLEN : 0), MYF(0))))
{
if (my_init_dynamic_array(&info->buffpek, sizeof(BUFFPEK),
maxbuffer, maxbuffer/2))
my_free((gptr) sort_keys,MYF(0));
else
break;
}
old_memavl=memavl;
if ((memavl=memavl/4*3) < MIN_SORT_MEMORY && old_memavl > MIN_SORT_MEMORY)
memavl=MIN_SORT_MEMORY;
}
if (memavl < MIN_SORT_MEMORY)
{
mi_check_print_error(info->sort_info->param,"Sort buffer to small"); /* purecov: tested */
goto err; /* purecov: tested */
}
// (*info->lock_in_memory)(info->sort_info->param);/* Everything is allocated */
if (info->sort_info->param->testflag & T_VERBOSE)
printf("Key %d - Allocating buffer for %d keys\n",info->key+1,keys);
info->sort_keys=sort_keys;
idx=error=0;
sort_keys[0]=(uchar*) (sort_keys+keys);
while(!(error=info->sort_info->got_error) ||
!(error=(*info->key_read)(info,sort_keys[idx])))
{
if (info->real_key_length > info->key_length)
{
if (write_key(info,sort_keys[idx],& info->tempfile_for_exceptions))
goto err;
continue;
}
if (++idx == keys)
{
if (write_keys(info,sort_keys,idx-1,
(BUFFPEK *)alloc_dynamic(&info->buffpek), &info->tempfile))
goto err;
sort_keys[0]=(uchar*) (sort_keys+keys);
memcpy(sort_keys[0],sort_keys[idx-1],(size_t) info->key_length);
idx=1;
}
sort_keys[idx]=sort_keys[idx-1]+info->key_length;
}
if (error > 0)
goto err;
if (info->buffpek.elements)
{
if (write_keys(info,sort_keys,idx,(BUFFPEK *)
alloc_dynamic(&info->buffpek),&info->tempfile))
goto err;
info->keys=(info->buffpek.elements-1)*(keys-1)+idx;
}
else
info->keys=idx;
info->sort_keys_length=keys;
goto ok;
err:
info->sort_info->got_error=1; /* no need to protect this with a mutex */
if (sort_keys)
my_free((gptr) sort_keys,MYF(0));
info->sort_keys=0;
delete_dynamic(& info->buffpek);
close_cached_file(& info->tempfile);
close_cached_file(& info->tempfile_for_exceptions);
ok:
remove_io_thread(& info->read_cache);
pthread_mutex_lock(& info->sort_info->mutex);
info->sort_info->threads_running--;
pthread_cond_signal(& info->sort_info->cond);
pthread_mutex_unlock(& info->sort_info->mutex);
return NULL;
} /* _thr_find_all_keys */
int _thr_write_keys(MI_SORT_PARAM *sort_param)
{
SORT_INFO *sort_info=sort_param->sort_info;
MI_CHECK *param=sort_info->param;
ulong length, keys;
ulong *rec_per_key_part=param->rec_per_key_part;
int i, got_error=sort_info->got_error;
MI_INFO *info=sort_info->info;
MYISAM_SHARE *share=info->s;
MI_SORT_PARAM *sinfo;
byte *mergebuf=0;
for (i=0, sinfo=sort_param ; i<sort_info->total_keys ; i++, sinfo++,
rec_per_key_part+=sinfo->keyinfo->keysegs)
{
if (!sinfo->sort_keys)
{
got_error=1;
continue;
}
share->state.key_map|=(ulonglong) 1 << sinfo->key;
if (param->testflag & T_STATISTICS)
update_key_parts(sinfo->keyinfo, rec_per_key_part,
sinfo->unique, (ulonglong) info->state->records);
if (!sinfo->buffpek.elements)
{
if (param->testflag & T_VERBOSE)
printf("Key %d - Dumping %lu keys\n",sinfo->key+1, sinfo->keys);
if (write_index(sinfo,sinfo->sort_keys,(uint) sinfo->keys) ||
flush_pending_blocks(sinfo))
got_error=1;
}
my_free((gptr) sinfo->sort_keys,MYF(0));
sinfo->sort_keys=0;
}
for (i=0, sinfo=sort_param ; i<sort_info->total_keys ; i++, sinfo++,
delete_dynamic(& sinfo->buffpek),
close_cached_file(& sinfo->tempfile),
close_cached_file(& sinfo->tempfile_for_exceptions))
{
if (got_error) continue;
if (sinfo->buffpek.elements)
{
uint maxbuffer=sinfo->buffpek.elements-1;
if (!mergebuf)
{
length=param->sort_buffer_length;
while (length >= MIN_SORT_MEMORY && !mergebuf)
{
mergebuf=my_malloc(length, MYF(0));
length=length*3/4;
}
if (!mergebuf)
{
got_error=1;
continue;
}
}
keys=length/sinfo->key_length;
if (maxbuffer >= MERGEBUFF2)
{
if (param->testflag & T_VERBOSE)
printf("Key %d - Merging %lu keys\n",sinfo->key+1, sinfo->keys);
if (merge_many_buff(sinfo, keys, (uchar **)mergebuf,
dynamic_element(&sinfo->buffpek, 0, BUFFPEK *),
&maxbuffer, &sinfo->tempfile))
{
got_error=1;
continue;
}
}
if (flush_io_cache(&sinfo->tempfile) ||
reinit_io_cache(&sinfo->tempfile,READ_CACHE,0L,0,0))
{
got_error=1;
continue;
}
if (param->testflag & T_VERBOSE)
printf("Key %d - Last merge and dumping keys", sinfo->key+1);
if (merge_index(sinfo, keys, (uchar **)mergebuf,
dynamic_element(&sinfo->buffpek,0,BUFFPEK *),
maxbuffer,&sinfo->tempfile)
|| flush_pending_blocks(sinfo))
{
got_error=1;
continue;
}
}
if (my_b_inited(&sinfo->tempfile_for_exceptions))
{
uint key_length;
if (param->testflag & T_VERBOSE)
printf("Key %d - Dumping 'long' keys", sinfo->key+1);
if (flush_io_cache(&sinfo->tempfile_for_exceptions) ||
reinit_io_cache(&sinfo->tempfile_for_exceptions,READ_CACHE,0L,0,0))
{
got_error=1;
continue;
}
while (!got_error
&& !my_b_read(&sinfo->tempfile_for_exceptions,(byte*)&key_length,
sizeof(key_length))
&& !my_b_read(&sinfo->tempfile_for_exceptions,(byte*)mergebuf,
(uint) key_length))
{
if (_mi_ck_write(info,sinfo->key,(uchar*) mergebuf,
key_length - info->s->rec_reflength))
got_error=1;
}
}
}
my_free((gptr) mergebuf,MYF(MY_ALLOW_ZERO_PTR));
return got_error;
}
/* Write all keys in memory to file for later merge */
static int NEAR_F write_keys(MI_SORT_PARAM *info, register uchar **sort_keys, static int NEAR_F write_keys(MI_SORT_PARAM *info, register uchar **sort_keys,
uint count, BUFFPEK *buffpek, uint count, BUFFPEK *buffpek, IO_CACHE *tempfile)
IO_CACHE *tempfile)
{ {
uchar **end; uchar **end;
uint sort_length=info->key_length; uint sort_length=info->key_length;
DBUG_ENTER("write_keys"); DBUG_ENTER("write_keys");
qsort2((byte*) sort_keys,count,sizeof(byte*),(qsort2_cmp) info->key_cmp, qsort2((byte*) sort_keys,count,sizeof(byte*),(qsort2_cmp) info->key_cmp,
info->sort_info); info);
if (!my_b_inited(tempfile) && if (!my_b_inited(tempfile) &&
open_cached_file(tempfile, info->tmpdir, "ST", DISK_BUFFER_SIZE, open_cached_file(tempfile, info->tmpdir, "ST", DISK_BUFFER_SIZE,
info->myf_rw)) info->sort_info->param->myf_rw))
DBUG_RETURN(1); /* purecov: inspected */ DBUG_RETURN(1); /* purecov: inspected */
buffpek->file_pos=my_b_tell(tempfile); buffpek->file_pos=my_b_tell(tempfile);
@ -288,12 +528,12 @@ static int NEAR_F write_keys(MI_SORT_PARAM *info, register uchar **sort_keys,
static int NEAR_F write_key(MI_SORT_PARAM *info, uchar *key, IO_CACHE *tempfile) static int NEAR_F write_key(MI_SORT_PARAM *info, uchar *key, IO_CACHE *tempfile)
{ {
uint key_length=info->sort_info->real_key_length; uint key_length=info->real_key_length;
DBUG_ENTER("write_key"); DBUG_ENTER("write_key");
if (!my_b_inited(tempfile) && if (!my_b_inited(tempfile) &&
open_cached_file(tempfile, info->tmpdir, "ST", DISK_BUFFER_SIZE, open_cached_file(tempfile, info->tmpdir, "ST", DISK_BUFFER_SIZE,
info->myf_rw)) info->sort_info->param->myf_rw))
DBUG_RETURN(1); DBUG_RETURN(1);
if (my_b_write(tempfile,(byte*)&key_length,sizeof(key_length)) || if (my_b_write(tempfile,(byte*)&key_length,sizeof(key_length)) ||
@ -302,27 +542,27 @@ static int NEAR_F write_key(MI_SORT_PARAM *info, uchar *key, IO_CACHE *tempfile)
DBUG_RETURN(0); DBUG_RETURN(0);
} /* write_key */ } /* write_key */
/* Write index */ /* Write index */
static int NEAR_F write_index(MI_SORT_PARAM *info, register uchar **sort_keys, static int NEAR_F write_index(MI_SORT_PARAM *info, register uchar **sort_keys,
register uint count) register uint count)
{ {
DBUG_ENTER("write_index"); DBUG_ENTER("write_index");
qsort2((gptr) sort_keys,(size_t) count,sizeof(byte*), qsort2((gptr) sort_keys,(size_t) count,sizeof(byte*),
(qsort2_cmp) info->key_cmp,info->sort_info); (qsort2_cmp) info->key_cmp,info);
while (count--) while (count--)
if ((*info->key_write)(info->sort_info,*sort_keys++)) if ((*info->key_write)(info,*sort_keys++))
DBUG_RETURN(-1); /* purecov: inspected */ DBUG_RETURN(-1); /* purecov: inspected */
DBUG_RETURN(0); DBUG_RETURN(0);
} /* write_index */ } /* write_index */
/* Merge buffers to make < MERGEBUFF2 buffers */ /* Merge buffers to make < MERGEBUFF2 buffers */
static int NEAR_F merge_many_buff(MI_SORT_PARAM *info, uint keys, static int NEAR_F merge_many_buff(MI_SORT_PARAM *info, uint keys,
uchar **sort_keys, BUFFPEK *buffpek, uchar **sort_keys, BUFFPEK *buffpek,
int *maxbuffer, IO_CACHE *t_file) int *maxbuffer, IO_CACHE *t_file)
{ {
register int i; register int i;
IO_CACHE t_file2, *from_file, *to_file, *temp; IO_CACHE t_file2, *from_file, *to_file, *temp;
@ -330,11 +570,11 @@ static int NEAR_F merge_many_buff(MI_SORT_PARAM *info, uint keys,
DBUG_ENTER("merge_many_buff"); DBUG_ENTER("merge_many_buff");
if (*maxbuffer < MERGEBUFF2) if (*maxbuffer < MERGEBUFF2)
DBUG_RETURN(0); /* purecov: inspected */ DBUG_RETURN(0); /* purecov: inspected */
if (flush_io_cache(t_file) || if (flush_io_cache(t_file) ||
open_cached_file(&t_file2,info->tmpdir,"ST",DISK_BUFFER_SIZE, open_cached_file(&t_file2,info->tmpdir,"ST",DISK_BUFFER_SIZE,
info->myf_rw)) info->sort_info->param->myf_rw))
DBUG_RETURN(1); /* purecov: inspected */ DBUG_RETURN(1); /* purecov: inspected */
from_file= t_file ; to_file= &t_file2; from_file= t_file ; to_file= &t_file2;
while (*maxbuffer >= MERGEBUFF2) while (*maxbuffer >= MERGEBUFF2)
@ -345,30 +585,30 @@ static int NEAR_F merge_many_buff(MI_SORT_PARAM *info, uint keys,
for (i=0 ; i <= *maxbuffer-MERGEBUFF*3/2 ; i+=MERGEBUFF) for (i=0 ; i <= *maxbuffer-MERGEBUFF*3/2 ; i+=MERGEBUFF)
{ {
if (merge_buffers(info,keys,from_file,to_file,sort_keys,lastbuff++, if (merge_buffers(info,keys,from_file,to_file,sort_keys,lastbuff++,
buffpek+i,buffpek+i+MERGEBUFF-1)) buffpek+i,buffpek+i+MERGEBUFF-1))
break; /* purecov: inspected */ break; /* purecov: inspected */
} }
if (merge_buffers(info,keys,from_file,to_file,sort_keys,lastbuff++, if (merge_buffers(info,keys,from_file,to_file,sort_keys,lastbuff++,
buffpek+i,buffpek+ *maxbuffer)) buffpek+i,buffpek+ *maxbuffer))
break; /* purecov: inspected */ break; /* purecov: inspected */
if (flush_io_cache(to_file)) if (flush_io_cache(to_file))
break; /* purecov: inspected */ break; /* purecov: inspected */
temp=from_file; from_file=to_file; to_file=temp; temp=from_file; from_file=to_file; to_file=temp;
*maxbuffer= (int) (lastbuff-buffpek)-1; *maxbuffer= (int) (lastbuff-buffpek)-1;
} }
close_cached_file(to_file); /* This holds old result */ close_cached_file(to_file); /* This holds old result */
if (to_file == t_file) if (to_file == t_file)
*t_file=t_file2; /* Copy result file */ *t_file=t_file2; /* Copy result file */
DBUG_RETURN(*maxbuffer >= MERGEBUFF2); /* Return 1 if interrupted */ DBUG_RETURN(*maxbuffer >= MERGEBUFF2); /* Return 1 if interrupted */
} /* merge_many_buff */ } /* merge_many_buff */
/* Read data to buffer */ /* Read data to buffer */
/* This returns (uint) -1 if something goes wrong */ /* This returns (uint) -1 if something goes wrong */
static uint NEAR_F read_to_buffer(IO_CACHE *fromfile, BUFFPEK *buffpek, static uint NEAR_F read_to_buffer(IO_CACHE *fromfile, BUFFPEK *buffpek,
uint sort_length) uint sort_length)
{ {
register uint count; register uint count;
uint length; uint length;
@ -376,24 +616,24 @@ static uint NEAR_F read_to_buffer(IO_CACHE *fromfile, BUFFPEK *buffpek,
if ((count=(uint) min((ha_rows) buffpek->max_keys,buffpek->count))) if ((count=(uint) min((ha_rows) buffpek->max_keys,buffpek->count)))
{ {
if (my_pread(fromfile->file,(byte*) buffpek->base, if (my_pread(fromfile->file,(byte*) buffpek->base,
(length= sort_length*count),buffpek->file_pos,MYF_RW)) (length= sort_length*count),buffpek->file_pos,MYF_RW))
return((uint) -1); /* purecov: inspected */ return((uint) -1); /* purecov: inspected */
buffpek->key=buffpek->base; buffpek->key=buffpek->base;
buffpek->file_pos+= length; /* New filepos */ buffpek->file_pos+= length; /* New filepos */
buffpek->count-= count; buffpek->count-= count;
buffpek->mem_count= count; buffpek->mem_count= count;
} }
return (count*sort_length); return (count*sort_length);
} /* read_to_buffer */ } /* read_to_buffer */
/* Merge buffers to one buffer */ /* Merge buffers to one buffer */
/* If to_file == 0 then use info->key_write */ /* If to_file == 0 then use info->key_write */
static int NEAR_F static int NEAR_F
merge_buffers(MI_SORT_PARAM *info, uint keys, IO_CACHE *from_file, merge_buffers(MI_SORT_PARAM *info, uint keys, IO_CACHE *from_file,
IO_CACHE *to_file, uchar **sort_keys, BUFFPEK *lastbuff, IO_CACHE *to_file, uchar **sort_keys, BUFFPEK *lastbuff,
BUFFPEK *Fb, BUFFPEK *Tb) BUFFPEK *Fb, BUFFPEK *Tb)
{ {
int error; int error;
uint sort_length,maxcount; uint sort_length,maxcount;
@ -413,8 +653,8 @@ merge_buffers(MI_SORT_PARAM *info, uint keys, IO_CACHE *from_file,
sort_length=info->key_length; sort_length=info->key_length;
if (init_queue(&queue,(uint) (Tb-Fb)+1,offsetof(BUFFPEK,key),0, if (init_queue(&queue,(uint) (Tb-Fb)+1,offsetof(BUFFPEK,key),0,
(int (*)(void*, byte *,byte*)) info->key_cmp, (int (*)(void*, byte *,byte*)) info->key_cmp,
(void*) info->sort_info)) (void*) info->sort_info))
DBUG_RETURN(1); /* purecov: inspected */ DBUG_RETURN(1); /* purecov: inspected */
for (buffpek= Fb ; buffpek <= Tb ; buffpek++) for (buffpek= Fb ; buffpek <= Tb ; buffpek++)
@ -423,7 +663,7 @@ merge_buffers(MI_SORT_PARAM *info, uint keys, IO_CACHE *from_file,
buffpek->base= strpos; buffpek->base= strpos;
buffpek->max_keys=maxcount; buffpek->max_keys=maxcount;
strpos+= (uint) (error=(int) read_to_buffer(from_file,buffpek, strpos+= (uint) (error=(int) read_to_buffer(from_file,buffpek,
sort_length)); sort_length));
if (error == -1) if (error == -1)
goto err; /* purecov: inspected */ goto err; /* purecov: inspected */
queue_insert(&queue,(char*) buffpek); queue_insert(&queue,(char*) buffpek);
@ -436,52 +676,52 @@ merge_buffers(MI_SORT_PARAM *info, uint keys, IO_CACHE *from_file,
buffpek=(BUFFPEK*) queue_top(&queue); buffpek=(BUFFPEK*) queue_top(&queue);
if (to_file) if (to_file)
{ {
if (my_b_write(to_file,(byte*) buffpek->key,(uint) sort_length)) if (my_b_write(to_file,(byte*) buffpek->key,(uint) sort_length))
{ {
error=1; goto err; /* purecov: inspected */ error=1; goto err; /* purecov: inspected */
} }
} }
else else
{ {
if ((*info->key_write)(info->sort_info,(void*) buffpek->key)) if ((*info->key_write)(info,(void*) buffpek->key))
{ {
error=1; goto err; /* purecov: inspected */ error=1; goto err; /* purecov: inspected */
} }
} }
buffpek->key+=sort_length; buffpek->key+=sort_length;
if (! --buffpek->mem_count) if (! --buffpek->mem_count)
{ {
if (!(error=(int) read_to_buffer(from_file,buffpek,sort_length))) if (!(error=(int) read_to_buffer(from_file,buffpek,sort_length)))
{ {
uchar *base=buffpek->base; uchar *base=buffpek->base;
uint max_keys=buffpek->max_keys; uint max_keys=buffpek->max_keys;
VOID(queue_remove(&queue,0)); VOID(queue_remove(&queue,0));
/* Put room used by buffer to use in other buffer */ /* Put room used by buffer to use in other buffer */
for (refpek= (BUFFPEK**) &queue_top(&queue); for (refpek= (BUFFPEK**) &queue_top(&queue);
refpek <= (BUFFPEK**) &queue_end(&queue); refpek <= (BUFFPEK**) &queue_end(&queue);
refpek++) refpek++)
{ {
buffpek= *refpek; buffpek= *refpek;
if (buffpek->base+buffpek->max_keys*sort_length == base) if (buffpek->base+buffpek->max_keys*sort_length == base)
{ {
buffpek->max_keys+=max_keys; buffpek->max_keys+=max_keys;
break; break;
} }
else if (base+max_keys*sort_length == buffpek->base) else if (base+max_keys*sort_length == buffpek->base)
{ {
buffpek->base=base; buffpek->base=base;
buffpek->max_keys+=max_keys; buffpek->max_keys+=max_keys;
break; break;
} }
} }
break; /* One buffer have been removed */ break; /* One buffer have been removed */
} }
} }
else if (error == -1) else if (error == -1)
goto err; /* purecov: inspected */ goto err; /* purecov: inspected */
queue_replaced(&queue); /* Top element has been replaced */ queue_replaced(&queue); /* Top element has been replaced */
} }
} }
buffpek=(BUFFPEK*) queue_top(&queue); buffpek=(BUFFPEK*) queue_top(&queue);
@ -492,9 +732,9 @@ merge_buffers(MI_SORT_PARAM *info, uint keys, IO_CACHE *from_file,
if (to_file) if (to_file)
{ {
if (my_b_write(to_file,(byte*) buffpek->key, if (my_b_write(to_file,(byte*) buffpek->key,
(sort_length*buffpek->mem_count))) (sort_length*buffpek->mem_count)))
{ {
error=1; goto err; /* purecov: inspected */ error=1; goto err; /* purecov: inspected */
} }
} }
else else
@ -502,18 +742,18 @@ merge_buffers(MI_SORT_PARAM *info, uint keys, IO_CACHE *from_file,
register uchar *end; register uchar *end;
strpos= buffpek->key; strpos= buffpek->key;
for (end=strpos+buffpek->mem_count*sort_length; for (end=strpos+buffpek->mem_count*sort_length;
strpos != end ; strpos != end ;
strpos+=sort_length) strpos+=sort_length)
{ {
if ((*info->key_write)(info->sort_info,(void*) strpos)) if ((*info->key_write)(info,(void*) strpos))
{ {
error=1; goto err; /* purecov: inspected */ error=1; goto err; /* purecov: inspected */
} }
} }
} }
} }
while ((error=(int) read_to_buffer(from_file,buffpek,sort_length)) != -1 && while ((error=(int) read_to_buffer(from_file,buffpek,sort_length)) != -1 &&
error != 0); error != 0);
lastbuff->count=count; lastbuff->count=count;
if (to_file) if (to_file)
@ -524,15 +764,15 @@ err:
} /* merge_buffers */ } /* merge_buffers */
/* Do a merge to output-file (save only positions) */ /* Do a merge to output-file (save only positions) */
static int NEAR_F static int NEAR_F
merge_index(MI_SORT_PARAM *info, uint keys, uchar **sort_keys, merge_index(MI_SORT_PARAM *info, uint keys, uchar **sort_keys,
BUFFPEK *buffpek, int maxbuffer, IO_CACHE *tempfile) BUFFPEK *buffpek, int maxbuffer, IO_CACHE *tempfile)
{ {
DBUG_ENTER("merge_index"); DBUG_ENTER("merge_index");
if (merge_buffers(info,keys,tempfile,(IO_CACHE*) 0,sort_keys,buffpek,buffpek, if (merge_buffers(info,keys,tempfile,(IO_CACHE*) 0,sort_keys,buffpek,buffpek,
buffpek+maxbuffer)) buffpek+maxbuffer))
DBUG_RETURN(1); /* purecov: inspected */ DBUG_RETURN(1); /* purecov: inspected */
DBUG_RETURN(0); DBUG_RETURN(0);
} /* merge_index */ } /* merge_index */

118
mysys/mf_iocache.c
View File

@ -85,7 +85,11 @@ init_functions(IO_CACHE* info, enum cache_type type)
info->write_function = 0; /* Force a core if used */ info->write_function = 0; /* Force a core if used */
break; break;
default: default:
info->read_function = _my_b_read; info->read_function =
#ifdef THREAD
info->share ? _my_b_read_r :
#endif
_my_b_read;
info->write_function = _my_b_write; info->write_function = _my_b_write;
} }
@ -127,6 +131,9 @@ int init_io_cache(IO_CACHE *info, File file, uint cachesize,
info->alloced_buffer = 0; info->alloced_buffer = 0;
info->buffer=0; info->buffer=0;
info->seek_not_done= test(file >= 0); info->seek_not_done= test(file >= 0);
#ifdef THREAD
info->share=0;
#endif
if (!cachesize) if (!cachesize)
if (! (cachesize= my_default_record_cache_size)) if (! (cachesize= my_default_record_cache_size))
@ -214,7 +221,6 @@ int init_io_cache(IO_CACHE *info, File file, uint cachesize,
DBUG_RETURN(0); DBUG_RETURN(0);
} /* init_io_cache */ } /* init_io_cache */
/* Wait until current request is ready */ /* Wait until current request is ready */
#ifdef HAVE_AIOWAIT #ifdef HAVE_AIOWAIT
@ -419,6 +425,90 @@ int _my_b_read(register IO_CACHE *info, byte *Buffer, uint Count)
DBUG_RETURN(0); DBUG_RETURN(0);
} }
#ifdef THREAD
int init_io_cache_share(IO_CACHE *info, IO_CACHE_SHARE *s, uint num_threads)
{
DBUG_ASSERT(info->type == READ_CACHE);
pthread_mutex_init(& s->mutex, MY_MUTEX_INIT_FAST);
pthread_cond_init (& s->cond, 0);
s->count=num_threads;
s->active=0; /* to catch errors */
info->share=s;
info->read_function=_my_b_read_r;
}
int remove_io_thread(IO_CACHE *info)
{
if (errno=pthread_mutex_lock(& info->share->mutex))
return -1;
if (! info->share->count--)
pthread_cond_signal(& info->share->cond);
pthread_mutex_unlock(& info->share->mutex);
return 0;
}
int _my_b_read_r(register IO_CACHE *info, byte *Buffer, uint Count)
{
my_off_t pos_in_file;
int length,diff_length,read_len;
DBUG_ENTER("_my_b_read_r");
if ((read_len=(uint) (info->read_end-info->read_pos)))
{
DBUG_ASSERT(Count >= read_len); /* User is not using my_b_read() */
memcpy(Buffer,info->read_pos, (size_t) (read_len));
Buffer+=read_len;
Count-=read_len;
}
#define IO_ROUND_UP(X) (((X)+IO_SIZE-1) & ~(IO_SIZE-1))
#define IO_ROUND_DN(X) ( (X) & ~(IO_SIZE-1))
while (Count) {
int cnt, len;
pos_in_file= info->pos_in_file + (uint)(info->read_end - info->buffer);
diff_length= pos_in_file & (IO_SIZE-1);
length=IO_ROUND_UP(Count+diff_length)-diff_length;
length=(length <= info->read_length) ?
length + IO_ROUND_DN(info->read_length - length) :
length - IO_ROUND_UP(length - info->read_length) ;
if (lock_io_cache(info))
{
info->share->active=info;
if (info->seek_not_done) /* File touched, do seek */
VOID(my_seek(info->file,pos_in_file,MY_SEEK_SET,MYF(0)));
len=my_read(info->file,info->buffer, length, info->myflags);
info->read_end=info->buffer + (len == -1 ? 0 : len);
info->error=(len == length ? 0 : len);
info->pos_in_file=pos_in_file;
unlock_io_cache(info);
}
else
{
info->error= info->share->active->error;
info->read_end= info->share->active->read_end;
info->pos_in_file= info->share->active->pos_in_file;
len= (info->error == -1 ? -1 : info->read_end-info->buffer);
}
info->read_pos=info->buffer;
info->seek_not_done=0;
if (info->error)
{
info->error=read_len;
DBUG_RETURN(1);
}
cnt=(len > Count) ? Count : len;
memcpy(Buffer,info->read_pos, (size_t)cnt);
Count -=cnt;
Buffer+=cnt;
read_len+=cnt;
info->read_pos+=cnt;
}
DBUG_RETURN(0);
}
#endif
/* /*
Do sequential read from the SEQ_READ_APPEND cache Do sequential read from the SEQ_READ_APPEND cache
we do this in three stages: we do this in three stages:
@ -454,7 +544,7 @@ int _my_b_seq_read(register IO_CACHE *info, byte *Buffer, uint Count)
*/ */
VOID(my_seek(info->file,pos_in_file,MY_SEEK_SET,MYF(0))); VOID(my_seek(info->file,pos_in_file,MY_SEEK_SET,MYF(0)));
info->seek_not_done=0; info->seek_not_done=0;
diff_length=(uint) (pos_in_file & (IO_SIZE-1)); diff_length=(uint) (pos_in_file & (IO_SIZE-1));
/* now the second stage begins - read from file descriptor */ /* now the second stage begins - read from file descriptor */
@ -510,7 +600,7 @@ int _my_b_seq_read(register IO_CACHE *info, byte *Buffer, uint Count)
memcpy(Buffer,info->buffer,(size_t) length); memcpy(Buffer,info->buffer,(size_t) length);
Count -= length; Count -= length;
Buffer += length; Buffer += length;
/* /*
added the line below to make added the line below to make
DBUG_ASSERT(pos_in_file==info->end_of_file) pass. DBUG_ASSERT(pos_in_file==info->end_of_file) pass.
@ -544,7 +634,7 @@ read_append_buffer:
/* /*
TODO: figure out if the assert below is needed or correct. TODO: figure out if the assert below is needed or correct.
*/ */
DBUG_ASSERT(pos_in_file == info->end_of_file); DBUG_ASSERT(pos_in_file == info->end_of_file);
copy_len=min(Count, len_in_buff); copy_len=min(Count, len_in_buff);
memcpy(Buffer, info->append_read_pos, copy_len); memcpy(Buffer, info->append_read_pos, copy_len);
info->append_read_pos += copy_len; info->append_read_pos += copy_len;
@ -910,7 +1000,7 @@ int _flush_io_cache(IO_CACHE *info, int need_append_buffer_lock)
if (!(append_cache = (info->type == SEQ_READ_APPEND))) if (!(append_cache = (info->type == SEQ_READ_APPEND)))
need_append_buffer_lock=0; need_append_buffer_lock=0;
if (info->type == WRITE_CACHE || append_cache) if (info->type == WRITE_CACHE || append_cache)
{ {
if (info->file == -1) if (info->file == -1)
@ -919,7 +1009,7 @@ int _flush_io_cache(IO_CACHE *info, int need_append_buffer_lock)
DBUG_RETURN((info->error= -1)); DBUG_RETURN((info->error= -1));
} }
LOCK_APPEND_BUFFER; LOCK_APPEND_BUFFER;
if ((length=(uint) (info->write_pos - info->write_buffer))) if ((length=(uint) (info->write_pos - info->write_buffer)))
{ {
pos_in_file=info->pos_in_file; pos_in_file=info->pos_in_file;
@ -956,7 +1046,7 @@ int _flush_io_cache(IO_CACHE *info, int need_append_buffer_lock)
info->end_of_file+=(info->write_pos-info->append_read_pos); info->end_of_file+=(info->write_pos-info->append_read_pos);
DBUG_ASSERT(info->end_of_file == my_tell(info->file,MYF(0))); DBUG_ASSERT(info->end_of_file == my_tell(info->file,MYF(0)));
} }
info->append_read_pos=info->write_pos=info->write_buffer; info->append_read_pos=info->write_pos=info->write_buffer;
UNLOCK_APPEND_BUFFER; UNLOCK_APPEND_BUFFER;
DBUG_RETURN(info->error); DBUG_RETURN(info->error);
@ -980,6 +1070,18 @@ int end_io_cache(IO_CACHE *info)
IO_CACHE_CALLBACK pre_close; IO_CACHE_CALLBACK pre_close;
DBUG_ENTER("end_io_cache"); DBUG_ENTER("end_io_cache");
#ifdef THREAD
/* simple protection against multi-close: destroying share first */
if (info->share)
if (pthread_cond_destroy (& info->share->cond) |
pthread_mutex_destroy(& info->share->mutex))
{
DBUG_RETURN(1);
}
else
info->share=0;
#endif
if ((pre_close=info->pre_close)) if ((pre_close=info->pre_close))
(*pre_close)(info); (*pre_close)(info);
if (info->alloced_buffer) if (info->alloced_buffer)