tstrx.c

Mysql for Windows最新源码

	
	mutex_exit(&kernel_mutex);

	oldtm = ut_clock();
	
	que_run_threads(thr);

	tm = ut_clock();
	printf("Wall time for commit %lu milliseconds\n", tm - oldtm);

	dict_table_print_by_name("TS_TABLE1");
	/*-------------------------------------*/
	/* UPDATE ROWS */
	ut_a(trx_start(trx, ULINT_UNDEFINED));	
#endif
	fork = que_fork_create(NULL, NULL, QUE_FORK_UPDATE, heap);
	fork->trx = trx;

	thr = que_thr_create(fork, fork, heap);

	table = dict_table_get("TS_TABLE1", trx);

	row = dtuple_create(heap, 3 + DATA_N_SYS_COLS);

	dict_table_copy_types(row, table);
	
	update = upd_create(1, heap);

	node = upd_node_create(fork, thr, table, &pcur, update, heap);
	thr->child = node;

	node->cmpl_info = 0;

	mutex_enter(&kernel_mutex);

	que_graph_publish(fork, trx->sess);

	trx->graph = fork;

	mutex_exit(&kernel_mutex);

   for (j = 0; j < 2; j++) {
    for (i = 0; i < *((ulint*)arg); i++) {
	
	dtuple_gen_test_tuple3(row, i, DTUPLE_TEST_FIXED30, buf);
	entry = dtuple_create(heap, 2);
	dfield_copy(dtuple_get_nth_field(entry, 0),
					dtuple_get_nth_field(row, 0));
	dfield_copy(dtuple_get_nth_field(entry, 1),
					dtuple_get_nth_field(row, 1));

	index = dict_table_get_first_index(table);
	tree = dict_index_get_tree(index);

	btr_pcur_set_mtr(&pcur, &mtr);
	
	mtr_start(&mtr);
	
	btr_pcur_open(tree, entry, PAGE_CUR_G, BTR_SEARCH_LEAF, &pcur, &mtr);

	btr_pcur_store_position(&pcur, &mtr);

	err = lock_clust_rec_read_check_and_lock(0, btr_pcur_get_rec(&pcur),
							index, LOCK_X, thr);
	ut_a(err == DB_SUCCESS);
	
	btr_pcur_commit(&pcur);	

	ufield = upd_get_nth_field(update, 0);

	ufield->col_no = 2;
	dfield_set_data(&(ufield->new_val), "updated field", 14);

	mutex_enter(&kernel_mutex);	

	ut_a(
	thr == que_fork_start_command(fork, SESS_COMM_EXECUTE, 0));

	mutex_exit(&kernel_mutex);

	que_run_threads(thr);

    } /* for (i = ... */
   }
	mtr_start(&mtr);

	ut_a(btr_pcur_restore_position(BTR_SEARCH_LEAF, &pcur, &mtr));

	ptr = rec_get_nth_field(btr_pcur_get_rec(&pcur), 5, &len);

	ut_a(ut_memcmp(ptr, "updated field", 14) == 0);

	btr_pcur_commit(&pcur);

	dict_table_print_by_name("TS_TABLE1");

	ufield = upd_get_nth_field(update, 0);

	ufield->col_no = 0;
	dfield_set_data(&(ufield->new_val), "31415926", 9);

	mutex_enter(&kernel_mutex);	

	ut_a(
	thr == que_fork_start_command(fork, SESS_COMM_EXECUTE, 0));

	mutex_exit(&kernel_mutex);

	que_run_threads(thr);

	dict_table_print_by_name("TS_TABLE1");
	/*-------------------------------------*/
	/* ROLLBACK */
	
	fork = que_fork_create(NULL, NULL, QUE_FORK_EXECUTE, heap);
	fork->trx = trx;

	thr = que_thr_create(fork, fork, heap);

	thr->child = roll_node_create(fork, thr, heap);

	mutex_enter(&kernel_mutex);	

	que_graph_publish(fork, trx->sess);

	trx->graph = fork;

	ut_a(thr == que_fork_start_command(fork, SESS_COMM_EXECUTE, 0));
	
	mutex_exit(&kernel_mutex);

	oldtm = ut_clock();

	que_run_threads(thr);

	tm = ut_clock();
	printf("Wall time for rollback of %lu updates %lu milliseconds\n",
		i, tm - oldtm);
