mirror/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-01-18 13:02:28 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge mleich@bk-internal.mysql.com:/home/bk/mysql-4.1

Browse source 
into three.local.lan:/home/matthias/Arbeit/mysql-4.1/src
This commit is contained in:
unknown 2004-12-17 11:02:44 +01:00
commit 9aa76331a0
19 changed files with 323 additions and 689 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
configure.in
View file

@ -16,7 +16,7 @@ SHARED_LIB_VERSION=14:0:0
# ndb version
NDB_VERSION_MAJOR=4
NDB_VERSION_MINOR=1
NDB_VERSION_BUILD=8
NDB_VERSION_BUILD=9
NDB_VERSION_STATUS=""
# Set all version vars based on $VERSION. How do we do this more elegant ?

70
mysql-test/r/rpl_insert_ignore.result Normal file
View file

@ -0,0 +1,70 @@
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;
CREATE TABLE t1 (
a int unsigned not null auto_increment primary key,
b int unsigned,
unique (b)
) ENGINE=innodb;
CREATE TABLE t2 (
a int unsigned, # to force INSERT SELECT to have a certain order
b int unsigned
) ENGINE=innodb;
INSERT INTO t1 VALUES (NULL, 1);
INSERT INTO t1 VALUES (NULL, 2);
INSERT INTO t1 VALUES (NULL, 3);
INSERT INTO t1 VALUES (NULL, 4);
INSERT INTO t2 VALUES (1, 1);
INSERT INTO t2 VALUES (2, 2);
INSERT INTO t2 VALUES (3, 5);
INSERT INTO t2 VALUES (4, 3);
INSERT INTO t2 VALUES (5, 4);
INSERT INTO t2 VALUES (6, 6);
INSERT IGNORE INTO t1 SELECT NULL, t2.b FROM t2 ORDER BY t2.a;
SELECT * FROM t1 ORDER BY a;
a b
1 1
2 2
3 3
4 4
5 5
6 6
SELECT * FROM t1 ORDER BY a;
a b
1 1
2 2
3 3
4 4
5 5
6 6
drop table t1;
CREATE TABLE t1 (
a int unsigned not null auto_increment primary key,
b int unsigned,
unique (b)
) ENGINE=myisam;
INSERT INTO t1 VALUES (1, 1);
INSERT INTO t1 VALUES (2, 2);
INSERT INTO t1 VALUES (3, 3);
INSERT INTO t1 VALUES (4, 4);
INSERT IGNORE INTO t1 SELECT NULL, t2.b FROM t2 ORDER BY t2.a;
SELECT * FROM t1 ORDER BY a;
a b
1 1
2 2
3 3
4 4
5 5
6 6
SELECT * FROM t1 ORDER BY a;
a b
1 1
2 2
3 3
4 4
5 5
6 6
drop table t1, t2;

1
mysql-test/t/rpl_insert_ignore-slave.opt Normal file
View file

@ -0,0 +1 @@
--innodb

71
mysql-test/t/rpl_insert_ignore.test Normal file
View file

@ -0,0 +1,71 @@
# Testcase for BUG#6287 "Slave skips auto_increment values in Replication with InnoDB"
# The bug was that if on master, INSERT IGNORE ignored some
# rows, and the table was InnoDB with auto_inc column, then on slave
# some rows received an auto_inc bigger than on master.
# Slave needs to be started with --innodb to store table in InnoDB.
# Same test for MyISAM (which had no bug).
-- source include/have_innodb.inc
-- source include/master-slave.inc
CREATE TABLE t1 (
a int unsigned not null auto_increment primary key,
b int unsigned,
unique (b)
) ENGINE=innodb;
CREATE TABLE t2 (
a int unsigned, # to force INSERT SELECT to have a certain order
b int unsigned
) ENGINE=innodb;
INSERT INTO t1 VALUES (NULL, 1);
INSERT INTO t1 VALUES (NULL, 2);
INSERT INTO t1 VALUES (NULL, 3);
INSERT INTO t1 VALUES (NULL, 4);
# An alternation of values which will conflict in t1 and will not.
INSERT INTO t2 VALUES (1, 1);
INSERT INTO t2 VALUES (2, 2);
INSERT INTO t2 VALUES (3, 5);
INSERT INTO t2 VALUES (4, 3);
INSERT INTO t2 VALUES (5, 4);
INSERT INTO t2 VALUES (6, 6);
INSERT IGNORE INTO t1 SELECT NULL, t2.b FROM t2 ORDER BY t2.a;
# Compare results
SELECT * FROM t1 ORDER BY a;
sync_slave_with_master;
SELECT * FROM t1 ORDER BY a;
# Now do the same for MyISAM
connection master;
drop table t1;
CREATE TABLE t1 (
a int unsigned not null auto_increment primary key,
b int unsigned,
unique (b)
) ENGINE=myisam;
INSERT INTO t1 VALUES (1, 1);
INSERT INTO t1 VALUES (2, 2);
INSERT INTO t1 VALUES (3, 3);
INSERT INTO t1 VALUES (4, 4);
INSERT IGNORE INTO t1 SELECT NULL, t2.b FROM t2 ORDER BY t2.a;
SELECT * FROM t1 ORDER BY a;
sync_slave_with_master;
SELECT * FROM t1 ORDER BY a;
connection master;
drop table t1, t2;
sync_slave_with_master;

11
ndb/include/ndbapi/Ndb.hpp
View file

