"myisamchk -p" for parallel recover works (no extensive testing though) (53ac59b9) · Commits · Software / OSDI20 Artifacts / mariadb

include/my_sys.h

+2 −25

Original line number	Diff line number	Diff line
		@@ -322,29 +322,6 @@ typedef struct st_io_cache_share
		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 */
		@@ -686,9 +663,9 @@ extern my_bool reinit_io_cache(IO_CACHE *info,enum cache_type type,
		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,
		extern void init_io_cache_share(IO_CACHE *info,
		IO_CACHE_SHARE *s, uint num_threads);
		extern int remove_io_thread(IO_CACHE *info);
		extern void remove_io_thread(IO_CACHE *info);
		#endif
		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);

include/myisam.h

+3 −1

Original line number	Diff line number	Diff line
		@@ -382,7 +382,7 @@ typedef struct st_mi_sort_param
		IO_CACHE tempfile, tempfile_for_exceptions;
		DYNAMIC_ARRAY buffpek;
		my_off_t pos,max_pos,filepos,start_recpos;
		my_bool fix_datafile;
		my_bool fix_datafile, master;
		char *record;
		char *tmpdir;
		int (key_cmp)(struct st_mi_sort_param , const void , const void );
		@@ -403,6 +403,8 @@ int mi_repair(MI_CHECK param, register MI_INFO info,
		int mi_sort_index(MI_CHECK param, register MI_INFO info, my_string name);
		int mi_repair_by_sort(MI_CHECK param, register MI_INFO info,
		const char * name, int rep_quick);
		int mi_repair_parallel(MI_CHECK param, register MI_INFO info,
		const char * name, int rep_quick);
		int change_to_newfile(const char * filename, const char * old_ext,
		const char * new_ext, uint raid_chunks,
		myf myflags);

myisam/mi_check.c

+29 −18

Original line number	Diff line number	Diff line
		@@ -14,7 +14,7 @@
		along with this program; if not, write to the Free Software
		Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */

		/* Descript, check and repair of MyISAM tables */
		/* Describe, check and repair of MyISAM tables */

		#include "ftdefs.h"
		#include <m_ctype.h>
		@@ -1187,6 +1187,7 @@ int mi_repair(MI_CHECK param, register MI_INFO info,
		my_seek(info->dfile,0L,MY_SEEK_END,MYF(0));
		sort_info.dupp=0;
		sort_param.fix_datafile= (my_bool) (!rep_quick);
		sort_param.master=1;
		sort_info.max_records= ~(ha_rows) 0;

		set_data_file_type(&sort_info, share);
		@@ -1873,6 +1874,7 @@ int mi_repair_by_sort(MI_CHECK param, register MI_INFO info,
		sort_param.tmpdir=param->tmpdir;
		sort_param.sort_info=&sort_info;
		sort_param.fix_datafile= (my_bool) (! rep_quick);
		sort_param.master =1;

		del=info->state->del;
		param->glob_crc=0;
		@@ -2082,7 +2084,7 @@ int mi_repair_by_sort(MI_CHECK param, register MI_INFO info,
		/* same as mi_repair_by_sort */
		/* but do it multithreaded */

		int mi_repair_by_sort_r(MI_CHECK param, register MI_INFO info,
		int mi_repair_parallel(MI_CHECK param, register MI_INFO info,
		const char * name, int rep_quick)
		{
		int got_error;
		@@ -2091,7 +2093,7 @@ int mi_repair_by_sort_r(MI_CHECK param, register MI_INFO info,
		ha_rows start_records;
		my_off_t new_header_length,del;
		File new_file;
		MI_SORT_PARAM sort_param=0, sinfo;
		MI_SORT_PARAM *sort_param=0;
		MYISAM_SHARE *share=info->s;
		ulong *rec_per_key_part;
		MI_KEYSEG *keyseg;
		@@ -2099,7 +2101,7 @@ int mi_repair_by_sort_r(MI_CHECK param, register MI_INFO info,
		IO_CACHE_SHARE io_share;
		SORT_INFO sort_info;
		ulonglong key_map=share->state.key_map;
		DBUG_ENTER("mi_repair_by_sort_r");
		DBUG_ENTER("mi_repair_parallel");

		start_records=info->state->records;
		got_error=1;
		@@ -2244,6 +2246,7 @@ int mi_repair_by_sort_r(MI_CHECK param, register MI_INFO info,
		sort_param[i].lock_in_memory=lock_memory;
		sort_param[i].tmpdir=param->tmpdir;
		sort_param[i].sort_info=&sort_info;
		sort_param[i].master=0;
		sort_param[i].fix_datafile=0;

		sort_param[i].filepos=new_header_length;
		@@ -2271,7 +2274,8 @@ int mi_repair_by_sort_r(MI_CHECK param, register MI_INFO info,
		sort_param[i].key_length+=ft_max_word_len_for_sort-ft_max_word_len;
		}
		sort_info.total_keys=i;
		sort_param[0].fix_datafile= ! rep_quick;
		sort_param[0].master= 1;
		sort_param[0].fix_datafile= (my_bool)(! rep_quick);

		sort_info.got_error=0;
		pthread_mutex_init(& sort_info.mutex, MY_MUTEX_INIT_FAST);
		@@ -2289,7 +2293,7 @@ int mi_repair_by_sort_r(MI_CHECK param, register MI_INFO info,
		In the second one all the threads will fill their sort_buffers
		(and call write_keys) at the same time, putting more stress on i/o.
		*/
		#if 1
		#ifndef USING_FIRST_APPROACH
		param->sort_buffer_length/sort_info.total_keys;
		#else
		param->sort_buffer_length*sort_param[i].key_length/length;
		@@ -2454,7 +2458,8 @@ static int sort_key_read(MI_SORT_PARAM sort_param, void key)
		if (info->state->records == sort_info->max_records)
		{
		mi_check_print_error(sort_info->param,
		"Found too many records; Can`t continue");
		"Key %d - Found too many records; Can't continue",
		sort_param->key+1);
		DBUG_RETURN(1);
		}
		sort_param->real_key_length=(info->s->rec_reflength+
		@@ -2543,6 +2548,7 @@ static int sort_get_next_record(MI_SORT_PARAM *sort_param)
		if (!sort_param->fix_datafile)
		{
		sort_param->filepos=sort_param->pos;
		if (sort_param->master)
		share->state.split++;
		}
		sort_param->max_pos=(sort_param->pos+=share->base.pack_reclength);
		@@ -2553,7 +2559,7 @@ static int sort_get_next_record(MI_SORT_PARAM *sort_param)
		mi_static_checksum(info,sort_param->record));
		DBUG_RETURN(0);
		}
		if (!sort_param->fix_datafile)
		if (!sort_param->fix_datafile && sort_param->master)
		{
		info->state->del++;
		info->state->empty+=share->base.pack_reclength;
		@@ -2699,7 +2705,8 @@ static int sort_get_next_record(MI_SORT_PARAM *sort_param)
		}
		if (b_type & (BLOCK_DELETED \| BLOCK_SYNC_ERROR))
		{
		if (!sort_param->fix_datafile && (b_type & BLOCK_DELETED))
		if (!sort_param->fix_datafile && sort_param->master &&
		(b_type & BLOCK_DELETED))
		{
		info->state->empty+=block_info.block_len;
		info->state->del++;
		@@ -2718,7 +2725,7 @@ static int sort_get_next_record(MI_SORT_PARAM *sort_param)
		continue;
		}

		if (!sort_param->fix_datafile)
		if (!sort_param->fix_datafile && sort_param->master)
		share->state.split++;
		if (! found_record++)
		{
		@@ -2860,6 +2867,7 @@ static int sort_get_next_record(MI_SORT_PARAM *sort_param)
		if (!sort_param->fix_datafile)
		{
		sort_param->filepos=sort_param->pos;
		if (sort_param->master)
		share->state.split++;
		}
		sort_param->max_pos=(sort_param->pos=block_info.filepos+
		@@ -2968,6 +2976,8 @@ int sort_write_record(MI_SORT_PARAM *sort_param)
		break;
		}
		}
		if (sort_param->master)
		{
		info->state->records++;
		if ((param->testflag & T_WRITE_LOOP) &&
		(info->state->records % WRITE_COUNT) == 0)
		@@ -2975,6 +2985,7 @@ int sort_write_record(MI_SORT_PARAM *sort_param)
		char llbuff[22];
		printf("%s\r", llstr(info->state->records,llbuff)); VOID(fflush(stdout));
		}
		}
		DBUG_RETURN(0);
		} /* sort_write_record */

myisam/myisamchk.c

+17 −7

Original line number	Diff line number	Diff line
		@@ -216,9 +216,15 @@ static struct my_option my_long_options[] =
		{"recover", 'r',
		"Can fix almost anything except unique keys that aren't unique.",
		0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
		{"paraller-recover", 'p',
		"Same as '-r' but creates all the keys in parallel",
		0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
		{"safe-recover", 'o',
		"Uses old recovery method; Slower than '-r' but can handle a couple of cases where '-r' reports that it can't fix the data file.",
		0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
		{"sort-recover", 'n',
		"Force recovering with sorting even if the temporary file was very big.",
		0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
		{"start-check-pos", OPT_START_CHECK_POS,
		"No help available.",
		0, 0, 0, GET_ULL, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
		@@ -244,9 +250,6 @@ static struct my_option my_long_options[] =
		(gptr*) &check_param.opt_sort_key,
		(gptr*) &check_param.opt_sort_key,
		0, GET_UINT, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
		{"sort-recover", 'n',
		"Force recovering with sorting even if the temporary file was very big.",
		0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
		{"tmpdir", 't',
		"Path for temporary files.",
		(gptr*) &check_param.tmpdir,
		@@ -514,6 +517,11 @@ get_one_option(int optid,
		if (argument != disabled_my_option)
		check_param.testflag\|= T_REP_BY_SORT;
		break;
		case 'p':
		check_param.testflag&= ~T_REP_ANY;
		if (argument != disabled_my_option)
		check_param.testflag\|= T_REP_PARALLEL;
		break;
		case 'o':
		check_param.testflag&= ~T_REP_ANY;
		check_param.force_sort= 0;
		@@ -864,8 +872,7 @@ static int myisamchk(MI_CHECK *param, my_string filename)
		if (tmp != share->state.key_map)
		info->update\|=HA_STATE_CHANGED;
		}
		if (rep_quick && chk_del(param, info,
		param->testflag & ~T_VERBOSE))
		if (rep_quick && chk_del(param, info, param->testflag & ~T_VERBOSE))
		{
		if (param->testflag & T_FORCE_CREATE)
		{
		@@ -881,14 +888,17 @@ static int myisamchk(MI_CHECK *param, my_string filename)
		}
		if (!error)
		{
		if ((param->testflag & T_REP_BY_SORT) &&
		if ((param->testflag & (T_REP_BY_SORT \| T_REP_PARALLEL)) &&
		(share->state.key_map \|\|
		(rep_quick && !param->keys_in_use && !recreate)) &&
		mi_test_if_sort_rep(info, info->state->records,
		info->s->state.key_map,
		param->force_sort))
		{
		if (param->testflag & T_REP_BY_SORT)
		error=mi_repair_by_sort(param,info,filename,rep_quick);
		else
		error=mi_repair_parallel(param,info,filename,rep_quick);
		state_updated=1;
		}
		else if (param->testflag & T_REP_ANY)

myisam/sort.c

+21 −11

Original line number	Diff line number	Diff line
		@@ -268,17 +268,23 @@ void _thr_find_all_keys(MI_SORT_PARAM info)
		uchar **sort_keys;
		MI_KEYSEG *keyseg;

		error=1;

		if (my_thread_init())
		goto err;
		if (info->sort_info->got_error)
		goto err;

		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;
		sort_keys= (uchar **) NULL;

		memavl=max(info->sortbuff_size, MIN_SORT_MEMORY);
		idx= info->sort_info->max_records;
		sort_length= info->key_length;
		maxbuffer= 1;

		while (memavl >= MIN_SORT_MEMORY)
		{
		@@ -327,7 +333,7 @@ void _thr_find_all_keys(MI_SORT_PARAM info)
		idx=error=0;
		sort_keys[0]=(uchar*) (sort_keys+keys);

		while(!(error=info->sort_info->got_error) \|\|
		while(!(error=info->sort_info->got_error) &&
		!(error=(*info->key_read)(info,sort_keys[idx])))
		{
		if (info->real_key_length > info->key_length)
		@@ -342,7 +348,6 @@ void _thr_find_all_keys(MI_SORT_PARAM info)
		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;
		@@ -378,6 +383,7 @@ void _thr_find_all_keys(MI_SORT_PARAM info)
		info->sort_info->threads_running--;
		pthread_cond_signal(& info->sort_info->cond);
		pthread_mutex_unlock(& info->sort_info->mutex);
		my_thread_end();
		return NULL;
		} /* _thr_find_all_keys */

		@@ -393,8 +399,9 @@ int _thr_write_keys(MI_SORT_PARAM *sort_param)
		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)
		for (i=0, sinfo=sort_param ; i<sort_info->total_keys ; i++,
		rec_per_key_part+=sinfo->keyinfo->keysegs,
		sinfo++)
		{
		if (!sinfo->sort_keys)
		{
		@@ -417,10 +424,11 @@ int _thr_write_keys(MI_SORT_PARAM *sort_param)
		sinfo->sort_keys=0;
		}

		for (i=0, sinfo=sort_param ; i<sort_info->total_keys ; i++, sinfo++,
		for (i=0, sinfo=sort_param ; i<sort_info->total_keys ; i++,
		delete_dynamic(& sinfo->buffpek),
		close_cached_file(& sinfo->tempfile),
		close_cached_file(& sinfo->tempfile_for_exceptions))
		close_cached_file(& sinfo->tempfile_for_exceptions),
		sinfo++)
		{
		if (got_error) continue;
		if (sinfo->buffpek.elements)
		@@ -520,8 +528,10 @@ static int NEAR_F write_keys(MI_SORT_PARAM info, register uchar *sort_keys,
		buffpek->count=count;

		for (end=sort_keys+count ; sort_keys != end ; sort_keys++)
		{
		if (my_b_write(tempfile,(byte) sort_keys,(uint) sort_length))
		DBUG_RETURN(1); /* purecov: inspected */
		}
		DBUG_RETURN(0);
		} /* write_keys */

		@@ -654,7 +664,7 @@ merge_buffers(MI_SORT_PARAM info, uint keys, IO_CACHE from_file,

		if (init_queue(&queue,(uint) (Tb-Fb)+1,offsetof(BUFFPEK,key),0,
		(int ()(void, byte ,byte)) info->key_cmp,
		(void*) info->sort_info))
		(void*) info))
		DBUG_RETURN(1); /* purecov: inspected */

		for (buffpek= Fb ; buffpek <= Tb ; buffpek++)