#ifdef notdefined
	/*-------------------------------------*/
	/* COMMIT */
	fork = que_fork_create(NULL, NULL, QUE_FORK_EXECUTE, heap);
	fork->trx = trx;

	thr = que_thr_create(fork, fork, heap);

	thr->child = commit_node_create(fork, thr, heap);

	mutex_enter(&kernel_mutex);	

	que_graph_publish(fork, trx->sess);

	trx->graph = fork;

	ut_a(thr == que_fork_start_command(fork, SESS_COMM_EXECUTE, 0));
	
	mutex_exit(&kernel_mutex);

	oldtm = ut_clock();
	
	que_run_threads(thr);

	tm = ut_clock();
	printf("Wall time for commit %lu milliseconds\n", tm - oldtm);

	/*-------------------------------------*/
#endif
	dict_table_print_by_name("TS_TABLE1");
	count++;

	if (count < 1) {
		goto loop;
	}
	return(0);
}

/*********************************************************************
Init TS_TABLE2 for TPC-A transaction. */

ulint
test4_5(
/*====*/
	void*	arg)
{
	sess_t*		sess;
	com_endpoint_t*	com_endpoint;
	mem_heap_t*	heap;
	que_fork_t*	fork;
	dict_table_t*	table;
	que_thr_t*	thr;
	trx_t*		trx;
	ulint		i;
	dtuple_t*	row;
	byte		buf[100];
	
	arg = arg;

	printf("-------------------------------------------------\n");
	printf("TEST 4_5. INIT FOR TPC-A\n");

	heap = mem_heap_create(512);

	com_endpoint = (com_endpoint_t*)heap;	/* This is a dummy non-NULL
						value */
	mutex_enter(&kernel_mutex);	

	sess = sess_open(ut_dulint_zero, com_endpoint, (byte*)"user1", 6);
	
	trx = sess->trx;

	mutex_exit(&kernel_mutex);

	ut_a(trx_start(trx, ULINT_UNDEFINED));	

	/*-------------------------------------*/
	/* INSERT INTO TABLE TO UPDATE */

    for (i = 0; i < 100; i++) {
	fork = que_fork_create(NULL, NULL, QUE_FORK_INSERT, heap);
	fork->trx = trx;

	thr = que_thr_create(fork, fork, heap);

	table = dict_table_get("TS_TABLE2", trx);

	row = dtuple_create(heap, 3 + DATA_N_SYS_COLS);

	dict_table_copy_types(row, table);
	
	thr->child = ins_node_create(fork, thr, row, table, heap);

	mutex_enter(&kernel_mutex);	

	que_graph_publish(fork, trx->sess);

	trx->graph = fork;

	mutex_exit(&kernel_mutex);

	dtuple_gen_test_tuple3(row, i, DTUPLE_TEST_FIXED30, buf);

	mutex_enter(&kernel_mutex);	

	ut_a(thr == que_fork_start_command(fork, SESS_COMM_EXECUTE, 0));

	mutex_exit(&kernel_mutex);

	que_run_threads(thr);

    }
/*	dict_table_print_by_name("TS_TABLE2"); */

	/*-------------------------------------*/
	/* COMMIT */
	fork = que_fork_create(NULL, NULL, QUE_FORK_EXECUTE, heap);
	fork->trx = trx;

	thr = que_thr_create(fork, fork, heap);

	thr->child = commit_node_create(fork, thr, heap);

	mutex_enter(&kernel_mutex);	

	que_graph_publish(fork, trx->sess);

	trx->graph = fork;

	ut_a(thr == que_fork_start_command(fork, SESS_COMM_EXECUTE, 0));
	
	mutex_exit(&kernel_mutex);

	que_run_threads(thr);

	/*-----------------------------------*/

	return(0);
}

