mirror/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-01-19 05:22:25 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

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

Browse source 
into poseidon.ndb.mysql.com:/home/tomas/mysql-4.1
This commit is contained in:
tomas@poseidon.ndb.mysql.com 2004-12-22 16:30:16 +01:00
commit 99622f3fa1
15 changed files with 194 additions and 84 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
ndb/include/ndbapi/ndb_cluster_connection.hpp
View file

@ -82,7 +82,7 @@ public:
void set_optimized_node_selection(int val);
int no_db_nodes();
unsigned no_db_nodes();
#endif
private:

7
ndb/include/transporter/TransporterRegistry.hpp
View file

@ -99,7 +99,12 @@ public:
unsigned sizeOfLongSignalMemory = 100);
bool init(NodeId localNodeId);
/**
* after a connect from client, perform connection using correct transporter
*/
bool connect_server(NDB_SOCKET_TYPE sockfd);
/**
* Remove all transporters
*/

2
ndb/include/util/ndb_opts.h
View file

@ -34,7 +34,7 @@ OPT_NDB_OPTIMIZED_NODE_SELECTION
#define OPT_NDB_CONNECTSTRING 'c'
#ifdef NDB_SHM_TRANSPORTER
#if defined(NDB_SHM_TRANSPORTER) && MYSQL_VERSION_ID >= 50000
#define OPT_NDB_SHM_DEFAULT 1
#else
#define OPT_NDB_SHM_DEFAULT 0

2
ndb/src/common/transporter/Makefile.am
View file

@ -13,7 +13,7 @@ EXTRA_libtransporter_la_SOURCES = SHM_Transporter.cpp SHM_Transporter.unix.cpp S
libtransporter_la_LIBADD = @ndb_transporter_opt_objs@
libtransporter_la_DEPENDENCIES = @ndb_transporter_opt_objs@
INCLUDES_LOC = -I$(top_srcdir)/ndb/include/kernel -I$(top_srcdir)/ndb/include/transporter @NDB_SCI_INCLUDES@
INCLUDES_LOC = -I$(top_srcdir)/ndb/include/mgmapi -I$(top_srcdir)/ndb/include/debugger -I$(top_srcdir)/ndb/include/kernel -I$(top_srcdir)/ndb/include/transporter @NDB_SCI_INCLUDES@
include $(top_srcdir)/ndb/config/common.mk.am
include $(top_srcdir)/ndb/config/type_util.mk.am

3
ndb/src/common/transporter/SCI_Transporter.cpp
View file

@ -44,7 +44,8 @@ SCI_Transporter::SCI_Transporter(TransporterRegistry &t_reg,
bool chksm,
bool signalId,
Uint32 reportFreq) :
Transporter(t_reg, lHostName, rHostName, r_port, _localNodeId,
Transporter(t_reg, tt_SCI_TRANSPORTER,
lHostName, rHostName, r_port, _localNodeId,
_remoteNodeId, 0, false, chksm, signalId)
{
DBUG_ENTER("SCI_Transporter::SCI_Transporter");

6
ndb/src/common/transporter/SHM_Transporter.cpp
View file

@ -38,7 +38,8 @@ SHM_Transporter::SHM_Transporter(TransporterRegistry &t_reg,
bool signalId,
key_t _shmKey,
Uint32 _shmSize) :
Transporter(t_reg, lHostName, rHostName, r_port, lNodeId, rNodeId,
Transporter(t_reg, tt_SHM_TRANSPORTER,
lHostName, rHostName, r_port, lNodeId, rNodeId,
0, false, checksum, signalId),
shmKey(_shmKey),
shmSize(_shmSize)
@ -256,6 +257,9 @@ SHM_Transporter::connect_client_impl(NDB_SOCKET_TYPE sockfd)
SocketOutputStream s_output(sockfd);
char buf[256];
#if 1
#endif
// Wait for server to create and attach
if (s_input.gets(buf, 256) == 0) {
NDB_CLOSE_SOCKET(sockfd);

3
ndb/src/common/transporter/TCP_Transporter.cpp
View file

@ -72,7 +72,8 @@ TCP_Transporter::TCP_Transporter(TransporterRegistry &t_reg,
NodeId rNodeId,
bool chksm, bool signalId,
Uint32 _reportFreq) :
Transporter(t_reg, lHostName, rHostName, r_port, lNodeId, rNodeId,
Transporter(t_reg, tt_TCP_TRANSPORTER,
lHostName, rHostName, r_port, lNodeId, rNodeId,
0, false, chksm, signalId),
m_sendBuffer(sendBufSize)
{

57
ndb/src/common/transporter/Transporter.cpp
View file

@ -24,7 +24,11 @@
#include <InputStream.hpp>
#include <OutputStream.hpp>
#include <EventLogger.hpp>
extern EventLogger g_eventLogger;
Transporter::Transporter(TransporterRegistry &t_reg,
TransporterType _type,
const char *lHostName,
const char *rHostName,
int r_port,
@ -35,6 +39,7 @@ Transporter::Transporter(TransporterRegistry &t_reg,
: m_r_port(r_port), remoteNodeId(rNodeId), localNodeId(lNodeId),
isServer(lNodeId < rNodeId),
m_packer(_signalId, _checksum),
m_type(_type),
m_transporter_registry(t_reg)
{
DBUG_ENTER("Transporter::Transporter");
@ -73,7 +78,8 @@ Transporter::Transporter(TransporterRegistry &t_reg,
m_socket_client= 0;
else
m_socket_client= new SocketClient(remoteHostName, r_port,
new SocketAuthSimple("ndbd", "ndbd passwd"));
new SocketAuthSimple("ndbd",
"ndbd passwd"));
DBUG_VOID_RETURN;
}
@ -84,7 +90,9 @@ Transporter::~Transporter(){
bool
Transporter::connect_server(NDB_SOCKET_TYPE sockfd) {
// all initial negotiation is done in TransporterRegistry::connect_server
DBUG_ENTER("Transporter::connect_server");
if(m_connected)
{
DBUG_RETURN(true); // TODO assert(0);
@ -108,27 +116,60 @@ Transporter::connect_client() {
if (sockfd == NDB_INVALID_SOCKET)
return false;
// send info about own id
DBUG_ENTER("Transporter::connect_client");
// send info about own id
// send info about own transporter type
SocketOutputStream s_output(sockfd);
s_output.println("%d", localNodeId);
s_output.println("%d %d", localNodeId, m_type);
// get remote id
int nodeId;
int nodeId, remote_transporter_type= -1;
SocketInputStream s_input(sockfd);
char buf[256];
if (s_input.gets(buf, 256) == 0) {
NDB_CLOSE_SOCKET(sockfd);
return false;
DBUG_RETURN(false);
}
if (sscanf(buf, "%d", &nodeId) != 1) {
int r= sscanf(buf, "%d %d", &nodeId, &remote_transporter_type);
switch (r) {
case 2:
break;
case 1:
// we're running version prior to 4.1.9
// ok, but with no checks on transporter configuration compatability
break;
default:
NDB_CLOSE_SOCKET(sockfd);
return false;
DBUG_RETURN(false);
}
DBUG_PRINT("info", ("nodeId=%d remote_transporter_type=%d",
nodeId, remote_transporter_type));
if (remote_transporter_type != -1)
{
if (remote_transporter_type != m_type)
{
DBUG_PRINT("error", ("Transporter types mismatch this=%d remote=%d",
m_type, remote_transporter_type));
NDB_CLOSE_SOCKET(sockfd);
g_eventLogger.error("Incompatible configuration: transporter type "
"mismatch with node %d", nodeId);
DBUG_RETURN(false);
}
}
else if (m_type == tt_SHM_TRANSPORTER)
{
g_eventLogger.warning("Unable to verify transporter compatability with node %d", nodeId);
}
bool res = connect_client_impl(sockfd);
if(res){
m_connected = true;
m_errorCount = 0;
}
return res;
DBUG_RETURN(res);
}
void

2
ndb/src/common/transporter/Transporter.hpp
View file

@ -71,6 +71,7 @@ public:
protected:
Transporter(TransporterRegistry &,
TransporterType,
const char *lHostName,
const char *rHostName,
int r_port,
@ -127,6 +128,7 @@ protected:
protected:
bool m_connected; // Are we connected
TransporterType m_type;
TransporterRegistry &m_transporter_registry;
void *get_callback_obj() { return m_transporter_registry.callbackObj; };

133
ndb/src/common/transporter/TransporterRegistry.cpp
View file

@ -47,6 +47,9 @@
#include <InputStream.hpp>
#include <OutputStream.hpp>
#include <EventLogger.hpp>
extern EventLogger g_eventLogger;
int g_shm_pid = 0;
SocketServer::Session * TransporterService::newSession(NDB_SOCKET_TYPE sockfd)
@ -57,49 +60,10 @@ SocketServer::Session * TransporterService::newSession(NDB_SOCKET_TYPE sockfd)
DBUG_RETURN(0);
}
if (!m_transporter_registry->connect_server(sockfd))
{
// read node id from client
int nodeId;
SocketInputStream s_input(sockfd);
char buf[256];
if (s_input.gets(buf, 256) == 0) {
NDB_CLOSE_SOCKET(sockfd);
DBUG_PRINT("error", ("Could not get node id from client"));
DBUG_RETURN(0);
}
if (sscanf(buf, "%d", &nodeId) != 1) {
NDB_CLOSE_SOCKET(sockfd);
DBUG_PRINT("error", ("Error in node id from client"));
DBUG_RETURN(0);
}
//check that nodeid is valid and that there is an allocated transporter
if ( nodeId < 0 || nodeId >= (int)m_transporter_registry->maxTransporters) {
NDB_CLOSE_SOCKET(sockfd);
DBUG_PRINT("error", ("Node id out of range from client"));
DBUG_RETURN(0);
}
if (m_transporter_registry->theTransporters[nodeId] == 0) {
NDB_CLOSE_SOCKET(sockfd);
DBUG_PRINT("error", ("No transporter for this node id from client"));
DBUG_RETURN(0);
}
//check that the transporter should be connected
if (m_transporter_registry->performStates[nodeId] != TransporterRegistry::CONNECTING) {
NDB_CLOSE_SOCKET(sockfd);
DBUG_PRINT("error", ("Transporter in wrong state for this node id from client"));
DBUG_RETURN(0);
}
Transporter *t= m_transporter_registry->theTransporters[nodeId];
// send info about own id (just as response to acknowledge connection)
SocketOutputStream s_output(sockfd);
s_output.println("%d", t->getLocalNodeId());
// setup transporter (transporter responsible for closing sockfd)
t->connect_server(sockfd);
NDB_CLOSE_SOCKET(sockfd);
DBUG_RETURN(0);
}
DBUG_RETURN(0);
@ -195,6 +159,91 @@ TransporterRegistry::init(NodeId nodeId) {
return true;
}
bool
TransporterRegistry::connect_server(NDB_SOCKET_TYPE sockfd)
{
DBUG_ENTER("TransporterRegistry::connect_server");
// read node id from client
// read transporter type
int nodeId, remote_transporter_type= -1;
SocketInputStream s_input(sockfd);
char buf[256];
if (s_input.gets(buf, 256) == 0) {
DBUG_PRINT("error", ("Could not get node id from client"));
DBUG_RETURN(false);
}
int r= sscanf(buf, "%d %d", &nodeId, &remote_transporter_type);
switch (r) {
case 2:
break;
case 1:
// we're running version prior to 4.1.9
// ok, but with no checks on transporter configuration compatability
break;
default:
DBUG_PRINT("error", ("Error in node id from client"));
DBUG_RETURN(false);
}
DBUG_PRINT("info", ("nodeId=%d remote_transporter_type=%d",
nodeId,remote_transporter_type));
//check that nodeid is valid and that there is an allocated transporter
if ( nodeId < 0 || nodeId >= (int)maxTransporters) {
DBUG_PRINT("error", ("Node id out of range from client"));
DBUG_RETURN(false);
}
if (theTransporters[nodeId] == 0) {
DBUG_PRINT("error", ("No transporter for this node id from client"));
DBUG_RETURN(false);
}
//check that the transporter should be connected
if (performStates[nodeId] != TransporterRegistry::CONNECTING) {
DBUG_PRINT("error", ("Transporter in wrong state for this node id from client"));
DBUG_RETURN(false);
}
Transporter *t= theTransporters[nodeId];
// send info about own id (just as response to acknowledge connection)
// send info on own transporter type
SocketOutputStream s_output(sockfd);
s_output.println("%d %d", t->getLocalNodeId(), t->m_type);
if (remote_transporter_type != -1)
{
if (remote_transporter_type != t->m_type)
{
DBUG_PRINT("error", ("Transporter types mismatch this=%d remote=%d",
t->m_type, remote_transporter_type));
g_eventLogger.error("Incompatible configuration: Transporter type "
"mismatch with node %d", nodeId);
// wait for socket close for 1 second to let message arrive at client
{
fd_set a_set;
FD_ZERO(&a_set);
FD_SET(sockfd, &a_set);
struct timeval timeout;
timeout.tv_sec = 1; timeout.tv_usec = 0;
select(sockfd+1, &a_set, 0, 0, &timeout);
}
DBUG_RETURN(false);
}
}
else if (t->m_type == tt_SHM_TRANSPORTER)
{
g_eventLogger.warning("Unable to verify transporter compatability with node %d", nodeId);
}
// setup transporter (transporter responsible for closing sockfd)
t->connect_server(sockfd);
DBUG_RETURN(true);
}
bool
TransporterRegistry::createTransporter(TCP_TransporterConfiguration *config) {
#ifdef NDB_TCP_TRANSPORTER

2
ndb/src/kernel/main.cpp
View file

@ -58,7 +58,7 @@ int main(int argc, char** argv)
// Print to stdout/console
g_eventLogger.createConsoleHandler();
g_eventLogger.setCategory("NDB");
g_eventLogger.enable(Logger::LL_INFO, Logger::LL_ALERT); // Log INFO to ALERT
g_eventLogger.enable(Logger::LL_ON, Logger::LL_ERROR);
globalEmulatorData.create();

37
ndb/src/mgmsrv/MgmtSrvr.cpp
View file

@ -133,8 +133,7 @@ MgmtSrvr::signalRecvThreadRun()
}
}
EventLogger g_EventLogger;
extern EventLogger g_eventLogger;
static NdbOut&
operator<<(NdbOut& out, const LogLevel & ll)
@ -200,7 +199,7 @@ MgmtSrvr::logLevelThreadRun()
void
MgmtSrvr::startEventLog()
{
g_EventLogger.setCategory("MgmSrvr");
g_eventLogger.setCategory("MgmSrvr");
ndb_mgm_configuration_iterator * iter = ndb_mgm_create_configuration_iterator
((ndb_mgm_configuration*)_config->m_configValues, CFG_SECTION_NODE);
@ -226,7 +225,7 @@ MgmtSrvr::startEventLog()
logdest.assfmt("FILE:filename=%s,maxsize=1000000,maxfiles=6",
clusterLog);
}
if(!g_EventLogger.addHandler(logdest)) {
if(!g_eventLogger.addHandler(logdest)) {
ndbout << "Warning: could not add log destination \""
<< logdest.c_str() << "\"" << endl;
}
@ -250,21 +249,21 @@ MgmtSrvr::setEventLogFilter(int severity, int enable)
{
Logger::LoggerLevel level = (Logger::LoggerLevel)severity;
if (enable > 0) {
g_EventLogger.enable(level);
g_eventLogger.enable(level);
} else if (enable == 0) {
g_EventLogger.disable(level);
} else if (g_EventLogger.isEnable(level)) {
g_EventLogger.disable(level);
g_eventLogger.disable(level);
} else if (g_eventLogger.isEnable(level)) {
g_eventLogger.disable(level);
} else {
g_EventLogger.enable(level);
g_eventLogger.enable(level);
}
return g_EventLogger.isEnable(level);
return g_eventLogger.isEnable(level);
}
bool
MgmtSrvr::isEventLogFilterEnabled(int severity)
{
return g_EventLogger.isEnable((Logger::LoggerLevel)severity);
return g_eventLogger.isEnable((Logger::LoggerLevel)severity);
}
static ErrorItem errorTable[] =
@ -1990,7 +1989,7 @@ MgmtSrvr::handleReceivedSignal(NdbApiSignal* signal)
}
default:
g_EventLogger.error("Unknown signal received. SignalNumber: "
g_eventLogger.error("Unknown signal received. SignalNumber: "
"%i from (%d, %x)",
gsn,
refToNode(signal->theSendersBlockRef),
@ -2066,7 +2065,7 @@ MgmtSrvr::handleStopReply(NodeId nodeId, Uint32 errCode)
error:
if(errCode != 0){
g_EventLogger.error("Unexpected signal received. SignalNumber: %i from %d",
g_eventLogger.error("Unexpected signal received. SignalNumber: %i from %d",
GSN_STOP_REF, nodeId);
}
}
@ -2286,7 +2285,7 @@ MgmtSrvr::alloc_node_id(NodeId * nodeId,
m_reserved_nodes.set(id_found);
char tmp_str[128];
m_reserved_nodes.getText(tmp_str);
g_EventLogger.info("Mgmt server state: nodeid %d reserved for ip %s, m_reserved_nodes %s.",
g_eventLogger.info("Mgmt server state: nodeid %d reserved for ip %s, m_reserved_nodes %s.",
id_found, get_connect_address(id_found), tmp_str);
DBUG_RETURN(true);
}
@ -2346,7 +2345,7 @@ MgmtSrvr::alloc_node_id(NodeId * nodeId,
*nodeId);
}
g_EventLogger.warning("Allocate nodeid (%d) failed. Connection from ip %s. "
g_eventLogger.warning("Allocate nodeid (%d) failed. Connection from ip %s. "
"Returned error string \"%s\"",
*nodeId,
client_addr != 0 ? inet_ntoa(((struct sockaddr_in *)(client_addr))->sin_addr) : "<none>",
@ -2369,10 +2368,10 @@ MgmtSrvr::alloc_node_id(NodeId * nodeId,
}
}
if (tmp_connected.length() > 0)
g_EventLogger.info("Mgmt server state: node id's %s connected but not reserved",
g_eventLogger.info("Mgmt server state: node id's %s connected but not reserved",
tmp_connected.c_str());
if (tmp_not_connected.length() > 0)
g_EventLogger.info("Mgmt server state: node id's %s not connected but reserved",
g_eventLogger.info("Mgmt server state: node id's %s not connected but reserved",
tmp_not_connected.c_str());
}
DBUG_RETURN(false);
@ -2404,7 +2403,7 @@ MgmtSrvr::eventReport(NodeId nodeId, const Uint32 * theData)
EventReport::EventType type = eventReport->getEventType();
// Log event
g_EventLogger.log(type, theData, nodeId,
g_eventLogger.log(type, theData, nodeId,
&m_event_listner[0].m_logLevel);
m_event_listner.log(type, theData, nodeId);
}
@ -2647,7 +2646,7 @@ MgmtSrvr::Allocated_resources::~Allocated_resources()
char tmp_str[128];
m_mgmsrv.m_reserved_nodes.getText(tmp_str);
g_EventLogger.info("Mgmt server state: nodeid %d freed, m_reserved_nodes %s.",
g_eventLogger.info("Mgmt server state: nodeid %d freed, m_reserved_nodes %s.",
get_nodeid(), tmp_str);
}
}

10
ndb/src/mgmsrv/main.cpp
View file

@ -86,7 +86,7 @@ static MgmGlobals glob;
* Global variables
*/
bool g_StopServer;
extern EventLogger g_EventLogger;
extern EventLogger g_eventLogger;
extern int global_mgmt_server_check;
@ -303,12 +303,12 @@ int main(int argc, char** argv)
BaseString::snprintf(msg, sizeof(msg),
"NDB Cluster Management Server. %s", NDB_VERSION_STRING);
ndbout_c(msg);
g_EventLogger.info(msg);
g_eventLogger.info(msg);
BaseString::snprintf(msg, 256, "Id: %d, Command port: %d",
glob.localNodeId, glob.port);
ndbout_c(msg);
g_EventLogger.info(msg);
g_eventLogger.info(msg);
g_StopServer = false;
glob.socketServer->startServer();
@ -324,10 +324,10 @@ int main(int argc, char** argv)
NdbSleep_MilliSleep(500);
}
g_EventLogger.info("Shutting down server...");
g_eventLogger.info("Shutting down server...");
glob.socketServer->stopServer();
glob.socketServer->stopSessions();
g_EventLogger.info("Shutdown complete");
g_eventLogger.info("Shutdown complete");
return 0;
error_end:
return 1;

10
ndb/src/ndbapi/ndb_cluster_connection.cpp
View file

@ -31,6 +31,9 @@
#include <Vector.hpp>
#include <md5_hash.hpp>
#include <EventLogger.hpp>
EventLogger g_eventLogger;
static int g_run_connect_thread= 0;
#include <NdbMutex.h>
@ -174,7 +177,7 @@ Ndb_cluster_connection_impl::get_next_node(Ndb_cluster_connection_node_iter &ite
return node.id;
}
int
unsigned
Ndb_cluster_connection::no_db_nodes()
{
return m_impl.m_all_nodes.size();
@ -248,6 +251,11 @@ Ndb_cluster_connection_impl::Ndb_cluster_connection_impl(const char *
{
DBUG_ENTER("Ndb_cluster_connection");
DBUG_PRINT("enter",("Ndb_cluster_connection this=0x%x", this));
g_eventLogger.createConsoleHandler();
g_eventLogger.setCategory("NdbApi");
g_eventLogger.enable(Logger::LL_ON, Logger::LL_ERROR);
m_transporter_facade=
TransporterFacade::theFacadeInstance= new TransporterFacade();

2
sql/mysqld.cc
View file

@ -53,7 +53,7 @@
#endif
#ifdef HAVE_NDBCLUSTER_DB
#define OPT_NDBCLUSTER_DEFAULT 0
#ifdef NDB_SHM_TRANSPORTER
#if defined(NDB_SHM_TRANSPORTER) && MYSQL_VERSION_ID >= 50000
#define OPT_NDB_SHM_DEFAULT 1
#else
#define OPT_NDB_SHM_DEFAULT 0