Commit eb9d5675 authored by unknown's avatar unknown

Merge mysql.com:/windows/Linux_space/MySQL/mysql-5.1-new-ndb

into  mysql.com:/windows/Linux_space/MySQL/mysql-5.1

parents d57163fc 41e68be6
--require r/have_ndbapi_examples.require
disable_query_log;
eval select LENGTH("$MY_NDB_EXAMPLES_BINARY") > 0 as "have_ndb_example";
enable_query_log;
...@@ -323,6 +323,8 @@ our $opt_skip_slave_binlog= 0; ...@@ -323,6 +323,8 @@ our $opt_skip_slave_binlog= 0;
our $exe_ndb_mgm; our $exe_ndb_mgm;
our $exe_ndb_waiter; our $exe_ndb_waiter;
our $path_ndb_tools_dir; our $path_ndb_tools_dir;
our $path_ndb_examples_dir;
our $exe_ndb_example;
our $file_ndb_testrun_log; our $file_ndb_testrun_log;
our @data_dir_lst; our @data_dir_lst;
...@@ -1231,6 +1233,9 @@ sub executable_setup () { ...@@ -1231,6 +1233,9 @@ sub executable_setup () {
mtr_script_exists("$glob_basedir/scripts/mysql_fix_privilege_tables", mtr_script_exists("$glob_basedir/scripts/mysql_fix_privilege_tables",
"/usr/bin/false"); "/usr/bin/false");
$path_ndb_tools_dir= mtr_path_exists("$glob_basedir/storage/ndb/tools"); $path_ndb_tools_dir= mtr_path_exists("$glob_basedir/storage/ndb/tools");
$path_ndb_examples_dir= mtr_path_exists("$glob_basedir/storage/ndb/ndbapi-examples");
$exe_ndb_example= mtr_exe_exists("$path_ndb_examples_dir/ndbapi_simple/ndbapi_simple",
$exe_mysqld);
$exe_ndb_mgm= "$glob_basedir/storage/ndb/src/mgmclient/ndb_mgm"; $exe_ndb_mgm= "$glob_basedir/storage/ndb/src/mgmclient/ndb_mgm";
$exe_ndb_waiter= "$glob_basedir/storage/ndb/tools/ndb_waiter"; $exe_ndb_waiter= "$glob_basedir/storage/ndb/tools/ndb_waiter";
$exe_ndbd= "$glob_basedir/storage/ndb/src/kernel/ndbd"; $exe_ndbd= "$glob_basedir/storage/ndb/src/kernel/ndbd";
...@@ -1295,6 +1300,7 @@ sub executable_setup () { ...@@ -1295,6 +1300,7 @@ sub executable_setup () {
} }
$path_ndb_tools_dir= "$glob_basedir/bin"; $path_ndb_tools_dir= "$glob_basedir/bin";
$path_ndb_examples_dir= "$glob_basedir/ndbapi-examples";
$exe_ndb_mgm= "$glob_basedir/bin/ndb_mgm"; $exe_ndb_mgm= "$glob_basedir/bin/ndb_mgm";
$exe_ndb_waiter= "$glob_basedir/bin/ndb_waiter"; $exe_ndb_waiter= "$glob_basedir/bin/ndb_waiter";
$exe_ndbd= "$glob_basedir/bin/ndbd"; $exe_ndbd= "$glob_basedir/bin/ndbd";
...@@ -1346,6 +1352,20 @@ sub environment_setup () { ...@@ -1346,6 +1352,20 @@ sub environment_setup () {
$ENV{'DYLD_LIBRARY_PATH'}= $ENV{'DYLD_LIBRARY_PATH'}=
"$extra_ld_library_paths" . "$extra_ld_library_paths" .
($ENV{'DYLD_LIBRARY_PATH'} ? ":$ENV{'DYLD_LIBRARY_PATH'}" : ""); ($ENV{'DYLD_LIBRARY_PATH'} ? ":$ENV{'DYLD_LIBRARY_PATH'}" : "");
# --------------------------------------------------------------------------
# Add the path where libndbclient can be found
# --------------------------------------------------------------------------
$ENV{'LD_LIBRARY_PATH'}=
(mtr_path_exists("$glob_basedir/storage/ndb/src/.libs") ? "$glob_basedir/storage/ndb/src/.libs" : "") .
($ENV{'LD_LIBRARY_PATH'} ? ":$ENV{'LD_LIBRARY_PATH'}" : "");
# --------------------------------------------------------------------------
# Add the path where libmysqlclient can be found
# --------------------------------------------------------------------------
$ENV{'LD_LIBRARY_PATH'}=
(mtr_path_exists("$glob_basedir/libmysql_r/.libs") ? "$glob_basedir/libmysql_r/.libs" : "") .
($ENV{'LD_LIBRARY_PATH'} ? ":$ENV{'LD_LIBRARY_PATH'}" : "");
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
# Also command lines in .opt files may contain env vars # Also command lines in .opt files may contain env vars
...@@ -1362,6 +1382,7 @@ sub environment_setup () { ...@@ -1362,6 +1382,7 @@ sub environment_setup () {
$ENV{'MASTER_MYSOCK1'}= $master->[1]->{'path_sock'}; $ENV{'MASTER_MYSOCK1'}= $master->[1]->{'path_sock'};
$ENV{'MASTER_MYPORT'}= $master->[0]->{'port'}; $ENV{'MASTER_MYPORT'}= $master->[0]->{'port'};
$ENV{'MASTER_MYPORT1'}= $master->[1]->{'port'}; $ENV{'MASTER_MYPORT1'}= $master->[1]->{'port'};
$ENV{'SLAVE_MYSOCK'}= $slave->[0]->{'path_sock'};
$ENV{'SLAVE_MYPORT'}= $slave->[0]->{'port'}; $ENV{'SLAVE_MYPORT'}= $slave->[0]->{'port'};
$ENV{'SLAVE_MYPORT1'}= $slave->[1]->{'port'}; $ENV{'SLAVE_MYPORT1'}= $slave->[1]->{'port'};
$ENV{'SLAVE_MYPORT2'}= $slave->[2]->{'port'}; $ENV{'SLAVE_MYPORT2'}= $slave->[2]->{'port'};
...@@ -3567,7 +3588,10 @@ sub run_mysqltest ($) { ...@@ -3567,7 +3588,10 @@ sub run_mysqltest ($) {
$ENV{'NDB_BACKUP_DIR'}= $clusters->[0]->{'data_dir'}; $ENV{'NDB_BACKUP_DIR'}= $clusters->[0]->{'data_dir'};
$ENV{'NDB_DATA_DIR'}= $clusters->[0]->{'data_dir'}; $ENV{'NDB_DATA_DIR'}= $clusters->[0]->{'data_dir'};
$ENV{'NDB_TOOLS_DIR'}= $path_ndb_tools_dir; $ENV{'NDB_TOOLS_DIR'}= $path_ndb_tools_dir;
$ENV{'NDB_EXAMPLES_DIR'}= $path_ndb_examples_dir;
$ENV{'MY_NDB_EXAMPLES_BINARY'}= ($exe_ndb_example eq "$path_ndb_examples_dir/ndbapi_simple/ndbapi_simple")?$exe_ndb_example:"";
$ENV{'NDB_TOOLS_OUTPUT'}= $file_ndb_testrun_log; $ENV{'NDB_TOOLS_OUTPUT'}= $file_ndb_testrun_log;
$ENV{'NDB_EXAMPLES_OUTPUT'}= $file_ndb_testrun_log;
$ENV{'NDB_CONNECTSTRING'}= $opt_ndbconnectstring; $ENV{'NDB_CONNECTSTRING'}= $opt_ndbconnectstring;
my $exe= $exe_mysqltest; my $exe= $exe_mysqltest;
......
DROP TABLE IF EXISTS t0;
drop database if exists mysqltest;
Running ndbapi_simple
Running ndbapi_simple_index
Running ndbapi_scan
Running ndbapi_retries
Running ndbapi_async
Running ndbapi_async1
use TEST_DB;
create table t0(c0 int, c1 int, c2 char(4), c3 char(4), c4 text,
primary key(c0, c2)) engine ndb charset latin1;
insert into t0 values (1, 2, 'a', 'b', null);
insert into t0 values (3, 4, 'c', 'd', null);
update t0 set c3 = 'e' where c0 = 1 and c2 = 'a';
update t0 set c3 = 'f';
update t0 set c3 = 'F';
update t0 set c2 = 'g' where c0 = 1;
update t0 set c2 = 'G' where c0 = 1;
update t0 set c0 = 5, c2 = 'H' where c0 = 3;
delete from t0;
drop table t0;
Running mgmapi_logevent
stop slave;
drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
reset master;
reset slave;
drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
start slave;
Running ndbapi_simple_dual
Running mgmapi_logevent
-- source include/have_ndb.inc
-- source include/have_ndbapi_examples.inc
--disable_warnings
DROP TABLE IF EXISTS t0;
drop database if exists mysqltest;
--enable_warnings
--exec echo Running ndbapi_simple
--exec $NDB_EXAMPLES_DIR/ndbapi_simple/ndbapi_simple $MASTER_MYSOCK "localhost:$NDBCLUSTER_PORT" >> $NDB_EXAMPLES_OUTPUT
--exec echo Running ndbapi_simple_index
--exec $NDB_EXAMPLES_DIR/ndbapi_simple_index/ndbapi_simple_index $MASTER_MYSOCK "localhost:$NDBCLUSTER_PORT" >> $NDB_EXAMPLES_OUTPUT
--exec echo Running ndbapi_scan
--exec $NDB_EXAMPLES_DIR/ndbapi_scan/ndbapi_scan $MASTER_MYSOCK "localhost:$NDBCLUSTER_PORT" >> $NDB_EXAMPLES_OUTPUT
--exec echo Running ndbapi_retries
--exec $NDB_EXAMPLES_DIR/ndbapi_retries/ndbapi_retries $MASTER_MYSOCK "localhost:$NDBCLUSTER_PORT" >> $NDB_EXAMPLES_OUTPUT
--exec echo Running ndbapi_async
--exec $NDB_EXAMPLES_DIR/ndbapi_async/ndbapi_async $MASTER_MYSOCK "localhost:$NDBCLUSTER_PORT" >> $NDB_EXAMPLES_OUTPUT
--exec echo Running ndbapi_async1
--exec $NDB_EXAMPLES_DIR/ndbapi_async1/ndbapi_async1 $MASTER_MYSOCK "localhost:$NDBCLUSTER_PORT" >> $NDB_EXAMPLES_OUTPUT
use TEST_DB;
create table t0(c0 int, c1 int, c2 char(4), c3 char(4), c4 text,
primary key(c0, c2)) engine ndb charset latin1;
#--exec echo Running ndbapi_event
#--exec $NDB_EXAMPLES_DIR/ndbapi_event/ndbapi_event "localhost:$NDBCLUSTER_PORT" 1 >> $NDB_EXAMPLES_OUTPUT
insert into t0 values (1, 2, 'a', 'b', null);
insert into t0 values (3, 4, 'c', 'd', null);
update t0 set c3 = 'e' where c0 = 1 and c2 = 'a'; -- use pk
update t0 set c3 = 'f'; -- use scan
update t0 set c3 = 'F'; -- use scan update to 'same'
update t0 set c2 = 'g' where c0 = 1; -- update pk part
update t0 set c2 = 'G' where c0 = 1; -- update pk part to 'same'
update t0 set c0 = 5, c2 = 'H' where c0 = 3; -- update full PK
delete from t0;
drop table t0;
--exec echo Running mgmapi_logevent
--exec $NDB_EXAMPLES_DIR/mgmapi_logevent/mgmapi_logevent "localhost:$NDBCLUSTER_PORT" 1 >> $NDB_EXAMPLES_OUTPUT
-- source include/have_ndb.inc
-- source include/have_binlog_format_row.inc
-- source include/master-slave.inc
-- source include/have_ndbapi_examples.inc
--exec echo Running ndbapi_simple_dual
--exec $NDB_EXAMPLES_DIR/ndbapi_simple_dual/ndbapi_simple_dual $MASTER_MYSOCK "localhost:$NDBCLUSTER_PORT" $SLAVE_MYSOCK "localhost:$NDBCLUSTER_PORT_SLAVE" >> $NDB_EXAMPLES_OUTPUT
--exec echo Running mgmapi_logevent
--exec $NDB_EXAMPLES_DIR/mgmapi_logevent/mgmapi_logevent "localhost:$NDBCLUSTER_PORT" "localhost:$NDBCLUSTER_PORT_SLAVE" 1 >> $NDB_EXAMPLES_OUTPUT
...@@ -49,6 +49,16 @@ extern my_bool opt_ndb_optimized_node_selection; ...@@ -49,6 +49,16 @@ extern my_bool opt_ndb_optimized_node_selection;
extern const char *opt_ndbcluster_connectstring; extern const char *opt_ndbcluster_connectstring;
extern ulong opt_ndb_cache_check_time; extern ulong opt_ndb_cache_check_time;
// ndb interface initialization/cleanup
#ifdef __cplusplus
extern "C" {
#endif
extern void ndb_init_internal();
extern void ndb_end_internal();
#ifdef __cplusplus
}
#endif
const char *ndb_distribution_names[]= {"KEYHASH", "LINHASH", NullS}; const char *ndb_distribution_names[]= {"KEYHASH", "LINHASH", NullS};
TYPELIB ndb_distribution_typelib= { array_elements(ndb_distribution_names)-1, TYPELIB ndb_distribution_typelib= { array_elements(ndb_distribution_names)-1,
"", ndb_distribution_names, NULL }; "", ndb_distribution_names, NULL };
...@@ -6394,6 +6404,9 @@ static int ndbcluster_init() ...@@ -6394,6 +6404,9 @@ static int ndbcluster_init()
if (have_ndbcluster != SHOW_OPTION_YES) if (have_ndbcluster != SHOW_OPTION_YES)
DBUG_RETURN(0); // nothing else to do DBUG_RETURN(0); // nothing else to do
// Initialize ndb interface
ndb_init_internal();
// Set connectstring if specified // Set connectstring if specified
if (opt_ndbcluster_connectstring != 0) if (opt_ndbcluster_connectstring != 0)
DBUG_PRINT("connectstring", ("%s", opt_ndbcluster_connectstring)); DBUG_PRINT("connectstring", ("%s", opt_ndbcluster_connectstring));
...@@ -6543,6 +6556,9 @@ static int ndbcluster_end(ha_panic_function type) ...@@ -6543,6 +6556,9 @@ static int ndbcluster_end(ha_panic_function type)
delete g_ndb_cluster_connection; delete g_ndb_cluster_connection;
g_ndb_cluster_connection= NULL; g_ndb_cluster_connection= NULL;
// cleanup ndb interface
ndb_end_internal();
pthread_mutex_destroy(&ndbcluster_mutex); pthread_mutex_destroy(&ndbcluster_mutex);
pthread_mutex_destroy(&LOCK_ndb_util_thread); pthread_mutex_destroy(&LOCK_ndb_util_thread);
pthread_cond_destroy(&COND_ndb_util_thread); pthread_cond_destroy(&COND_ndb_util_thread);
......
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "slave.h" #include "slave.h"
#include "ha_ndbcluster_binlog.h" #include "ha_ndbcluster_binlog.h"
#include "NdbDictionary.hpp" #include "NdbDictionary.hpp"
#include "ndb_cluster_connection.hpp"
#include <util/NdbAutoPtr.hpp> #include <util/NdbAutoPtr.hpp>
#ifdef ndb_dynamite #ifdef ndb_dynamite
...@@ -111,8 +112,7 @@ static NDB_SCHEMA_OBJECT *ndb_get_schema_object(const char *key, ...@@ -111,8 +112,7 @@ static NDB_SCHEMA_OBJECT *ndb_get_schema_object(const char *key,
static void ndb_free_schema_object(NDB_SCHEMA_OBJECT **ndb_schema_object, static void ndb_free_schema_object(NDB_SCHEMA_OBJECT **ndb_schema_object,
bool have_lock); bool have_lock);
/* instantiated in storage/ndb/src/ndbapi/Ndbif.cpp */ static Uint64 *p_latest_trans_gci= 0;
extern Uint64 g_latest_trans_gci;
/* /*
Global variables for holding the binlog_index table reference Global variables for holding the binlog_index table reference
...@@ -439,7 +439,7 @@ static void ndbcluster_binlog_wait(THD *thd) ...@@ -439,7 +439,7 @@ static void ndbcluster_binlog_wait(THD *thd)
{ {
DBUG_ENTER("ndbcluster_binlog_wait"); DBUG_ENTER("ndbcluster_binlog_wait");
const char *save_info= thd ? thd->proc_info : 0; const char *save_info= thd ? thd->proc_info : 0;
ulonglong wait_epoch= g_latest_trans_gci; ulonglong wait_epoch= *p_latest_trans_gci;
int count= 30; int count= 30;
if (thd) if (thd)
thd->proc_info= "Waiting for ndbcluster binlog update to " thd->proc_info= "Waiting for ndbcluster binlog update to "
...@@ -3284,6 +3284,7 @@ static void ndb_free_schema_object(NDB_SCHEMA_OBJECT **ndb_schema_object, ...@@ -3284,6 +3284,7 @@ static void ndb_free_schema_object(NDB_SCHEMA_OBJECT **ndb_schema_object,
DBUG_VOID_RETURN; DBUG_VOID_RETURN;
} }
pthread_handler_t ndb_binlog_thread_func(void *arg) pthread_handler_t ndb_binlog_thread_func(void *arg)
{ {
THD *thd; /* needs to be first for thread_stack */ THD *thd; /* needs to be first for thread_stack */
...@@ -3292,6 +3293,7 @@ pthread_handler_t ndb_binlog_thread_func(void *arg) ...@@ -3292,6 +3293,7 @@ pthread_handler_t ndb_binlog_thread_func(void *arg)
Thd_ndb *thd_ndb=0; Thd_ndb *thd_ndb=0;
int ndb_update_binlog_index= 1; int ndb_update_binlog_index= 1;
injector *inj= injector::instance(); injector *inj= injector::instance();
#ifdef RUN_NDB_BINLOG_TIMER #ifdef RUN_NDB_BINLOG_TIMER
Timer main_timer; Timer main_timer;
#endif #endif
...@@ -3380,6 +3382,8 @@ pthread_handler_t ndb_binlog_thread_func(void *arg) ...@@ -3380,6 +3382,8 @@ pthread_handler_t ndb_binlog_thread_func(void *arg)
*/ */
injector_thd= thd; injector_thd= thd;
injector_ndb= i_ndb; injector_ndb= i_ndb;
p_latest_trans_gci=
injector_ndb->get_ndb_cluster_connection().get_latest_trans_gci();
schema_ndb= s_ndb; schema_ndb= s_ndb;
ndb_binlog_thread_running= 1; ndb_binlog_thread_running= 1;
if (opt_bin_log) if (opt_bin_log)
...@@ -3476,7 +3480,7 @@ restart: ...@@ -3476,7 +3480,7 @@ restart:
"ndb_latest_handled_binlog_epoch: %u, while current epoch: %u. " "ndb_latest_handled_binlog_epoch: %u, while current epoch: %u. "
"RESET MASTER should be issued. Resetting ndb_latest_handled_binlog_epoch.", "RESET MASTER should be issued. Resetting ndb_latest_handled_binlog_epoch.",
(unsigned) ndb_latest_handled_binlog_epoch, (unsigned) schema_gci); (unsigned) ndb_latest_handled_binlog_epoch, (unsigned) schema_gci);
g_latest_trans_gci= 0; *p_latest_trans_gci= 0;
ndb_latest_handled_binlog_epoch= 0; ndb_latest_handled_binlog_epoch= 0;
ndb_latest_applied_binlog_epoch= 0; ndb_latest_applied_binlog_epoch= 0;
ndb_latest_received_binlog_epoch= 0; ndb_latest_received_binlog_epoch= 0;
...@@ -3503,7 +3507,7 @@ restart: ...@@ -3503,7 +3507,7 @@ restart:
} }
do_ndbcluster_binlog_close_connection= BCCC_running; do_ndbcluster_binlog_close_connection= BCCC_running;
for ( ; !((abort_loop || do_ndbcluster_binlog_close_connection) && for ( ; !((abort_loop || do_ndbcluster_binlog_close_connection) &&
ndb_latest_handled_binlog_epoch >= g_latest_trans_gci) && ndb_latest_handled_binlog_epoch >= *p_latest_trans_gci) &&
do_ndbcluster_binlog_close_connection != BCCC_restart; ) do_ndbcluster_binlog_close_connection != BCCC_restart; )
{ {
#ifndef DBUG_OFF #ifndef DBUG_OFF
...@@ -3511,8 +3515,8 @@ restart: ...@@ -3511,8 +3515,8 @@ restart:
{ {
DBUG_PRINT("info", ("do_ndbcluster_binlog_close_connection: %d, " DBUG_PRINT("info", ("do_ndbcluster_binlog_close_connection: %d, "
"ndb_latest_handled_binlog_epoch: %llu, " "ndb_latest_handled_binlog_epoch: %llu, "
"g_latest_trans_gci: %llu", do_ndbcluster_binlog_close_connection, "*p_latest_trans_gci: %llu", do_ndbcluster_binlog_close_connection,
ndb_latest_handled_binlog_epoch, g_latest_trans_gci)); ndb_latest_handled_binlog_epoch, *p_latest_trans_gci));
} }
#endif #endif
#ifdef RUN_NDB_BINLOG_TIMER #ifdef RUN_NDB_BINLOG_TIMER
...@@ -3548,7 +3552,7 @@ restart: ...@@ -3548,7 +3552,7 @@ restart:
} }
if ((abort_loop || do_ndbcluster_binlog_close_connection) && if ((abort_loop || do_ndbcluster_binlog_close_connection) &&
(ndb_latest_handled_binlog_epoch >= g_latest_trans_gci || (ndb_latest_handled_binlog_epoch >= *p_latest_trans_gci ||
!ndb_binlog_running)) !ndb_binlog_running))
break; /* Shutting down server */ break; /* Shutting down server */
...@@ -3598,11 +3602,11 @@ restart: ...@@ -3598,11 +3602,11 @@ restart:
{ {
DBUG_PRINT("info", ("do_ndbcluster_binlog_close_connection= BCCC_restart")); DBUG_PRINT("info", ("do_ndbcluster_binlog_close_connection= BCCC_restart"));
do_ndbcluster_binlog_close_connection= BCCC_restart; do_ndbcluster_binlog_close_connection= BCCC_restart;
if (ndb_latest_received_binlog_epoch < g_latest_trans_gci && ndb_binlog_running) if (ndb_latest_received_binlog_epoch < *p_latest_trans_gci && ndb_binlog_running)
{ {
sql_print_error("NDB Binlog: latest transaction in epoch %lld not in binlog " sql_print_error("NDB Binlog: latest transaction in epoch %lld not in binlog "
"as latest received epoch is %lld", "as latest received epoch is %lld",
g_latest_trans_gci, ndb_latest_received_binlog_epoch); *p_latest_trans_gci, ndb_latest_received_binlog_epoch);
} }
} }
} }
...@@ -3784,11 +3788,11 @@ restart: ...@@ -3784,11 +3788,11 @@ restart:
{ {
DBUG_PRINT("info", ("do_ndbcluster_binlog_close_connection= BCCC_restart")); DBUG_PRINT("info", ("do_ndbcluster_binlog_close_connection= BCCC_restart"));
do_ndbcluster_binlog_close_connection= BCCC_restart; do_ndbcluster_binlog_close_connection= BCCC_restart;
if (ndb_latest_received_binlog_epoch < g_latest_trans_gci && ndb_binlog_running) if (ndb_latest_received_binlog_epoch < *p_latest_trans_gci && ndb_binlog_running)
{ {
sql_print_error("NDB Binlog: latest transaction in epoch %lld not in binlog " sql_print_error("NDB Binlog: latest transaction in epoch %lld not in binlog "
"as latest received epoch is %lld", "as latest received epoch is %lld",
g_latest_trans_gci, ndb_latest_received_binlog_epoch); *p_latest_trans_gci, ndb_latest_received_binlog_epoch);
} }
} }
} }
...@@ -3865,6 +3869,7 @@ err: ...@@ -3865,6 +3869,7 @@ err:
ndb_obj_cnt+= ndbcluster_util_inited ? 1 : 0; ndb_obj_cnt+= ndbcluster_util_inited ? 1 : 0;
injector_thd= 0; injector_thd= 0;
injector_ndb= 0; injector_ndb= 0;
p_latest_trans_gci= 0;
schema_ndb= 0; schema_ndb= 0;
pthread_mutex_unlock(&injector_mutex); pthread_mutex_unlock(&injector_mutex);
thd->db= 0; // as not to try to free memory thd->db= 0; // as not to try to free memory
...@@ -3985,7 +3990,7 @@ ndbcluster_show_status_binlog(THD* thd, stat_print_fn *stat_print, ...@@ -3985,7 +3990,7 @@ ndbcluster_show_status_binlog(THD* thd, stat_print_fn *stat_print,
"latest_handled_binlog_epoch=%s, " "latest_handled_binlog_epoch=%s, "
"latest_applied_binlog_epoch=%s", "latest_applied_binlog_epoch=%s",
llstr(ndb_latest_epoch, buff1), llstr(ndb_latest_epoch, buff1),
llstr(g_latest_trans_gci, buff2), llstr(*p_latest_trans_gci, buff2),
llstr(ndb_latest_received_binlog_epoch, buff3), llstr(ndb_latest_received_binlog_epoch, buff3),
llstr(ndb_latest_handled_binlog_epoch, buff4), llstr(ndb_latest_handled_binlog_epoch, buff4),
llstr(ndb_latest_applied_binlog_epoch, buff5)); llstr(ndb_latest_applied_binlog_epoch, buff5));
......
...@@ -1093,6 +1093,15 @@ public: ...@@ -1093,6 +1093,15 @@ public:
~Ndb(); ~Ndb();
#ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
/**
* The current ndb_cluster_connection get_ndb_cluster_connection.
*
* @return the current connection
*/
Ndb_cluster_connection& get_ndb_cluster_connection();
#endif
#ifndef DOXYGEN_SHOULD_SKIP_INTERNAL #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
/** /**
* The current catalog name can be fetched by getCatalogName. * The current catalog name can be fetched by getCatalogName.
......
...@@ -735,6 +735,7 @@ private: ...@@ -735,6 +735,7 @@ private:
Uint32 theTCConPtr; // Transaction Co-ordinator connection pointer. Uint32 theTCConPtr; // Transaction Co-ordinator connection pointer.
Uint64 theTransactionId; // theTransactionId of the transaction Uint64 theTransactionId; // theTransactionId of the transaction
Uint32 theGlobalCheckpointId; // The gloabl checkpoint identity of the transaction Uint32 theGlobalCheckpointId; // The gloabl checkpoint identity of the transaction
Uint64 *p_latest_trans_gci; // Reference to latest gci for connection
ConStatusType theStatus; // The status of the connection ConStatusType theStatus; // The status of the connection
enum CompletionStatus { enum CompletionStatus {
NotCompleted, NotCompleted,
...@@ -753,7 +754,7 @@ private: ...@@ -753,7 +754,7 @@ private:
bool theTransactionIsStarted; bool theTransactionIsStarted;
bool theInUseState; bool theInUseState;
bool theSimpleState; bool theSimpleState;
Uint8 m_abortOption; // Type of commit Uint8 m_abortOption; // Type of commi
enum ListState { enum ListState {
NotInList, NotInList,
......
...@@ -115,6 +115,8 @@ public: ...@@ -115,6 +115,8 @@ public:
void init_get_next_node(Ndb_cluster_connection_node_iter &iter); void init_get_next_node(Ndb_cluster_connection_node_iter &iter);
unsigned int get_next_node(Ndb_cluster_connection_node_iter &iter); unsigned int get_next_node(Ndb_cluster_connection_node_iter &iter);
unsigned get_active_ndb_objects() const; unsigned get_active_ndb_objects() const;
Uint64 *get_latest_trans_gci();
#endif #endif
private: private:
......
...@@ -361,6 +361,7 @@ private: ...@@ -361,6 +361,7 @@ private:
Uint32 poll_SHM(Uint32 timeOutMillis); Uint32 poll_SHM(Uint32 timeOutMillis);
int m_shm_own_pid; int m_shm_own_pid;
int m_transp_count;
}; };
#endif // Define of TransporterRegistry_H #endif // Define of TransporterRegistry_H
...@@ -5,7 +5,9 @@ BIN_DIRS := ndbapi_simple \ ...@@ -5,7 +5,9 @@ BIN_DIRS := ndbapi_simple \
ndbapi_simple_index \ ndbapi_simple_index \
ndbapi_event \ ndbapi_event \
ndbapi_scan \ ndbapi_scan \
mgmapi_logevent mgmapi_logevent \
ndbapi_simple_dual \
mgmapi_logevent_dual
bins: $(patsubst %, _bins_%, $(BIN_DIRS)) bins: $(patsubst %, _bins_%, $(BIN_DIRS))
......
...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../.. ...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../..
INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include
LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \ LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \
-L$(TOP_SRCDIR)/libmysql_r/.libs \ -L$(TOP_SRCDIR)/libmysql_r/.libs \
-L$(TOP_SRCDIR)/zlib/.libs \
-L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings -L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings
SYS_LIB = SYS_LIB =
......
...@@ -39,7 +39,7 @@ ...@@ -39,7 +39,7 @@
exit(-1); \ exit(-1); \
} }
int main() int main(int argc, char** argv)
{ {
NdbMgmHandle h; NdbMgmHandle h;
NdbLogEventHandle le; NdbLogEventHandle le;
...@@ -51,6 +51,15 @@ int main() ...@@ -51,6 +51,15 @@ int main()
0 }; 0 };
struct ndb_logevent event; struct ndb_logevent event;
if (argc < 2)
{
printf("Arguments are <connect_string cluster> [<iterations>].\n");
exit(-1);
}
const char *connectstring = argv[1];
int iterations = -1;
if (argc > 2)
iterations = atoi(argv[2]);
ndb_init(); ndb_init();
h= ndb_mgm_create_handle(); h= ndb_mgm_create_handle();
...@@ -59,14 +68,19 @@ int main() ...@@ -59,14 +68,19 @@ int main()
printf("Unable to create handle\n"); printf("Unable to create handle\n");
exit(-1); exit(-1);
} }
if (ndb_mgm_set_connectstring(h, connectstring) == -1)
{
printf("Unable to set connectstring\n");
exit(-1);
}
if (ndb_mgm_connect(h,0,0,0)) MGMERROR(h); if (ndb_mgm_connect(h,0,0,0)) MGMERROR(h);
le= ndb_mgm_create_logevent_handle(h, filter); le= ndb_mgm_create_logevent_handle(h, filter);
if ( le == 0 ) MGMERROR(h); if ( le == 0 ) MGMERROR(h);
while (1) while (iterations-- != 0)
{ {
int timeout= 5000; int timeout= 1000;
int r= ndb_logevent_get_next(le,&event,timeout); int r= ndb_logevent_get_next(le,&event,timeout);
if (r == 0) if (r == 0)
printf("No event within %d milliseconds\n", timeout); printf("No event within %d milliseconds\n", timeout);
......
TARGET = mgmapi_logevent_dual
SRCS = $(TARGET).cpp
OBJS = $(TARGET).o
CXX = g++
CFLAGS = -c -Wall -fno-rtti -fno-exceptions
CXXFLAGS =
DEBUG =
LFLAGS = -Wall
TOP_SRCDIR = ../../../..
INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include
LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \
-L$(TOP_SRCDIR)/libmysql_r/.libs \
-L$(TOP_SRCDIR)/zlib/.libs \
-L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings
SYS_LIB =
$(TARGET): $(OBJS)
$(CXX) $(CXXFLAGS) $(LFLAGS) $(LIB_DIR) $(OBJS) -lndbclient -lmysqlclient_r -lmysys -lmystrings -lz $(SYS_LIB) -o $(TARGET)
$(TARGET).o: $(SRCS)
$(CXX) $(CFLAGS) -I$(TOP_SRCDIR)/include -I$(INCLUDE_DIR) -I$(INCLUDE_DIR)/mgmapi -I$(INCLUDE_DIR)/ndbapi $(SRCS)
clean:
rm -f *.o $(TARGET)
/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include <mysql.h>
#include <ndbapi/NdbApi.hpp>
#include <mgmapi.h>
#include <stdio.h>
/*
* export LD_LIBRARY_PATH=../../../libmysql_r/.libs:../../../ndb/src/.libs
*/
#define MGMERROR(h) \
{ \
fprintf(stderr, "code: %d msg: %s\n", \
ndb_mgm_get_latest_error(h), \
ndb_mgm_get_latest_error_msg(h)); \
exit(-1); \
}
#define LOGEVENTERROR(h) \
{ \
fprintf(stderr, "code: %d msg: %s\n", \
ndb_logevent_get_latest_error(h), \
ndb_logevent_get_latest_error_msg(h)); \
exit(-1); \
}
int main(int argc, char** argv)
{
NdbMgmHandle h1,h2;
NdbLogEventHandle le1,le2;
int filter[] = { 15, NDB_MGM_EVENT_CATEGORY_BACKUP,
15, NDB_MGM_EVENT_CATEGORY_CONNECTION,
15, NDB_MGM_EVENT_CATEGORY_NODE_RESTART,
15, NDB_MGM_EVENT_CATEGORY_STARTUP,
15, NDB_MGM_EVENT_CATEGORY_ERROR,
0 };
struct ndb_logevent event1, event2;
if (argc < 3)
{
printf("Arguments are <connect_string cluster 1> <connect_string cluster 2> [<iterations>].\n");
exit(-1);
}
const char *connectstring1 = argv[1];
const char *connectstring2 = argv[2];
int iterations = -1;
if (argc > 3)
iterations = atoi(argv[3]);
ndb_init();
h1= ndb_mgm_create_handle();
h2= ndb_mgm_create_handle();
if ( h1 == 0 || h2 == 0 )
{
printf("Unable to create handle\n");
exit(-1);
}
if (ndb_mgm_set_connectstring(h1, connectstring1) == -1 ||
ndb_mgm_set_connectstring(h2, connectstring1))
{
printf("Unable to set connectstring\n");
exit(-1);
}
if (ndb_mgm_connect(h1,0,0,0)) MGMERROR(h1);
if (ndb_mgm_connect(h2,0,0,0)) MGMERROR(h2);
if ((le1= ndb_mgm_create_logevent_handle(h1, filter)) == 0) MGMERROR(h1);
if ((le2= ndb_mgm_create_logevent_handle(h1, filter)) == 0) MGMERROR(h2);
while (iterations-- != 0)
{
int timeout= 1000;
int r1= ndb_logevent_get_next(le1,&event1,timeout);
if (r1 == 0)
printf("No event within %d milliseconds\n", timeout);
else if (r1 < 0)
LOGEVENTERROR(le1)
else
{
switch (event1.type) {
case NDB_LE_BackupStarted:
printf("Node %d: BackupStarted\n", event1.source_nodeid);
printf(" Starting node ID: %d\n", event1.BackupStarted.starting_node);
printf(" Backup ID: %d\n", event1.BackupStarted.backup_id);
break;
case NDB_LE_BackupCompleted:
printf("Node %d: BackupCompleted\n", event1.source_nodeid);
printf(" Backup ID: %d\n", event1.BackupStarted.backup_id);
break;
case NDB_LE_BackupAborted:
printf("Node %d: BackupAborted\n", event1.source_nodeid);
break;
case NDB_LE_BackupFailedToStart:
printf("Node %d: BackupFailedToStart\n", event1.source_nodeid);
break;
case NDB_LE_NodeFailCompleted:
printf("Node %d: NodeFailCompleted\n", event1.source_nodeid);
break;
case NDB_LE_ArbitResult:
printf("Node %d: ArbitResult\n", event1.source_nodeid);
printf(" code %d, arbit_node %d\n",
event1.ArbitResult.code & 0xffff,
event1.ArbitResult.arbit_node);
break;
case NDB_LE_DeadDueToHeartbeat:
printf("Node %d: DeadDueToHeartbeat\n", event1.source_nodeid);
printf(" node %d\n", event1.DeadDueToHeartbeat.node);
break;
case NDB_LE_Connected:
printf("Node %d: Connected\n", event1.source_nodeid);
printf(" node %d\n", event1.Connected.node);
break;
case NDB_LE_Disconnected:
printf("Node %d: Disconnected\n", event1.source_nodeid);
printf(" node %d\n", event1.Disconnected.node);
break;
case NDB_LE_NDBStartCompleted:
printf("Node %d: StartCompleted\n", event1.source_nodeid);
printf(" version %d.%d.%d\n",
event1.NDBStartCompleted.version >> 16 & 0xff,
event1.NDBStartCompleted.version >> 8 & 0xff,
event1.NDBStartCompleted.version >> 0 & 0xff);
break;
case NDB_LE_ArbitState:
printf("Node %d: ArbitState\n", event1.source_nodeid);
printf(" code %d, arbit_node %d\n",
event1.ArbitState.code & 0xffff,
event1.ArbitResult.arbit_node);
break;
default:
break;
}
}
int r2= ndb_logevent_get_next(le1,&event2,timeout);
if (r2 == 0)
printf("No event within %d milliseconds\n", timeout);
else if (r2 < 0)
LOGEVENTERROR(le2)
else
{
switch (event2.type) {
case NDB_LE_BackupStarted:
printf("Node %d: BackupStarted\n", event2.source_nodeid);
printf(" Starting node ID: %d\n", event2.BackupStarted.starting_node);
printf(" Backup ID: %d\n", event2.BackupStarted.backup_id);
break;
case NDB_LE_BackupCompleted:
printf("Node %d: BackupCompleted\n", event2.source_nodeid);
printf(" Backup ID: %d\n", event2.BackupStarted.backup_id);
break;
case NDB_LE_BackupAborted:
printf("Node %d: BackupAborted\n", event2.source_nodeid);
break;
case NDB_LE_BackupFailedToStart:
printf("Node %d: BackupFailedToStart\n", event2.source_nodeid);
break;
case NDB_LE_NodeFailCompleted:
printf("Node %d: NodeFailCompleted\n", event2.source_nodeid);
break;
case NDB_LE_ArbitResult:
printf("Node %d: ArbitResult\n", event2.source_nodeid);
printf(" code %d, arbit_node %d\n",
event2.ArbitResult.code & 0xffff,
event2.ArbitResult.arbit_node);
break;
case NDB_LE_DeadDueToHeartbeat:
printf("Node %d: DeadDueToHeartbeat\n", event2.source_nodeid);
printf(" node %d\n", event2.DeadDueToHeartbeat.node);
break;
case NDB_LE_Connected:
printf("Node %d: Connected\n", event2.source_nodeid);
printf(" node %d\n", event2.Connected.node);
break;
case NDB_LE_Disconnected:
printf("Node %d: Disconnected\n", event2.source_nodeid);
printf(" node %d\n", event2.Disconnected.node);
break;
case NDB_LE_NDBStartCompleted:
printf("Node %d: StartCompleted\n", event2.source_nodeid);
printf(" version %d.%d.%d\n",
event2.NDBStartCompleted.version >> 16 & 0xff,
event2.NDBStartCompleted.version >> 8 & 0xff,
event2.NDBStartCompleted.version >> 0 & 0xff);
break;
case NDB_LE_ArbitState:
printf("Node %d: ArbitState\n", event2.source_nodeid);
printf(" code %d, arbit_node %d\n",
event2.ArbitState.code & 0xffff,
event2.ArbitResult.arbit_node);
break;
default:
break;
}
}
}
ndb_mgm_destroy_logevent_handle(&le1);
ndb_mgm_destroy_logevent_handle(&le2);
ndb_mgm_destroy_handle(&h1);
ndb_mgm_destroy_handle(&h2);
ndb_end(0);
return 0;
}
...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../.. ...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../..
INCLUDE_DIR = $(TOP_SRCDIR) INCLUDE_DIR = $(TOP_SRCDIR)
LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \ LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \
-L$(TOP_SRCDIR)/libmysql_r/.libs \ -L$(TOP_SRCDIR)/libmysql_r/.libs \
-L$(TOP_SRCDIR)/zlib/.libs \
-L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings -L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings
SYS_LIB = SYS_LIB =
......
...@@ -136,7 +136,12 @@ void closeTransaction(Ndb * ndb , async_callback_t * cb); ...@@ -136,7 +136,12 @@ void closeTransaction(Ndb * ndb , async_callback_t * cb);
/** /**
* Function to create table * Function to create table
*/ */
int create_table(Ndb * myNdb); void create_table(MYSQL &mysql);
/**
* Function to drop table
*/
void drop_table(MYSQL &mysql);
/** /**
* stat. variables * stat. variables
...@@ -193,7 +198,7 @@ callback(int result, NdbTransaction* trans, void* aObject) ...@@ -193,7 +198,7 @@ callback(int result, NdbTransaction* trans, void* aObject)
/** /**
* Create table "GARAGE" * Create table "GARAGE"
*/ */
int create_table(MYSQL &mysql) void create_table(MYSQL &mysql)
{ {
while (mysql_query(&mysql, while (mysql_query(&mysql,
"CREATE TABLE" "CREATE TABLE"
...@@ -208,15 +213,21 @@ int create_table(MYSQL &mysql) ...@@ -208,15 +213,21 @@ int create_table(MYSQL &mysql)
MYSQLERROR(mysql); MYSQLERROR(mysql);
std::cout << "MySQL Cluster already has example table: GARAGE. " std::cout << "MySQL Cluster already has example table: GARAGE. "
<< "Dropping it..." << std::endl; << "Dropping it..." << std::endl;
/************** drop_table(mysql);
* Drop table * create_table(mysql);
**************/ }
}
/**
* Drop table GARAGE
*/
void drop_table(MYSQL &mysql)
{
if (mysql_query(&mysql, "DROP TABLE GARAGE")) if (mysql_query(&mysql, "DROP TABLE GARAGE"))
MYSQLERROR(mysql); MYSQLERROR(mysql);
}
return 1;
} }
void asynchExitHandler(Ndb * m_ndb) void asynchExitHandler(Ndb * m_ndb)
{ {
if (m_ndb != NULL) if (m_ndb != NULL)
...@@ -339,8 +350,6 @@ int populate(Ndb * myNdb, int data, async_callback_t * cbData) ...@@ -339,8 +350,6 @@ int populate(Ndb * myNdb, int data, async_callback_t * cbData)
{ {
transaction[current].conn = myNdb->startTransaction(); transaction[current].conn = myNdb->startTransaction();
if (transaction[current].conn == NULL) { if (transaction[current].conn == NULL) {
if (asynchErrorHandler(transaction[current].conn, myNdb))
{
/** /**
* no transaction to close since conn == null * no transaction to close since conn == null
*/ */
...@@ -348,8 +357,6 @@ int populate(Ndb * myNdb, int data, async_callback_t * cbData) ...@@ -348,8 +357,6 @@ int populate(Ndb * myNdb, int data, async_callback_t * cbData)
retries++; retries++;
continue; continue;
} }
asynchExitHandler(myNdb);
}
myNdbOperation = transaction[current].conn->getNdbOperation(myTable); myNdbOperation = transaction[current].conn->getNdbOperation(myTable);
if (myNdbOperation == NULL) if (myNdbOperation == NULL)
{ {
...@@ -406,8 +413,15 @@ int populate(Ndb * myNdb, int data, async_callback_t * cbData) ...@@ -406,8 +413,15 @@ int populate(Ndb * myNdb, int data, async_callback_t * cbData)
return -1; return -1;
} }
int main() int main(int argc, char** argv)
{ {
if (argc != 3)
{
std::cout << "Arguments are <socket mysqld> <connect_string cluster>.\n";
exit(-1);
}
char * mysqld_sock = argv[1];
const char *connectstring = argv[2];
ndb_init(); ndb_init();
MYSQL mysql; MYSQL mysql;
...@@ -420,7 +434,7 @@ int main() ...@@ -420,7 +434,7 @@ int main()
exit(-1); exit(-1);
} }
if ( !mysql_real_connect(&mysql, "localhost", "root", "", "", if ( !mysql_real_connect(&mysql, "localhost", "root", "", "",
3306, "/tmp/mysql.sock", 0) ) 0, mysqld_sock, 0) )
MYSQLERROR(mysql); MYSQLERROR(mysql);
mysql_query(&mysql, "CREATE DATABASE TEST_DB"); mysql_query(&mysql, "CREATE DATABASE TEST_DB");
...@@ -432,7 +446,7 @@ int main() ...@@ -432,7 +446,7 @@ int main()
/************************************************************** /**************************************************************
* Connect to ndb cluster * * Connect to ndb cluster *
**************************************************************/ **************************************************************/
Ndb_cluster_connection cluster_connection; Ndb_cluster_connection cluster_connection(connectstring);
if (cluster_connection.connect(4, 5, 1)) if (cluster_connection.connect(4, 5, 1))
{ {
std::cout << "Unable to connect to cluster within 30 secs." << std::endl; std::cout << "Unable to connect to cluster within 30 secs." << std::endl;
...@@ -454,7 +468,7 @@ int main() ...@@ -454,7 +468,7 @@ int main()
/** /**
* Initialise transaction array * Initialise transaction array
*/ */
for(int i = 0 ; i < 1024 ; i++) for(int i = 0 ; i < 10 ; i++)
{ {
transaction[i].used = 0; transaction[i].used = 0;
transaction[i].conn = 0; transaction[i].conn = 0;
...@@ -462,9 +476,9 @@ int main() ...@@ -462,9 +476,9 @@ int main()
} }
int i=0; int i=0;
/** /**
* Do 20000 insert transactions. * Do 10 insert transactions.
*/ */
while(i < 20000) while(i < 10)
{ {
while(populate(myNdb,i,0)<0) // <0, no space on free list. Sleep and try again. while(populate(myNdb,i,0)<0) // <0, no space on free list. Sleep and try again.
milliSleep(10); milliSleep(10);
...@@ -473,4 +487,6 @@ int main() ...@@ -473,4 +487,6 @@ int main()
} }
std::cout << "Number of temporary errors: " << tempErrors << std::endl; std::cout << "Number of temporary errors: " << tempErrors << std::endl;
delete myNdb; delete myNdb;
drop_table(mysql);
} }
...@@ -6,9 +6,10 @@ CFLAGS = -c -Wall -fno-rtti -fno-exceptions ...@@ -6,9 +6,10 @@ CFLAGS = -c -Wall -fno-rtti -fno-exceptions
DEBUG = DEBUG =
LFLAGS = -Wall LFLAGS = -Wall
TOP_SRCDIR = ../../../.. TOP_SRCDIR = ../../../..
INCLUDE_DIR = $(TOP_SRCDIR) INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include
LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \ LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \
-L$(TOP_SRCDIR)/libmysql_r/.libs \ -L$(TOP_SRCDIR)/libmysql_r/.libs \
-L$(TOP_SRCDIR)/zlib/.libs \
-L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings -L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings
SYS_LIB = SYS_LIB =
...@@ -16,7 +17,7 @@ $(TARGET): $(OBJS) ...@@ -16,7 +17,7 @@ $(TARGET): $(OBJS)
$(CXX) $(LFLAGS) $(LIB_DIR) $(OBJS) -lndbclient -lmysqlclient_r -lmysys -lmystrings -lz $(SYS_LIB) -o $(TARGET) $(CXX) $(LFLAGS) $(LIB_DIR) $(OBJS) -lndbclient -lmysqlclient_r -lmysys -lmystrings -lz $(SYS_LIB) -o $(TARGET)
$(TARGET).o: $(SRCS) $(TARGET).o: $(SRCS)
$(CXX) $(CFLAGS) -I$(INCLUDE_DIR)/storage/ndb/include -I$(INCLUDE_DIR)/storage/ndb/include/ndbapi $(SRCS) $(CXX) $(CFLAGS) -I$(TOP_SRCDIR)/include -I$(INCLUDE_DIR) -I$(INCLUDE_DIR)/ndbapi $(SRCS)
clean: clean:
rm -f *.o $(TARGET) rm -f *.o $(TARGET)
...@@ -17,38 +17,48 @@ ...@@ -17,38 +17,48 @@
// //
// ndbapi_async1.cpp: Using asynchronous transactions in NDB API // ndbapi_async1.cpp: Using asynchronous transactions in NDB API
// //
// Execute ndbapi_example1 to create the table "MYTABLENAME"
// before executing this program.
// //
// Correct output from this program is: // Correct output from this program is:
// //
// Successful insert. // Successful insert.
// Successful insert. // Successful insert.
#include <mysql.h>
#include <NdbApi.hpp> #include <NdbApi.hpp>
// Used for cout // Used for cout
#include <iostream> #include <iostream>
#define PRINT_ERROR(code,msg) \
std::cout << "Error in " << __FILE__ << ", line: " << __LINE__ \
<< ", code: " << code \
<< ", msg: " << msg << "." << std::endl
#define MYSQLERROR(mysql) { \
PRINT_ERROR(mysql_errno(&mysql),mysql_error(&mysql)); \
exit(-1); }
#define APIERROR(error) \ #define APIERROR(error) \
{ std::cout << "Error in " << __FILE__ << ", line:" << __LINE__ << ", code:" \ { std::cout << "Error in " << __FILE__ << ", line:" << __LINE__ << ", code:" \
<< error.code << ", msg: " << error.message << "." << std::endl; \ << error.code << ", msg: " << error.message << "." << std::endl; \
exit(-1); } exit(-1); }
static void create_table(MYSQL &);
static void drop_table(MYSQL &);
static void callback(int result, NdbTransaction* NdbObject, void* aObject); static void callback(int result, NdbTransaction* NdbObject, void* aObject);
int main() int main(int argc, char** argv)
{ {
ndb_init(); if (argc != 3)
Ndb_cluster_connection *cluster_connection=
new Ndb_cluster_connection(); // Object representing the cluster
if (cluster_connection->wait_until_ready(30,30))
{ {
std::cout << "Cluster was not ready within 30 secs." << std::endl; std::cout << "Arguments are <socket mysqld> <connect_string cluster>.\n";
exit(-1); exit(-1);
} }
char * mysqld_sock = argv[1];
const char *connectstring = argv[2];
ndb_init();
Ndb_cluster_connection *cluster_connection=
new Ndb_cluster_connection(connectstring); // Object representing the cluster
int r= cluster_connection->connect(5 /* retries */, int r= cluster_connection->connect(5 /* retries */,
3 /* delay between retries */, 3 /* delay between retries */,
...@@ -66,14 +76,31 @@ int main() ...@@ -66,14 +76,31 @@ int main()
exit(-1); exit(-1);
} }
if (cluster_connection->wait_until_ready(30,30)) if (cluster_connection->wait_until_ready(30,0))
{ {
std::cout << "Cluster was not ready within 30 secs." << std::endl; std::cout << "Cluster was not ready within 30 secs." << std::endl;
exit(-1); exit(-1);
} }
// connect to mysql server
MYSQL mysql;
if ( !mysql_init(&mysql) ) {
std::cout << "mysql_init failed\n";
exit(-1);
}
if ( !mysql_real_connect(&mysql, "localhost", "root", "", "",
0, mysqld_sock, 0) )
MYSQLERROR(mysql);
/********************************************
* Connect to database via mysql-c *
********************************************/
mysql_query(&mysql, "CREATE DATABASE TEST_DB_1");
if (mysql_query(&mysql, "USE TEST_DB_1") != 0) MYSQLERROR(mysql);
create_table(mysql);
Ndb* myNdb = new Ndb( cluster_connection, Ndb* myNdb = new Ndb( cluster_connection,
"TEST_DB_2" ); // Object representing the database "TEST_DB_1" ); // Object representing the database
NdbTransaction* myNdbTransaction[2]; // For transactions NdbTransaction* myNdbTransaction[2]; // For transactions
NdbOperation* myNdbOperation; // For operations NdbOperation* myNdbOperation; // For operations
...@@ -119,10 +146,38 @@ int main() ...@@ -119,10 +146,38 @@ int main()
delete myNdb; delete myNdb;
delete cluster_connection; delete cluster_connection;
drop_table(mysql);
ndb_end(0); ndb_end(0);
return 0; return 0;
} }
/*********************************************************
* Create a table named MYTABLENAME if it does not exist *
*********************************************************/
static void create_table(MYSQL &mysql)
{
if (mysql_query(&mysql,
"CREATE TABLE"
" MYTABLENAME"
" (ATTR1 INT UNSIGNED NOT NULL PRIMARY KEY,"
" ATTR2 INT UNSIGNED NOT NULL)"
" ENGINE=NDB"))
MYSQLERROR(mysql);
}
/***********************************
* Drop a table named MYTABLENAME
***********************************/
static void drop_table(MYSQL &mysql)
{
if (mysql_query(&mysql,
"DROP TABLE"
" MYTABLENAME"))
MYSQLERROR(mysql);
}
/* /*
* callback : This is called when the transaction is polled * callback : This is called when the transaction is polled
* *
......
...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../.. ...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../..
INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include
LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \ LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \
-L$(TOP_SRCDIR)/libmysql_r/.libs \ -L$(TOP_SRCDIR)/libmysql_r/.libs \
-L$(TOP_SRCDIR)/zlib/.libs \
-L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings -L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings
SYS_LIB = SYS_LIB =
......
...@@ -117,16 +117,23 @@ int myCreateEvent(Ndb* myNdb, ...@@ -117,16 +117,23 @@ int myCreateEvent(Ndb* myNdb,
int main(int argc, char** argv) int main(int argc, char** argv)
{ {
if (argc < 3)
{
std::cout << "Arguments are <connect_string cluster> <timeout> [m(merge events)|d(debug)].\n";
exit(-1);
}
const char *connectstring = argv[1];
int timeout = atoi(argv[2]);
ndb_init(); ndb_init();
bool merge_events = argc > 1 && strchr(argv[1], 'm') != 0; bool merge_events = argc > 3 && strchr(argv[3], 'm') != 0;
#ifdef VM_TRACE #ifdef VM_TRACE
bool dbug = argc > 1 && strchr(argv[1], 'd') != 0; bool dbug = argc > 3 && strchr(argv[3], 'd') != 0;
if (dbug) DBUG_PUSH("d:t:"); if (dbug) DBUG_PUSH("d:t:");
if (dbug) putenv("API_SIGNAL_LOG=-"); if (dbug) putenv("API_SIGNAL_LOG=-");
#endif #endif
Ndb_cluster_connection *cluster_connection= Ndb_cluster_connection *cluster_connection=
new Ndb_cluster_connection(); // Object representing the cluster new Ndb_cluster_connection(connectstring); // Object representing the cluster
int r= cluster_connection->connect(5 /* retries */, int r= cluster_connection->connect(5 /* retries */,
3 /* delay between retries */, 3 /* delay between retries */,
...@@ -179,7 +186,7 @@ int main(int argc, char** argv) ...@@ -179,7 +186,7 @@ int main(int argc, char** argv)
int i, j, k, l; int i, j, k, l;
j = 0; j = 0;
while (j < 99) { while (j < timeout) {
// Start "transaction" for handling events // Start "transaction" for handling events
NdbEventOperation* op; NdbEventOperation* op;
...@@ -211,7 +218,7 @@ int main(int argc, char** argv) ...@@ -211,7 +218,7 @@ int main(int argc, char** argv)
NdbEventOperation* the_op = op; NdbEventOperation* the_op = op;
i= 0; i= 0;
while (i < 40) { while (i < timeout) {
// printf("now waiting for event...\n"); // printf("now waiting for event...\n");
int r = myNdb->pollEvents(1000); // wait for event or 1000 ms int r = myNdb->pollEvents(1000); // wait for event or 1000 ms
if (r > 0) { if (r > 0) {
...@@ -287,7 +294,7 @@ int main(int argc, char** argv) ...@@ -287,7 +294,7 @@ int main(int argc, char** argv)
} }
} }
} else } else
;//printf("timed out\n"); printf("timed out (%i)\n", timeout);
} }
// don't want to listen to events anymore // don't want to listen to events anymore
if (myNdb->dropEventOperation(the_op)) APIERROR(myNdb->getNdbError()); if (myNdb->dropEventOperation(the_op)) APIERROR(myNdb->getNdbError());
......
...@@ -6,9 +6,10 @@ CFLAGS = -c -Wall -fno-rtti -fno-exceptions ...@@ -6,9 +6,10 @@ CFLAGS = -c -Wall -fno-rtti -fno-exceptions
DEBUG = DEBUG =
LFLAGS = -Wall LFLAGS = -Wall
TOP_SRCDIR = ../../../.. TOP_SRCDIR = ../../../..
INCLUDE_DIR = ../../include INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include
LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \ LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \
-L$(TOP_SRCDIR)/libmysql_r/.libs \ -L$(TOP_SRCDIR)/libmysql_r/.libs \
-L$(TOP_SRCDIR)/zlib/.libs \
-L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings -L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings
SYS_LIB = SYS_LIB =
...@@ -16,7 +17,7 @@ $(TARGET): $(OBJS) ...@@ -16,7 +17,7 @@ $(TARGET): $(OBJS)
$(CXX) $(LFLAGS) $(LIB_DIR) $(OBJS) -lndbclient -lmysqlclient_r -lmysys -lmystrings -lz $(SYS_LIB) -o $(TARGET) $(CXX) $(LFLAGS) $(LIB_DIR) $(OBJS) -lndbclient -lmysqlclient_r -lmysys -lmystrings -lz $(SYS_LIB) -o $(TARGET)
$(TARGET).o: $(SRCS) $(TARGET).o: $(SRCS)
$(CXX) $(CFLAGS) -I$(INCLUDE_DIR) -I$(INCLUDE_DIR)/ndbapi $(SRCS) $(CXX) $(CFLAGS) -I$(TOP_SRCDIR)/include -I$(INCLUDE_DIR) -I$(INCLUDE_DIR)/ndbapi $(SRCS)
clean: clean:
rm -f *.o $(TARGET) rm -f *.o $(TARGET)
...@@ -17,9 +17,6 @@ ...@@ -17,9 +17,6 @@
// //
// ndbapi_retries.cpp: Error handling and transaction retries // ndbapi_retries.cpp: Error handling and transaction retries
// //
// Execute ndbapi_simple to create the table "MYTABLENAME"
// before executing this program.
//
// There are many ways to program using the NDB API. In this example // There are many ways to program using the NDB API. In this example
// we execute two inserts in the same transaction using // we execute two inserts in the same transaction using
// NdbConnection::execute(NoCommit). // NdbConnection::execute(NoCommit).
...@@ -29,6 +26,7 @@ ...@@ -29,6 +26,7 @@
// Application errors (i.e. errors at points marked with APIERROR) // Application errors (i.e. errors at points marked with APIERROR)
// should be handled by the application programmer. // should be handled by the application programmer.
#include <mysql.h>
#include <NdbApi.hpp> #include <NdbApi.hpp>
// Used for cout // Used for cout
...@@ -38,6 +36,14 @@ ...@@ -38,6 +36,14 @@
#include <unistd.h> #include <unistd.h>
#define TIME_TO_SLEEP_BETWEEN_TRANSACTION_RETRIES 1 #define TIME_TO_SLEEP_BETWEEN_TRANSACTION_RETRIES 1
#define PRINT_ERROR(code,msg) \
std::cout << "Error in " << __FILE__ << ", line: " << __LINE__ \
<< ", code: " << code \
<< ", msg: " << msg << "." << std::endl
#define MYSQLERROR(mysql) { \
PRINT_ERROR(mysql_errno(&mysql),mysql_error(&mysql)); \
exit(-1); }
// //
// APIERROR prints an NdbError object // APIERROR prints an NdbError object
// //
...@@ -176,13 +182,44 @@ int executeInsertTransaction(int transactionId, Ndb* myNdb, ...@@ -176,13 +182,44 @@ int executeInsertTransaction(int transactionId, Ndb* myNdb,
return result; return result;
} }
/*********************************************************
* Create a table named MYTABLENAME if it does not exist *
*********************************************************/
static void create_table(MYSQL &mysql)
{
if (mysql_query(&mysql,
"CREATE TABLE"
" MYTABLENAME"
" (ATTR1 INT UNSIGNED NOT NULL PRIMARY KEY,"
" ATTR2 INT UNSIGNED NOT NULL)"
" ENGINE=NDB"))
MYSQLERROR(mysql);
}
/***********************************
* Drop a table named MYTABLENAME
***********************************/
static void drop_table(MYSQL &mysql)
{
if (mysql_query(&mysql,
"DROP TABLE"
" MYTABLENAME"))
MYSQLERROR(mysql);
}
int main() int main(int argc, char** argv)
{ {
if (argc != 3)
{
std::cout << "Arguments are <socket mysqld> <connect_string cluster>.\n";
exit(-1);
}
char * mysqld_sock = argv[1];
const char *connectstring = argv[2];
ndb_init(); ndb_init();
Ndb_cluster_connection *cluster_connection= Ndb_cluster_connection *cluster_connection=
new Ndb_cluster_connection(); // Object representing the cluster new Ndb_cluster_connection(connectstring); // Object representing the cluster
int r= cluster_connection->connect(5 /* retries */, int r= cluster_connection->connect(5 /* retries */,
3 /* delay between retries */, 3 /* delay between retries */,
...@@ -205,6 +242,22 @@ int main() ...@@ -205,6 +242,22 @@ int main()
std::cout << "Cluster was not ready within 30 secs." << std::endl; std::cout << "Cluster was not ready within 30 secs." << std::endl;
exit(-1); exit(-1);
} }
// connect to mysql server
MYSQL mysql;
if ( !mysql_init(&mysql) ) {
std::cout << "mysql_init failed\n";
exit(-1);
}
if ( !mysql_real_connect(&mysql, "localhost", "root", "", "",
0, mysqld_sock, 0) )
MYSQLERROR(mysql);
/********************************************
* Connect to database via mysql-c *
********************************************/
mysql_query(&mysql, "CREATE DATABASE TEST_DB_1");
if (mysql_query(&mysql, "USE TEST_DB_1") != 0) MYSQLERROR(mysql);
create_table(mysql);
Ndb* myNdb= new Ndb( cluster_connection, Ndb* myNdb= new Ndb( cluster_connection,
"TEST_DB_1" ); // Object representing the database "TEST_DB_1" ); // Object representing the database
...@@ -231,6 +284,8 @@ int main() ...@@ -231,6 +284,8 @@ int main()
delete myNdb; delete myNdb;
delete cluster_connection; delete cluster_connection;
drop_table(mysql);
ndb_end(0); ndb_end(0);
return 0; return 0;
} }
...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../.. ...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../..
INCLUDE_DIR = $(TOP_SRCDIR) INCLUDE_DIR = $(TOP_SRCDIR)
LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \ LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \
-L$(TOP_SRCDIR)/libmysql_r/.libs \ -L$(TOP_SRCDIR)/libmysql_r/.libs \
-L$(TOP_SRCDIR)/zlib/.libs \
-L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings -L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings
SYS_LIB = SYS_LIB =
......
...@@ -113,10 +113,20 @@ struct Car ...@@ -113,10 +113,20 @@ struct Car
char color[20]; char color[20];
}; };
/**
* Function to drop table
*/
void drop_table(MYSQL &mysql)
{
if (mysql_query(&mysql, "DROP TABLE GARAGE"))
MYSQLERROR(mysql);
}
/** /**
* Function to create table * Function to create table
*/ */
int create_table(MYSQL &mysql) void create_table(MYSQL &mysql)
{ {
while (mysql_query(&mysql, while (mysql_query(&mysql,
"CREATE TABLE" "CREATE TABLE"
...@@ -131,16 +141,14 @@ int create_table(MYSQL &mysql) ...@@ -131,16 +141,14 @@ int create_table(MYSQL &mysql)
MYSQLERROR(mysql); MYSQLERROR(mysql);
std::cout << "MySQL Cluster already has example table: GARAGE. " std::cout << "MySQL Cluster already has example table: GARAGE. "
<< "Dropping it..." << std::endl; << "Dropping it..." << std::endl;
/************** /******************
* Drop table * * Recreate table *
**************/ ******************/
if (mysql_query(&mysql, "DROP TABLE GARAGE")) drop_table(mysql);
MYSQLERROR(mysql); create_table(mysql);
} }
return 1;
} }
int populate(Ndb * myNdb) int populate(Ndb * myNdb)
{ {
int i; int i;
...@@ -721,8 +729,15 @@ int scan_print(Ndb * myNdb) ...@@ -721,8 +729,15 @@ int scan_print(Ndb * myNdb)
} }
int main() int main(int argc, char** argv)
{ {
if (argc != 3)
{
std::cout << "Arguments are <socket mysqld> <connect_string cluster>.\n";
exit(-1);
}
char * mysqld_sock = argv[1];
const char *connectstring = argv[2];
ndb_init(); ndb_init();
MYSQL mysql; MYSQL mysql;
...@@ -735,7 +750,7 @@ int main() ...@@ -735,7 +750,7 @@ int main()
exit(-1); exit(-1);
} }
if ( !mysql_real_connect(&mysql, "localhost", "root", "", "", if ( !mysql_real_connect(&mysql, "localhost", "root", "", "",
3306, "/tmp/mysql.sock", 0) ) 0, mysqld_sock, 0) )
MYSQLERROR(mysql); MYSQLERROR(mysql);
mysql_query(&mysql, "CREATE DATABASE TEST_DB"); mysql_query(&mysql, "CREATE DATABASE TEST_DB");
...@@ -748,7 +763,7 @@ int main() ...@@ -748,7 +763,7 @@ int main()
* Connect to ndb cluster * * Connect to ndb cluster *
**************************************************************/ **************************************************************/
Ndb_cluster_connection cluster_connection; Ndb_cluster_connection cluster_connection(connectstring);
if (cluster_connection.connect(4, 5, 1)) if (cluster_connection.connect(4, 5, 1))
{ {
std::cout << "Unable to connect to cluster within 30 secs." << std::endl; std::cout << "Unable to connect to cluster within 30 secs." << std::endl;
...@@ -821,5 +836,10 @@ int main() ...@@ -821,5 +836,10 @@ int main()
if(scan_print(&myNdb) > 0) if(scan_print(&myNdb) > 0)
std::cout << "scan_print: Success!" << std::endl << std::endl; std::cout << "scan_print: Success!" << std::endl << std::endl;
/**
* Drop table
*/
drop_table(mysql);
return 0; return 0;
} }
...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../.. ...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../..
INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include
LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \ LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \
-L$(TOP_SRCDIR)/libmysql_r/.libs \ -L$(TOP_SRCDIR)/libmysql_r/.libs \
-L$(TOP_SRCDIR)/zlib/.libs \
-L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings -L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings
SYS_LIB = SYS_LIB =
......
...@@ -52,15 +52,22 @@ static void run_application(MYSQL &, Ndb_cluster_connection &); ...@@ -52,15 +52,22 @@ static void run_application(MYSQL &, Ndb_cluster_connection &);
PRINT_ERROR(error.code,error.message); \ PRINT_ERROR(error.code,error.message); \
exit(-1); } exit(-1); }
int main() int main(int argc, char** argv)
{ {
if (argc != 3)
{
std::cout << "Arguments are <socket mysqld> <connect_string cluster>.\n";
exit(-1);
}
// ndb_init must be called first // ndb_init must be called first
ndb_init(); ndb_init();
// connect to mysql server and cluster and run application // connect to mysql server and cluster and run application
{ {
char * mysqld_sock = argv[1];
const char *connectstring = argv[2];
// Object representing the cluster // Object representing the cluster
Ndb_cluster_connection cluster_connection; Ndb_cluster_connection cluster_connection(connectstring);
// Connect to cluster management server (ndb_mgmd) // Connect to cluster management server (ndb_mgmd)
if (cluster_connection.connect(4 /* retries */, if (cluster_connection.connect(4 /* retries */,
...@@ -85,7 +92,7 @@ int main() ...@@ -85,7 +92,7 @@ int main()
exit(-1); exit(-1);
} }
if ( !mysql_real_connect(&mysql, "localhost", "root", "", "", if ( !mysql_real_connect(&mysql, "localhost", "root", "", "",
3306, "/tmp/mysql.sock", 0) ) 0, mysqld_sock, 0) )
MYSQLERROR(mysql); MYSQLERROR(mysql);
// run the application code // run the application code
...@@ -94,13 +101,11 @@ int main() ...@@ -94,13 +101,11 @@ int main()
ndb_end(0); ndb_end(0);
std::cout << "\nTo drop created table use:\n"
<< "echo \"drop table MYTABLENAME\" | mysql TEST_DB_1 -u root\n";
return 0; return 0;
} }
static void create_table(MYSQL &); static void create_table(MYSQL &);
static void drop_table(MYSQL &);
static void do_insert(Ndb &); static void do_insert(Ndb &);
static void do_update(Ndb &); static void do_update(Ndb &);
static void do_delete(Ndb &); static void do_delete(Ndb &);
...@@ -130,6 +135,8 @@ static void run_application(MYSQL &mysql, ...@@ -130,6 +135,8 @@ static void run_application(MYSQL &mysql,
do_update(myNdb); do_update(myNdb);
do_delete(myNdb); do_delete(myNdb);
do_read(myNdb); do_read(myNdb);
drop_table(mysql);
mysql_query(&mysql, "DROP DATABASE TEST_DB_1");
} }
/********************************************************* /*********************************************************
...@@ -146,6 +153,17 @@ static void create_table(MYSQL &mysql) ...@@ -146,6 +153,17 @@ static void create_table(MYSQL &mysql)
MYSQLERROR(mysql); MYSQLERROR(mysql);
} }
/***********************************
* Drop a table named MYTABLENAME
***********************************/
static void drop_table(MYSQL &mysql)
{
if (mysql_query(&mysql,
"DROP TABLE"
" MYTABLENAME"))
MYSQLERROR(mysql);
}
/************************************************************************** /**************************************************************************
* Using 5 transactions, insert 10 tuples in table: (0,0),(1,1),...,(9,9) * * Using 5 transactions, insert 10 tuples in table: (0,0),(1,1),...,(9,9) *
**************************************************************************/ **************************************************************************/
......
TARGET = ndbapi_simple_dual
SRCS = $(TARGET).cpp
OBJS = $(TARGET).o
CXX = g++
CFLAGS = -c -Wall -fno-rtti -fno-exceptions
CXXFLAGS =
DEBUG =
LFLAGS = -Wall
TOP_SRCDIR = ../../../..
INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include
LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \
-L$(TOP_SRCDIR)/libmysql_r/.libs \
-L$(TOP_SRCDIR)/zlib/.libs \
-L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings
SYS_LIB =
$(TARGET): $(OBJS)
$(CXX) $(CXXFLAGS) $(LFLAGS) $(LIB_DIR) $(OBJS) -lndbclient -lmysqlclient_r -lmysys -lmystrings -lz $(SYS_LIB) -o $(TARGET)
$(TARGET).o: $(SRCS)
$(CXX) $(CFLAGS) -I$(TOP_SRCDIR)/include -I$(INCLUDE_DIR) -I$(INCLUDE_DIR)/ndbapi $(SRCS)
clean:
rm -f *.o $(TARGET)
/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
/*
* ndbapi_simple_dual.cpp: Using synchronous transactions in NDB API
*
* Correct output from this program is:
*
* ATTR1 ATTR2
* 0 10
* 1 1
* 2 12
* Detected that deleted tuple doesn't exist!
* 4 14
* 5 5
* 6 16
* 7 7
* 8 18
* 9 9
* ATTR1 ATTR2
* 0 10
* 1 1
* 2 12
* Detected that deleted tuple doesn't exist!
* 4 14
* 5 5
* 6 16
* 7 7
* 8 18
* 9 9
*
*/
#include <mysql.h>
#include <NdbApi.hpp>
// Used for cout
#include <stdio.h>
#include <iostream>
static void run_application(MYSQL &, Ndb_cluster_connection &, const char* table, const char* db);
#define PRINT_ERROR(code,msg) \
std::cout << "Error in " << __FILE__ << ", line: " << __LINE__ \
<< ", code: " << code \
<< ", msg: " << msg << "." << std::endl
#define MYSQLERROR(mysql) { \
PRINT_ERROR(mysql_errno(&mysql),mysql_error(&mysql)); \
exit(-1); }
#define APIERROR(error) { \
PRINT_ERROR(error.code,error.message); \
exit(-1); }
int main(int argc, char** argv)
{
if (argc != 5)
{
std::cout << "Arguments are <socket mysqld1> <connect_string cluster 1> <socket mysqld2> <connect_string cluster 2>.\n";
exit(-1);
}
// ndb_init must be called first
ndb_init();
{
char * mysqld1_sock = argv[1];
const char *connectstring1 = argv[2];
char * mysqld2_sock = argv[3];
const char *connectstring2 = argv[4];
// Object representing the cluster 1
Ndb_cluster_connection cluster1_connection(connectstring1);
MYSQL mysql1;
// Object representing the cluster 2
Ndb_cluster_connection cluster2_connection(connectstring2);
MYSQL mysql2;
// connect to mysql server and cluster 1 and run application
// Connect to cluster 1 management server (ndb_mgmd)
if (cluster1_connection.connect(4 /* retries */,
5 /* delay between retries */,
1 /* verbose */))
{
std::cout << "Cluster 1 management server was not ready within 30 secs.\n";
exit(-1);
}
// Optionally connect and wait for the storage nodes (ndbd's)
if (cluster1_connection.wait_until_ready(30,0) < 0)
{
std::cout << "Cluster 1 was not ready within 30 secs.\n";
exit(-1);
}
// connect to mysql server in cluster 1
if ( !mysql_init(&mysql1) ) {
std::cout << "mysql_init failed\n";
exit(-1);
}
if ( !mysql_real_connect(&mysql1, "localhost", "root", "", "",
0, mysqld1_sock, 0) )
MYSQLERROR(mysql1);
// connect to mysql server and cluster 2 and run application
// Connect to cluster management server (ndb_mgmd)
if (cluster2_connection.connect(4 /* retries */,
5 /* delay between retries */,
1 /* verbose */))
{
std::cout << "Cluster 2 management server was not ready within 30 secs.\n";
exit(-1);
}
// Optionally connect and wait for the storage nodes (ndbd's)
if (cluster2_connection.wait_until_ready(30,0) < 0)
{
std::cout << "Cluster 2 was not ready within 30 secs.\n";
exit(-1);
}
// connect to mysql server in cluster 2
if ( !mysql_init(&mysql2) ) {
std::cout << "mysql_init failed\n";
exit(-1);
}
if ( !mysql_real_connect(&mysql2, "localhost", "root", "", "",
0, mysqld2_sock, 0) )
MYSQLERROR(mysql2);
// run the application code
run_application(mysql1, cluster1_connection, "MYTABLENAME1", "TEST_DB_1");
run_application(mysql2, cluster2_connection, "MYTABLENAME2", "TEST_DB_2");
}
// Note: all connections must have been destroyed before calling ndb_end()
ndb_end(0);
return 0;
}
static void create_table(MYSQL &, const char* table);
static void drop_table(MYSQL &, const char* table);
static void do_insert(Ndb &, const char* table);
static void do_update(Ndb &, const char* table);
static void do_delete(Ndb &, const char* table);
static void do_read(Ndb &, const char* table);
static void run_application(MYSQL &mysql,
Ndb_cluster_connection &cluster_connection,
const char* table,
const char* db)
{
/********************************************
* Connect to database via mysql-c *
********************************************/
char db_stmt[256];
sprintf(db_stmt, "CREATE DATABASE %s\n", db);
mysql_query(&mysql, db_stmt);
sprintf(db_stmt, "USE %s", db);
if (mysql_query(&mysql, db_stmt) != 0) MYSQLERROR(mysql);
create_table(mysql, table);
/********************************************
* Connect to database via NdbApi *
********************************************/
// Object representing the database
Ndb myNdb( &cluster_connection, db );
if (myNdb.init()) APIERROR(myNdb.getNdbError());
/*
* Do different operations on database
*/
do_insert(myNdb, table);
do_update(myNdb, table);
do_delete(myNdb, table);
do_read(myNdb, table);
/*
* Drop the table
*/
drop_table(mysql, table);
sprintf(db_stmt, "DROP DATABASE %s\n", db);
mysql_query(&mysql, db_stmt);
}
/*********************************************************
* Create a table named by table if it does not exist *
*********************************************************/
static void create_table(MYSQL &mysql, const char* table)
{
char create_stmt[256];
sprintf(create_stmt, "CREATE TABLE %s \
(ATTR1 INT UNSIGNED NOT NULL PRIMARY KEY,\
ATTR2 INT UNSIGNED NOT NULL)\
ENGINE=NDB", table);
if (mysql_query(&mysql, create_stmt))
MYSQLERROR(mysql);
}
/*******************************
* Drop a table named by table
*******************************/
static void drop_table(MYSQL &mysql, const char* table)
{
char drop_stmt[256];
sprintf(drop_stmt, "DROP TABLE IF EXISTS %s", table);
if (mysql_query(&mysql, drop_stmt))
MYSQLERROR(mysql);
}
/**************************************************************************
* Using 5 transactions, insert 10 tuples in table: (0,0),(1,1),...,(9,9) *
**************************************************************************/
static void do_insert(Ndb &myNdb, const char* table)
{
const NdbDictionary::Dictionary* myDict= myNdb.getDictionary();
const NdbDictionary::Table *myTable= myDict->getTable(table);
if (myTable == NULL)
APIERROR(myDict->getNdbError());
for (int i = 0; i < 5; i++) {
NdbTransaction *myTransaction= myNdb.startTransaction();
if (myTransaction == NULL) APIERROR(myNdb.getNdbError());
NdbOperation *myOperation= myTransaction->getNdbOperation(myTable);
if (myOperation == NULL) APIERROR(myTransaction->getNdbError());
myOperation->insertTuple();
myOperation->equal("ATTR1", i);
myOperation->setValue("ATTR2", i);
myOperation= myTransaction->getNdbOperation(myTable);
if (myOperation == NULL) APIERROR(myTransaction->getNdbError());
myOperation->insertTuple();
myOperation->equal("ATTR1", i+5);
myOperation->setValue("ATTR2", i+5);
if (myTransaction->execute( NdbTransaction::Commit ) == -1)
APIERROR(myTransaction->getNdbError());
myNdb.closeTransaction(myTransaction);
}
}
/*****************************************************************
* Update the second attribute in half of the tuples (adding 10) *
*****************************************************************/
static void do_update(Ndb &myNdb, const char* table)
{
const NdbDictionary::Dictionary* myDict= myNdb.getDictionary();
const NdbDictionary::Table *myTable= myDict->getTable(table);
if (myTable == NULL)
APIERROR(myDict->getNdbError());
for (int i = 0; i < 10; i+=2) {
NdbTransaction *myTransaction= myNdb.startTransaction();
if (myTransaction == NULL) APIERROR(myNdb.getNdbError());
NdbOperation *myOperation= myTransaction->getNdbOperation(myTable);
if (myOperation == NULL) APIERROR(myTransaction->getNdbError());
myOperation->updateTuple();
myOperation->equal( "ATTR1", i );
myOperation->setValue( "ATTR2", i+10);
if( myTransaction->execute( NdbTransaction::Commit ) == -1 )
APIERROR(myTransaction->getNdbError());
myNdb.closeTransaction(myTransaction);
}
}
/*************************************************
* Delete one tuple (the one with primary key 3) *
*************************************************/
static void do_delete(Ndb &myNdb, const char* table)
{
const NdbDictionary::Dictionary* myDict= myNdb.getDictionary();
const NdbDictionary::Table *myTable= myDict->getTable(table);
if (myTable == NULL)
APIERROR(myDict->getNdbError());
NdbTransaction *myTransaction= myNdb.startTransaction();
if (myTransaction == NULL) APIERROR(myNdb.getNdbError());
NdbOperation *myOperation= myTransaction->getNdbOperation(myTable);
if (myOperation == NULL) APIERROR(myTransaction->getNdbError());
myOperation->deleteTuple();
myOperation->equal( "ATTR1", 3 );
if (myTransaction->execute(NdbTransaction::Commit) == -1)
APIERROR(myTransaction->getNdbError());
myNdb.closeTransaction(myTransaction);
}
/*****************************
* Read and print all tuples *
*****************************/
static void do_read(Ndb &myNdb, const char* table)
{
const NdbDictionary::Dictionary* myDict= myNdb.getDictionary();
const NdbDictionary::Table *myTable= myDict->getTable(table);
if (myTable == NULL)
APIERROR(myDict->getNdbError());
std::cout << "ATTR1 ATTR2" << std::endl;
for (int i = 0; i < 10; i++) {
NdbTransaction *myTransaction= myNdb.startTransaction();
if (myTransaction == NULL) APIERROR(myNdb.getNdbError());
NdbOperation *myOperation= myTransaction->getNdbOperation(myTable);
if (myOperation == NULL) APIERROR(myTransaction->getNdbError());
myOperation->readTuple(NdbOperation::LM_Read);
myOperation->equal("ATTR1", i);
NdbRecAttr *myRecAttr= myOperation->getValue("ATTR2", NULL);
if (myRecAttr == NULL) APIERROR(myTransaction->getNdbError());
if(myTransaction->execute( NdbTransaction::Commit ) == -1)
if (i == 3) {
std::cout << "Detected that deleted tuple doesn't exist!" << std::endl;
} else {
APIERROR(myTransaction->getNdbError());
}
if (i != 3) {
printf(" %2d %2d\n", i, myRecAttr->u_32_value());
}
myNdb.closeTransaction(myTransaction);
}
}
...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../.. ...@@ -10,6 +10,7 @@ TOP_SRCDIR = ../../../..
INCLUDE_DIR = $(TOP_SRCDIR) INCLUDE_DIR = $(TOP_SRCDIR)
LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \ LIB_DIR = -L$(TOP_SRCDIR)/storage/ndb/src/.libs \
-L$(TOP_SRCDIR)/libmysql_r/.libs \ -L$(TOP_SRCDIR)/libmysql_r/.libs \
-L$(TOP_SRCDIR)/zlib/.libs \
-L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings -L$(TOP_SRCDIR)/mysys -L$(TOP_SRCDIR)/strings
SYS_LIB = SYS_LIB =
......
...@@ -49,8 +49,15 @@ ...@@ -49,8 +49,15 @@
PRINT_ERROR(error.code,error.message); \ PRINT_ERROR(error.code,error.message); \
exit(-1); } exit(-1); }
int main() int main(int argc, char** argv)
{ {
if (argc != 3)
{
std::cout << "Arguments are <socket mysqld> <connect_string cluster>.\n";
exit(-1);
}
char * mysqld_sock = argv[1];
const char *connectstring = argv[2];
ndb_init(); ndb_init();
MYSQL mysql; MYSQL mysql;
...@@ -63,7 +70,7 @@ int main() ...@@ -63,7 +70,7 @@ int main()
exit(-1); exit(-1);
} }
if ( !mysql_real_connect(&mysql, "localhost", "root", "", "", if ( !mysql_real_connect(&mysql, "localhost", "root", "", "",
3306, "/tmp/mysql.sock", 0) ) 0, mysqld_sock, 0) )
MYSQLERROR(mysql); MYSQLERROR(mysql);
mysql_query(&mysql, "CREATE DATABASE TEST_DB_1"); mysql_query(&mysql, "CREATE DATABASE TEST_DB_1");
...@@ -85,7 +92,7 @@ int main() ...@@ -85,7 +92,7 @@ int main()
**************************************************************/ **************************************************************/
Ndb_cluster_connection *cluster_connection= Ndb_cluster_connection *cluster_connection=
new Ndb_cluster_connection(); // Object representing the cluster new Ndb_cluster_connection(connectstring); // Object representing the cluster
if (cluster_connection->connect(5,3,1)) if (cluster_connection->connect(5,3,1))
{ {
...@@ -110,7 +117,7 @@ int main() ...@@ -110,7 +117,7 @@ int main()
const NdbDictionary::Table *myTable= myDict->getTable("MYTABLENAME"); const NdbDictionary::Table *myTable= myDict->getTable("MYTABLENAME");
if (myTable == NULL) if (myTable == NULL)
APIERROR(myDict->getNdbError()); APIERROR(myDict->getNdbError());
const NdbDictionary::Index *myIndex= myDict->getIndex("MYINDEXNAME","MYTABLENAME"); const NdbDictionary::Index *myIndex= myDict->getIndex("MYINDEXNAME$unique","MYTABLENAME");
if (myIndex == NULL) if (myIndex == NULL)
APIERROR(myDict->getNdbError()); APIERROR(myDict->getNdbError());
......
...@@ -213,8 +213,8 @@ TransporterRegistry::unpack(Uint32 * readPtr, ...@@ -213,8 +213,8 @@ TransporterRegistry::unpack(Uint32 * readPtr,
Uint32 * eodPtr, Uint32 * eodPtr,
NodeId remoteNodeId, NodeId remoteNodeId,
IOState state) { IOState state) {
static SignalHeader signalHeader; SignalHeader signalHeader;
static LinearSectionPtr ptr[3]; LinearSectionPtr ptr[3];
Uint32 loop_count = 0; Uint32 loop_count = 0;
if(state == NoHalt || state == HaltOutput){ if(state == NoHalt || state == HaltOutput){
while ((readPtr < eodPtr) && (loop_count < MAX_RECEIVED_SIGNALS)) { while ((readPtr < eodPtr) && (loop_count < MAX_RECEIVED_SIGNALS)) {
......
...@@ -80,14 +80,15 @@ SocketServer::Session * TransporterService::newSession(NDB_SOCKET_TYPE sockfd) ...@@ -80,14 +80,15 @@ SocketServer::Session * TransporterService::newSession(NDB_SOCKET_TYPE sockfd)
TransporterRegistry::TransporterRegistry(void * callback, TransporterRegistry::TransporterRegistry(void * callback,
unsigned _maxTransporters, unsigned _maxTransporters,
unsigned sizeOfLongSignalMemory) unsigned sizeOfLongSignalMemory) :
m_mgm_handle(0),
m_transp_count(0)
{ {
DBUG_ENTER("TransporterRegistry::TransporterRegistry"); DBUG_ENTER("TransporterRegistry::TransporterRegistry");
nodeIdSpecified = false; nodeIdSpecified = false;
maxTransporters = _maxTransporters; maxTransporters = _maxTransporters;
sendCounter = 1; sendCounter = 1;
m_mgm_handle= 0;
callbackObj=callback; callbackObj=callback;
...@@ -1002,7 +1003,6 @@ TransporterRegistry::performReceive() ...@@ -1002,7 +1003,6 @@ TransporterRegistry::performReceive()
#endif #endif
} }
static int x = 0;
void void
TransporterRegistry::performSend() TransporterRegistry::performSend()
{ {
...@@ -1070,7 +1070,7 @@ TransporterRegistry::performSend() ...@@ -1070,7 +1070,7 @@ TransporterRegistry::performSend()
} }
#endif #endif
#ifdef NDB_TCP_TRANSPORTER #ifdef NDB_TCP_TRANSPORTER
for (i = x; i < nTCPTransporters; i++) for (i = m_transp_count; i < nTCPTransporters; i++)
{ {
TCP_Transporter *t = theTCPTransporters[i]; TCP_Transporter *t = theTCPTransporters[i];
if (t && t->hasDataToSend() && t->isConnected() && if (t && t->hasDataToSend() && t->isConnected() &&
...@@ -1079,7 +1079,7 @@ TransporterRegistry::performSend() ...@@ -1079,7 +1079,7 @@ TransporterRegistry::performSend()
t->doSend(); t->doSend();
} }
} }
for (i = 0; i < x && i < nTCPTransporters; i++) for (i = 0; i < m_transp_count && i < nTCPTransporters; i++)
{ {
TCP_Transporter *t = theTCPTransporters[i]; TCP_Transporter *t = theTCPTransporters[i];
if (t && t->hasDataToSend() && t->isConnected() && if (t && t->hasDataToSend() && t->isConnected() &&
...@@ -1088,8 +1088,8 @@ TransporterRegistry::performSend() ...@@ -1088,8 +1088,8 @@ TransporterRegistry::performSend()
t->doSend(); t->doSend();
} }
} }
x++; m_transp_count++;
if (x == nTCPTransporters) x = 0; if (m_transp_count == nTCPTransporters) m_transp_count = 0;
#endif #endif
#endif #endif
#ifdef NDB_SCI_TRANSPORTER #ifdef NDB_SCI_TRANSPORTER
......
...@@ -16,6 +16,16 @@ ...@@ -16,6 +16,16 @@
#include <ndb_global.h> #include <ndb_global.h>
#include <my_sys.h> #include <my_sys.h>
#include <NdbMutex.h>
NdbMutex *g_ndb_connection_mutex = NULL;
void
ndb_init_internal()
{
if (!g_ndb_connection_mutex)
g_ndb_connection_mutex = NdbMutex_Create();
}
int int
ndb_init() ndb_init()
...@@ -25,11 +35,20 @@ ndb_init() ...@@ -25,11 +35,20 @@ ndb_init()
write(2, err, strlen(err)); write(2, err, strlen(err));
exit(1); exit(1);
} }
ndb_init_internal();
return 0; return 0;
} }
void
ndb_end_internal()
{
if (g_ndb_connection_mutex)
NdbMutex_Destroy(g_ndb_connection_mutex);
}
void void
ndb_end(int flags) ndb_end(int flags)
{ {
my_end(flags); my_end(flags);
ndb_end_internal();
} }
...@@ -1194,7 +1194,7 @@ const unsigned int * ...@@ -1194,7 +1194,7 @@ const unsigned int *
ndb_mgm_get_clusterlog_severity_filter(NdbMgmHandle handle) ndb_mgm_get_clusterlog_severity_filter(NdbMgmHandle handle)
{ {
SET_ERROR(handle, NDB_MGM_NO_ERROR, "Executing: ndb_mgm_get_clusterlog_severity_filter"); SET_ERROR(handle, NDB_MGM_NO_ERROR, "Executing: ndb_mgm_get_clusterlog_severity_filter");
static unsigned int enabled[(int)NDB_MGM_EVENT_SEVERITY_ALL]= unsigned int enabled[(int)NDB_MGM_EVENT_SEVERITY_ALL]=
{0,0,0,0,0,0,0}; {0,0,0,0,0,0,0};
const ParserRow<ParserDummy> getinfo_reply[] = { const ParserRow<ParserDummy> getinfo_reply[] = {
MGM_CMD("clusterlog", NULL, ""), MGM_CMD("clusterlog", NULL, ""),
......
...@@ -1172,6 +1172,12 @@ convertEndian(Uint32 Data) ...@@ -1172,6 +1172,12 @@ convertEndian(Uint32 Data)
} }
// <internal> // <internal>
Ndb_cluster_connection &
Ndb::get_ndb_cluster_connection()
{
return theImpl->m_ndb_cluster_connection;
}
const char * Ndb::getCatalogName() const const char * Ndb::getCatalogName() const
{ {
return theImpl->m_dbname.c_str(); return theImpl->m_dbname.c_str();
......
...@@ -1303,8 +1303,6 @@ NdbDictionaryImpl::NdbDictionaryImpl(Ndb &ndb, ...@@ -1303,8 +1303,6 @@ NdbDictionaryImpl::NdbDictionaryImpl(Ndb &ndb,
m_local_table_data_size= 0; m_local_table_data_size= 0;
} }
static int f_dictionary_count = 0;
NdbDictionaryImpl::~NdbDictionaryImpl() NdbDictionaryImpl::~NdbDictionaryImpl()
{ {
NdbElement_t<Ndb_local_table_info> * curr = m_localHash.m_tableHash.getNext(0); NdbElement_t<Ndb_local_table_info> * curr = m_localHash.m_tableHash.getNext(0);
...@@ -1317,33 +1315,6 @@ NdbDictionaryImpl::~NdbDictionaryImpl() ...@@ -1317,33 +1315,6 @@ NdbDictionaryImpl::~NdbDictionaryImpl()
curr = m_localHash.m_tableHash.getNext(curr); curr = m_localHash.m_tableHash.getNext(curr);
} }
m_globalHash->lock();
if(--f_dictionary_count == 0){
delete NdbDictionary::Column::FRAGMENT;
delete NdbDictionary::Column::FRAGMENT_FIXED_MEMORY;
delete NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY;
delete NdbDictionary::Column::ROW_COUNT;
delete NdbDictionary::Column::COMMIT_COUNT;
delete NdbDictionary::Column::ROW_SIZE;
delete NdbDictionary::Column::RANGE_NO;
delete NdbDictionary::Column::DISK_REF;
delete NdbDictionary::Column::RECORDS_IN_RANGE;
delete NdbDictionary::Column::ROWID;
delete NdbDictionary::Column::ROW_GCI;
NdbDictionary::Column::FRAGMENT= 0;
NdbDictionary::Column::FRAGMENT_FIXED_MEMORY= 0;
NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY= 0;
NdbDictionary::Column::ROW_COUNT= 0;
NdbDictionary::Column::COMMIT_COUNT= 0;
NdbDictionary::Column::ROW_SIZE= 0;
NdbDictionary::Column::RANGE_NO= 0;
NdbDictionary::Column::DISK_REF= 0;
NdbDictionary::Column::RECORDS_IN_RANGE= 0;
NdbDictionary::Column::ROWID= 0;
NdbDictionary::Column::ROW_GCI= 0;
}
m_globalHash->unlock();
} else { } else {
assert(curr == 0); assert(curr == 0);
} }
...@@ -1486,32 +1457,6 @@ NdbDictionaryImpl::setTransporter(class Ndb* ndb, ...@@ -1486,32 +1457,6 @@ NdbDictionaryImpl::setTransporter(class Ndb* ndb,
{ {
m_globalHash = &tf->m_globalDictCache; m_globalHash = &tf->m_globalDictCache;
if(m_receiver.setTransporter(ndb, tf)){ if(m_receiver.setTransporter(ndb, tf)){
m_globalHash->lock();
if(f_dictionary_count++ == 0){
NdbDictionary::Column::FRAGMENT=
NdbColumnImpl::create_pseudo("NDB$FRAGMENT");
NdbDictionary::Column::FRAGMENT_FIXED_MEMORY=
NdbColumnImpl::create_pseudo("NDB$FRAGMENT_FIXED_MEMORY");
NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY=
NdbColumnImpl::create_pseudo("NDB$FRAGMENT_VARSIZED_MEMORY");
NdbDictionary::Column::ROW_COUNT=
NdbColumnImpl::create_pseudo("NDB$ROW_COUNT");
NdbDictionary::Column::COMMIT_COUNT=
NdbColumnImpl::create_pseudo("NDB$COMMIT_COUNT");
NdbDictionary::Column::ROW_SIZE=
NdbColumnImpl::create_pseudo("NDB$ROW_SIZE");
NdbDictionary::Column::RANGE_NO=
NdbColumnImpl::create_pseudo("NDB$RANGE_NO");
NdbDictionary::Column::DISK_REF=
NdbColumnImpl::create_pseudo("NDB$DISK_REF");
NdbDictionary::Column::RECORDS_IN_RANGE=
NdbColumnImpl::create_pseudo("NDB$RECORDS_IN_RANGE");
NdbDictionary::Column::ROWID=
NdbColumnImpl::create_pseudo("NDB$ROWID");
NdbDictionary::Column::ROW_GCI=
NdbColumnImpl::create_pseudo("NDB$ROW_GCI");
}
m_globalHash->unlock();
return true; return true;
} }
return false; return false;
......
...@@ -927,8 +927,6 @@ NdbEventOperationImpl::printAll() ...@@ -927,8 +927,6 @@ NdbEventOperationImpl::printAll()
* Each Ndb object has a Object. * Each Ndb object has a Object.
*/ */
// ToDo ref count this so it get's destroyed
NdbMutex *NdbEventBuffer::p_add_drop_mutex= 0;
NdbEventBuffer::NdbEventBuffer(Ndb *ndb) : NdbEventBuffer::NdbEventBuffer(Ndb *ndb) :
m_system_nodes(ndb->theImpl->theNoOfDBnodes), m_system_nodes(ndb->theImpl->theNoOfDBnodes),
...@@ -940,7 +938,8 @@ NdbEventBuffer::NdbEventBuffer(Ndb *ndb) : ...@@ -940,7 +938,8 @@ NdbEventBuffer::NdbEventBuffer(Ndb *ndb) :
m_max_free_thresh(100), m_max_free_thresh(100),
m_gci_slip_thresh(3), m_gci_slip_thresh(3),
m_dropped_ev_op(0), m_dropped_ev_op(0),
m_active_op_count(0) m_active_op_count(0),
m_add_drop_mutex(0)
{ {
#ifdef VM_TRACE #ifdef VM_TRACE
m_latest_command= "NdbEventBuffer::NdbEventBuffer"; m_latest_command= "NdbEventBuffer::NdbEventBuffer";
...@@ -952,16 +951,6 @@ NdbEventBuffer::NdbEventBuffer(Ndb *ndb) : ...@@ -952,16 +951,6 @@ NdbEventBuffer::NdbEventBuffer(Ndb *ndb) :
exit(-1); exit(-1);
} }
m_mutex= ndb->theImpl->theWaiter.m_mutex; m_mutex= ndb->theImpl->theWaiter.m_mutex;
lock();
if (p_add_drop_mutex == 0)
{
if ((p_add_drop_mutex = NdbMutex_Create()) == NULL) {
ndbout_c("NdbEventBuffer: NdbMutex_Create() failed");
exit(-1);
}
}
unlock();
// ToDo set event buffer size // ToDo set event buffer size
// pre allocate event data array // pre allocate event data array
m_sz= 0; m_sz= 0;
...@@ -971,6 +960,10 @@ NdbEventBuffer::NdbEventBuffer(Ndb *ndb) : ...@@ -971,6 +960,10 @@ NdbEventBuffer::NdbEventBuffer(Ndb *ndb) :
m_free_data= 0; m_free_data= 0;
m_free_data_sz= 0; m_free_data_sz= 0;
// get reference to mutex managed by current connection
m_add_drop_mutex=
m_ndb->theImpl->m_ndb_cluster_connection.m_event_add_drop_mutex;
// initialize lists // initialize lists
bzero(&g_empty_gci_container, sizeof(Gci_container)); bzero(&g_empty_gci_container, sizeof(Gci_container));
init_gci_containers(); init_gci_containers();
...@@ -1008,14 +1001,6 @@ NdbEventBuffer::~NdbEventBuffer() ...@@ -1008,14 +1001,6 @@ NdbEventBuffer::~NdbEventBuffer()
} }
NdbCondition_Destroy(p_cond); NdbCondition_Destroy(p_cond);
lock();
if (p_add_drop_mutex)
{
NdbMutex_Destroy(p_add_drop_mutex);
p_add_drop_mutex = 0;
}
unlock();
} }
void void
......
...@@ -407,8 +407,8 @@ public: ...@@ -407,8 +407,8 @@ public:
void dropEventOperation(NdbEventOperation *); void dropEventOperation(NdbEventOperation *);
static NdbEventOperationImpl* getEventOperationImpl(NdbEventOperation* tOp); static NdbEventOperationImpl* getEventOperationImpl(NdbEventOperation* tOp);
void add_drop_lock() { NdbMutex_Lock(p_add_drop_mutex); } void add_drop_lock() { NdbMutex_Lock(m_add_drop_mutex); }
void add_drop_unlock() { NdbMutex_Unlock(p_add_drop_mutex); } void add_drop_unlock() { NdbMutex_Unlock(m_add_drop_mutex); }
void lock() { NdbMutex_Lock(m_mutex); } void lock() { NdbMutex_Lock(m_mutex); }
void unlock() { NdbMutex_Unlock(m_mutex); } void unlock() { NdbMutex_Unlock(m_mutex); }
...@@ -510,6 +510,7 @@ private: ...@@ -510,6 +510,7 @@ private:
NdbEventOperationImpl *m_dropped_ev_op; NdbEventOperationImpl *m_dropped_ev_op;
Uint32 m_active_op_count; Uint32 m_active_op_count;
NdbMutex *m_add_drop_mutex;
}; };
inline inline
......
...@@ -32,8 +32,6 @@ ...@@ -32,8 +32,6 @@
#include <signaldata/TcKeyFailConf.hpp> #include <signaldata/TcKeyFailConf.hpp>
#include <signaldata/TcHbRep.hpp> #include <signaldata/TcHbRep.hpp>
Uint64 g_latest_trans_gci = 0;
/***************************************************************************** /*****************************************************************************
NdbTransaction( Ndb* aNdb ); NdbTransaction( Ndb* aNdb );
...@@ -64,6 +62,7 @@ NdbTransaction::NdbTransaction( Ndb* aNdb ) : ...@@ -64,6 +62,7 @@ NdbTransaction::NdbTransaction( Ndb* aNdb ) :
theTCConPtr(0), theTCConPtr(0),
theTransactionId(0), theTransactionId(0),
theGlobalCheckpointId(0), theGlobalCheckpointId(0),
p_latest_trans_gci(0),
theStatus(NotConnected), theStatus(NotConnected),
theCompletionStatus(NotCompleted), theCompletionStatus(NotCompleted),
theCommitStatus(NotStarted), theCommitStatus(NotStarted),
...@@ -129,6 +128,8 @@ NdbTransaction::init() ...@@ -129,6 +128,8 @@ NdbTransaction::init()
theCompletedLastOp = NULL; theCompletedLastOp = NULL;
theGlobalCheckpointId = 0; theGlobalCheckpointId = 0;
p_latest_trans_gci =
theNdb->theImpl->m_ndb_cluster_connection.get_latest_trans_gci();
theCommitStatus = Started; theCommitStatus = Started;
theCompletionStatus = NotCompleted; theCompletionStatus = NotCompleted;
m_abortOption = AbortOnError; m_abortOption = AbortOnError;
...@@ -1572,7 +1573,7 @@ NdbTransaction::receiveTC_COMMITCONF(const TcCommitConf * commitConf) ...@@ -1572,7 +1573,7 @@ NdbTransaction::receiveTC_COMMITCONF(const TcCommitConf * commitConf)
theGlobalCheckpointId = commitConf->gci; theGlobalCheckpointId = commitConf->gci;
// theGlobalCheckpointId == 0 if NoOp transaction // theGlobalCheckpointId == 0 if NoOp transaction
if (theGlobalCheckpointId) if (theGlobalCheckpointId)
g_latest_trans_gci = theGlobalCheckpointId; *p_latest_trans_gci = theGlobalCheckpointId;
return 0; return 0;
} else { } else {
#ifdef NDB_NO_DROPPED_SIGNAL #ifdef NDB_NO_DROPPED_SIGNAL
...@@ -1752,7 +1753,7 @@ from other transactions. ...@@ -1752,7 +1753,7 @@ from other transactions.
theCommitStatus = Committed; theCommitStatus = Committed;
theGlobalCheckpointId = tGCI; theGlobalCheckpointId = tGCI;
assert(tGCI); assert(tGCI);
g_latest_trans_gci = tGCI; *p_latest_trans_gci = tGCI;
} else if ((tNoComp >= tNoSent) && } else if ((tNoComp >= tNoSent) &&
(theLastExecOpInList->theCommitIndicator == 1)){ (theLastExecOpInList->theCommitIndicator == 1)){
...@@ -1930,7 +1931,7 @@ NdbTransaction::receiveTCINDXCONF(const TcIndxConf * indxConf, ...@@ -1930,7 +1931,7 @@ NdbTransaction::receiveTCINDXCONF(const TcIndxConf * indxConf,
theCommitStatus = Committed; theCommitStatus = Committed;
theGlobalCheckpointId = tGCI; theGlobalCheckpointId = tGCI;
assert(tGCI); assert(tGCI);
g_latest_trans_gci = tGCI; *p_latest_trans_gci = tGCI;
} else if ((tNoComp >= tNoSent) && } else if ((tNoComp >= tNoSent) &&
(theLastExecOpInList->theCommitIndicator == 1)){ (theLastExecOpInList->theCommitIndicator == 1)){
/**********************************************************************/ /**********************************************************************/
......
...@@ -35,8 +35,6 @@ ...@@ -35,8 +35,6 @@
#include <EventLogger.hpp> #include <EventLogger.hpp>
EventLogger g_eventLogger; EventLogger g_eventLogger;
static int g_run_connect_thread= 0;
#include <NdbMutex.h> #include <NdbMutex.h>
#ifdef VM_TRACE #ifdef VM_TRACE
NdbMutex *ndb_print_state_mutex= NULL; NdbMutex *ndb_print_state_mutex= NULL;
...@@ -87,8 +85,9 @@ const char *Ndb_cluster_connection::get_connectstring(char *buf, ...@@ -87,8 +85,9 @@ const char *Ndb_cluster_connection::get_connectstring(char *buf,
pthread_handler_t run_ndb_cluster_connection_connect_thread(void *me) pthread_handler_t run_ndb_cluster_connection_connect_thread(void *me)
{ {
g_run_connect_thread= 1; Ndb_cluster_connection_impl* connection= (Ndb_cluster_connection_impl*) me;
((Ndb_cluster_connection_impl*) me)->connect_thread(); connection->m_run_connect_thread= 1;
connection->connect_thread();
return me; return me;
} }
...@@ -258,9 +257,6 @@ unsigned Ndb_cluster_connection::get_connect_count() const ...@@ -258,9 +257,6 @@ unsigned Ndb_cluster_connection::get_connect_count() const
return m_impl.get_connect_count(); return m_impl.get_connect_count();
} }
/* /*
* Ndb_cluster_connection_impl * Ndb_cluster_connection_impl
*/ */
...@@ -269,11 +265,17 @@ Ndb_cluster_connection_impl::Ndb_cluster_connection_impl(const char * ...@@ -269,11 +265,17 @@ Ndb_cluster_connection_impl::Ndb_cluster_connection_impl(const char *
connect_string) connect_string)
: Ndb_cluster_connection(*this), : Ndb_cluster_connection(*this),
m_optimized_node_selection(1), m_optimized_node_selection(1),
m_name(0) m_name(0),
m_run_connect_thread(0),
m_event_add_drop_mutex(0),
m_latest_trans_gci(0)
{ {
DBUG_ENTER("Ndb_cluster_connection"); DBUG_ENTER("Ndb_cluster_connection");
DBUG_PRINT("enter",("Ndb_cluster_connection this=0x%x", this)); DBUG_PRINT("enter",("Ndb_cluster_connection this=0x%x", this));
if (!m_event_add_drop_mutex)
m_event_add_drop_mutex= NdbMutex_Create();
g_eventLogger.createConsoleHandler(); g_eventLogger.createConsoleHandler();
g_eventLogger.setCategory("NdbApi"); g_eventLogger.setCategory("NdbApi");
g_eventLogger.enable(Logger::LL_ON, Logger::LL_ERROR); g_eventLogger.enable(Logger::LL_ON, Logger::LL_ERROR);
...@@ -301,6 +303,33 @@ Ndb_cluster_connection_impl::Ndb_cluster_connection_impl(const char * ...@@ -301,6 +303,33 @@ Ndb_cluster_connection_impl::Ndb_cluster_connection_impl(const char *
} }
m_transporter_facade= new TransporterFacade(); m_transporter_facade= new TransporterFacade();
NdbMutex_Lock(g_ndb_connection_mutex);
if(g_ndb_connection_count++ == 0){
NdbDictionary::Column::FRAGMENT=
NdbColumnImpl::create_pseudo("NDB$FRAGMENT");
NdbDictionary::Column::FRAGMENT_FIXED_MEMORY=
NdbColumnImpl::create_pseudo("NDB$FRAGMENT_FIXED_MEMORY");
NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY=
NdbColumnImpl::create_pseudo("NDB$FRAGMENT_VARSIZED_MEMORY");
NdbDictionary::Column::ROW_COUNT=
NdbColumnImpl::create_pseudo("NDB$ROW_COUNT");
NdbDictionary::Column::COMMIT_COUNT=
NdbColumnImpl::create_pseudo("NDB$COMMIT_COUNT");
NdbDictionary::Column::ROW_SIZE=
NdbColumnImpl::create_pseudo("NDB$ROW_SIZE");
NdbDictionary::Column::RANGE_NO=
NdbColumnImpl::create_pseudo("NDB$RANGE_NO");
NdbDictionary::Column::DISK_REF=
NdbColumnImpl::create_pseudo("NDB$DISK_REF");
NdbDictionary::Column::RECORDS_IN_RANGE=
NdbColumnImpl::create_pseudo("NDB$RECORDS_IN_RANGE");
NdbDictionary::Column::ROWID=
NdbColumnImpl::create_pseudo("NDB$ROWID");
NdbDictionary::Column::ROW_GCI=
NdbColumnImpl::create_pseudo("NDB$ROW_GCI");
}
NdbMutex_Unlock(g_ndb_connection_mutex);
DBUG_VOID_RETURN; DBUG_VOID_RETURN;
} }
...@@ -314,7 +343,7 @@ Ndb_cluster_connection_impl::~Ndb_cluster_connection_impl() ...@@ -314,7 +343,7 @@ Ndb_cluster_connection_impl::~Ndb_cluster_connection_impl()
if (m_connect_thread) if (m_connect_thread)
{ {
void *status; void *status;
g_run_connect_thread= 0; m_run_connect_thread= 0;
NdbThread_WaitFor(m_connect_thread, &status); NdbThread_WaitFor(m_connect_thread, &status);
NdbThread_Destroy(&m_connect_thread); NdbThread_Destroy(&m_connect_thread);
m_connect_thread= 0; m_connect_thread= 0;
...@@ -339,6 +368,36 @@ Ndb_cluster_connection_impl::~Ndb_cluster_connection_impl() ...@@ -339,6 +368,36 @@ Ndb_cluster_connection_impl::~Ndb_cluster_connection_impl()
if (m_name) if (m_name)
free(m_name); free(m_name);
NdbMutex_Lock(g_ndb_connection_mutex);
if(--g_ndb_connection_count == 0){
delete NdbDictionary::Column::FRAGMENT;
delete NdbDictionary::Column::FRAGMENT_FIXED_MEMORY;
delete NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY;
delete NdbDictionary::Column::ROW_COUNT;
delete NdbDictionary::Column::COMMIT_COUNT;
delete NdbDictionary::Column::ROW_SIZE;
delete NdbDictionary::Column::RANGE_NO;
delete NdbDictionary::Column::DISK_REF;
delete NdbDictionary::Column::RECORDS_IN_RANGE;
delete NdbDictionary::Column::ROWID;
delete NdbDictionary::Column::ROW_GCI;
NdbDictionary::Column::FRAGMENT= 0;
NdbDictionary::Column::FRAGMENT_FIXED_MEMORY= 0;
NdbDictionary::Column::FRAGMENT_VARSIZED_MEMORY= 0;
NdbDictionary::Column::ROW_COUNT= 0;
NdbDictionary::Column::COMMIT_COUNT= 0;
NdbDictionary::Column::ROW_SIZE= 0;
NdbDictionary::Column::RANGE_NO= 0;
NdbDictionary::Column::DISK_REF= 0;
NdbDictionary::Column::RECORDS_IN_RANGE= 0;
NdbDictionary::Column::ROWID= 0;
NdbDictionary::Column::ROW_GCI= 0;
}
NdbMutex_Unlock(g_ndb_connection_mutex);
if (m_event_add_drop_mutex)
NdbMutex_Destroy(m_event_add_drop_mutex);
DBUG_VOID_RETURN; DBUG_VOID_RETURN;
} }
...@@ -576,17 +635,23 @@ void Ndb_cluster_connection_impl::connect_thread() ...@@ -576,17 +635,23 @@ void Ndb_cluster_connection_impl::connect_thread()
if (r == -1) { if (r == -1) {
printf("Ndb_cluster_connection::connect_thread error\n"); printf("Ndb_cluster_connection::connect_thread error\n");
DBUG_ASSERT(false); DBUG_ASSERT(false);
g_run_connect_thread= 0; m_run_connect_thread= 0;
} else { } else {
// Wait before making a new connect attempt // Wait before making a new connect attempt
NdbSleep_SecSleep(1); NdbSleep_SecSleep(1);
} }
} while (g_run_connect_thread); } while (m_run_connect_thread);
if (m_connect_callback) if (m_connect_callback)
(*m_connect_callback)(); (*m_connect_callback)();
DBUG_VOID_RETURN; DBUG_VOID_RETURN;
} }
Uint64 *
Ndb_cluster_connection::get_latest_trans_gci()
{
return m_impl.get_latest_trans_gci();
}
void void
Ndb_cluster_connection::init_get_next_node(Ndb_cluster_connection_node_iter &iter) Ndb_cluster_connection::init_get_next_node(Ndb_cluster_connection_node_iter &iter)
{ {
......
...@@ -20,6 +20,10 @@ ...@@ -20,6 +20,10 @@
#include <ndb_cluster_connection.hpp> #include <ndb_cluster_connection.hpp>
#include <Vector.hpp> #include <Vector.hpp>
#include <NdbMutex.h>
extern NdbMutex *g_ndb_connection_mutex;
static int g_ndb_connection_count = 0;
class TransporterFacade; class TransporterFacade;
class ConfigRetriever; class ConfigRetriever;
...@@ -41,6 +45,9 @@ class Ndb_cluster_connection_impl : public Ndb_cluster_connection ...@@ -41,6 +45,9 @@ class Ndb_cluster_connection_impl : public Ndb_cluster_connection
Uint32 get_next_node(Ndb_cluster_connection_node_iter &iter); Uint32 get_next_node(Ndb_cluster_connection_node_iter &iter);
inline unsigned get_connect_count() const; inline unsigned get_connect_count() const;
public:
inline Uint64 *get_latest_trans_gci() { return &m_latest_trans_gci; }
private: private:
friend class Ndb; friend class Ndb;
friend class NdbImpl; friend class NdbImpl;
...@@ -72,6 +79,9 @@ private: ...@@ -72,6 +79,9 @@ private:
int m_optimized_node_selection; int m_optimized_node_selection;
char *m_name; char *m_name;
int m_run_connect_thread;
NdbMutex *m_event_add_drop_mutex;
Uint64 m_latest_trans_gci;
}; };
#endif #endif
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment