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

Merge branch 'develop' into MXS-1075

Browse Source
This commit is contained in:
MassimilianoPinto
2017-03-07 16:11:18 +01:00
parent 5ea049cffc f18a40ce73
commit 8bacc91153
29 changed files with 878 additions and 530 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
server/modules/routing/CMakeLists.txt
View File

@ -7,6 +7,7 @@ endif()
add_subdirectory(cli)
add_subdirectory(debugcli)
add_subdirectory(hintrouter)
add_subdirectory(maxinfo)
add_subdirectory(readconnroute)
add_subdirectory(readwritesplit)

14
server/modules/routing/avrorouter/avro.c
View File

@ -193,6 +193,7 @@ MXS_MODULE* MXS_CREATE_MODULE()
{"group_rows", MXS_MODULE_PARAM_COUNT, "1000"},
{"group_trx", MXS_MODULE_PARAM_COUNT, "1"},
{"start_index", MXS_MODULE_PARAM_COUNT, "1"},
{"block_size", MXS_MODULE_PARAM_COUNT, "0"},
{"codec", MXS_MODULE_PARAM_ENUM, "null", MXS_MODULE_OPT_ENUM_UNIQUE, codec_values},
{MXS_END_MODULE_PARAMS}
}
@ -416,6 +417,7 @@ createInstance(SERVICE *service, char **options)
inst->trx_target = config_get_integer(params, "group_trx");
inst->codec = config_get_enum(params, "codec", codec_values);
int first_file = config_get_integer(params, "start_index");
inst->block_size = config_get_integer(params, "block_size");
MXS_CONFIG_PARAMETER *param = config_get_param(params, "source");
bool err = false;
@ -490,6 +492,10 @@ createInstance(SERVICE *service, char **options)
{
first_file = MXS_MAX(1, atoi(value));
}
else if (strcmp(options[i], "block_size") == 0)
{
inst->block_size = atoi(value);
}
else
{
MXS_WARNING("Unknown router option: '%s'", options[i]);
@ -1065,14 +1071,20 @@ void converter_func(void* data)
while (!router->service->svc_do_shutdown && ok && binlog_end == AVRO_OK)
{
uint64_t start_pos = router->current_pos;
char binlog_name[BINLOG_FNAMELEN + 1];
strcpy(binlog_name, router->binlog_name);
if (avro_open_binlog(router->binlogdir, router->binlog_name, &router->binlog_fd))
{
binlog_end = avro_read_all_events(router);
if (router->current_pos != start_pos)
if (router->current_pos != start_pos || strcmp(binlog_name, router->binlog_name) != 0)
{
/** We processed some data, reset the conversion task delay */
router->task_delay = 1;
/** Update the GTID index */
avro_update_index(router);
}
avro_close_binlog(router->binlog_fd);

10
server/modules/routing/avrorouter/avro_file.c
View File

@ -106,7 +106,7 @@ void avro_close_binlog(int fd)
* @param filepath Path to the created file
* @param json_schema The schema of the table in JSON format
*/
AVRO_TABLE* avro_table_alloc(const char* filepath, const char* json_schema, const char *codec)
AVRO_TABLE* avro_table_alloc(const char* filepath, const char* json_schema, const char *codec, size_t block_size)
{
AVRO_TABLE *table = MXS_CALLOC(1, sizeof(AVRO_TABLE));
if (table)
@ -128,7 +128,7 @@ AVRO_TABLE* avro_table_alloc(const char* filepath, const char* json_schema, cons
else
{
rc = avro_file_writer_create_with_codec(filepath, table->avro_schema,
&table->avro_file, codec, 0);
&table->avro_file, codec, block_size);
}
if (rc)
@ -884,12 +884,6 @@ void avro_flush_all_tables(AVRO_INSTANCE *router, enum avrorouter_file_op flush)
}
hashtable_iterator_free(iter);
}
/** Update the GTID index */
if (flush == AVROROUTER_FLUSH)
{
avro_update_index(router);
}
}
/**

43
server/modules/routing/avrorouter/avro_rbr.c
View File

@ -122,7 +122,8 @@ bool handle_table_map_event(AVRO_INSTANCE *router, REP_HEADER *hdr, uint8_t *ptr
/** Close the file and open a new one */
hashtable_delete(router->open_tables, table_ident);
AVRO_TABLE *avro_table = avro_table_alloc(filepath, json_schema,
codec_to_string(router->codec));
codec_to_string(router->codec),
router->block_size);
if (avro_table)
{
@ -306,14 +307,19 @@ bool handle_row_event(AVRO_INSTANCE *router, REP_HEADER *hdr, uint8_t *ptr)
* beforehand so we must continue processing them until we reach the end
* of the event. */
int rows = 0;
while (ptr - start < hdr->event_size - BINLOG_EVENT_HDR_LEN)
{
/** Add the current GTID and timestamp */
uint8_t *end = ptr + hdr->event_size;
uint8_t *end = ptr + hdr->event_size - BINLOG_EVENT_HDR_LEN;
int event_type = get_event_type(hdr->event_type);
prepare_record(router, hdr, event_type, &record);
ptr = process_row_event_data(map, create, &record, ptr, col_present, end);
avro_file_writer_append_value(table->avro_file, &record);
if (avro_file_writer_append_value(table->avro_file, &record))
{
MXS_ERROR("Failed to write value at position %ld: %s",
router->current_pos, avro_strerror());
}
/** Update rows events have the before and after images of the
* affected rows so we'll process them as another record with
@ -322,7 +328,11 @@ bool handle_row_event(AVRO_INSTANCE *router, REP_HEADER *hdr, uint8_t *ptr)
{
prepare_record(router, hdr, UPDATE_EVENT_AFTER, &record);
ptr = process_row_event_data(map, create, &record, ptr, col_present, end);
avro_file_writer_append_value(table->avro_file, &record);
if (avro_file_writer_append_value(table->avro_file, &record))
{
MXS_ERROR("Failed to write value at position %ld: %s",
router->current_pos, avro_strerror());
}
}
rows++;
@ -518,14 +528,23 @@ uint8_t* process_row_event_data(TABLE_MAP *map, TABLE_CREATE *create, avro_value
for (long i = 0; i < map->columns && npresent < ncolumns; i++)
{
ss_dassert(create->columns == map->columns);
avro_value_get_by_name(record, create->column_names[i], &field, NULL);
ss_debug(int rc = )avro_value_get_by_name(record, create->column_names[i], &field, NULL);
ss_dassert(rc == 0);
if (bit_is_set(columns_present, ncolumns, i))
{
npresent++;
if (bit_is_set(null_bitmap, ncolumns, i))
{
avro_value_set_null(&field);
if (column_is_blob(map->column_types[i]))
{
uint8_t nullvalue = 0;
avro_value_set_bytes(&field, &nullvalue, 1);
}
else
{
avro_value_set_null(&field);
}
}
else if (column_is_fixed_string(map->column_types[i]))
{
@ -614,8 +633,16 @@ uint8_t* process_row_event_data(TABLE_MAP *map, TABLE_CREATE *create, avro_value
uint64_t len = 0;
memcpy(&len, ptr, bytes);
ptr += bytes;
avro_value_set_bytes(&field, ptr, len);
ptr += len;
if (len)
{
avro_value_set_bytes(&field, ptr, len);
ptr += len;
}
else
{
uint8_t nullvalue = 0;
avro_value_set_bytes(&field, &nullvalue, 1);
}
ss_dassert(ptr < end);
}
else if (column_is_temporal(map->column_types[i]))

4
server/modules/routing/avrorouter/avrorouter.h
View File

@ -281,6 +281,7 @@ typedef struct avro_instance
uint64_t row_count; /*< Row events processed */
uint64_t row_target; /*< Minimum about of row events that will trigger
* a flush of all tables */
uint64_t block_size; /**< Avro datablock size */
enum mxs_avro_codec_type codec; /**< Avro codec type, defaults to `null` */
struct avro_instance *next;
} AVRO_INSTANCE;
@ -299,7 +300,8 @@ extern void avro_client_rotate(AVRO_INSTANCE *router, AVRO_CLIENT *client, uint8
extern bool avro_open_binlog(const char *binlogdir, const char *file, int *fd);
extern void avro_close_binlog(int fd);
extern avro_binlog_end_t avro_read_all_events(AVRO_INSTANCE *router);
extern AVRO_TABLE* avro_table_alloc(const char* filepath, const char* json_schema, const char *codec);
extern AVRO_TABLE* avro_table_alloc(const char* filepath, const char* json_schema,
const char *codec, size_t block_size);
extern void avro_table_free(AVRO_TABLE *table);
extern char* json_new_schema_from_table(TABLE_MAP *map);
extern void save_avro_schema(const char *path, const char* schema, TABLE_MAP *map);

71
server/modules/routing/binlogrouter/blr.c
View File

@ -1784,6 +1784,7 @@ errorReply(MXS_ROUTER *instance,
mxs_error_action_t action,
bool *succp)
{
ss_dassert(backend_dcb->dcb_role == DCB_ROLE_BACKEND_HANDLER);
ROUTER_INSTANCE *router = (ROUTER_INSTANCE *)instance;
int error;
socklen_t len;
@ -1794,54 +1795,46 @@ errorReply(MXS_ROUTER *instance,
mysql_errno = (unsigned long) extract_field(((uint8_t *)GWBUF_DATA(message) + 5), 16);
errmsg = extract_message(message);
/** Don't handle same error twice on same DCB */
if (backend_dcb->dcb_errhandle_called)
/** Check router state and set errno an message */
if (router->master_state < BLRM_BINLOGDUMP || router->master_state != BLRM_SLAVE_STOPPED)
{
/** Check router state and set errno an message */
if (router->master_state < BLRM_BINLOGDUMP || router->master_state != BLRM_SLAVE_STOPPED)
/* Authentication failed */
if (router->master_state == BLRM_TIMESTAMP)
{
/* Authentication failed */
if (router->master_state == BLRM_TIMESTAMP)
spinlock_acquire(&router->lock);
/* set io error message */
if (router->m_errmsg)
{
spinlock_acquire(&router->lock);
/* set io error message */
if (router->m_errmsg)
{
free(router->m_errmsg);
}
router->m_errmsg = mxs_strdup("#28000 Authentication with master server failed");
/* set mysql_errno */
router->m_errno = 1045;
/* Stop replication */
router->master_state = BLRM_SLAVE_STOPPED;
spinlock_release(&router->lock);
/* Force backend DCB close */
dcb_close(backend_dcb);
MXS_ERROR("%s: Master connection error %lu '%s' in state '%s', "
"%s while connecting to master %s:%d",
router->service->name, router->m_errno, router->m_errmsg,
blrm_states[BLRM_TIMESTAMP], msg,
router->service->dbref->server->name,
router->service->dbref->server->port);
free(router->m_errmsg);
}
}
if (errmsg)
{
free(errmsg);
}
router->m_errmsg = mxs_strdup("#28000 Authentication with master server failed");
/* set mysql_errno */
router->m_errno = 1045;
/** we optimistically assume that previous call succeed */
*succp = true;
return;
/* Stop replication */
router->master_state = BLRM_SLAVE_STOPPED;
spinlock_release(&router->lock);
/* Force backend DCB close */
dcb_close(backend_dcb);
MXS_ERROR("%s: Master connection error %lu '%s' in state '%s', "
"%s while connecting to master %s:%d",
router->service->name, router->m_errno, router->m_errmsg,
blrm_states[BLRM_TIMESTAMP], msg,
router->service->dbref->server->name,
router->service->dbref->server->port);
}
}
else
if (errmsg)
{
backend_dcb->dcb_errhandle_called = true;
free(errmsg);
}
/** we optimistically assume that previous call succeed */
*succp = true;
return;
len = sizeof(error);
if (router->master &&
getsockopt(router->master->fd, SOL_SOCKET, SO_ERROR, &error, &len) == 0 &&

8
server/modules/routing/hintrouter/CMakeLists.txt Normal file
View File

@ -0,0 +1,8 @@
add_library(hintrouter SHARED
hintrouter.cc
hintroutersession.cc
)
target_link_libraries(hintrouter maxscale-common)
set_target_properties(hintrouter PROPERTIES VERSION "1.0.0")
install_module(hintrouter core)

68
server/modules/routing/hintrouter/hintrouter.cc Normal file
View File

@ -0,0 +1,68 @@
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2019-07-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#define MXS_MODULE_NAME "hintrouter"
#include "hintrouter.hh"
#include <maxscale/log_manager.h>
HintRouter::HintRouter(SERVICE* pService)
: maxscale::Router<HintRouter, HintRouterSession>(pService)
{
MXS_NOTICE("Hint router [%s] created.", pService->name);
}
//static
HintRouter* HintRouter::create(SERVICE* pService, char** pzOptions)
{
return new HintRouter(pService);
}
HintRouterSession* HintRouter::newSession(MXS_SESSION *pSession)
{
return new HintRouterSession(pSession, this);
}
void HintRouter::diagnostics(DCB* pOut)
{
}
uint64_t HintRouter::getCapabilities()
{
return 0;
}
extern "C" MXS_MODULE* MXS_CREATE_MODULE()
{
static MXS_MODULE module =
{
MXS_MODULE_API_ROUTER, /* Module type */
MXS_MODULE_BETA_RELEASE, /* Release status */
MXS_ROUTER_VERSION, /* Implemented module API version */
"A hint router", /* Description */
"V1.0.0", /* Module version */
&HintRouter::s_object,
NULL, /* Process init, can be null */
NULL, /* Process finish, can be null */
NULL, /* Thread init */
NULL, /* Thread finish */
{
{MXS_END_MODULE_PARAMS}
}
};
return &module;
}

36
server/modules/routing/hintrouter/hintrouter.hh Normal file
View File

@ -0,0 +1,36 @@
#pragma once
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2019-07-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#include <maxscale/cppdefs.hh>
#include <maxscale/router.hh>
#include "hintroutersession.hh"
class HintRouter : public maxscale::Router<HintRouter, HintRouterSession>
{
public:
static HintRouter* create(SERVICE* pService, char** pzOptions);
HintRouterSession* newSession(MXS_SESSION *pSession);
void diagnostics(DCB* pOut);
uint64_t getCapabilities();
private:
HintRouter(SERVICE* pService);
private:
HintRouter(const HintRouter&);
HintRouter& operator = (const HintRouter&);
};

56
server/modules/routing/hintrouter/hintroutersession.cc Normal file
View File

@ -0,0 +1,56 @@
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2019-07-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#define MXS_MODULE_NAME "hintrouter"
#include "hintroutersession.hh"
#include <maxscale/log_manager.h>
HintRouterSession::HintRouterSession(MXS_SESSION* pSession, HintRouter* pRouter)
: maxscale::RouterSession(pSession)
, m_pRouter(pRouter)
{
}
HintRouterSession::~HintRouterSession()
{
}
void HintRouterSession::close()
{
}
int32_t HintRouterSession::routeQuery(GWBUF* pPacket)
{
MXS_ERROR("routeQuery not implemented yet.");
return 0;
}
void HintRouterSession::clientReply(GWBUF* pPacket, DCB* pBackend)
{
MXS_ERROR("clientReply not implemented yet.");
}
void HintRouterSession::handleError(GWBUF* pMessage,
DCB* pProblem,
mxs_error_action_t action,
bool* pSuccess)
{
ss_dassert(pProblem->dcb_role == DCB_ROLE_BACKEND_HANDLER);
MXS_ERROR("handleError not implemented yet.");
}

45
server/modules/routing/hintrouter/hintroutersession.hh Normal file
View File

@ -0,0 +1,45 @@
#pragma once
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2019-07-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#include <maxscale/cppdefs.hh>
#include <maxscale/router.hh>
class HintRouter;
class HintRouterSession : public maxscale::RouterSession
{
public:
HintRouterSession(MXS_SESSION* pSession,
HintRouter* pRouter);
~HintRouterSession();
void close();
int32_t routeQuery(GWBUF* pPacket);
void clientReply(GWBUF* pPacket, DCB* pBackend);
void handleError(GWBUF* pMessage,
DCB* pProblem,
mxs_error_action_t action,
bool* pSuccess);
private:
HintRouterSession(const HintRouterSession&);
HintRouterSession& operator = (const HintRouterSession&);
private:
HintRouter* m_pRouter;
};

17
server/modules/routing/maxinfo/maxinfo.c
View File

@ -276,26 +276,13 @@ static void handleError(MXS_ROUTER *instance,
bool *succp)
{
ss_dassert(backend_dcb->dcb_role == DCB_ROLE_BACKEND_HANDLER);
DCB *client_dcb;
MXS_SESSION *session = backend_dcb->session;
mxs_session_state_t sesstate;
/** Don't handle same error twice on same DCB */
if (backend_dcb->dcb_errhandle_called)
{
/** we optimistically assume that previous call succeed */
*succp = true;
return;
}
else
{
backend_dcb->dcb_errhandle_called = true;
}
sesstate = session->state;
client_dcb = session->client_dcb;
if (sesstate == SESSION_STATE_ROUTER_READY)
if (session->state == SESSION_STATE_ROUTER_READY)
{
CHK_DCB(client_dcb);
client_dcb->func.write(client_dcb, gwbuf_clone(errbuf));

19
server/modules/routing/readconnroute/readconnroute.c
View File

@ -679,23 +679,12 @@ static void handleError(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session
DCB *problem_dcb, mxs_error_action_t action, bool *succp)
{
ss_dassert(problem_dcb->dcb_role == DCB_ROLE_BACKEND_HANDLER);
DCB *client_dcb;
MXS_SESSION *session = problem_dcb->session;
mxs_session_state_t sesstate;
ROUTER_CLIENT_SES *router_cli_ses = (ROUTER_CLIENT_SES *) router_session;
/** Don't handle same error twice on same DCB */
if (problem_dcb->dcb_errhandle_called)
{
/** we optimistically assume that previous call succeed */
*succp = true;
return;
}
else
{
problem_dcb->dcb_errhandle_called = true;
}
sesstate = session->state;
client_dcb = session->client_dcb;
@ -706,11 +695,7 @@ static void handleError(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session
client_dcb->func.write(client_dcb, gwbuf_clone(errbuf));
}
if (DCB_ROLE_CLIENT_HANDLER == problem_dcb->dcb_role)
{
dcb_close(problem_dcb);
}
else if (router_cli_ses && problem_dcb == router_cli_ses->backend_dcb)
if (router_cli_ses && problem_dcb == router_cli_ses->backend_dcb)
{
router_cli_ses->backend_dcb = NULL;
dcb_close(problem_dcb);

207
server/modules/routing/readwritesplit/readwritesplit.c
View File

@ -1180,6 +1180,7 @@ static void handleError(MXS_ROUTER *instance,
mxs_error_action_t action,
bool *succp)
{
ss_dassert(problem_dcb->dcb_role == DCB_ROLE_BACKEND_HANDLER);
ROUTER_INSTANCE *inst = (ROUTER_INSTANCE *)instance;
ROUTER_CLIENT_SES *rses = (ROUTER_CLIENT_SES *)router_session;
CHK_CLIENT_RSES(rses);
@ -1187,150 +1188,124 @@ static void handleError(MXS_ROUTER *instance,
if (rses->rses_closed)
{
/** Session is already closed */
problem_dcb->dcb_errhandle_called = true;
*succp = false;
return;
}
/** Don't handle same error twice on same DCB */
if (problem_dcb->dcb_errhandle_called)
{
/** we optimistically assume that previous call succeed */
/*
* The return of true is potentially misleading, but appears to
* be safe with the code as it stands on 9 Sept 2015 - MNB
*/
*succp = true;
return;
}
else
{
problem_dcb->dcb_errhandle_called = true;
}
MXS_SESSION *session = problem_dcb->session;
ss_dassert(session);
if (problem_dcb->dcb_role == DCB_ROLE_CLIENT_HANDLER)
{
dcb_close(problem_dcb);
*succp = false;
}
else
{
backend_ref_t *bref = get_bref_from_dcb(rses, problem_dcb);
backend_ref_t *bref = get_bref_from_dcb(rses, problem_dcb);
switch (action)
{
switch (action)
{
case ERRACT_NEW_CONNECTION:
{
/**
* If master has lost its Master status error can't be
* handled so that session could continue.
*/
if (rses->rses_master_ref && rses->rses_master_ref->bref_dcb == problem_dcb)
{
/**
* If master has lost its Master status error can't be
* handled so that session could continue.
*/
if (rses->rses_master_ref && rses->rses_master_ref->bref_dcb == problem_dcb)
SERVER *srv = rses->rses_master_ref->ref->server;
bool can_continue = false;
if (rses->rses_config.master_failure_mode != RW_FAIL_INSTANTLY &&
(bref == NULL || !BREF_IS_WAITING_RESULT(bref)))
{
SERVER *srv = rses->rses_master_ref->ref->server;
bool can_continue = false;
if (rses->rses_config.master_failure_mode != RW_FAIL_INSTANTLY &&
(bref == NULL || !BREF_IS_WAITING_RESULT(bref)))
{
/** The failure of a master is not considered a critical
* failure as partial functionality still remains. Reads
* are allowed as long as slave servers are available
* and writes will cause an error to be returned.
*
* If we were waiting for a response from the master, we
* can't be sure whether it was executed or not. In this
* case the safest thing to do is to close the client
* connection. */
can_continue = true;
}
else if (!SERVER_IS_MASTER(srv) && !srv->master_err_is_logged)
{
MXS_ERROR("Server %s:%d lost the master status. Readwritesplit "
"service can't locate the master. Client sessions "
"will be closed.", srv->name, srv->port);
srv->master_err_is_logged = true;
}
*succp = can_continue;
if (bref != NULL)
{
CHK_BACKEND_REF(bref);
RW_CHK_DCB(bref, problem_dcb);
dcb_close(problem_dcb);
RW_CLOSE_BREF(bref);
close_failed_bref(bref, true);
}
else
{
MXS_ERROR("Server %s:%d lost the master status but could not locate the "
"corresponding backend ref.", srv->name, srv->port);
}
/** The failure of a master is not considered a critical
* failure as partial functionality still remains. Reads
* are allowed as long as slave servers are available
* and writes will cause an error to be returned.
*
* If we were waiting for a response from the master, we
* can't be sure whether it was executed or not. In this
* case the safest thing to do is to close the client
* connection. */
can_continue = true;
}
else if (bref)
else if (!SERVER_IS_MASTER(srv) && !srv->master_err_is_logged)
{
/** Check whether problem_dcb is same as dcb of rses->forced_node
* and within READ ONLY transaction:
* if true reset rses->forced_node and close session
*/
if (rses->forced_node &&
(rses->forced_node->bref_dcb == problem_dcb &&
session_trx_is_read_only(problem_dcb->session)))
{
MXS_ERROR("forced_node SLAVE %s in opened READ ONLY transaction has failed:"
" closing session",
problem_dcb->server->unique_name);
rses->forced_node = NULL;
*succp = false;
break;
}
/** We should reconnect only if we find a backend for this
* DCB. If this DCB is an older DCB that has been closed,
* we can ignore it. */
*succp = handle_error_new_connection(inst, &rses, problem_dcb, errmsgbuf);
MXS_ERROR("Server %s:%d lost the master status. Readwritesplit "
"service can't locate the master. Client sessions "
"will be closed.", srv->name, srv->port);
srv->master_err_is_logged = true;
}
if (bref)
*succp = can_continue;
if (bref != NULL)
{
/** This is a valid DCB for a backend ref */
if (BREF_IS_IN_USE(bref) && bref->bref_dcb == problem_dcb)
{
ss_dassert(false);
MXS_ERROR("Backend '%s' is still in use and points to the problem DCB.",
bref->ref->server->unique_name);
}
CHK_BACKEND_REF(bref);
RW_CHK_DCB(bref, problem_dcb);
dcb_close(problem_dcb);
RW_CLOSE_BREF(bref);
close_failed_bref(bref, true);
}
else
{
const char *remote = problem_dcb->state == DCB_STATE_POLLING &&
problem_dcb->server ? problem_dcb->server->unique_name : "CLOSED";
MXS_ERROR("DCB connected to '%s' is not in use by the router "
"session, not closing it. DCB is in state '%s'",
remote, STRDCBSTATE(problem_dcb->state));
MXS_ERROR("Server %s:%d lost the master status but could not locate the "
"corresponding backend ref.", srv->name, srv->port);
}
break;
}
else if (bref)
{
/** Check whether problem_dcb is same as dcb of rses->forced_node
* and within READ ONLY transaction:
* if true reset rses->forced_node and close session
*/
if (rses->forced_node &&
(rses->forced_node->bref_dcb == problem_dcb &&
session_trx_is_read_only(problem_dcb->session)))
{
MXS_ERROR("forced_node SLAVE %s in opened READ ONLY transaction has failed:"
" closing session",
problem_dcb->server->unique_name);
rses->forced_node = NULL;
*succp = false;
break;
}
/** We should reconnect only if we find a backend for this
* DCB. If this DCB is an older DCB that has been closed,
* we can ignore it. */
*succp = handle_error_new_connection(inst, &rses, problem_dcb, errmsgbuf);
}
if (bref)
{
/** This is a valid DCB for a backend ref */
if (BREF_IS_IN_USE(bref) && bref->bref_dcb == problem_dcb)
{
ss_dassert(false);
MXS_ERROR("Backend '%s' is still in use and points to the problem DCB.",
bref->ref->server->unique_name);
}
}
else
{
const char *remote = problem_dcb->state == DCB_STATE_POLLING &&
problem_dcb->server ? problem_dcb->server->unique_name : "CLOSED";
MXS_ERROR("DCB connected to '%s' is not in use by the router "
"session, not closing it. DCB is in state '%s'",
remote, STRDCBSTATE(problem_dcb->state));
}
break;
}
case ERRACT_REPLY_CLIENT:
{
handle_error_reply_client(session, rses, problem_dcb, errmsgbuf);
*succp = false; /*< no new backend servers were made available */
break;
}
{
handle_error_reply_client(session, rses, problem_dcb, errmsgbuf);
*succp = false; /*< no new backend servers were made available */
break;
}
default:
ss_dassert(!true);
*succp = false;
break;
}
}
}

16
server/modules/routing/schemarouter/schemarouter.c
View File

@ -3557,33 +3557,19 @@ static void handleError(MXS_ROUTER* instance,
mxs_error_action_t action,
bool* succp)
{
ss_dassert(problem_dcb->dcb_role == DCB_ROLE_BACKEND_HANDLER);
MXS_SESSION* session;
ROUTER_INSTANCE* inst = (ROUTER_INSTANCE *)instance;
ROUTER_CLIENT_SES* rses = (ROUTER_CLIENT_SES *)router_session;
CHK_DCB(problem_dcb);
/** Don't handle same error twice on same DCB */
if (problem_dcb->dcb_errhandle_called)
{
/** we optimistically assume that previous call succeed */
*succp = true;
return;
}
else
{
problem_dcb->dcb_errhandle_called = true;
}
session = problem_dcb->session;
if (session == NULL || rses == NULL)
{
*succp = false;
}
else if (DCB_ROLE_CLIENT_HANDLER == problem_dcb->dcb_role)
{
*succp = false;
}
else
{
CHK_SESSION(session);