hcq/MaxScale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MXS-2196: Allocate a session before allocating DCBs

Browse Source 
Allocating the session before a DCB guarantees that at no point will a DCB
have a null session. This further clarifies the concept of the session and
also allows the listener reference to be moved there.

Ideally, the session itself would allocate and assign the client DCB but
since the Listener is the only one who does it, it's acceptable for now.
This commit is contained in:
Markus Mäkelä
2018-12-03 19:25:26 +02:00
parent 692127a2cb
commit 43c33e9f4a
20 changed files with 106 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
server/core/dcb.cc
View File

@ -147,21 +147,20 @@ static MXB_WORKER* get_dcb_owner(dcb_role_t role)
return RoutingWorker::get_current();
}
DCB::DCB(dcb_role_t role, const SListener& listener, SERVICE* service)
DCB::DCB(dcb_role_t role, MXS_SESSION* session)
: MXB_POLL_DATA{dcb_poll_handler, get_dcb_owner(role)}
, dcb_role(role)
, session(nullptr)
, listener(listener)
, session(session)
, high_water(config_writeq_high_water())
, low_water(config_writeq_low_water())
, service(service)
, service(session->service)
, last_read(mxs_clock())
{
// TODO: Remove DCB_ROLE_INTERNAL to always have a valid listener
if (listener)
if (session->listener)
{
func = listener->protocol_func();
authfunc = listener->auth_func();
func = session->listener->protocol_func();
authfunc = session->listener->auth_func();
}
if (high_water && low_water)
@ -221,9 +220,9 @@ DCB::~DCB()
*
* @return An available DCB or NULL if none could be allocated.
*/
DCB* dcb_alloc(dcb_role_t role, const SListener& listener, SERVICE* service)
DCB* dcb_alloc(dcb_role_t role, MXS_SESSION* session)
{
return new(std::nothrow) DCB(role, listener, service);
return new(std::nothrow) DCB(role, session);
}
/**
@ -351,7 +350,7 @@ DCB* dcb_connect(SERVER* server, MXS_SESSION* session, const char* protocol)
}
}
if ((dcb = dcb_alloc(DCB_ROLE_BACKEND_HANDLER, NULL, session->service)) == NULL)
if ((dcb = dcb_alloc(DCB_ROLE_BACKEND_HANDLER, session)) == NULL)
{
return NULL;
}
@ -1280,9 +1279,9 @@ void printDCB(DCB* dcb)
{
printf("\tUsername: %s\n", dcb->user);
}
if (dcb->listener)
if (dcb->session->listener)
{
printf("\tProtocol: %s\n", dcb->listener->protocol());
printf("\tProtocol: %s\n", dcb->session->listener->protocol());
}
if (dcb->writeq)
{
@ -1388,9 +1387,9 @@ void dprintOneDCB(DCB* pdcb, DCB* dcb)
"\tUsername: %s\n",
dcb->user);
}
if (dcb->listener)
if (dcb->session->listener)
{
dcb_printf(pdcb, "\tProtocol: %s\n", dcb->listener->protocol());
dcb_printf(pdcb, "\tProtocol: %s\n", dcb->session->listener->protocol());
}
if (dcb->writeq)
{
@ -1544,9 +1543,9 @@ void dprintDCB(DCB* pdcb, DCB* dcb)
"\tUsername: %s\n",
dcb->user);
}
if (dcb->listener)
if (dcb->session->listener)
{
dcb_printf(pdcb, "\tProtocol: %s\n", dcb->listener->protocol());
dcb_printf(pdcb, "\tProtocol: %s\n", dcb->session->listener->protocol());
}
if (dcb->session)
@ -2122,8 +2121,8 @@ static int dcb_create_SSL(DCB* dcb, SSL_LISTENER* ssl)
*/
int dcb_accept_SSL(DCB* dcb)
{
if ((NULL == dcb->listener || NULL == dcb->listener->ssl())
|| (NULL == dcb->ssl && dcb_create_SSL(dcb, dcb->listener->ssl()) != 0))
if (NULL == dcb->session->listener->ssl()
|| (NULL == dcb->ssl && dcb_create_SSL(dcb, dcb->session->listener->ssl()) != 0))
{
return -1;
}
@ -2435,8 +2434,7 @@ void dcb_process_idle_sessions(int thr)
{
if (dcb->dcb_role == DCB_ROLE_CLIENT_HANDLER)
{
mxb_assert(dcb->listener);
SERVICE* service = dcb->listener->service();
SERVICE* service = dcb->session->service;
if (service->conn_idle_timeout && dcb->state == DCB_STATE_POLLING)
{

5
server/core/internal/session.hh
View File

@ -164,9 +164,12 @@ public:
using FilterList = std::vector<SessionFilter>;
Session(SERVICE* service, DCB* client_dcb);
Session(const SListener& listener);
~Session();
// Links a client DCB to a session
void set_client_dcb(DCB* dcb);
bool setup_filters(Service* service);
const FilterList& get_filters() const

56
server/core/listener.cc
View File

@ -38,6 +38,7 @@
#include <maxscale/routingworker.hh>
#include "internal/modules.hh"
#include "internal/session.hh"
static std::list<SListener> all_listeners;
static std::mutex listener_lock;
@ -675,24 +676,24 @@ const char* Listener::state() const
{
switch (m_state)
{
case CREATED:
return "Created";
case CREATED:
return "Created";
case STARTED:
return "Running";
case STARTED:
return "Running";
case STOPPED:
return "Stopped";
case STOPPED:
return "Stopped";
case FAILED:
return "Failed";
case FAILED:
return "Failed";
case DESTROYED:
return "Destroyed";
case DESTROYED:
return "Destroyed";
default:
mxb_assert(!true);
return "Unknown";
default:
mxb_assert(!true);
return "Unknown";
}
}
@ -864,17 +865,21 @@ DCB* Listener::accept_one_dcb()
{
configure_network_socket(c_sock, client_conn.ss_family);
client_dcb = dcb_alloc(DCB_ROLE_CLIENT_HANDLER, m_self, m_service);
mxs::Session* session = new(std::nothrow) mxs::Session(m_self);
client_dcb = dcb_alloc(DCB_ROLE_CLIENT_HANDLER, session);
if (client_dcb == NULL)
if (!session || !client_dcb)
{
MXS_ERROR("Failed to create DCB object for client connection.");
MXS_ERROR("Failed to create session objects for client connection.");
close(c_sock);
delete session;
dcb_close(client_dcb);
return NULL;
}
else
{
session->set_client_dcb(client_dcb);
client_dcb->fd = c_sock;
client_dcb->session = session_alloc(m_service, client_dcb);
// get client address
if (client_conn.ss_family == AF_UNIX)
@ -907,25 +912,28 @@ DCB* Listener::accept_one_dcb()
}
/** Allocate DCB specific authentication data */
if (client_dcb->authfunc.create
&& (client_dcb->authenticator_data =
client_dcb->authfunc.create(client_dcb->listener->auth_instance())) == NULL)
if (m_auth_func.create
&& (client_dcb->authenticator_data = m_auth_func.create(m_auth_instance)) == NULL)
{
MXS_ERROR("Failed to create authenticator for client DCB");
delete session;
dcb_close(client_dcb);
return NULL;
}
if (client_dcb->service->max_connections
&& client_dcb->service->client_count >= client_dcb->service->max_connections)
if (m_service->max_connections && m_service->client_count >= m_service->max_connections)
{
// TODO: If connections can be queued, this is the place to put the
// TODO: connection on that queue.
if (client_dcb->func.connlimit)
if (m_proto_func.connlimit)
{
client_dcb->func.connlimit(client_dcb, client_dcb->service->max_connections);
m_proto_func.connlimit(client_dcb, m_service->max_connections);
}
// TODO: This is never used as the client connection is not up yet
client_dcb->session->close_reason = SESSION_CLOSE_TOO_MANY_CONNECTIONS;
delete session;
dcb_close(client_dcb);
client_dcb = NULL;
}

3
server/core/server.cc
View File

@ -264,13 +264,14 @@ DCB* server_get_persistent(SERVER* server, const char* user, const char* ip, con
dcb = server->persistent[id];
while (dcb)
{
// TODO: Fix this, it won't work (DCB in pool has no session)
if (dcb->user
&& dcb->remote
&& ip
&& !dcb->dcb_errhandle_called
&& 0 == strcmp(dcb->user, user)
&& 0 == strcmp(dcb->remote, ip)
&& 0 == strcmp(dcb->listener->protocol(), protocol))
&& 0 == strcmp(dcb->session->listener->protocol(), protocol))
{
if (NULL == previous)
{

29
server/core/session.cc
View File

@ -77,7 +77,7 @@ static void session_simple_free(MXS_SESSION* session, DCB* dcb);
static void session_add_to_all_list(MXS_SESSION* session);
static MXS_SESSION* session_find_free();
static void session_final_free(MXS_SESSION* session);
static void session_deliver_response(MXS_SESSION* session);
static void session_deliver_response(MXS_SESSION* session);
/**
* The clientReply of the session.
@ -88,13 +88,14 @@ static void session_deliver_response(MXS_SESSION* session);
*/
static int session_reply(MXS_FILTER* inst, MXS_FILTER_SESSION* session, GWBUF* data);
MXS_SESSION::MXS_SESSION(DCB* client_dcb)
MXS_SESSION::MXS_SESSION(const SListener& listener)
: state(SESSION_STATE_READY)
, ses_id(session_get_next_id())
, client_dcb(client_dcb)
, client_dcb(nullptr)
, listener(listener)
, router_session(nullptr)
, stats{time(0)}
, service(client_dcb->service)
, service(listener ? listener->service() : nullptr)
, head{}
, tail{}
, refcount(1)
@ -112,11 +113,6 @@ MXS_SESSION::~MXS_SESSION()
{
}
MXS_SESSION* session_alloc(SERVICE* service, DCB* client_dcb)
{
return new(std::nothrow) Session(service, client_dcb);
}
bool session_start(MXS_SESSION* session)
{
session->router_session = session->service->router->newSession(session->service->router_instance,
@ -179,7 +175,6 @@ void session_link_backend_dcb(MXS_SESSION* session, DCB* dcb)
mxb::atomic::add(&session->refcount, 1);
dcb->session = session;
dcb->service = session->service;
/** Move this DCB under the same thread */
dcb->owner = session->client_dcb->owner;
@ -1097,10 +1092,10 @@ const char* session_get_close_reason(const MXS_SESSION* session)
}
}
Session::Session(SERVICE* service, DCB* client_dcb)
: MXS_SESSION(client_dcb)
Session::Session(const SListener& listener)
: MXS_SESSION(listener)
{
if (service->retain_last_statements != -1) // Explicitly set for the service
if (service->retain_last_statements != -1) // Explicitly set for the service
{
m_retain_last_statements = service->retain_last_statements;
}
@ -1124,6 +1119,13 @@ Session::~Session()
}
}
void Session::set_client_dcb(DCB* dcb)
{
mxb_assert(client_dcb == nullptr);
mxb_assert(dcb->dcb_role == DCB_ROLE_CLIENT_HANDLER);
client_dcb = dcb;
}
namespace
{
@ -1172,7 +1174,6 @@ bool get_cmd_and_stmt(GWBUF* pBuffer, const char** ppCmd, char** ppStmt, int* pL
return deallocate;
}
}
void Session::dump_statements() const

12
server/core/ssl.cc
View File

@ -54,7 +54,7 @@ int ssl_authenticate_client(DCB* dcb, bool is_capable)
const char* remote = dcb->remote ? dcb->remote : "";
const char* service = (dcb->service && dcb->service->name) ? dcb->service->name : "";
if (NULL == dcb->listener || NULL == dcb->listener->ssl())
if (NULL == dcb->session->listener || NULL == dcb->session->listener->ssl())
{
/* Not an SSL connection on account of listener configuration */
return SSL_AUTH_CHECKS_OK;
@ -133,8 +133,8 @@ bool ssl_is_connection_healthy(DCB* dcb)
* then everything is as we wish. Otherwise, either there is a problem or
* more to be done.
*/
return NULL == dcb->listener
|| NULL == dcb->listener->ssl()
return NULL == dcb->session->listener
|| NULL == dcb->session->listener->ssl()
|| dcb->ssl_state == SSL_ESTABLISHED;
}
@ -171,7 +171,7 @@ bool ssl_check_data_to_process(DCB* dcb)
*/
bool ssl_required_by_dcb(DCB* dcb)
{
return NULL != dcb->listener && NULL != dcb->listener->ssl();
return NULL != dcb->session->listener && NULL != dcb->session->listener->ssl();
}
/**
@ -186,8 +186,8 @@ bool ssl_required_by_dcb(DCB* dcb)
*/
bool ssl_required_but_not_negotiated(DCB* dcb)
{
return NULL != dcb->listener
&& NULL != dcb->listener->ssl()
return NULL != dcb->session->listener
&& NULL != dcb->session->listener->ssl()
&& SSL_HANDSHAKE_UNKNOWN == dcb->ssl_state;
}

2
server/core/test/test_dcb.cc
View File

@ -50,7 +50,7 @@ static int test1()
DCB* dcb;
/* Single buffer tests */
fprintf(stderr, "testdcb : creating buffer with type DCB_ROLE_INTERNAL");
dcb = dcb_alloc(DCB_ROLE_INTERNAL, nullptr, nullptr);
dcb = dcb_alloc(DCB_ROLE_INTERNAL, nullptr);
printDCB(dcb);
fprintf(stderr, "\t..done\nAllocated dcb.");
// TODO: Without running workers, the following will hang. As it does not

8
server/core/test/test_poll.cc
View File

@ -49,15 +49,12 @@ static int test1()
DCB* dcb;
int eno = 0;
SERVICE service;
service.routerModule = (char*)"required by a check in dcb.cc";
/* Poll tests */
fprintf(stderr,
"testpoll : Initialise the polling system.");
init_test_env(NULL);
fprintf(stderr, "\t..done\nAdd a DCB");
dcb = dcb_alloc(DCB_ROLE_CLIENT_HANDLER, nullptr, nullptr);
dcb = dcb_alloc(DCB_ROLE_CLIENT_HANDLER, nullptr);
if (dcb == NULL)
{
@ -66,7 +63,6 @@ static int test1()
}
dcb->fd = socket(AF_UNIX, SOCK_STREAM, 0);
dcb->service = &service;
if (dcb->fd < 0)
{
@ -101,8 +97,6 @@ static int test1()
sleep(10);
// TODO, fix this for workers: poll_shutdown();
fprintf(stderr, "\t..done\nTidy up.");
SERVICE my_service = {};
dcb->service = &my_service;
dcb_close(dcb);
fprintf(stderr, "\t..done\n");

7
server/modules/authenticator/CDCPlainAuth/cdc_plain_auth.cc
View File

@ -219,7 +219,7 @@ static int cdc_auth_check(DCB* dcb,
{
int rval = CDC_STATE_AUTH_FAILED;
if (dcb->listener->users())
if (dcb->session->listener->users())
{
/* compute SHA1 of auth_data */
uint8_t sha1_step1[SHA_DIGEST_LENGTH] = "";
@ -228,7 +228,7 @@ static int cdc_auth_check(DCB* dcb,
gw_sha1_str(auth_data, SHA_DIGEST_LENGTH, sha1_step1);
gw_bin2hex(hex_step1, sha1_step1, SHA_DIGEST_LENGTH);
if (users_auth(dcb->listener->users(), username, hex_step1))
if (users_auth(dcb->session->listener->users(), username, hex_step1))
{
rval = CDC_STATE_AUTH_OK;
}
@ -263,7 +263,8 @@ static int cdc_auth_authenticate(DCB* dcb)
cdc_auth_check(dcb, protocol, client_data->user, client_data->auth_data, client_data->flags);
/* On failed authentication try to reload users and authenticate again */
if (CDC_STATE_AUTH_OK != auth_ret && cdc_replace_users(dcb->listener.get()) == MXS_AUTH_LOADUSERS_OK)
if (CDC_STATE_AUTH_OK != auth_ret
&& cdc_replace_users(dcb->session->listener.get()) == MXS_AUTH_LOADUSERS_OK)
{
auth_ret = cdc_auth_check(dcb,
protocol,

2
server/modules/authenticator/GSSAPI/GSSAPIAuth/gssapi_auth.cc
View File

@ -491,7 +491,7 @@ int gssapi_auth_authenticate(DCB* dcb)
{
int rval = MXS_AUTH_FAILED;
gssapi_auth_t* auth = (gssapi_auth_t*)dcb->authenticator_data;
GSSAPI_INSTANCE* instance = (GSSAPI_INSTANCE*)dcb->listener->auth_instance();
GSSAPI_INSTANCE* instance = (GSSAPI_INSTANCE*)dcb->session->listener->auth_instance();
if (auth->state == GSSAPI_AUTH_INIT)
{

4
server/modules/authenticator/MySQLAuth/mysql_auth.cc
View File

@ -293,7 +293,7 @@ static int mysql_auth_authenticate(DCB* dcb)
client_data->user,
client_data->db);
MYSQL_AUTH* instance = (MYSQL_AUTH*)dcb->listener->auth_instance();
MYSQL_AUTH* instance = (MYSQL_AUTH*)dcb->session->listener->auth_instance();
MySQLProtocol* protocol = DCB_PROTOCOL(dcb, MySQLProtocol);
auth_ret = validate_mysql_user(instance,
dcb,
@ -684,7 +684,7 @@ int mysql_auth_reauthenticate(DCB* dcb,
temp.auth_token = token;
temp.auth_token_len = token_len;
MYSQL_AUTH* instance = (MYSQL_AUTH*)dcb->listener->auth_instance();
MYSQL_AUTH* instance = (MYSQL_AUTH*)dcb->session->listener->auth_instance();
int rc = validate_mysql_user(instance, dcb, &temp, scramble, scramble_len);
if (rc != MXS_AUTH_SUCCEEDED && service_refresh_users(dcb->service) == 0)

2
server/modules/filter/test/mock_backend.cc
View File

@ -174,7 +174,7 @@ class ResultSetDCB : public DCB
{
public:
ResultSetDCB()
: DCB(DCB_ROLE_CLIENT_HANDLER, nullptr, nullptr)
: DCB(DCB_ROLE_CLIENT_HANDLER, nullptr)
{
DCB* pDcb = this;

2
server/modules/filter/test/mock_dcb.cc
View File

@ -34,7 +34,7 @@ Dcb::Dcb(MXS_SESSION* pSession,
const char* zUser,
const char* zHost,
Handler* pHandler)
: DCB(DCB_ROLE_CLIENT_HANDLER, nullptr, nullptr)
: DCB(DCB_ROLE_CLIENT_HANDLER, nullptr)
, m_user(zUser)
, m_host(zHost)
, m_pHandler(pHandler)

3
server/modules/filter/test/mock_session.cc
View File

@ -17,7 +17,6 @@ namespace
{
SERVICE dummy_service;
}
namespace maxscale
@ -27,7 +26,7 @@ namespace mock
{
Session::Session(Client* pClient)
: mxs::Session(&dummy_service, &m_client_dcb)
: mxs::Session(nullptr)
, m_client(*pClient)
, m_client_dcb(this, pClient->user(), pClient->host(), pClient)
{

2
server/modules/protocol/HTTPD/httpd.cc
View File

@ -204,7 +204,7 @@ static int httpd_read_event(DCB* dcb)
/** If listener->authenticator is the default authenticator, it means that
* we don't need to check the user credentials. All other authenticators
* cause a 401 Unauthorized to be returned on the first try. */
bool auth_ok = httpd_default_auth() == std::string(dcb->listener->authenticator());
bool auth_ok = httpd_default_auth() == std::string(dcb->session->listener->authenticator());
/**
* Get the request headers

8
server/modules/protocol/MySQL/mariadbclient/mysql_client.cc
View File

@ -673,7 +673,7 @@ static void check_packet(DCB* dcb, GWBUF* buf, int bytes)
if (bytes == MYSQL_AUTH_PACKET_BASE_SIZE)
{
/** This is an SSL request packet */
mxb_assert(dcb->listener->ssl());
mxb_assert(dcb->session->listener->ssl());
mxb_assert(buflen == bytes && pktlen >= buflen);
}
else
@ -1463,10 +1463,8 @@ static int gw_client_close(DCB* dcb)
{
MXS_SESSION* target = dcb->session;
if (target->state != SESSION_STATE_TO_BE_FREED)
if (target->state == SESSION_STATE_ROUTER_READY || target->state == SESSION_STATE_STOPPING)
{
mxb_assert(target->state == SESSION_STATE_ROUTER_READY
|| target->state == SESSION_STATE_STOPPING);
MXB_AT_DEBUG(bool removed = ) mxs_rworker_deregister_session(target->ses_id);
mxb_assert(removed);
session_close(target);
@ -1648,7 +1646,7 @@ static int route_by_statement(MXS_SESSION* session, uint64_t capabilities, GWBUF
if (rcap_type_required(capabilities, RCAP_TYPE_CONTIGUOUS_INPUT))
{
mxb_assert(GWBUF_IS_CONTIGUOUS(packetbuf));
SERVICE* service = session->client_dcb->service;
SERVICE* service = session->service;
if (rcap_type_required(capabilities, RCAP_TYPE_TRANSACTION_TRACKING)
&& !service->session_track_trx_state

10
server/modules/routing/binlogrouter/blr_master.cc
View File

@ -55,6 +55,8 @@
#include <maxscale/session.hh>
#include <maxscale/utils.h>
#include "../../../core/internal/session.hh"
static GWBUF* blr_make_query(DCB* dcb, char* query);
static GWBUF* blr_make_registration(ROUTER_INSTANCE* router);
static GWBUF* blr_make_binlog_dump(ROUTER_INSTANCE* router);
@ -186,7 +188,8 @@ static void blr_start_master(void* data)
pthread_mutex_unlock(&router->lock);
DCB* client = dcb_alloc(DCB_ROLE_INTERNAL, NULL, NULL);
// TODO: Fix this
DCB* client = dcb_alloc(DCB_ROLE_INTERNAL, NULL);
/* Create fake 'client' DCB */
if (client == NULL)
@ -201,7 +204,8 @@ static void blr_start_master(void* data)
/* Create MySQL Athentication from configured user/passwd */
client->data = CreateMySQLAuthData(router->user, router->password, "");
client->session = session_alloc(router->service, client);
// TODO: Fix this
client->session = new mxs::Session(nullptr);
router->session = client->session;
/* Create a session for dummy client DCB */
@ -210,7 +214,6 @@ static void blr_start_master(void* data)
MXS_ERROR("failed to create session for connection to master");
return;
}
client->service = router->service;
/**
* 'client' is the fake DCB that emulates a client session:
@ -237,7 +240,6 @@ static void blr_start_master(void* data)
return;
}
router->master->remote = MXS_STRDUP_A(router->service->dbref->server->address);
router->master->service = router->service;
MXS_NOTICE("%s: attempting to connect to master"
" server [%s]:%d, binlog='%s', pos=%lu%s%s",

1
server/modules/routing/hintrouter/hintrouter.cc
View File

@ -205,7 +205,6 @@ Dcb HintRouter::connect_to_backend(MXS_SESSION* session,
{
HR_DEBUG("Connected.");
mxb::atomic::add(&sref->connections, 1, mxb::atomic::RELAXED);
new_connection->service = session->service;
result = Dcb(new_connection);
string name(new_connection->server->name);