/*********************************************************************
Test for TPC-A transaction. */

ulint
test5(
/*==*/
	void*	arg)
{
	ulint		tm, oldtm;
	sess_t*		sess;
	com_endpoint_t*	com_endpoint;
	mem_heap_t*	heap;
	que_fork_t*	fork1;
	que_fork_t*	fork2;
	que_fork_t*	cfork;
	dict_table_t*	table;
	dict_table_t*	table2;
	que_thr_t*	thr;
	trx_t*		trx;
	ulint		i;
	dtuple_t*	row;
	dtuple_t*	entry;
	byte		buf[100];
	ulint		count 		= 0;
	btr_pcur_t	pcur;
	upd_t*		update;
	upd_field_t*	ufield;
	dict_tree_t*	tree;
	dict_index_t*	index;
	mtr_t		mtr;
	upd_node_t*	node;
	ulint		err;
	ins_node_t*	inode;
	
	arg = arg;

	printf("-------------------------------------------------\n");
	printf("TEST 5. TPC-A %lu \n", *((ulint*)arg));

	oldtm = ut_clock();
	
	heap = mem_heap_create(512);

	com_endpoint = (com_endpoint_t*)heap;	/* This is a dummy non-NULL
						value */
	mutex_enter(&kernel_mutex);	

	sess = sess_open(ut_dulint_zero, com_endpoint, (byte*)"user1", 6);
	
	trx = sess->trx;

	mutex_exit(&kernel_mutex);

	ut_a(trx_start(trx, ULINT_UNDEFINED));	
	/*-----------------------------------*/

	fork1 = que_fork_create(NULL, NULL, QUE_FORK_INSERT, heap);
	fork1->trx = trx;

	thr = que_thr_create(fork1, fork1, heap);

	table = dict_table_get("TS_TABLE3", trx);

	row = dtuple_create(heap, 3 + DATA_N_SYS_COLS);

	dict_table_copy_types(row, table);

	inode = ins_node_create(fork1, thr, row, table, heap);
	
	thr->child = inode;

	row_ins_init_sys_fields_at_sql_compile(inode->row, inode->table, heap);
	row_ins_init_sys_fields_at_sql_prepare(inode->row, inode->table, trx);

	inode->init_all_sys_fields = FALSE;

	mutex_enter(&kernel_mutex);	

	que_graph_publish(fork1, trx->sess);

	trx->graph = fork1;

	mutex_exit(&kernel_mutex);

	fork2 = que_fork_create(NULL, NULL, QUE_FORK_UPDATE, heap);
	fork2->trx = trx;

	thr = que_thr_create(fork2, fork2, heap);

	table2 = dict_table_get("TS_TABLE2", trx);

	update = upd_create(1, heap);

	entry = dtuple_create(heap, 2);
	dfield_copy(dtuple_get_nth_field(entry, 0),
					dtuple_get_nth_field(row, 0));
	dfield_copy(dtuple_get_nth_field(entry, 1),
					dtuple_get_nth_field(row, 1));
	
	node = upd_node_create(fork2, thr, table2, &pcur, update, heap);
	thr->child = node;

	node->cmpl_info = UPD_NODE_NO_ORD_CHANGE | UPD_NODE_NO_SIZE_CHANGE;

	mutex_enter(&kernel_mutex);

	que_graph_publish(fork2, trx->sess);

	trx->graph = fork2;

	mutex_exit(&kernel_mutex);

	cfork = que_fork_create(NULL, NULL, QUE_FORK_EXECUTE, heap);
	cfork->trx = trx;

	thr = que_thr_create(cfork, cfork, heap);

	thr->child = commit_node_create(cfork, thr, heap);

	oldtm = ut_clock();
loop:
	/*-------------------------------------*/
	/* INSERT */

/*	printf("Trx %lu %lu starts, thr %lu\n",
						ut_dulint_get_low(trx->id),
						(ulint)trx,
						*((ulint*)arg)); */

	dtuple_gen_test_tuple3(row, count, DTUPLE_TEST_FIXED30, buf);

	mutex_enter(&kernel_mutex);	

	thr = que_fork_start_command(fork1, SESS_COMM_EXECUTE, 0);

	mutex_exit(&kernel_mutex);

	que_run_threads(thr);

	/*-------------------------------------*/
	/* 3 UPDATES */

    for (i = 0; i < 3; i++) {
	
	dtuple_gen_search_tuple3(entry, *((ulint*)arg), buf);

	index = dict_table_get_first_index(table2);
	tree = dict_index_get_tree(index);

	btr_pcur_set_mtr(&pcur, &mtr);
	
	mtr_start(&mtr);
	
	btr_pcur_open(tree, entry, PAGE_CUR_G, BTR_MODIFY_LEAF, &pcur, &mtr);

/*	btr_pcur_store_position(&pcur, &mtr); */

	err = lock_clust_rec_read_check_and_lock(0, btr_pcur_get_rec(&pcur),
							index, LOCK_X, thr);
	ut_a(err == DB_SUCCESS);
	
	ufield = upd_get_nth_field(update, 0);

	ufield->col_no = 2;
	dfield_set_data(&(ufield->new_val),
				"updated field1234567890123456", 30);

	mutex_enter(&kernel_mutex);	

	thr = que_fork_start_command(fork2, SESS_COMM_EXECUTE, 0);

	mutex_exit(&kernel_mutex);

	que_run_threads(thr);

    } /* for (i = ... */

	/*-------------------------------------*/
	/* COMMIT */
#ifdef notdefined
	mutex_enter(&kernel_mutex);	

	thr = que_fork_start_command(cfork, SESS_COMM_EXECUTE, 0);
	
	mutex_exit(&kernel_mutex);

	que_run_threads(thr);

/*	printf("Trx %lu %lu committed\n", ut_dulint_get_low(trx->id),
								(ulint)trx); */
#endif
	count++;

	if (count < 1000) {
		ut_a(trx_start(trx, ULINT_UNDEFINED));	

		goto loop;
	}

	tm = ut_clock();
	printf("Wall time for TPC-A %lu trxs %lu milliseconds\n",
					count, tm - oldtm);

	/*-------------------------------------*/
/*	dict_table_print_by_name("TS_TABLE2"); 
	dict_table_print_by_name("TS_TABLE3"); */

	return(0);
}

/********************************************************************
Main test function. */

void 
main(void) 
/*======*/
{
	ulint	tm, oldtm;
	os_thread_id_t	id[5];
	ulint	n1000[5];
	ulint	i;
	ulint	n5000	= 500;

	srv_boot("initfile");
	os_aio_init(160, 5);
	fil_init(25);
	buf_pool_init(POOL_SIZE, POOL_SIZE);
	fsp_init();
	log_init();

	create_files();
	init_space();

	sess_sys_init_at_db_start();
	
	trx_sys_create();
	
	lock_sys_create(1024);
	
	dict_create();

	oldtm = ut_clock();
	
	ut_rnd_set_seed(19);

	test1(NULL);
	test1_5(NULL);
	test1_6(NULL);
	test4_5(NULL);

	for (i = 1; i < 5; i++) {
		n1000[i] = i;
		id[i] = id[i];
/*		os_thread_create(test5, n1000 + i, id + i); */
	}

/*	mem_print_info(); */

/*	test2(&n5000); */

	n5000 = 30;

	test5(&n5000);

	n5000 = 30;
/*	test5(&n5000); */
	
/*	mem_print_info(); */
	
/*	dict_table_print_by_name("TS_TABLE1"); */

	tm = ut_clock();
	printf("Wall time for test %lu milliseconds\n", tm - oldtm);
	printf("TESTS COMPLETED SUCCESSFULLY!\n");
}