@ -1586,7 +1586,6 @@ private:
/******************************************************************************
* These are the private variables in this class.
*****************************************************************************/
NdbObjectIdMap* theNdbObjectIdMap;
Ndb_cluster_connection *m_ndb_cluster_connection;
NdbConnection** thePreparedTransactionsArray;
@ -1637,10 +1636,6 @@ private:
Uint32 theMyRef; // My block reference
Uint32 theNode; // The node number of our node
Uint32 theNoOfDBnodes; // The number of DB nodes
Uint32 * theDBnodes; // The node number of the DB nodes
Uint8 *the_release_ind;// 1 indicates to release all connections to node
Uint64 the_last_check_time;
Uint64 theFirstTransId;
@ -1663,10 +1658,6 @@ private:
InitConfigError
} theInitState;
// Ensure good distribution of connects
Uint32 theCurrentConnectIndex;
Uint32 theCurrentConnectCounter;
/**
* Computes fragement id for primary key
*
@ -1692,7 +1683,7 @@ private:
Uint32 noOfFragments;
Uint32 * fragment2PrimaryNodeMap;
void init(Uint32 noOfNodes, Uint32 nodeIds[]);
void init(Uint32 noOfNodes, Uint8 nodeIds[]);
void release();
} startTransactionNodeSelectionData;

42
ndb/src/common/util/version.c
View file

@ -17,6 +17,8 @@
#include <ndb_global.h>
#include <ndb_version.h>
#include <version.h>
#include <NdbEnv.h>
#include <NdbOut.hpp>
Uint32 getMajor(Uint32 version) {
return (version >> 16) & 0xFF;
@ -67,8 +69,27 @@ struct NdbUpGradeCompatible {
/*#define TEST_VERSION*/
#define HAVE_NDB_SETVERSION
#ifdef HAVE_NDB_SETVERSION
Uint32 ndbOwnVersionTesting = 0;
void
ndbSetOwnVersion() {
char buf[256];
if (NdbEnv_GetEnv("NDB_SETVERSION", buf, sizeof(buf))) {
Uint32 _v1,_v2,_v3;
if (sscanf(buf, "%u.%u.%u", &_v1, &_v2, &_v3) == 3) {
ndbOwnVersionTesting = MAKE_VERSION(_v1,_v2,_v3);
ndbout_c("Testing: Version set to 0x%x", ndbOwnVersionTesting);
}
}
}
#else
void ndbSetOwnVersion() {}
#endif
#ifndef TEST_VERSION
struct NdbUpGradeCompatible ndbCompatibleTable_full[] = {
{ MAKE_VERSION(4,1,9), MAKE_VERSION(4,1,8), UG_Exact },
{ MAKE_VERSION(3,5,2), MAKE_VERSION(3,5,1), UG_Exact },
{ 0, 0, UG_Null }
};
@ -78,8 +99,6 @@ struct NdbUpGradeCompatible ndbCompatibleTable_upgrade[] = {
{ 0, 0, UG_Null }
};
void ndbSetOwnVersion() {}
#else /* testing purposes */
struct NdbUpGradeCompatible ndbCompatibleTable_full[] = {
@ -100,19 +119,6 @@ struct NdbUpGradeCompatible ndbCompatibleTable_upgrade[] = {
};
Uint32 ndbOwnVersionTesting = 0;
void
ndbSetOwnVersion() {
char buf[256];
if (NdbEnv_GetEnv("NDB_SETVERSION", buf, sizeof(buf))) {
Uint32 _v1,_v2,_v3;
if (sscanf(buf, "%u.%u.%u", &_v1, &_v2, &_v3) == 3) {
ndbOwnVersionTesting = MAKE_VERSION(_v1,_v2,_v3);
ndbout_c("Testing: Version set to 0x%x", ndbOwnVersionTesting);
}
}
}
#endif
void ndbPrintVersion()
@ -126,13 +132,13 @@ void ndbPrintVersion()
Uint32
ndbGetOwnVersion()
{
#ifndef TEST_VERSION
return NDB_VERSION_D;
#else /* testing purposes */
#ifdef HAVE_NDB_SETVERSION
if (ndbOwnVersionTesting == 0)
return NDB_VERSION_D;
else
return ndbOwnVersionTesting;
#else
return NDB_VERSION_D;
#endif
}

537
ndb/src/mgmsrv/CommandInterpreter.cpp
View file

@ -30,64 +30,10 @@
#include <version.h>
static const char* helpTexts[] = {
"HELP Print help text",
"HELP SHOW Help for the SHOW command",
#ifdef VM_TRACE // DEBUG ONLY
"HELP DEBUG Help for debug compiled version",
#endif
"SHOW Print information about cluster",
"SHOW CONFIG Print configuration",
"SHOW PARAMETERS Print configuration parameters",
"START BACKUP Start backup\n"
"ABORT BACKUP <backup id> Aborts backup\n"
"CLUSTERLOG ON Enable Cluster logging",
"CLUSTERLOG OFF Disable Cluster logging",
"CLUSTERLOG FILTER <severity> Toggle severity filter on/off",
"CLUSTERLOG INFO Print cluster log information",
"{<id>|ALL} START Start DB node (started with -n)",
"{<id>|ALL} RESTART [-n] [-i] Restart DB node",
"{<id>|ALL} STOP Stop DB node",
"{<id>|ALL} STATUS Print status",
"{<id>|ALL} CLUSTERLOG {<category>=<level>}+ Set log level for cluster log",
"QUIT Quit management server",
};
static const unsigned noOfHelpTexts = sizeof(helpTexts)/sizeof(const char*);
static const char* helpTextShow =
"SHOW prints NDB Cluster information\n\n"
"SHOW Print information about cluster\n"
"SHOW CONFIG Print configuration (in initial config file format)\n"
"SHOW PARAMETERS Print information about configuration parameters\n\n"
;
#ifdef VM_TRACE // DEBUG ONLY
static const char* helpTextDebug =
"SHOW PROPERTIES Print config properties object\n"
"{<id>|ALL} LOGLEVEL {<category>=<level>}+ Set log level\n"
"{<id>|ALL} ERROR <errorNo> Inject error into NDB node\n"
"{<id>|ALL} TRACE <traceNo> Set trace number\n"
"{<id>|ALL} LOG [BLOCK = {ALL|<block>+}] Set logging on in & out signals\n"
"{<id>|ALL} LOGIN [BLOCK = {ALL|<block>+}] Set logging on in signals\n"
"{<id>|ALL} LOGOUT [BLOCK = {ALL|<block>+}] Set logging on out signals\n"
"{<id>|ALL} LOGOFF [BLOCK = {ALL|<block>+}] Unset signal logging\n"
"{<id>|ALL} TESTON Start signal logging\n"
"{<id>|ALL} TESTOFF Stop signal logging\n"
"{<id>|ALL} SET <configParamName> <value> Update configuration variable\n"
"{<id>|ALL} DUMP <arg> Dump system state to cluster.log\n"
"{<id>|ALL} GETSTAT Print statistics\n"
"\n"
;
#endif
//******************************************************************************
//******************************************************************************
CommandInterpreter::CommandInterpreter(MgmtSrvr& mgmtSrvr) :
_mgmtSrvr(mgmtSrvr) {
// _mgmtSrvr.setCallback(CmdBackupCallback);
}
@ -145,48 +91,7 @@ int CommandInterpreter::readAndExecute() {
char* firstToken = strtok(line, " ");
char* allAfterFirstToken = strtok(NULL, "\0");
if (strcmp(firstToken, "HELP") == 0) {
executeHelp(allAfterFirstToken);
return true;
}
else if (strcmp(firstToken, "?") == 0) {
executeHelp(allAfterFirstToken);
return true;
}
else if (strcmp(firstToken, "SHOW") == 0) {
executeShow(allAfterFirstToken);
return true;
}
else if(strcmp(firstToken, "START") == 0 &&
allAfterFirstToken != 0 &&
strncmp(allAfterFirstToken, "BACKUP", sizeof("BACKUP") - 1) == 0){
executeStartBackup(allAfterFirstToken);
return true;
}
else if(strcmp(firstToken, "ABORT") == 0 &&
allAfterFirstToken != 0 &&
strncmp(allAfterFirstToken, "BACKUP", sizeof("BACKUP") - 1) == 0){
executeAbortBackup(allAfterFirstToken);
return true;
}
else if(strcmp(firstToken, "ENTER") == 0 &&
allAfterFirstToken != 0 &&
strncmp(allAfterFirstToken, "SINGLE USER MODE ",
sizeof("SINGLE USER MODE") - 1) == 0){
executeEnterSingleUser(allAfterFirstToken);
return true;
}
else if(strcmp(firstToken, "EXIT") == 0 &&
allAfterFirstToken != 0 &&
strncmp(allAfterFirstToken, "SINGLE USER MODE ",
sizeof("SINGLE USER MODE") - 1) == 0){
executeExitSingleUser(allAfterFirstToken);
return true;
}
else if (strcmp(firstToken, "ALL") == 0) {
if (strcmp(firstToken, "ALL") == 0) {
analyseAfterFirstToken(-1, allAfterFirstToken);
}
else if(strcmp(firstToken, "QUIT") == 0 ||
@ -218,8 +123,6 @@ static const CommandInterpreter::CommandFunctionPair commands[] = {
{ "START", &CommandInterpreter::executeStart }
,{ "RESTART", &CommandInterpreter::executeRestart }
,{ "STOP", &CommandInterpreter::executeStop }
,{ "STATUS", &CommandInterpreter::executeStatus }
,{ "LOGLEVEL", &CommandInterpreter::executeLogLevel }
#ifdef ERROR_INSERT
,{ "ERROR", &CommandInterpreter::executeError }
#endif
@ -230,9 +133,7 @@ static const CommandInterpreter::CommandFunctionPair commands[] = {
,{ "LOGOFF", &CommandInterpreter::executeLogOff }
,{ "TESTON", &CommandInterpreter::executeTestOn }
,{ "TESTOFF", &CommandInterpreter::executeTestOff }
,{ "CLUSTERLOG", &CommandInterpreter::executeEventReporting }
,{ "DUMP", &CommandInterpreter::executeDumpState }
,{ "JONAS", &CommandInterpreter::jonas }
};
@ -370,104 +271,9 @@ bool CommandInterpreter::parseBlockSpecification(const char* allAfterLog,
return true;
}
//******************************************************************************
//******************************************************************************
void CommandInterpreter::executeHelp(char* parameters) {
(void)parameters; // Don't want compiler warning
if (emptyString(parameters)) {
unsigned i;
for (i = 0; i<noOfHelpTexts; i++) {
ndbout << helpTexts[i] << endl;
}
ndbout << endl
<< "<severity> = "
<< "ALERT | CRITICAL | ERROR | WARNING | INFO | DEBUG"
<< endl;
ndbout << "<category> = ";
for(i = 0; i<CFG_MIN_LOGLEVEL; i++){
ndbout << ndb_mgm_get_event_category_string((ndb_mgm_event_category)i);
if (i < CFG_MIN_LOGLEVEL - 1) {
ndbout << " | ";
}
}
ndbout << endl;
ndbout << "<level> = " << "0 - 15"
<< endl;
ndbout << endl;
} else if (strcmp(parameters, "SHOW") == 0) {
ndbout << helpTextShow;
#ifdef VM_TRACE // DEBUG ONLY
} else if (strcmp(parameters, "DEBUG") == 0) {
ndbout << helpTextDebug;
#endif
} else {
ndbout << "Invalid argument." << endl;
}
}
//*****************************************************************************
//*****************************************************************************
void CommandInterpreter::executeShow(char* parameters) {
if (emptyString(parameters)) {
ndbout << "Cluster Configuration" << endl
<< "---------------------" << endl;
NodeId nodeId = 0;
ndbout << _mgmtSrvr.getNodeCount(NDB_MGM_NODE_TYPE_NDB)
<< " NDB Node(s) with"
<< endl;
while (_mgmtSrvr.getNextNodeId(&nodeId, NDB_MGM_NODE_TYPE_NDB)){
ndbout << " Node Id = " << nodeId << endl;
}
ndbout << endl;
nodeId = 0;
ndbout << _mgmtSrvr.getNodeCount(NDB_MGM_NODE_TYPE_API)
<< " API Node(s) with"
<< endl;
while (_mgmtSrvr.getNextNodeId(&nodeId, NDB_MGM_NODE_TYPE_API)){
ndbout << " Node Id = " << nodeId << endl;
}
ndbout << endl;
nodeId = 0;
ndbout << _mgmtSrvr.getNodeCount(NDB_MGM_NODE_TYPE_MGM)
<< " MGM Node(s) with"
<< endl;
while (_mgmtSrvr.getNextNodeId(&nodeId, NDB_MGM_NODE_TYPE_MGM)){
ndbout << " Node Id = " << nodeId << endl;
}
ndbout << endl;
ndbout << helpTextShow;
return;
} else if (strcmp(parameters, "PROPERTIES") == 0 ||
strcmp(parameters, "PROP") == 0) {
ndbout << "_mgmtSrvr.getConfig()->print();" << endl; /* XXX */
} else if (strcmp(parameters, "CONFIGURATION") == 0 ||
strcmp(parameters, "CONFIG") == 0){
ndbout << "_mgmtSrvr.getConfigFile()->print();" << endl; /* XXX */
_mgmtSrvr.getConfig()->printConfigFile();
} else if (strcmp(parameters, "PARAMETERS") == 0 ||
strcmp(parameters, "PARAMS") == 0 ||
strcmp(parameters, "PARAM") == 0) {
ndbout << "_mgmtSrvr.getConfigInfo()->print();" << endl; /* XXX */
} else {
ndbout << "Invalid argument." << endl;
}
}
void
stopCallback(int nodeId, void * anyData, int errCode){
if(errCode == 0){
@ -483,59 +289,6 @@ stopCallback(int nodeId, void * anyData, int errCode){
}
}
void
versionCallback(int nodeId, int version, void * anyData, int errCode){
if(errCode == 0){
MgmtSrvr * mgm = (MgmtSrvr *)anyData;
switch(mgm->getNodeType(nodeId)){
case NDB_MGM_NODE_TYPE_MGM:
{
ndbout << "MGMT node:\t" << nodeId << " ";
ndbout_c(" (Version %d.%d.%d)",
getMajor(version) ,
getMinor(version),
getBuild(version));
}
break;
case NDB_MGM_NODE_TYPE_NDB:
{
ndbout << "DB node:\t" << nodeId << " ";
if(version == 0)
ndbout << "(no version information available)" << endl;
else {
ndbout_c(" (Version %d.%d.%d)",
getMajor(version) ,
getMinor(version),
getBuild(version));
}
}
break;
case NDB_MGM_NODE_TYPE_API:
{
ndbout << "API node:\t" << nodeId << " ";
if(version == 0)
ndbout << "(no version information available)" << endl;
else {
ndbout_c(" (Version %d.%d.%d)",
getMajor(version) ,
getMinor(version),
getBuild(version));
}
}
break;
case NDB_MGM_NODE_TYPE_UNKNOWN:
case NDB_MGM_NODE_TYPE_REP:
abort();
};
} else {
MgmtSrvr * mgm = (MgmtSrvr *)anyData;
char err_str[1024];
ndbout << mgm->getErrorText(errCode,err_str,sizeof(err_str)) << endl;
}
}
//*****************************************************************************
//*****************************************************************************
void CommandInterpreter::executeStop(int processId,
@ -643,124 +396,6 @@ CommandInterpreter::executeDumpState(int processId, const char* parameters,
}
}
void CommandInterpreter::executeStatus(int processId,
const char* parameters, bool all) {
(void)all; // Don't want compiler warning
if (! emptyString(parameters)) {
ndbout << "No parameters expected to this command." << endl;
return;
}
ndb_mgm_node_status status;
Uint32 startPhase, version, dynamicId, nodeGroup, connectCount;
bool system;
int result = _mgmtSrvr.status(processId,
&status, &version, &startPhase, &system,
&dynamicId, &nodeGroup, &connectCount);
if(result != 0){
ndbout << get_error_text(result) << endl;
return;
}
ndbout << "Node " << processId << ": ";
switch(status){
case NDB_MGM_NODE_STATUS_NO_CONTACT:
ndbout << "No contact" << endl;
break;
case NDB_MGM_NODE_STATUS_NOT_STARTED:
ndbout << "Not started" ;
break;
case NDB_MGM_NODE_STATUS_STARTING:
ndbout << "Starting (Start phase " << startPhase << ")" ;
break;
case NDB_MGM_NODE_STATUS_STARTED:
ndbout << "Started" ;
break;
case NDB_MGM_NODE_STATUS_SHUTTING_DOWN:
ndbout << "Shutting down " << (system == false ? "node" : "system")
<< " (Phase " << startPhase << ")"
;
break;
case NDB_MGM_NODE_STATUS_RESTARTING:
ndbout << "Restarting" ;
break;
case NDB_MGM_NODE_STATUS_SINGLEUSER:
ndbout << "Single user mode" ;
break;
default:
ndbout << "Unknown state" ;
break;
}
if(status != NDB_MGM_NODE_STATUS_NO_CONTACT){
ndbout_c(" (Version %d.%d.%d)",
getMajor(version) ,
getMinor(version),
getBuild(version));
// NOTE It's possible to print dynamicId and nodeGroup here ...
// ndbout << ", " <<dynamicId<<", "<<nodeGroup<<endl;
}
}
//*****************************************************************************
//*****************************************************************************
void CommandInterpreter::executeLogLevel(int processId,
const char* parameters, bool all) {
#if 0
(void)all; // Don't want compiler warning
SetLogLevelOrd logLevel; logLevel.clear();
if (emptyString(parameters) || (strcmp(parameters, "ALL") == 0)) {
for(Uint32 i = 0; i<EventLoggerBase::noOfEventCategoryNames; i++)
logLevel.setLogLevel(EventLoggerBase::eventCategoryNames[i].category, 7);
} else {
char * tmpString = strdup(parameters);
char * tmpPtr = 0;
char * item = strtok_r(tmpString, ", ", &tmpPtr);
while(item != NULL){
char categoryTxt[255];
int level;
const int m = sscanf(item, "%[^=]=%d", categoryTxt, &level);
if(m != 2){
free(tmpString);
ndbout << "Invalid loglevel specification category=level" << endl;
return;
}
LogLevel::EventCategory cat;
if(!EventLoggerBase::matchEventCategory(categoryTxt,
&cat)){
ndbout << "Invalid loglevel specification, unknown category: "
<< categoryTxt << endl;
free(tmpString);
return ;
}
if(level < 0 || level > 15){
ndbout << "Invalid loglevel specification row, level 0-15" << endl;
free(tmpString);
return ;
}
logLevel.setLogLevel(cat, level);
item = strtok_r(NULL, ", ", &tmpPtr);
}
free(tmpString);
}
int result = _mgmtSrvr.setNodeLogLevel(processId, logLevel);
if (result != 0) {
ndbout << get_error_text(result) << endl;
}
#endif
}
//*****************************************************************************
//*****************************************************************************
void CommandInterpreter::executeError(int processId,
@ -956,173 +591,3 @@ void CommandInterpreter::executeTestOff(int processId,
}
}
//*****************************************************************************
//*****************************************************************************
void CommandInterpreter::executeEventReporting(int processId,
const char* parameters,
bool all) {
#if 0
(void)all; // Don't want compiler warning
SetLogLevelOrd logLevel; logLevel.clear();
if (emptyString(parameters) || (strcmp(parameters, "ALL") == 0)) {
for(Uint32 i = 0; i<EventLoggerBase::noOfEventCategoryNames; i++)
logLevel.setLogLevel(EventLoggerBase::eventCategoryNames[i].category, 7);
} else {
char * tmpString = strdup(parameters);
char * tmpPtr = 0;
char * item = strtok_r(tmpString, ", ", &tmpPtr);
while(item != NULL){
char categoryTxt[255];
int level;
const int m = sscanf(item, "%[^=]=%d", categoryTxt, &level);
if(m != 2){
free(tmpString);
ndbout << "Invalid loglevel specification category=level" << endl;
return;
}
LogLevel::EventCategory cat;
if(!EventLoggerBase::matchEventCategory(categoryTxt,
&cat)){
ndbout << "Invalid loglevel specification, unknown category: "
<< categoryTxt << endl;
free(tmpString);
return ;
}
if(level < 0 || level > 15){
ndbout << "Invalid loglevel specification row, level 0-15" << endl;
free(tmpString);
return ;
}
logLevel.setLogLevel(cat, level);
item = strtok_r(NULL, ", ", &tmpPtr);
}
free(tmpString);
}
ndbout_c("processId %d", processId);
int result = _mgmtSrvr.setEventReportingLevel(processId, logLevel);
if (result != 0) {
ndbout << get_error_text(result) << endl;
}
#endif
}
void
CommandInterpreter::executeStartBackup(char* parameters) {
Uint32 backupId;
int result = _mgmtSrvr.startBackup(backupId);
if (result != 0) {
ndbout << get_error_text(result) << endl;
} else {
// ndbout << "Start of backup ordered" << endl;
}
}
void
CommandInterpreter::executeAbortBackup(char* parameters) {
strtok(parameters, " ");
char* id = strtok(NULL, "\0");
int bid = -1;
if(id == 0 || sscanf(id, "%d", &bid) != 1){
ndbout << "Invalid arguments: expected <BackupId>" << endl;
return;
}
int result = _mgmtSrvr.abortBackup(bid);
if (result != 0) {
ndbout << get_error_text(result) << endl;
} else {
ndbout << "Abort of backup " << bid << " ordered" << endl;
}
}
void
CommandInterpreter::executeEnterSingleUser(char* parameters) {
strtok(parameters, " ");
char* id = strtok(NULL, " ");
id = strtok(NULL, " ");
id = strtok(NULL, "\0");
int nodeId = -1;
if(id == 0 || sscanf(id, "%d", &nodeId) != 1){
ndbout << "Invalid arguments: expected <NodeId>" << endl;
return;
}
int result = _mgmtSrvr.enterSingleUser(0, nodeId,0,0);
if (result != 0) {
ndbout << get_error_text(result) << endl;
} else {
ndbout << "Entering single user mode, granting access for node "
<< nodeId << " OK." << endl;
}
}
void CommandInterpreter::executeExitSingleUser(char* parameters) {
_mgmtSrvr.exitSingleUser(0,0,0,0);
}
#include <NdbApiSignal.hpp>
void
CommandInterpreter::jonas(int processId, const char* parameters, bool all) {
MgmtSrvr::Area51 tmp = _mgmtSrvr.getStuff();
NdbApiSignal signal(0);
Uint32 * theData = signal.getDataPtrSend();
Uint32 data[25];
Uint32 sec0[70];
Uint32 sec1[123];
data[0] = 12;
data[1] = 13;
unsigned i;
for(i = 0; i<70; i++)
sec0[i] = i;
for(i = 0; i<123; i++)
sec1[i] = 70+i;
signal.set(0, CMVMI, GSN_TESTSIG, 3);
signal.m_noOfSections = 2;
signal.m_fragmentInfo = 1;
LinearSectionPtr ptr[3];
theData[0] = 3;
theData[1] = 0;
theData[2] = 7; // FragmentId
ptr[0].sz = 2;
ptr[0].p = &data[0];
ptr[1].sz = 60;
ptr[1].p = &sec0[0];
tmp.theFacade->lock_mutex();
tmp.theRegistry->prepareSend(&signal, 1, theData, processId, ptr);
tmp.theFacade->unlock_mutex();
signal.set(0, CMVMI, GSN_TESTSIG, 3);
signal.m_noOfSections = 2;
signal.m_fragmentInfo = 3;
theData[0] = 0;
theData[1] = 1;
theData[2] = 7; // FragmentId
ptr[0].sz = 10;
ptr[0].p = &sec0[60];
ptr[1].sz = 123;
ptr[1].p = &sec1[0];
tmp.theFacade->lock_mutex();
tmp.theRegistry->prepareSend(&signal, 1, theData, processId, ptr);
tmp.theFacade->unlock_mutex();
}

8
ndb/src/mgmsrv/CommandInterpreter.hpp
View file

@ -130,7 +130,6 @@ public:
void executeStop(int processId, const char* parameters, bool all);
void executeStart(int processId, const char* parameters, bool all);
void executeRestart(int processId, const char* parameters, bool all);
void executeLogLevel(int processId, const char* parameters, bool all);
void executeError(int processId, const char* parameters, bool all);
void executeTrace(int processId, const char* parameters, bool all);
void executeLog(int processId, const char* parameters, bool all);
@ -140,14 +139,7 @@ public:
void executeTestOn(int processId, const char* parameters, bool all);
void executeTestOff(int processId, const char* parameters, bool all);
void executeStatus(int processId, const char* parameters, bool all);
void executeEnterSingleUser(char* parameters);
void executeExitSingleUser(char* parameters);
void executeEventReporting(int processId, const char* parameters, bool all);
void executeDumpState(int processId, const char* parameters, bool all);
void executeStartBackup(char * pars);
void executeAbortBackup(char * pars);
void jonas(int processId, const char* parameters, bool all);
/**
* A execute function definition

37
ndb/src/ndbapi/Ndb.cpp
View file

@ -65,37 +65,25 @@ NdbConnection* Ndb::doConnect(Uint32 tConNode)
// We will connect to any node. Make sure that we have connections to all
// nodes.
//****************************************************************************
Uint32 tNoOfDbNodes = theNoOfDBnodes;
i = theCurrentConnectIndex;
Uint32 tNoOfDbNodes= theImpl->theNoOfDBnodes;
Uint32 &theCurrentConnectIndex= theImpl->theCurrentConnectIndex;
UintR Tcount = 0;
do {
if (i >= tNoOfDbNodes) {
i = 0;
theCurrentConnectIndex++;
if (theCurrentConnectIndex >= tNoOfDbNodes) {
theCurrentConnectIndex = 0;
}//if
Tcount++;
tNode = theDBnodes[i];
tNode = theImpl->theDBnodes[theCurrentConnectIndex];
TretCode = NDB_connect(tNode);
if ((TretCode == 1) || (TretCode == 2)) {
//****************************************************************************
// We have connections now to the desired node. Return
//****************************************************************************
if (theCurrentConnectIndex == i) {
theCurrentConnectCounter++;
if (theCurrentConnectCounter == 8) {
theCurrentConnectCounter = 1;
theCurrentConnectIndex++;
}//if
} else {
// Set to 2 because we have already connected to a node
// when we get here.
theCurrentConnectCounter = 2;
theCurrentConnectIndex = i;
}//if
return getConnectedNdbConnection(tNode);
} else if (TretCode != 0) {
tAnyAlive = 1;
}//if
i++;
} while (Tcount < tNoOfDbNodes);
//****************************************************************************
// We were unable to find a free connection. If no node alive we will report
@ -211,8 +199,9 @@ Ndb::doDisconnect()
NdbConnection* tNdbCon;
CHECK_STATUS_MACRO_VOID;
DBUG_PRINT("info", ("theNoOfDBnodes=%d", theNoOfDBnodes));
Uint32 tNoOfDbNodes = theNoOfDBnodes;
Uint32 tNoOfDbNodes = theImpl->theNoOfDBnodes;
Uint8 *theDBnodes= theImpl->theDBnodes;
DBUG_PRINT("info", ("theNoOfDBnodes=%d", tNoOfDbNodes));
UintR i;
for (i = 0; i < tNoOfDbNodes; i++) {
Uint32 tNode = theDBnodes[i];
@ -259,8 +248,8 @@ Ndb::waitUntilReady(int timeout)
unsigned int foundAliveNode = 0;
TransporterFacade *tp = TransporterFacade::instance();
tp->lock_mutex();
for (unsigned int i = 0; i < theNoOfDBnodes; i++) {
const NodeId nodeId = theDBnodes[i];
for (unsigned int i = 0; i < theImpl->theNoOfDBnodes; i++) {
const NodeId nodeId = theImpl->theDBnodes[i];
//************************************************
// If any node is answering, ndb is answering
//************************************************
@ -270,7 +259,7 @@ Ndb::waitUntilReady(int timeout)
}//for
tp->unlock_mutex();
if (foundAliveNode == theNoOfDBnodes) {
if (foundAliveNode == theImpl->theNoOfDBnodes) {
DBUG_RETURN(0);
}//if
if (foundAliveNode > 0) {
@ -1077,7 +1066,7 @@ Ndb::guessPrimaryNode(Uint32 fragmentId){
void
Ndb::StartTransactionNodeSelectionData::init(Uint32 noOfNodes,
Uint32 nodeIds[]) {
Uint8 nodeIds[]) {
kValue = 6;
noOfFragments = 2 * noOfNodes;

4
ndb/src/ndbapi/NdbConnection.cpp
View file

@ -83,7 +83,7 @@ NdbConnection::NdbConnection( Ndb* aNdb ) :
{
theListState = NotInList;
theError.code = 0;
theId = theNdb->theNdbObjectIdMap->map(this);
theId = theNdb->theImpl->theNdbObjectIdMap.map(this);
#define CHECK_SZ(mask, sz) assert((sizeof(mask)/sizeof(mask[0])) == sz)
@ -99,7 +99,7 @@ Remark: Deletes the connection object.
NdbConnection::~NdbConnection()
{
DBUG_ENTER("NdbConnection::~NdbConnection");
theNdb->theNdbObjectIdMap->unmap(theId, this);
theNdb->theImpl->theNdbObjectIdMap.unmap(theId, this);
DBUG_VOID_RETURN;
}//NdbConnection::~NdbConnection()

35
ndb/src/ndbapi/NdbImpl.hpp
View file

@ -17,7 +17,15 @@
#ifndef NDB_IMPL_HPP
#define NDB_IMPL_HPP
#include <Vector.hpp>
#include <Ndb.hpp>
#include <NdbError.hpp>
#include <NdbCondition.h>
#include <NdbReceiver.hpp>
#include <NdbOperation.hpp>
#include <kernel/ndb_limits.h>
#include <NdbTick.h>
#include "ObjectMap.hpp"
/**
@ -25,20 +33,21 @@
*/
class NdbImpl {
public:
Vector<class NdbTableImpl *> m_invalidTables;
NdbImpl();
~NdbImpl();
void checkErrorCode(Uint32 i);
void checkInvalidTable(class NdbDictionaryImpl * dict);
// Ensure good distribution of connects
Uint32 theCurrentConnectIndex;
NdbObjectIdMap theNdbObjectIdMap;
Uint32 theNoOfDBnodes; // The number of DB nodes
Uint8 theDBnodes[MAX_NDB_NODES]; // The node number of the DB nodes
// 1 indicates to release all connections to node
Uint32 the_release_ind[MAX_NDB_NODES];
};
#include <Ndb.hpp>
#include <NdbError.hpp>
#include <NdbCondition.h>
#include <NdbReceiver.hpp>
#include <NdbOperation.hpp>
#include <NdbTick.h>
#ifdef VM_TRACE
#define TRACE_DEBUG(x) ndbout << x << endl;
#else
@ -57,7 +66,7 @@ public:
inline
void *
Ndb::int2void(Uint32 val){
return theNdbObjectIdMap->getObject(val);
return theImpl->theNdbObjectIdMap.getObject(val);
}
inline

4
ndb/src/ndbapi/NdbReceiver.cpp
View file

@ -40,7 +40,7 @@ NdbReceiver::~NdbReceiver()
{
DBUG_ENTER("NdbReceiver::~NdbReceiver");
if (m_id != NdbObjectIdMap::InvalidId) {
m_ndb->theNdbObjectIdMap->unmap(m_id, this);
m_ndb->theImpl->theNdbObjectIdMap.unmap(m_id, this);
}
delete[] m_rows;
DBUG_VOID_RETURN;
@ -54,7 +54,7 @@ NdbReceiver::init(ReceiverType type, void* owner)
m_owner = owner;
if (m_id == NdbObjectIdMap::InvalidId) {
if (m_ndb)
m_id = m_ndb->theNdbObjectIdMap->map(this);
m_id = m_ndb->theImpl->theNdbObjectIdMap.map(this);
}
theFirstRecAttr = NULL;

21
ndb/src/ndbapi/Ndbif.cpp
View file

@ -92,8 +92,8 @@ Ndb::init(int aMaxNoOfTransactions)
theDictionary->setTransporter(this, theFacade);
aNrOfCon = theNoOfDBnodes;
aNrOfOp = 2*theNoOfDBnodes;
aNrOfCon = theImpl->theNoOfDBnodes;
aNrOfOp = 2*theImpl->theNoOfDBnodes;
// Create connection object in a linked list
if((createConIdleList(aNrOfCon)) == -1){
@ -192,14 +192,14 @@ void Ndb::connected(Uint32 ref)
}
TransporterFacade * theFacade = TransporterFacade::instance();
int i;
theNoOfDBnodes= 0;
int i, n= 0;
for (i = 1; i < MAX_NDB_NODES; i++){
if (theFacade->getIsDbNode(i)){
theDBnodes[theNoOfDBnodes] = i;
theNoOfDBnodes++;
theImpl->theDBnodes[n] = i;
n++;
}
}
theImpl->theNoOfDBnodes= n;
theFirstTransId = ((Uint64)tBlockNo << 52)+
((Uint64)tmpTheNode << 40);
theFirstTransId += theFacade->m_max_trans_id;
@ -207,9 +207,10 @@ void Ndb::connected(Uint32 ref)
DBUG_PRINT("info",("connected with ref=%x, id=%d, no_db_nodes=%d, first_trans_id=%lx",
theMyRef,
tmpTheNode,
theNoOfDBnodes,
theImpl->theNoOfDBnodes,
theFirstTransId));
startTransactionNodeSelectionData.init(theNoOfDBnodes, theDBnodes);
startTransactionNodeSelectionData.init(theImpl->theNoOfDBnodes,
theImpl->theDBnodes);
theCommitAckSignal = new NdbApiSignal(theMyRef);
theDictionary->m_receiver.m_reference= theMyRef;
@ -247,7 +248,9 @@ Ndb::report_node_failure(Uint32 node_id)
*
* This method is only called by ClusterMgr (via lots of methods)
*/
the_release_ind[node_id] = 1;
theImpl->the_release_ind[node_id] = 1;
// must come after
theImpl->the_release_ind[0] = 1;
theWaiter.nodeFail(node_id);
return;
}//Ndb::report_node_failure()

30
ndb/src/ndbapi/Ndbinit.cpp
View file

@ -82,7 +82,6 @@ void Ndb::setup(Ndb_cluster_connection *ndb_cluster_connection,
{
DBUG_ENTER("Ndb::setup");
theNdbObjectIdMap= 0;
m_ndb_cluster_connection= ndb_cluster_connection;
thePreparedTransactionsArray= NULL;
theSentTransactionsArray= NULL;
@ -110,9 +109,6 @@ void Ndb::setup(Ndb_cluster_connection *ndb_cluster_connection,
theCallList= NULL;
theScanList= NULL;
theNdbBlobIdleList= NULL;
theNoOfDBnodes= 0;
theDBnodes= NULL;
the_release_ind= NULL;
the_last_check_time= 0;
theFirstTransId= 0;
theRestartGCI= 0;
@ -134,19 +130,12 @@ void Ndb::setup(Ndb_cluster_connection *ndb_cluster_connection,
theError.code = 0;
theNdbObjectIdMap = new NdbObjectIdMap(1024,1024);
theConnectionArray = new NdbConnection * [MAX_NDB_NODES];
theDBnodes = new Uint32[MAX_NDB_NODES];
the_release_ind = new Uint8[MAX_NDB_NODES];
theCommitAckSignal = NULL;
theCurrentConnectCounter = 1;
theCurrentConnectIndex = 0;
int i;
for (i = 0; i < MAX_NDB_NODES ; i++) {
theConnectionArray[i] = NULL;
the_release_ind[i] = 0;
theDBnodes[i] = 0;
}//forg
for (i = 0; i < 2048 ; i++) {
theFirstTupleId[i] = 0;
@ -213,7 +202,6 @@ Ndb::~Ndb()
doDisconnect();
delete theDictionary;
delete theImpl;
NdbGlobalEventBuffer_drop(theGlobalEventBufferHandle);
@ -260,15 +248,12 @@ Ndb::~Ndb()
startTransactionNodeSelectionData.release();
delete []theConnectionArray;
delete []theDBnodes;
delete []the_release_ind;
if(theCommitAckSignal != NULL){
delete theCommitAckSignal;
theCommitAckSignal = NULL;
}
if(theNdbObjectIdMap != 0)
delete theNdbObjectIdMap;
delete theImpl;
/**
* This sleep is to make sure that the transporter
@ -307,4 +292,17 @@ NdbWaiter::~NdbWaiter(){
NdbCondition_Destroy(m_condition);
}
NdbImpl::NdbImpl() : theNdbObjectIdMap(1024,1024),
theCurrentConnectIndex(0),
theNoOfDBnodes(0)
{
int i;
for (i = 0; i < MAX_NDB_NODES; i++) {
the_release_ind[i] = 0;
}
}
NdbImpl::~NdbImpl()
{
}

39
ndb/src/ndbapi/Ndblist.cpp
View file

@ -30,10 +30,18 @@ void
Ndb::checkFailedNode()
{
DBUG_ENTER("Ndb::checkFailedNode");
DBUG_PRINT("enter", ("theNoOfDBnodes: %d", theNoOfDBnodes));
Uint32 *the_release_ind= theImpl->the_release_ind;
if (the_release_ind[0] == 0)
{
DBUG_VOID_RETURN;
}
Uint32 tNoOfDbNodes = theImpl->theNoOfDBnodes;
Uint8 *theDBnodes= theImpl->theDBnodes;
DBUG_ASSERT(theNoOfDBnodes < MAX_NDB_NODES);
for (Uint32 i = 0; i < theNoOfDBnodes; i++){
DBUG_PRINT("enter", ("theNoOfDBnodes: %d", tNoOfDbNodes));
DBUG_ASSERT(tNoOfDbNodes < MAX_NDB_NODES);
for (Uint32 i = 0; i < tNoOfDbNodes; i++){
const NodeId node_id = theDBnodes[i];
DBUG_PRINT("info", ("i: %d, node_id: %d", i, node_id));
@ -56,31 +64,6 @@ Ndb::checkFailedNode()
DBUG_VOID_RETURN;
}
#if 0
void
NdbImpl::checkInvalidTable(NdbDictionaryImpl * dict){
Uint32 sz = m_invalidTables.size();
for(Int32 i = sz - 1; i >= 0; i--){
NdbTableImpl * tab = m_invalidTables[i];
m_invalidTables.erase(i);
dict->tableDropped(* tab);
}
}
void
NdbImpl::checkErrorCode(Uint32 i, NdbTableImpl * tab){
switch(i){
case 241:
case 283:
case 284:
case 285:
case 1225:
case 1226:
}
}
#endif
/***************************************************************************
* int createConIdleList(int aNrOfCon);
*

3
sql/log_event.cc
View file

@ -373,6 +373,9 @@ int Log_event::exec_event(struct st_relay_log_info* rli)
Note that Rotate_log_event::exec_event() does not call this function,
so there is no chance that a fake rotate event resets
last_master_timestamp.
Note that we update without mutex (probably ok - except in some very
rare cases, only consequence is that value may take some time to
display in Seconds_Behind_Master - not critical).
*/
rli->last_master_timestamp= when;
}

62
sql/slave.cc
View file

@ -545,7 +545,7 @@ int terminate_slave_threads(MASTER_INFO* mi,int thread_mask,bool skip_lock)
int terminate_slave_thread(THD* thd, pthread_mutex_t* term_lock,
pthread_mutex_t *cond_lock,
pthread_cond_t* term_cond,
volatile bool* slave_running)
volatile uint *slave_running)
{
if (term_lock)
{
@ -583,7 +583,7 @@ int terminate_slave_thread(THD* thd, pthread_mutex_t* term_lock,
int start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock,
pthread_mutex_t *cond_lock,
pthread_cond_t *start_cond,
volatile bool *slave_running,
volatile uint *slave_running,
volatile ulong *slave_run_id,
MASTER_INFO* mi,
bool high_priority)
@ -963,7 +963,7 @@ void end_slave()
static bool io_slave_killed(THD* thd, MASTER_INFO* mi)
{
DBUG_ASSERT(mi->io_thd == thd);
DBUG_ASSERT(mi->slave_running == 1); // tracking buffer overrun
DBUG_ASSERT(mi->slave_running); // tracking buffer overrun
return mi->abort_slave || abort_loop || thd->killed;
}
@ -1767,19 +1767,13 @@ void init_master_info_with_options(MASTER_INFO* mi)
strmake(mi->ssl_key, master_ssl_key, sizeof(mi->ssl_key)-1);
}
static void clear_slave_error(RELAY_LOG_INFO* rli)
void clear_slave_error(RELAY_LOG_INFO* rli)
{
/* Clear the errors displayed by SHOW SLAVE STATUS */
rli->last_slave_error[0]= 0;
rli->last_slave_errno= 0;
}
void clear_slave_error_timestamp(RELAY_LOG_INFO* rli)
{
rli->last_master_timestamp= 0;
clear_slave_error(rli);
}
/*
Reset UNTIL condition for RELAY_LOG_INFO
SYNOPSYS
@ -2166,6 +2160,11 @@ int show_master_info(THD* thd, MASTER_INFO* mi)
String *packet= &thd->packet;
protocol->prepare_for_resend();
/*
TODO: we read slave_running without run_lock, whereas these variables
are updated under run_lock and not data_lock. In 5.0 we should lock
run_lock on top of data_lock (with good order).
*/
pthread_mutex_lock(&mi->data_lock);
pthread_mutex_lock(&mi->rli.data_lock);
@ -2226,7 +2225,12 @@ int show_master_info(THD* thd, MASTER_INFO* mi)
protocol->store(mi->ssl_cipher, &my_charset_bin);
protocol->store(mi->ssl_key, &my_charset_bin);
if (mi->rli.last_master_timestamp)
/*
Seconds_Behind_Master: if SQL thread is running and I/O thread is
connected, we can compute it otherwise show NULL (i.e. unknown).
*/
if ((mi->slave_running == MYSQL_SLAVE_RUN_CONNECT) &&
mi->rli.slave_running)
{
long tmp= (long)((time_t)time((time_t*) 0)
- mi->rli.last_master_timestamp)
@ -2246,9 +2250,13 @@ int show_master_info(THD* thd, MASTER_INFO* mi)
slave is 2. At SHOW SLAVE STATUS time, assume that the difference
between timestamp of slave and rli->last_master_timestamp is 0
(i.e. they are in the same second), then we get 0-(2-1)=-1 as a result.
This confuses users, so we don't go below 0.
This confuses users, so we don't go below 0: hence the max().
last_master_timestamp == 0 (an "impossible" timestamp 1970) is a
special marker to say "consider we have caught up".
*/
protocol->store((longlong)(max(0, tmp)));
protocol->store((longlong)(mi->rli.last_master_timestamp ? max(0, tmp)
: 0));
}
else
protocol->store_null();
@ -3041,6 +3049,8 @@ slave_begin:
connected:
// TODO: the assignment below should be under mutex (5.0)
mi->slave_running= MYSQL_SLAVE_RUN_CONNECT;
thd->slave_net = &mysql->net;
thd->proc_info = "Checking master version";
if (get_master_version_and_clock(mysql, mi))
@ -3072,6 +3082,7 @@ dump");
goto err;
}
mi->slave_running= MYSQL_SLAVE_RUN_NOT_CONNECT;
thd->proc_info= "Waiting to reconnect after a failed binlog dump request";
#ifdef SIGNAL_WITH_VIO_CLOSE
thd->clear_active_vio();
@ -3148,6 +3159,7 @@ max_allowed_packet",
mysql_error(mysql));
goto err;
}
mi->slave_running= MYSQL_SLAVE_RUN_NOT_CONNECT;
thd->proc_info = "Waiting to reconnect after a failed master event read";
#ifdef SIGNAL_WITH_VIO_CLOSE
thd->clear_active_vio();
@ -3323,6 +3335,14 @@ slave_begin:
pthread_mutex_lock(&LOCK_thread_count);
threads.append(thd);
pthread_mutex_unlock(&LOCK_thread_count);
/*
We are going to set slave_running to 1. Assuming slave I/O thread is
alive and connected, this is going to make Seconds_Behind_Master be 0
i.e. "caught up". Even if we're just at start of thread. Well it's ok, at
the moment we start we can think we are caught up, and the next second we
start receiving data so we realize we are not caught up and
Seconds_Behind_Master grows. No big deal.
*/
rli->slave_running = 1;
rli->abort_slave = 0;
pthread_mutex_unlock(&rli->run_lock);
@ -4169,6 +4189,21 @@ Before assert, my_b_tell(cur_log)=%s rli->event_relay_log_pos=%s",
*/
if (hot_log)
{
/*
We say in Seconds_Behind_Master that we have "caught up". Note that
for example if network link is broken but I/O slave thread hasn't
noticed it (slave_net_timeout not elapsed), then we'll say "caught
up" whereas we're not really caught up. Fixing that would require
internally cutting timeout in smaller pieces in network read, no
thanks. Another example: SQL has caught up on I/O, now I/O has read
a new event and is queuing it; the false "0" will exist until SQL
finishes executing the new event; it will be look abnormal only if
the events have old timestamps (then you get "many", 0, "many").
Transient phases like this can't really be fixed.
*/
time_t save_timestamp= rli->last_master_timestamp;
rli->last_master_timestamp= 0;
DBUG_ASSERT(rli->relay_log.get_open_count() == rli->cur_log_old_open_count);
/*
We can, and should release data_lock while we are waiting for
@ -4215,6 +4250,7 @@ Before assert, my_b_tell(cur_log)=%s rli->event_relay_log_pos=%s",
rli->relay_log.wait_for_update(rli->sql_thd, 1);
// re-acquire data lock since we released it earlier
pthread_mutex_lock(&rli->data_lock);
rli->last_master_timestamp= save_timestamp;
continue;
}
/*

27
sql/slave.h
View file

@ -98,6 +98,21 @@ enum enum_binlog_formats {
BINLOG_FORMAT_323_LESS_57,
BINLOG_FORMAT_323_GEQ_57 };
/*
3 possible values for MASTER_INFO::slave_running and
RELAY_LOG_INFO::slave_running.
The values 0,1,2 are very important: to keep the diff small, I didn't
substitute places where we use 0/1 with the newly defined symbols. So don't change
these values.
The same way, code is assuming that in RELAY_LOG_INFO we use only values
0/1.
I started with using an enum, but
enum_variable=1; is not legal so would have required many line changes.
*/
#define MYSQL_SLAVE_NOT_RUN 0
#define MYSQL_SLAVE_RUN_NOT_CONNECT 1
#define MYSQL_SLAVE_RUN_CONNECT 2
/****************************************************************************
Replication SQL Thread
@ -251,7 +266,8 @@ typedef struct st_relay_log_info
/* if not set, the value of other members of the structure are undefined */
bool inited;
volatile bool abort_slave, slave_running;
volatile bool abort_slave;
volatile uint slave_running;
/*
Condition and its parameters from START SLAVE UNTIL clause.
@ -385,7 +401,8 @@ typedef struct st_master_info
#endif
bool inited;
enum enum_binlog_formats old_format;
volatile bool abort_slave, slave_running;
volatile bool abort_slave;
volatile uint slave_running;
volatile ulong slave_run_id;
/*
The difference in seconds between the clock of the master and the clock of
@ -464,7 +481,7 @@ int terminate_slave_threads(MASTER_INFO* mi, int thread_mask,
int terminate_slave_thread(THD* thd, pthread_mutex_t* term_mutex,
pthread_mutex_t* cond_lock,
pthread_cond_t* term_cond,
volatile bool* slave_running);
volatile uint* slave_running);
int start_slave_threads(bool need_slave_mutex, bool wait_for_start,
MASTER_INFO* mi, const char* master_info_fname,
const char* slave_info_fname, int thread_mask);
@ -477,7 +494,7 @@ int start_slave_threads(bool need_slave_mutex, bool wait_for_start,
int start_slave_thread(pthread_handler h_func, pthread_mutex_t* start_lock,
pthread_mutex_t *cond_lock,
pthread_cond_t* start_cond,
volatile bool *slave_running,
volatile uint *slave_running,
volatile ulong *slave_run_id,
MASTER_INFO* mi,
bool high_priority);
@ -519,7 +536,7 @@ void slave_print_error(RELAY_LOG_INFO* rli, int err_code, const char* msg, ...);
void end_slave(); /* clean up */
void init_master_info_with_options(MASTER_INFO* mi);
void clear_until_condition(RELAY_LOG_INFO* rli);
void clear_slave_error_timestamp(RELAY_LOG_INFO* rli);
void clear_slave_error(RELAY_LOG_INFO* rli);
int init_master_info(MASTER_INFO* mi, const char* master_info_fname,
const char* slave_info_fname,
bool abort_if_no_master_info_file,

8
sql/sql_repl.cc
View file

@ -880,10 +880,10 @@ int reset_slave(THD *thd, MASTER_INFO* mi)
*/
init_master_info_with_options(mi);
/*
Reset errors, and master timestamp (the idea is that we forget about the
Reset errors (the idea is that we forget about the
old master).
*/
clear_slave_error_timestamp(&mi->rli);
clear_slave_error(&mi->rli);
clear_until_condition(&mi->rli);
// close master_info_file, relay_log_info_file, set mi->inited=rli->inited=0
@ -1143,8 +1143,8 @@ int change_master(THD* thd, MASTER_INFO* mi)
pthread_mutex_lock(&mi->rli.data_lock);
mi->rli.abort_pos_wait++; /* for MASTER_POS_WAIT() to abort */
/* Clear the errors, for a clean start, and master timestamp */
clear_slave_error_timestamp(&mi->rli);
/* Clear the errors, for a clean start */
clear_slave_error(&mi->rli);
clear_until_condition(&mi->rli);
/*
If we don't write new coordinates to disk now, then old will remain in