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

MXS-2067: Remove spinlock.h

Browse Source 
Removed the spinlock.h header and replaced with plain pthread types and
functions.
This commit is contained in:
Markus Mäkelä 2018-09-26 18:55:10 +03:00
parent c95adf1f2e
commit 9278da1f54
No known key found for this signature in database
GPG Key ID: 72D48FCE664F7B19
50 changed files with 392 additions and 472 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/maxscale/buffer.h
View File

@ -29,7 +29,6 @@
#include <maxscale/cdefs.h>
#include <string.h>
#include <maxscale/hint.h>
#include <maxscale/spinlock.h>
#include <stdint.h>
MXS_BEGIN_DECLS

1
include/maxscale/dcb.h
View File

@ -24,7 +24,6 @@
#include <maxscale/log.h>
#include <maxscale/modinfo.h>
#include <maxscale/protocol.h>
#include <maxscale/spinlock.h>
#include <maxscale/ssl.h>
MXS_BEGIN_DECLS

2
include/maxscale/listener.h
View File

@ -46,7 +46,7 @@ typedef struct servlistener
struct dcb* listener; /**< The DCB for the listener */
struct users* users; /**< The user data for this listener */
struct service* service; /**< The service which used by this listener */
SPINLOCK lock;
pthread_mutex_t lock;
int active; /**< True if the port has not been deleted */
struct servlistener* next; /**< Next service protocol */
} SERV_LISTENER; // TODO: Rename to LISTENER

2
include/maxscale/monitor.h
View File

@ -233,7 +233,7 @@ struct mxs_monitor
char* name; /**< The name of the monitor module */
char user[MAX_MONITOR_USER_LEN]; /*< Monitor username */
char password[MAX_MONITOR_PASSWORD_LEN]; /*< Monitor password */
SPINLOCK lock;
pthread_mutex_t lock;
MXS_CONFIG_PARAMETER* parameters; /*< configuration parameters */
MXS_MONITORED_SERVER* monitored_servers; /*< List of servers the monitor monitors */
monitor_state_t state; /**< The state of the monitor. This should ONLY be

1
include/maxscale/service.h
View File

@ -29,7 +29,6 @@
#include <maxbase/jansson.h>
#include <maxscale/protocol.h>
#include <maxscale/spinlock.h>
#include <maxscale/dcb.h>
#include <maxscale/server.h>
#include <maxscale/listener.h>

1
include/maxscale/session.h
View File

@ -25,7 +25,6 @@
#include <maxscale/dcb.h>
#include <maxscale/buffer.h>
#include <maxscale/log.h>
#include <maxscale/spinlock.h>
MXS_BEGIN_DECLS

39
include/maxscale/spinlock.h
View File

@ -1,39 +0,0 @@
/*
* Copyright (c) 2018 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: 2022-01-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.
*/
#pragma once
#include <maxscale/cdefs.h>
#include <stdbool.h>
#include <pthread.h>
MXS_BEGIN_DECLS
typedef pthread_mutex_t SPINLOCK;
#define SPINLOCK_INIT PTHREAD_MUTEX_INITIALIZER
static inline void spinlock_init(SPINLOCK* a)
{
pthread_mutex_init(a, NULL);
}
static inline void spinlock_acquire(const SPINLOCK* a)
{
pthread_mutex_lock((SPINLOCK*)a);
}
static inline void spinlock_release(const SPINLOCK* a)
{
pthread_mutex_unlock((SPINLOCK*)a);
}
MXS_END_DECLS

1
server/core/buffer.cc
View File

@ -22,7 +22,6 @@
#include <maxscale/alloc.h>
#include <maxscale/hint.h>
#include <maxscale/log.h>
#include <maxscale/spinlock.h>
#include <maxscale/utils.h>
#if defined (BUFFER_TRACE)

1
server/core/config.cc
View File

@ -53,7 +53,6 @@
#include <maxscale/pcre2.h>
#include <maxscale/router.h>
#include <maxscale/secrets.h>
#include <maxscale/spinlock.h>
#include <maxscale/utils.h>
#include <maxscale/utils.hh>
#include <maxscale/version.h>

1
server/core/dcb.cc
View File

@ -49,7 +49,6 @@
#include <maxscale/router.h>
#include <maxscale/server.h>
#include <maxscale/service.h>
#include <maxscale/spinlock.h>
#include <maxscale/utils.h>
#include <maxscale/routingworker.hh>

18
server/core/gateway.cc
View File

@ -235,17 +235,17 @@ const DEBUG_ARGUMENT debug_arguments[] =
#ifndef OPENSSL_1_1
/** SSL multi-threading functions and structures */
static SPINLOCK* ssl_locks;
static pthread_mutex_t* ssl_locks;
static void ssl_locking_function(int mode, int n, const char* file, int line)
{
if (mode & CRYPTO_LOCK)
{
spinlock_acquire(&ssl_locks[n]);
pthread_mutex_lock(&ssl_locks[n]);
}
else
{
spinlock_release(&ssl_locks[n]);
pthread_mutex_unlock(&ssl_locks[n]);
}
}
/**
@ -253,7 +253,7 @@ static void ssl_locking_function(int mode, int n, const char* file, int line)
*/
struct CRYPTO_dynlock_value
{
SPINLOCK lock;
pthread_mutex_t lock;
};
/**
@ -269,7 +269,7 @@ static struct CRYPTO_dynlock_value* ssl_create_dynlock(const char* file, int lin
(struct CRYPTO_dynlock_value*) MXS_MALLOC(sizeof(struct CRYPTO_dynlock_value));
if (lock)
{
spinlock_init(&lock->lock);
pthread_mutex_init(&lock->lock, NULL);
}
return lock;
}
@ -285,11 +285,11 @@ static void ssl_lock_dynlock(int mode, struct CRYPTO_dynlock_value* n, const cha
{
if (mode & CRYPTO_LOCK)
{
spinlock_acquire(&n->lock);
pthread_mutex_lock(&n->lock);
}
else
{
spinlock_release(&n->lock);
pthread_mutex_unlock(&n->lock);
}
}
@ -1866,7 +1866,7 @@ int main(int argc, char** argv)
#ifndef OPENSSL_1_1
numlocks = CRYPTO_num_locks();
if ((ssl_locks = (SPINLOCK*)MXS_MALLOC(sizeof(SPINLOCK) * (numlocks + 1))) == NULL)
if ((ssl_locks = (pthread_mutex_t*)MXS_MALLOC(sizeof(pthread_mutex_t) * (numlocks + 1))) == NULL)
{
rc = MAXSCALE_INTERNALERROR;
goto return_main;
@ -1874,7 +1874,7 @@ int main(int argc, char** argv)
for (int i = 0; i < numlocks + 1; i++)
{
spinlock_init(&ssl_locks[i]);
pthread_mutex_init(&ssl_locks[i], NULL);
}
CRYPTO_set_locking_callback(ssl_locking_function);
CRYPTO_set_dynlock_create_callback(ssl_create_dynlock);

1
server/core/housekeeper.cc
View File

@ -27,7 +27,6 @@
#include <maxscale/housekeeper.h>
#include <maxscale/json_api.h>
#include <maxscale/query_classifier.h>
#include <maxscale/spinlock.h>
/**
* @file housekeeper.cc Provide a mechanism to run periodic tasks

2
server/core/listener.cc
View File

@ -142,7 +142,7 @@ SERV_LISTENER* listener_alloc(struct service* service,
proto->users = NULL;
proto->next = NULL;
proto->auth_instance = auth_instance;
spinlock_init(&proto->lock);
pthread_mutex_init(&proto->lock, NULL);
return proto;
}

1
server/core/modulecmd.cc
View File

@ -18,7 +18,6 @@
#include <maxscale/alloc.h>
#include <maxscale/config.h>
#include <maxscale/pcre2.h>
#include <maxscale/spinlock.h>
#include "internal/filter.hh"
#include "internal/modules.h"

47
server/core/monitor.cc
View File

@ -39,7 +39,6 @@
#include <maxscale/pcre2.h>
#include <maxscale/routingworker.h>
#include <maxscale/secrets.h>
#include <maxscale/spinlock.h>
#include <maxscale/utils.hh>
#include <maxscale/json_api.h>
#include <mysqld_error.h>
@ -146,7 +145,7 @@ MXS_MONITOR* monitor_create(const char* name, const char* module, MXS_CONFIG_PAR
memset(mon->journal_hash, 0, sizeof(mon->journal_hash));
mon->disk_space_threshold = NULL;
mon->disk_space_check_interval = config_get_integer(params, CN_DISK_SPACE_CHECK_INTERVAL);
spinlock_init(&mon->lock);
pthread_mutex_init(&mon->lock, NULL);
for (auto& s : mxs::strtok(config_get_string(params, CN_SERVERS), ","))
{
@ -243,7 +242,7 @@ void monitor_start(MXS_MONITOR* monitor, const MXS_CONFIG_PARAMETER* params)
{
if (monitor)
{
spinlock_acquire(&monitor->lock);
pthread_mutex_lock(&monitor->lock);
// Only start the monitor if it's stopped.
if (monitor->state == MONITOR_STATE_STOPPED)
@ -264,7 +263,7 @@ void monitor_start(MXS_MONITOR* monitor, const MXS_CONFIG_PARAMETER* params)
}
}
spinlock_release(&monitor->lock);
pthread_mutex_unlock(&monitor->lock);
}
}
@ -296,7 +295,7 @@ void monitor_stop(MXS_MONITOR* monitor)
{
if (monitor)
{
spinlock_acquire(&monitor->lock);
pthread_mutex_lock(&monitor->lock);
/** Only stop the monitor if it is running */
if (monitor->state == MONITOR_STATE_RUNNING)
@ -315,7 +314,7 @@ void monitor_stop(MXS_MONITOR* monitor)
}
}
spinlock_release(&monitor->lock);
pthread_mutex_unlock(&monitor->lock);
}
}
@ -381,7 +380,7 @@ bool monitor_add_server(MXS_MONITOR* mon, SERVER* server)
monitor_stop(mon);
}
spinlock_acquire(&mon->lock);
pthread_mutex_lock(&mon->lock);
if (mon->monitored_servers == NULL)
{
@ -396,7 +395,7 @@ bool monitor_add_server(MXS_MONITOR* mon, SERVER* server)
}
ptr->next = db;
}
spinlock_release(&mon->lock);
pthread_mutex_unlock(&mon->lock);
if (old_state == MONITOR_STATE_RUNNING)
{
@ -448,7 +447,7 @@ void monitor_remove_server(MXS_MONITOR* mon, SERVER* server)
monitor_stop(mon);
}
spinlock_acquire(&mon->lock);
pthread_mutex_lock(&mon->lock);
MXS_MONITORED_SERVER* ptr = mon->monitored_servers;
@ -471,7 +470,7 @@ void monitor_remove_server(MXS_MONITOR* mon, SERVER* server)
ptr = ptr->next;
}
}
spinlock_release(&mon->lock);
pthread_mutex_unlock(&mon->lock);
if (ptr)
{
@ -828,7 +827,7 @@ bool check_monitor_permissions(MXS_MONITOR* monitor, const char* query)
*/
void monitor_add_parameters(MXS_MONITOR* monitor, MXS_CONFIG_PARAMETER* params)
{
spinlock_acquire(&monitor->lock);
pthread_mutex_lock(&monitor->lock);
while (params)
{
@ -849,7 +848,7 @@ void monitor_add_parameters(MXS_MONITOR* monitor, MXS_CONFIG_PARAMETER* params)
params = params->next;
}
spinlock_release(&monitor->lock);
pthread_mutex_unlock(&monitor->lock);
}
void monitor_set_parameter(MXS_MONITOR* monitor, const char* key, const char* value)
@ -866,7 +865,7 @@ bool monitor_remove_parameter(MXS_MONITOR* monitor, const char* key)
MXS_CONFIG_PARAMETER* prev = NULL;
bool rval = false;
spinlock_acquire(&monitor->lock);
pthread_mutex_lock(&monitor->lock);
for (MXS_CONFIG_PARAMETER* p = monitor->parameters; p; p = p->next)
{
@ -890,14 +889,14 @@ bool monitor_remove_parameter(MXS_MONITOR* monitor, const char* key)
prev = p;
}
spinlock_release(&monitor->lock);
pthread_mutex_unlock(&monitor->lock);
return rval;
}
void mon_alter_parameter(MXS_MONITOR* monitor, const char* key, const char* value)
{
spinlock_acquire(&monitor->lock);
pthread_mutex_lock(&monitor->lock);
for (MXS_CONFIG_PARAMETER* p = monitor->parameters; p; p = p->next)
{
@ -909,7 +908,7 @@ void mon_alter_parameter(MXS_MONITOR* monitor, const char* key, const char* valu
}
}
spinlock_release(&monitor->lock);
pthread_mutex_unlock(&monitor->lock);
}
/**
@ -1546,7 +1545,7 @@ MXS_MONITOR* monitor_server_in_use(const SERVER* server)
for (MXS_MONITOR* mon = allMonitors; mon && !rval; mon = mon->next)
{
spinlock_acquire(&mon->lock);
pthread_mutex_lock(&mon->lock);
if (mon->active)
{
@ -1559,7 +1558,7 @@ MXS_MONITOR* monitor_server_in_use(const SERVER* server)
}
}
spinlock_release(&mon->lock);
pthread_mutex_unlock(&mon->lock);
}
return rval;
@ -1579,7 +1578,7 @@ static bool create_monitor_config(const MXS_MONITOR* monitor, const char* filena
return false;
}
spinlock_acquire(&monitor->lock);
pthread_mutex_lock((pthread_mutex_t*)&monitor->lock);
dprintf(file, "[%s]\n", monitor->name);
dprintf(file, "%s=monitor\n", CN_TYPE);
@ -1606,7 +1605,7 @@ static bool create_monitor_config(const MXS_MONITOR* monitor, const char* filena
{CN_TYPE, CN_SERVERS},
config_monitor_params,
mod->parameters);
spinlock_release(&monitor->lock);
pthread_mutex_unlock((pthread_mutex_t*)&monitor->lock);
close(file);
return true;
@ -1787,7 +1786,7 @@ json_t* monitor_json_data(const MXS_MONITOR* monitor, const char* host)
{
json_t* rval = json_object();
spinlock_acquire(&monitor->lock);
pthread_mutex_lock((pthread_mutex_t*)&monitor->lock);
json_object_set_new(rval, CN_ID, json_string(monitor->name));
json_object_set_new(rval, CN_TYPE, json_string(CN_MONITORS));
@ -1827,7 +1826,7 @@ json_t* monitor_json_data(const MXS_MONITOR* monitor, const char* host)
json_object_set_new(rel, CN_SERVERS, mon_rel);
}
spinlock_release(&monitor->lock);
pthread_mutex_unlock((pthread_mutex_t*)&monitor->lock);
json_object_set_new(rval, CN_RELATIONSHIPS, rel);
json_object_set_new(rval, CN_ATTRIBUTES, attr);
@ -1874,7 +1873,7 @@ json_t* monitor_relations_to_server(const SERVER* server, const char* host)
for (MXS_MONITOR* mon = allMonitors; mon; mon = mon->next)
{
spinlock_acquire(&mon->lock);
pthread_mutex_lock(&mon->lock);
if (mon->active)
{
@ -1888,7 +1887,7 @@ json_t* monitor_relations_to_server(const SERVER* server, const char* host)
}
}
spinlock_release(&mon->lock);
pthread_mutex_unlock(&mon->lock);
}
guard.unlock();

1
server/core/session.cc
View File

@ -35,7 +35,6 @@
#include <maxscale/poll.h>
#include <maxscale/router.h>
#include <maxscale/service.h>
#include <maxscale/spinlock.h>
#include <maxscale/utils.h>
#include <maxscale/json_api.h>
#include <maxscale/protocol/mysql.h>

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

@ -556,7 +556,7 @@ int cdc_replace_users(SERV_LISTENER* listener)
int i = cdc_read_users(newusers, path);
USERS* oldusers = NULL;
spinlock_acquire(&listener->lock);
pthread_mutex_lock(&listener->lock);
if (i > 0)
{
@ -578,7 +578,7 @@ int cdc_replace_users(SERV_LISTENER* listener)
cdc_set_service_user(listener);
spinlock_release(&listener->lock);
pthread_mutex_unlock(&listener->lock);
if (oldusers)
{

3
server/modules/filter/luafilter/luafilter.cc
View File

@ -49,7 +49,6 @@
#include <maxscale/modutil.h>
#include <maxscale/query_classifier.h>
#include <maxscale/session.h>
#include <maxscale/spinlock.h>
/*
* The filter entry points
@ -249,7 +248,7 @@ void expose_functions(lua_State* state, GWBUF** active_buffer)
*/
static MXS_FILTER* createInstance(const char* name, MXS_CONFIG_PARAMETER* params)
{
LUA_INSTANCE* my_instance = new (std::nothrow) LUA_INSTANCE;
LUA_INSTANCE* my_instance = new( std::nothrow) LUA_INSTANCE;
if (my_instance == NULL)
{

37
server/modules/filter/mqfilter/mqfilter.cc
View File

@ -75,7 +75,6 @@
#include <maxscale/protocol/mysql.h>
#include <maxscale/log.h>
#include <maxscale/query_classifier.h>
#include <maxscale/spinlock.h>
#include <maxscale/session.h>
#include <maxscale/housekeeper.h>
#include <maxscale/alloc.h>
@ -569,8 +568,8 @@ static MXS_FILTER* createInstance(const char* name, MXS_CONFIG_PARAMETER* params
if (my_instance)
{
spinlock_init(&my_instance->rconn_lock);
spinlock_init(&my_instance->msg_lock);
pthread_mutex_init(&my_instance->rconn_lock, NULL);
pthread_mutex_init(&my_instance->msg_lock, NULL);
uid_gen = 0;
if ((my_instance->conn = amqp_new_connection()) == NULL)
@ -680,7 +679,7 @@ int declareQueue(MQ_INSTANCE* my_instance, MQ_SESSION* my_session, char* qname)
int success = 1;
amqp_rpc_reply_t reply;
spinlock_acquire(&my_instance->rconn_lock);
pthread_mutex_lock(&my_instance->rconn_lock);
amqp_queue_declare(my_instance->conn,
my_instance->channel,
@ -709,7 +708,7 @@ int declareQueue(MQ_INSTANCE* my_instance, MQ_SESSION* my_session, char* qname)
success = 0;
MXS_ERROR("Failed to bind queue to exchange.");
}
spinlock_release(&my_instance->rconn_lock);
pthread_mutex_unlock(&my_instance->rconn_lock);
return success;
}
@ -725,7 +724,7 @@ bool sendMessage(void* data)
mqmessage* tmp;
int err_num = AMQP_STATUS_OK;
spinlock_acquire(&instance->rconn_lock);
pthread_mutex_lock(&instance->rconn_lock);
if (instance->conn_stat != AMQP_STATUS_OK)
{
if (difftime(time(NULL), instance->last_rconn) > instance->rconn_intv)
@ -745,7 +744,7 @@ bool sendMessage(void* data)
}
err_num = instance->conn_stat;
}
spinlock_release(&instance->rconn_lock);
pthread_mutex_unlock(&instance->rconn_lock);
if (err_num != AMQP_STATUS_OK)
{
@ -753,17 +752,17 @@ bool sendMessage(void* data)
return true;
}
spinlock_acquire(&instance->msg_lock);
pthread_mutex_lock(&instance->msg_lock);
tmp = instance->messages;
if (tmp == NULL)
{
spinlock_release(&instance->msg_lock);
pthread_mutex_unlock(&instance->msg_lock);
return true;
}
instance->messages = instance->messages->next;
spinlock_release(&instance->msg_lock);
pthread_mutex_unlock(&instance->msg_lock);
while (tmp)
{
@ -776,9 +775,9 @@ bool sendMessage(void* data)
tmp->prop,
amqp_cstring_bytes(tmp->msg));
spinlock_acquire(&instance->rconn_lock);
pthread_mutex_lock(&instance->rconn_lock);
instance->conn_stat = err_num;
spinlock_release(&instance->rconn_lock);
pthread_mutex_unlock(&instance->rconn_lock);
if (err_num == AMQP_STATUS_OK)
{
@ -789,24 +788,24 @@ bool sendMessage(void* data)
atomic_add(&instance->stats.n_sent, 1);
atomic_add(&instance->stats.n_queued, -1);
spinlock_acquire(&instance->msg_lock);
pthread_mutex_lock(&instance->msg_lock);
tmp = instance->messages;
if (tmp == NULL)
{
spinlock_release(&instance->msg_lock);
pthread_mutex_unlock(&instance->msg_lock);
return true;
}
instance->messages = instance->messages->next;
spinlock_release(&instance->msg_lock);
pthread_mutex_unlock(&instance->msg_lock);
}
else
{
spinlock_acquire(&instance->msg_lock);
pthread_mutex_lock(&instance->msg_lock);
tmp->next = instance->messages;
instance->messages = tmp;
spinlock_release(&instance->msg_lock);
pthread_mutex_unlock(&instance->msg_lock);
return true;
}
}
@ -836,12 +835,12 @@ void pushMessage(MQ_INSTANCE* instance, amqp_basic_properties_t* prop, char* msg
return;
}
spinlock_acquire(&instance->msg_lock);
pthread_mutex_lock(&instance->msg_lock);
newmsg->next = instance->messages;
instance->messages = newmsg;
spinlock_release(&instance->msg_lock);
pthread_mutex_unlock(&instance->msg_lock);
atomic_add(&instance->stats.n_msg, 1);
atomic_add(&instance->stats.n_queued, 1);

18
server/modules/filter/regexfilter/regexfilter.cc
View File

@ -79,11 +79,11 @@ typedef struct
*/
typedef struct
{
MXS_DOWNSTREAM down; /* The downstream filter */
SPINLOCK lock;
int no_change; /* No. of unchanged requests */
int replacements;/* No. of changed requests */
int active; /* Is filter active */
MXS_DOWNSTREAM down; /* The downstream filter */
pthread_mutex_t lock;
int no_change; /* No. of unchanged requests */
int replacements; /* No. of changed requests */
int active; /* Is filter active */
} REGEX_SESSION;
void log_match(REGEX_INSTANCE* inst, char* re, char* old, char* newsql);
@ -374,17 +374,17 @@ static int routeQuery(MXS_FILTER* instance, MXS_FILTER_SESSION* session, GWBUF*
{
queue = modutil_replace_SQL(queue, newsql);
queue = gwbuf_make_contiguous(queue);
spinlock_acquire(&my_session->lock);
pthread_mutex_lock(&my_session->lock);
log_match(my_instance, my_instance->match, sql, newsql);
spinlock_release(&my_session->lock);
pthread_mutex_unlock(&my_session->lock);
MXS_FREE(newsql);
my_session->replacements++;
}
else
{
spinlock_acquire(&my_session->lock);
pthread_mutex_lock(&my_session->lock);
log_nomatch(my_instance, my_instance->match, sql);
spinlock_release(&my_session->lock);
pthread_mutex_unlock(&my_session->lock);
my_session->no_change++;
}
MXS_FREE(sql);

8
server/modules/include/cdc.h
View File

@ -78,10 +78,10 @@ typedef struct cdc_session
*/
typedef struct cdc_protocol
{
int state; /*< CDC protocol state */
char user[CDC_USER_MAXLEN + 1]; /*< username for authentication */
SPINLOCK lock; /*< Protocol structure lock */
char type[CDC_TYPE_LEN + 1]; /*< Request Type */
int state; /*< CDC protocol state */
char user[CDC_USER_MAXLEN + 1]; /*< username for authentication */
pthread_mutex_t lock; /*< Protocol structure lock */
char type[CDC_TYPE_LEN + 1]; /*< Request Type */
} CDC_protocol;
/* routines */

5
server/modules/include/debugcli.h
View File

@ -30,7 +30,6 @@
#include <maxscale/cdefs.h>
#include <maxscale/service.h>
#include <maxscale/session.h>
#include <maxscale/spinlock.h>
MXS_BEGIN_DECLS
@ -42,8 +41,8 @@ struct cli_session;
*/
typedef struct cli_instance
{
SPINLOCK lock; /*< The instance spinlock */
SERVICE* service; /*< The debug cli service */
pthread_mutex_t lock; /*< The instance spinlock */
SERVICE* service;/*< The debug cli service */
struct cli_session
* sessions; /*< Linked list of sessions within this instance */
struct cli_instance

1
server/modules/monitor/galeramon/galeramon.cc
View File

@ -30,7 +30,6 @@
#include <maxscale/modinfo.h>
#include <maxscale/mysql_utils.h>
#include <maxscale/secrets.h>
#include <maxscale/spinlock.h>
#define DONOR_NODE_NAME_MAX_LEN 60
#define DONOR_LIST_SET_VAR "SET GLOBAL wsrep_sst_donor = \""

6
server/modules/protocol/CDC/cdc.cc
View File

@ -390,7 +390,7 @@ static CDC_protocol* cdc_protocol_init(DCB* dcb)
p->state = CDC_ALLOC;
spinlock_init(&p->lock);
pthread_mutex_init(&p->lock, NULL);
/* memory allocation here */
p->state = CDC_STATE_WAIT_FOR_AUTH;
@ -415,13 +415,13 @@ static void cdc_protocol_done(DCB* dcb)
p = (CDC_protocol*) dcb->protocol;
spinlock_acquire(&p->lock);
pthread_mutex_lock(&p->lock);
/* deallocate memory here */
p->state = CDC_STATE_CLOSE;
spinlock_release(&p->lock);
pthread_mutex_unlock(&p->lock);
}
/**

6
server/modules/protocol/maxscaled/maxscaled.cc
View File

@ -376,7 +376,7 @@ static int maxscaled_accept(DCB* listener)
continue;
}
spinlock_init(&maxscaled_protocol->lock);
pthread_mutex_init(&maxscaled_protocol->lock, NULL);
client_dcb->protocol = (void*)maxscaled_protocol;
client_dcb->session = session_alloc(listener->session->service, client_dcb);
@ -407,13 +407,13 @@ static int maxscaled_close(DCB* dcb)
return 0;
}
spinlock_acquire(&maxscaled->lock);
pthread_mutex_lock(&maxscaled->lock);
if (maxscaled->username)
{
MXS_FREE(maxscaled->username);
maxscaled->username = NULL;
}
spinlock_release(&maxscaled->lock);
pthread_mutex_unlock(&maxscaled->lock);
return 0;
}

7
server/modules/protocol/maxscaled/maxscaled.h
View File

@ -26,7 +26,6 @@
* @endverbatim
*/
#include <maxscale/dcb.h>
#include <maxscale/spinlock.h>
#include <maxscale/housekeeper.h>
MXS_BEGIN_DECLS
@ -36,9 +35,9 @@ MXS_BEGIN_DECLS
*/
typedef struct maxscaled
{
SPINLOCK lock; /**< Protocol structure lock */
int state; /**< The connection state */
char* username; /**< The login name of the user */
pthread_mutex_t lock; /**< Protocol structure lock */
int state; /**< The connection state */
char* username; /**< The login name of the user */
} MAXSCALED;
#define MAXSCALED_STATE_LOGIN 1 /**< Waiting for user */

1
server/modules/routing/avrorouter/avro.cc
View File

@ -40,7 +40,6 @@
#include <maxscale/router.h>
#include <maxscale/server.h>
#include <maxscale/service.h>
#include <maxscale/spinlock.h>
#include <maxscale/utils.hh>
#include <maxscale/pcre2.h>
#include <maxscale/routingworker.h>

2
server/modules/routing/avrorouter/avro_client.cc
View File

@ -28,9 +28,7 @@
#include <maxscale/server.h>
#include <maxscale/router.h>
#include <maxbase/atomic.h>
#include <maxscale/spinlock.h>
#include <maxscale/dcb.h>
#include <maxscale/spinlock.h>
#include <maxscale/log.h>
#include <maxscale/version.h>
#include <maxscale/alloc.h>

1
server/modules/routing/avrorouter/avro_main.cc
View File

@ -36,7 +36,6 @@
#include <maxscale/router.h>
#include <maxscale/server.h>
#include <maxscale/service.h>
#include <maxscale/spinlock.h>
#include <maxscale/utils.h>
#include <maxscale/routingworker.h>
#include <binlog_common.h>

1
server/modules/routing/avrorouter/avrorouter.hh
View File

@ -22,7 +22,6 @@
#include <maxscale/alloc.h>
#include <maxscale/dcb.h>
#include <maxscale/service.h>
#include <maxscale/spinlock.h>
#include <maxscale/mysql_binlog.h>
#include <maxscale/users.h>
#include <cdc.h>

65
server/modules/routing/binlogrouter/blr.cc
View File

@ -47,7 +47,6 @@
#include <maxscale/secrets.h>
#include <maxscale/server.h>
#include <maxscale/service.h>
#include <maxscale/spinlock.h>
#include <maxscale/users.h>
#include <maxscale/utils.h>
#include <maxscale/utils.hh>
@ -100,7 +99,7 @@ GWBUF* blr_cache_read_response(ROUTER_INSTANCE* router,
extern bool blr_load_last_mariadb_gtid(ROUTER_INSTANCE* router,
MARIADB_GTID_INFO* result);
static SPINLOCK instlock;
static pthread_mutex_t instlock;
static ROUTER_INSTANCE* instances;
static const MXS_ENUM_VALUE enc_algo_values[] =
@ -130,7 +129,7 @@ static const MXS_ENUM_VALUE binlog_storage_values[] =
extern "C" MXS_MODULE* MXS_CREATE_MODULE()
{
MXS_NOTICE("Initialise binlog router module.");
spinlock_init(&instlock);
pthread_mutex_init(&instlock, NULL);
instances = NULL;
static MXS_ROUTER_OBJECT MyObject =
@ -294,10 +293,10 @@ static MXS_ROUTER* createInstance(SERVICE* service, MXS_CONFIG_PARAMETER* params
memset(&inst->saved_master, 0, sizeof(MASTER_RESPONSES));
inst->service = service;
spinlock_init(&inst->lock);
pthread_mutex_init(&inst->lock, NULL);
inst->files = NULL;
spinlock_init(&inst->fileslock);
spinlock_init(&inst->binlog_lock);
pthread_mutex_init(&inst->fileslock, NULL);
pthread_mutex_init(&inst->binlog_lock, NULL);
inst->binlog_fd = -1;
inst->master_chksum = true;
@ -975,10 +974,10 @@ static MXS_ROUTER* createInstance(SERVICE* service, MXS_CONFIG_PARAMETER* params
* insert this router instance into the linked list of routers
* that have been created with this module.
*/
spinlock_acquire(&instlock);
pthread_mutex_lock(&instlock);
inst->next = instances;
instances = inst;
spinlock_release(&instlock);
pthread_mutex_unlock(&instlock);
/*
* Initialise the binlog cache for this router instance
@ -1189,7 +1188,7 @@ static MXS_ROUTER_SESSION* newSession(MXS_ROUTER* instance, MXS_SESSION* session
slave->overrun = 0;
slave->uuid = NULL;
slave->hostname = NULL;
spinlock_init(&slave->catch_lock);
pthread_mutex_init(&slave->catch_lock, NULL);
slave->dcb = session->client_dcb;
slave->router = inst;
#ifdef BLFILE_IN_SLAVE
@ -1211,10 +1210,10 @@ static MXS_ROUTER_SESSION* newSession(MXS_ROUTER* instance, MXS_SESSION* session
/**
* Add this session to the list of active sessions.
*/
spinlock_acquire(&inst->lock);
pthread_mutex_lock(&inst->lock);
slave->next = inst->slaves;
inst->slaves = slave;
spinlock_release(&inst->lock);
pthread_mutex_unlock(&inst->lock);
return reinterpret_cast<MXS_ROUTER_SESSION*>(slave);
}
@ -1244,7 +1243,7 @@ static void freeSession(MXS_ROUTER* router_instance,
* Remove the slave session form the list of slaves that are using the
* router currently.
*/
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
if (router->slaves == slave)
{
router->slaves = slave->next;
@ -1263,7 +1262,7 @@ static void freeSession(MXS_ROUTER* router_instance,
ptr->next = slave->next;
}
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
MXS_DEBUG("%lu [freeSession] Unlinked router_client_session %p from "
"router %p. Connections : %d. ",
@ -1469,14 +1468,14 @@ static void diagnostics(MXS_ROUTER* router, DCB* dcb)
char buf[40];
struct tm tm;
spinlock_acquire(&router_inst->lock);
pthread_mutex_lock(&router_inst->lock);
session = router_inst->slaves;
while (session)
{
i++;
session = session->next;
}
spinlock_release(&router_inst->lock);
pthread_mutex_unlock(&router_inst->lock);
minno = router_inst->stats.minno;
min30 = 0.0;
@ -1652,7 +1651,7 @@ static void diagnostics(MXS_ROUTER* router, DCB* dcb)
router_inst->stats.n_reads != 0 ?
((double)router_inst->stats.n_binlogs / router_inst->stats.n_reads) : 0);
spinlock_acquire(&router_inst->lock);
pthread_mutex_lock(&router_inst->lock);
if (router_inst->stats.lastReply)
{
if (buf[strlen(buf) - 1] == '\n')
@ -1722,7 +1721,7 @@ static void diagnostics(MXS_ROUTER* router, DCB* dcb)
{
dcb_printf(dcb, "\tNo events received from master yet\n");
}
spinlock_release(&router_inst->lock);
pthread_mutex_unlock(&router_inst->lock);
if (router_inst->active_logs)
{
@ -1762,7 +1761,7 @@ static void diagnostics(MXS_ROUTER* router, DCB* dcb)
if (router_inst->slaves)
{
dcb_printf(dcb, "\tSlaves:\n");
spinlock_acquire(&router_inst->lock);
pthread_mutex_lock(&router_inst->lock);
session = router_inst->slaves;
while (session)
{
@ -1962,7 +1961,7 @@ static void diagnostics(MXS_ROUTER* router, DCB* dcb)
dcb_printf(dcb, "\t\t--------------------\n\n");
session = session->next;
}
spinlock_release(&router_inst->lock);
pthread_mutex_unlock(&router_inst->lock);
}
}
@ -2108,7 +2107,7 @@ static json_t* diagnostics_json(const MXS_ROUTER* router)
json_object_set_new(rval, "average_events_per_packets", json_real(average_packets));
spinlock_acquire(&router_inst->lock);
pthread_mutex_lock(&router_inst->lock);
if (router_inst->stats.lastReply)
{
if (buf[strlen(buf) - 1] == '\n')
@ -2166,7 +2165,7 @@ static json_t* diagnostics_json(const MXS_ROUTER* router)
json_object_set_new(rval, "latest_event_timestamp", json_string(buf));
}
}
spinlock_release(&router_inst->lock);
pthread_mutex_unlock(&router_inst->lock);
json_object_set_new(rval, "active_logs", json_boolean(router_inst->active_logs));
json_object_set_new(rval, "reconnect_pending", json_boolean(router_inst->reconnect_pending));
@ -2194,7 +2193,7 @@ static json_t* diagnostics_json(const MXS_ROUTER* router)
if (router_inst->slaves)
{
json_t* arr = json_array();
spinlock_acquire(&router_inst->lock);
pthread_mutex_lock(&router_inst->lock);
for (ROUTER_SLAVE* session = router_inst->slaves; session; session = session->next)
{
@ -2307,7 +2306,7 @@ static json_t* diagnostics_json(const MXS_ROUTER* router)
json_array_append_new(arr, slave);
}
spinlock_release(&router_inst->lock);
pthread_mutex_unlock(&router_inst->lock);
json_object_set_new(rval, "slaves", arr);
}
@ -2399,7 +2398,7 @@ static void errorReply(MXS_ROUTER* instance,
/* Authentication failed: stop replication */
if (router->master_state == BLRM_TIMESTAMP)
{
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
/* set io error message */
if (router->m_errmsg)
{
@ -2411,7 +2410,7 @@ static void errorReply(MXS_ROUTER* instance,
/* Stop replication */
router->master_state = BLRM_SLAVE_STOPPED;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
/* Force backend DCB close */
dcb_close(backend_dcb);
@ -2458,7 +2457,7 @@ static void errorReply(MXS_ROUTER* instance,
/** Check router state and set errno and message */
if (router->master_state != BLRM_SLAVE_STOPPED)
{
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
/* set mysql_errno */
router->m_errno = mysql_errno;
@ -2468,7 +2467,7 @@ static void errorReply(MXS_ROUTER* instance,
MXS_FREE(router->m_errmsg);
}
router->m_errmsg = MXS_STRDUP_A(errmsg);
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
MXS_ERROR("%s: Master connection error %lu '%s' in state '%s', "
"%sattempting reconnect to master [%s]:%d",
@ -2541,7 +2540,7 @@ static bool rses_begin_locked_router_action(ROUTER_SLAVE* rses)
{
bool succp = false;
spinlock_acquire(&rses->rses_lock);
pthread_mutex_lock(&rses->rses_lock);
succp = true;
return succp;
@ -2563,7 +2562,7 @@ static bool rses_begin_locked_router_action(ROUTER_SLAVE* rses)
*/
static void rses_end_locked_router_action(ROUTER_SLAVE* rses)
{
spinlock_release(&rses->rses_lock);
pthread_mutex_unlock(&rses->rses_lock);
}
@ -2590,7 +2589,7 @@ static bool stats_func(void* inst)
router->stats.minno = 0;
}
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
slave = router->slaves;
while (slave)
{
@ -2602,7 +2601,7 @@ static bool stats_func(void* inst)
}
slave = slave->next;
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
return true;
}
@ -2985,14 +2984,14 @@ static void destroyInstance(MXS_ROUTER* instance)
}
}
spinlock_acquire(&inst->lock);
pthread_mutex_lock(&inst->lock);
if (inst->master_state != BLRM_UNCONFIGURED)
{
inst->master_state = BLRM_SLAVE_STOPPED;
}
spinlock_release(&inst->lock);
pthread_mutex_unlock(&inst->lock);
if (inst->client)
{

140
server/modules/routing/binlogrouter/blr.hh
View File

@ -340,11 +340,11 @@ typedef struct mariadb_gtid_elems
/** MariaDB GTID info */
typedef struct mariadb_gtid_info
{
char gtid[GTID_MAX_LEN + 1]; /** MariaDB 10.x GTID, string value */
char binlog_name[BINLOG_FNAMELEN + 1]; /** The binlog file */
uint64_t start; /** The BEGIN pos: i.e the GTID event */
uint64_t end; /** The next_pos in COMMIT event */
MARIADB_GTID_ELEMS gtid_elms; /** MariaDB 10.x GTID components */
char gtid[GTID_MAX_LEN + 1]; /** MariaDB 10.x GTID, string value */
char binlog_name[BINLOG_FNAMELEN + 1];/** The binlog file */
uint64_t start; /** The BEGIN pos: i.e the GTID event */
uint64_t end; /** The next_pos in COMMIT event */
MARIADB_GTID_ELEMS gtid_elms; /** MariaDB 10.x GTID components */
} MARIADB_GTID_INFO;
/* Master Server configuration struct */
@ -499,22 +499,22 @@ typedef struct
*/
typedef struct
{
BLCACHE_RECORD** records; /*< The actual binlog records */
int current; /*< The next record that will be inserted */
int cnt; /*< The number of records in the cache */
SPINLOCK lock; /*< The spinlock for the cache */
BLCACHE_RECORD** records;/*< The actual binlog records */
int current;/*< The next record that will be inserted */
int cnt; /*< The number of records in the cache */
mutable pthread_mutex_t lock; /*< The spinlock for the cache */
} BLCACHE;
typedef struct blfile
{
char binlog_name[BINLOG_FNAMELEN + 1];
/*< Name of the binlog file */
int fd; /*< Actual file descriptor */
int refcnt; /*< Reference count for file */
BLCACHE* cache; /*< Record cache for this file */
SPINLOCK lock; /*< The file lock */
MARIADB_GTID_ELEMS gtid_elms; /*< Elements for file prefix */
struct blfile* next; /*< Next file in list */
int fd; /*< Actual file descriptor */
int refcnt; /*< Reference count for file */
BLCACHE* cache; /*< Record cache for this file */
mutable pthread_mutex_t lock; /*< The file lock */
MARIADB_GTID_ELEMS gtid_elms; /*< Elements for file prefix */
struct blfile* next; /*< Next file in list */
} BLFILE;
/**
@ -599,18 +599,18 @@ typedef struct router_slave
uint8_t seqno; /*< Replication dump sequence no */
uint32_t lastEventTimestamp;
/*< Last event timestamp sent */
SPINLOCK catch_lock; /*< Event catchup lock */
unsigned int cstate; /*< Catch up state */
bool mariadb10_compat; /*< MariaDB 10.0 compatibility */
SPINLOCK rses_lock; /*< Protects rses_deleted */
pthread_t pthread;
ROUTER_INSTANCE* router;/*< Pointer to the owning router */
struct router_slave* next;
SLAVE_STATS stats; /*< Slave statistics */
time_t connect_time; /*< Connect time of slave */
char* warning_msg; /*< Warning message */
int heartbeat; /*< Heartbeat in seconds */
uint8_t lastEventReceived;
mutable pthread_mutex_t catch_lock; /*< Event catchup lock */
unsigned int cstate; /*< Catch up state */
bool mariadb10_compat; /*< MariaDB 10.0 compatibility */
mutable pthread_mutex_t rses_lock; /*< Protects rses_deleted */
pthread_t pthread;
ROUTER_INSTANCE* router; /*< Pointer to the owning router */
struct router_slave* next;
SLAVE_STATS stats; /*< Slave statistics */
time_t connect_time; /*< Connect time of slave */
char* warning_msg; /*< Warning message */
int heartbeat; /*< Heartbeat in seconds */
uint8_t lastEventReceived;
/*< Last event received */
time_t lastReply; /*< Last event sent */
/*< lsi: Last Sent Information */
@ -744,19 +744,19 @@ typedef struct binlog_encryption_ctx
*/
struct ROUTER_INSTANCE : public MXS_ROUTER
{
SERVICE* service; /*< Pointer to the service using this router */
ROUTER_SLAVE* slaves; /*< Link list of all the slave connections */
SPINLOCK lock; /*< Spinlock for the instance data */
char* uuid; /*< UUID for the router to use w/master */
int orig_masterid;/*< Server ID of the master, internally used */
int masterid; /*< Set ID of the master, sent to slaves */
int serverid; /*< ID for the router to use w/master */
int initbinlog; /*< Initial binlog file number */
char* user; /*< User name to use with master */
char* password; /*< Password to use with master */
char* fileroot; /*< Root of binlog filename */
bool master_chksum;/*< Does the master provide checksums */
bool mariadb10_compat;
SERVICE* service; /*< Pointer to the service using this router */
ROUTER_SLAVE* slaves; /*< Link list of all the slave connections */
mutable pthread_mutex_t lock; /*< Spinlock for the instance data */
char* uuid; /*< UUID for the router to use w/master */
int orig_masterid; /*< Server ID of the master, internally used */
int masterid; /*< Set ID of the master, sent to slaves */
int serverid; /*< ID for the router to use w/master */
int initbinlog; /*< Initial binlog file number */
char* user; /*< User name to use with master */
char* password; /*< Password to use with master */
char* fileroot; /*< Root of binlog filename */
bool master_chksum; /*< Does the master provide checksums */
bool mariadb10_compat;
/*< MariaDB 10.0 compatibility */
bool maxwell_compat;/*< Zendesk's Maxwell compatibility */
char* master_uuid; /*< Set UUID of the master, sent to slaves */
@ -768,11 +768,11 @@ struct ROUTER_INSTANCE : public MXS_ROUTER
/*< Last event received */
uint32_t lastEventTimestamp;
/*< Timestamp from last event */
MASTER_RESPONSES saved_master; /*< Saved master responses */
char* binlogdir; /*< The directory with the binlog files */
SPINLOCK binlog_lock; /*< Lock to control update of the binlog position */
int trx_safe; /*< Detect and handle partial transactions */
PENDING_TRANSACTION pending_transaction;
MASTER_RESPONSES saved_master; /*< Saved master responses */
char* binlogdir; /*< The directory with the binlog files */
mutable pthread_mutex_t binlog_lock; /*< Lock to control update of the binlog position */
int trx_safe; /*< Detect and handle partial transactions */
PENDING_TRANSACTION pending_transaction;
/*< Pending transaction */
enum blr_event_state master_event_state;
/*< Packet read state */
@ -791,31 +791,31 @@ struct ROUTER_INSTANCE : public MXS_ROUTER
uint64_t last_event_pos; /*< Position of last event written */
uint64_t current_safe_event;
/*< Position of the latest safe event being sent to slaves */
char prevbinlog[BINLOG_FNAMELEN + 1];
int rotating; /*< Rotation in progress flag */
BLFILE* files; /*< Files used by the slaves */
SPINLOCK fileslock; /*< Lock for the files queue above */
unsigned int short_burst; /*< Short burst for slave catchup */
unsigned int long_burst; /*< Long burst for slave catchup */
unsigned long burst_size; /*< Maximum size of burst to send */
unsigned long heartbeat; /*< Configured heartbeat value */
ROUTER_STATS stats; /*< Statistics for this router */
int active_logs;
int reconnect_pending;
int retry_interval; /*< Connect retry interval */
int retry_count; /*< Connect retry counter */
int retry_limit; /*< Retry limit */
time_t connect_time;
int handling_threads;
unsigned long m_errno; /*< master response mysql errno */
char* m_errmsg; /*< master response mysql error message */
char* set_master_version; /*< Send custom Version to slaves */
char* set_master_hostname; /*< Send custom Hostname to slaves */
bool set_master_uuid; /*< Send custom Master UUID to slaves */
bool set_master_server_id; /*< Send custom Master server_id to slaves */
int send_slave_heartbeat; /*< Enable sending heartbeat to slaves */
bool ssl_enabled; /*< Use SSL connection to master */
int ssl_cert_verification_depth;
char prevbinlog[BINLOG_FNAMELEN + 1];
int rotating; /*< Rotation in progress flag */
BLFILE* files; /*< Files used by the slaves */
mutable pthread_mutex_t fileslock; /*< Lock for the files queue above */
unsigned int short_burst;/*< Short burst for slave catchup */
unsigned int long_burst; /*< Long burst for slave catchup */
unsigned long burst_size; /*< Maximum size of burst to send */
unsigned long heartbeat; /*< Configured heartbeat value */
ROUTER_STATS stats; /*< Statistics for this router */
int active_logs;
int reconnect_pending;
int retry_interval; /*< Connect retry interval */
int retry_count; /*< Connect retry counter */
int retry_limit; /*< Retry limit */
time_t connect_time;
int handling_threads;
unsigned long m_errno; /*< master response mysql errno */
char* m_errmsg; /*< master response mysql error message */
char* set_master_version; /*< Send custom Version to slaves */
char* set_master_hostname; /*< Send custom Hostname to slaves */
bool set_master_uuid; /*< Send custom Master UUID to slaves */
bool set_master_server_id; /*< Send custom Master server_id to slaves */
int send_slave_heartbeat; /*< Enable sending heartbeat to slaves */
bool ssl_enabled; /*< Use SSL connection to master */
int ssl_cert_verification_depth;
/*< The maximum length of the certificate
* authority chain that will be accepted.
*/

2
server/modules/routing/binlogrouter/blr_cache.cc
View File

@ -33,9 +33,7 @@
#include <maxscale/server.h>
#include <maxscale/router.h>
#include <maxbase/atomic.h>
#include <maxscale/spinlock.h>
#include <maxscale/dcb.h>
#include <maxscale/spinlock.h>
#include <maxscale/log.h>

36
server/modules/routing/binlogrouter/blr_event.cc
View File

@ -31,7 +31,7 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
* won't be updated to router->current_pos
*/
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
if (router->trx_safe == 0
|| (router->trx_safe
&& router->pending_transaction.state == BLRM_NO_TRANSACTION))
@ -40,7 +40,7 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
router->binlog_position = router->current_pos;
router->current_safe_event = router->current_pos;
}
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
/**
* Detect transactions in events if trx_safe is set:
@ -71,7 +71,7 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
domainid = extract_field(ptr + MYSQL_HEADER_LEN + 1 + BINLOG_EVENT_HDR_LEN + 8, 32);
flags = *(ptr + MYSQL_HEADER_LEN + 1 + BINLOG_EVENT_HDR_LEN + 8 + 4);
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
/**
* Detect whether it's a standalone transaction:
@ -126,7 +126,7 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
router->pending_transaction.start_pos = router->current_pos;
router->pending_transaction.end_pos = 0;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
}
// Query Event check
@ -147,7 +147,7 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
+ var_block_len + 1 + db_name_len,
statement_len);
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
/* Check for BEGIN (it comes for START TRANSACTION too) */
if (strncmp(statement_sql, "BEGIN", 5) == 0)
@ -181,7 +181,7 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
router->pending_transaction.state = BLRM_STANDALONE_SEEN;
}
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
MXS_FREE(statement_sql);
}
@ -189,13 +189,13 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
/* Check for COMMIT in Transactional engines, i.e InnoDB */
if (hdr.event_type == XID_EVENT)
{
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
if (router->pending_transaction.state >= BLRM_TRANSACTION_START)
{
router->pending_transaction.state = BLRM_XID_EVENT_SEEN;
}
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
}
}
@ -223,7 +223,7 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
hdr.event_type);
MXS_ERROR("%s", errmsg);
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
/* Handle error messages */
char* old_errmsg = router->m_errmsg;
@ -234,7 +234,7 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
router->master_state = BLRM_SLAVE_STOPPED;
router->stats.n_binlog_errors++;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
MXS_FREE(old_errmsg);
@ -279,9 +279,9 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
{
if (hdr.event_type == ROTATE_EVENT)
{
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
router->rotating = 1;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
}
uint32_t offset = MYSQL_HEADER_LEN + 1; // Skip header and OK byte
@ -334,7 +334,7 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
* may depend on pending transaction
*/
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
if (router->trx_safe == 0
|| (router->trx_safe
@ -343,7 +343,7 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
router->binlog_position = router->current_pos;
router->current_safe_event = router->last_event_pos;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
/* Notify clients events can be read */
blr_notify_all_slaves(router);
@ -384,13 +384,13 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
}
}
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
/* Notify clients events can be read */
blr_notify_all_slaves(router);
/* update binlog_position and set pending to NO_TRX */
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
router->binlog_position = router->current_pos;
@ -398,11 +398,11 @@ bool blr_handle_one_event(MXS_ROUTER* instance, REP_HEADER& hdr, uint8_t* ptr, u
router->pending_transaction.state = BLRM_NO_TRANSACTION;
router->pending_transaction.standalone = false;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
}
else
{
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
}
}
}

69
server/modules/routing/binlogrouter/blr_file.cc
View File

@ -40,7 +40,6 @@
#include <maxscale/secrets.h>
#include <maxscale/server.h>
#include <maxscale/service.h>
#include <maxscale/spinlock.h>
#include <maxscale/utils.h>
using std::string;
@ -527,7 +526,7 @@ static int blr_file_create(ROUTER_INSTANCE* router, char* orig_file)
if (blr_file_add_magic(fd))
{
close(router->binlog_fd);
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
/// Use an intermediate buffer in case the source and destination overlap
char new_binlog[strlen(file) + 1];
@ -540,7 +539,7 @@ static int blr_file_create(ROUTER_INSTANCE* router, char* orig_file)
router->binlog_position = BINLOG_MAGIC_SIZE;
router->current_safe_event = BINLOG_MAGIC_SIZE;
router->last_written = BINLOG_MAGIC_SIZE;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
created = 1;
@ -630,7 +629,7 @@ void blr_file_append(ROUTER_INSTANCE* router, char* file)
}
fsync(fd);
close(router->binlog_fd);
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
memmove(router->binlog_name, file, BINLOG_FNAMELEN);
router->current_pos = lseek(fd, 0L, SEEK_END);
if (router->current_pos < 4)
@ -659,12 +658,12 @@ void blr_file_append(ROUTER_INSTANCE* router, char* file)
path,
router->current_pos);
close(fd);
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
return;
}
}
router->binlog_fd = fd;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
}
/**
@ -765,11 +764,11 @@ int blr_write_binlog_record(ROUTER_INSTANCE* router,
}
/* Increment offsets */
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
router->current_pos = hdr->next_pos;
router->last_written += size;
router->last_event_pos = hdr->next_pos - hdr->event_size;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
/* Check whether adding the Start Encryption event into current binlog */
if (router->encryption.enabled && write_start_encryption_event)
@ -884,7 +883,7 @@ BLFILE* blr_open_binlog(ROUTER_INSTANCE* router,
char path[PATH_MAX + 1] = "";
BLFILE* file;
spinlock_acquire(&router->fileslock);
pthread_mutex_lock(&router->fileslock);
file = router->files;
while (file
@ -901,13 +900,13 @@ BLFILE* blr_open_binlog(ROUTER_INSTANCE* router,
{
/* Reuse 'file' */
file->refcnt++;
spinlock_release(&router->fileslock);
pthread_mutex_unlock(&router->fileslock);
return file;
}
if ((file = (BLFILE*)MXS_CALLOC(1, sizeof(BLFILE))) == NULL)
{
spinlock_release(&router->fileslock);
pthread_mutex_unlock(&router->fileslock);
return NULL;
}
strcpy(file->binlog_name, binlog);
@ -922,7 +921,7 @@ BLFILE* blr_open_binlog(ROUTER_INSTANCE* router,
sizeof(MARIADB_GTID_ELEMS));
}
spinlock_init(&file->lock);
pthread_mutex_init(&file->lock, NULL);
strcpy(path, router->binlogdir);
strcat(path, "/");
@ -945,13 +944,13 @@ BLFILE* blr_open_binlog(ROUTER_INSTANCE* router,
{
MXS_ERROR("Failed to open binlog file %s", path);
MXS_FREE(file);
spinlock_release(&router->fileslock);
pthread_mutex_unlock(&router->fileslock);
return NULL;
}
file->next = router->files;
router->files = file;
spinlock_release(&router->fileslock);
pthread_mutex_unlock(&router->fileslock);
return file;
}
@ -995,7 +994,7 @@ GWBUF* blr_read_binlog(ROUTER_INSTANCE* router,
return NULL;
}
spinlock_acquire(&file->lock);
pthread_mutex_lock(&file->lock);
if (fstat(file->fd, &statb) == 0)
{
filelen = statb.st_size;
@ -1009,16 +1008,16 @@ GWBUF* blr_read_binlog(ROUTER_INSTANCE* router,
BINLOG_ERROR_MSG_LEN,
"blr_read_binlog called with invalid file->fd, pos %lu",
pos);
spinlock_release(&file->lock);
pthread_mutex_unlock(&file->lock);
return NULL;
}
}
spinlock_release(&file->lock);
pthread_mutex_unlock(&file->lock);
if (pos > filelen)
{
spinlock_acquire(&router->binlog_lock);
spinlock_acquire(&file->lock);
pthread_mutex_lock(&router->binlog_lock);
pthread_mutex_lock(&file->lock);
/* Check whether is current router file */
if (!blr_compare_binlogs(router,
@ -1048,14 +1047,14 @@ GWBUF* blr_read_binlog(ROUTER_INSTANCE* router,
hdr->ok = SLAVE_POS_BEYOND_EOF;
}
spinlock_release(&file->lock);
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&file->lock);
pthread_mutex_unlock(&router->binlog_lock);
return NULL;
}
spinlock_acquire(&router->binlog_lock);
spinlock_acquire(&file->lock);
pthread_mutex_lock(&router->binlog_lock);
pthread_mutex_lock(&file->lock);
/* Check current router file and router position */
if (blr_compare_binlogs(router,
@ -1082,14 +1081,14 @@ GWBUF* blr_read_binlog(ROUTER_INSTANCE* router,
hdr->ok = SLAVE_POS_READ_OK;
}
spinlock_release(&file->lock);
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&file->lock);
pthread_mutex_unlock(&router->binlog_lock);
return NULL;
}
spinlock_release(&file->lock);
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&file->lock);
pthread_mutex_unlock(&router->binlog_lock);
/* Read the header information from the file */
if ((n = pread(file->fd,
@ -1414,7 +1413,7 @@ GWBUF* blr_read_binlog(ROUTER_INSTANCE* router,
void blr_close_binlog(ROUTER_INSTANCE* router, BLFILE* file)
{
mxb_assert(file);
spinlock_acquire(&router->fileslock);
pthread_mutex_lock(&router->fileslock);
file->refcnt--;
if (file->refcnt == 0)
{
@ -1439,7 +1438,7 @@ void blr_close_binlog(ROUTER_INSTANCE* router, BLFILE* file)
{
file = NULL;
}
spinlock_release(&router->fileslock);
pthread_mutex_unlock(&router->fileslock);
if (file)
{
@ -1723,7 +1722,7 @@ int blr_file_next_exists(ROUTER_INSTANCE* router,
MXS_DEBUG("The next Binlog file from GTID maps repo is [%s]",
bigbuf);
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
/**
* Update GTID elems in the slave->f_info struct:
@ -1734,7 +1733,7 @@ int blr_file_next_exists(ROUTER_INSTANCE* router,
slave->f_info.gtid_elms.domain_id = result.gtid_elms.domain_id;
slave->f_info.gtid_elms.server_id = result.gtid_elms.server_id;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
}
else
{
@ -3751,13 +3750,13 @@ int blr_write_special_event(ROUTER_INSTANCE* router,
MXS_FREE(new_event);
// Increment offsets, next event will be written after this special one
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
router->last_written += event_size;
router->current_pos = file_offset + event_size;
router->last_event_pos = file_offset;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
// Force write
fsync(router->binlog_fd);
@ -3842,7 +3841,7 @@ uint8_t* blr_create_start_encryption_event(ROUTER_INSTANCE* router,
/* Update the encryption context */
uint8_t* nonce_ptr = &(new_event[BINLOG_EVENT_HDR_LEN + 4 + 1]);
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
memcpy(new_encryption_ctx->nonce, nonce_ptr, BLRM_NONCE_LENGTH);
new_encryption_ctx->binlog_crypto_scheme = new_event[BINLOG_EVENT_HDR_LEN];
@ -3854,7 +3853,7 @@ uint8_t* blr_create_start_encryption_event(ROUTER_INSTANCE* router,
MXS_FREE(router->encryption_ctx);
router->encryption_ctx = new_encryption_ctx;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
return new_event;
}

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

@ -53,7 +53,6 @@
#include <maxscale/server.h>
#include <maxscale/service.h>
#include <maxscale/session.h>
#include <maxscale/spinlock.h>
#include <maxscale/utils.h>
static GWBUF* blr_make_query(DCB* dcb, char* query);
@ -139,7 +138,7 @@ static void blr_start_master(void* data)
}
router->stats.n_binlogs_ses = 0;
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
if (router->master_state != BLRM_UNCONNECTED)
{
@ -160,7 +159,7 @@ static void blr_start_master(void* data)
/* Return only if state is not BLRM_CONNECTING */
if (router->master_state != BLRM_CONNECTING)
{
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
return;
}
}
@ -171,7 +170,7 @@ static void blr_start_master(void* data)
{
/* Force stopped state */
router->master_state = BLRM_SLAVE_STOPPED;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
MXS_ERROR("%s: failure while connecting to master server '%s', "
"reached %d maximum number of retries. "
@ -185,7 +184,7 @@ static void blr_start_master(void* data)
/* Force connecting state */
router->master_state = BLRM_CONNECTING;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
DCB* client = dcb_alloc(DCB_ROLE_INTERNAL, NULL);
@ -223,9 +222,9 @@ static void blr_start_master(void* data)
router->session,
BLR_PROTOCOL)) == NULL)
{
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
router->retry_count++;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
blr_start_master_in_main(router, connect_retry);
@ -355,7 +354,7 @@ void blr_close_master_in_main(void* data)
static void blr_restart_master(ROUTER_INSTANCE* router)
{
/* Now it is safe to unleash other threads on this router instance */
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
router->reconnect_pending = 0;
router->active_logs = 0;
@ -366,7 +365,7 @@ static void blr_restart_master(ROUTER_INSTANCE* router)
{
/* Force stopped state */
router->master_state = BLRM_SLAVE_STOPPED;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
MXS_ERROR("%s: failed to connect to master server '%s', "
"reached %d maximum number of retries. "
@ -399,7 +398,7 @@ static void blr_restart_master(ROUTER_INSTANCE* router)
new_config.port);
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
blr_start_master_in_main(router, connect_retry);
@ -413,7 +412,7 @@ static void blr_restart_master(ROUTER_INSTANCE* router)
{
/* Force connecting state */
router->master_state = BLRM_CONNECTING;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
blr_start_master_in_main(router);
}
@ -438,7 +437,7 @@ void blr_master_reconnect(ROUTER_INSTANCE* router)
return;
}
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
if (router->active_logs)
{
/* Currently processing a response, set a flag
@ -453,13 +452,13 @@ void blr_master_reconnect(ROUTER_INSTANCE* router)
router->active_logs = 1;
do_reconnect = 1;
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
if (do_reconnect)
{
blr_restart_master(router);
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
router->active_logs = 0;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
}
}
@ -473,13 +472,13 @@ void blr_master_close(ROUTER_INSTANCE* router)
dcb_close(router->master);
router->master = NULL;
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
if (router->master_state != BLRM_SLAVE_STOPPED)
{
router->master_state = BLRM_UNCONNECTED;
}
router->master_event_state = BLR_EVENT_DONE;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
gwbuf_free(router->stored_event);
router->stored_event = NULL;
@ -498,9 +497,9 @@ void blr_master_response(ROUTER_INSTANCE* router, GWBUF* buf)
{
atomic_add(&router->handling_threads, 1);
mxb_assert(router->handling_threads == 1);
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
router->active_logs = 1;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
if (router->master_state > BLRM_MAXSTATE)
{
@ -508,11 +507,11 @@ void blr_master_response(ROUTER_INSTANCE* router, GWBUF* buf)
router->master_state);
gwbuf_free(buf);
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
if (router->reconnect_pending)
{
router->active_logs = 0;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
atomic_add(&router->handling_threads, -1);
MXS_ERROR("%s: Pending reconnect in state %s.",
router->service->name,
@ -521,7 +520,7 @@ void blr_master_response(ROUTER_INSTANCE* router, GWBUF* buf)
return;
}
router->active_logs = 0;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
atomic_add(&router->handling_threads, -1);
return;
}
@ -564,7 +563,7 @@ void blr_master_response(ROUTER_INSTANCE* router, GWBUF* buf)
blrm_states[router->master_state]);
gwbuf_free(buf);
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
/* set mysql errno */
router->m_errno = mysql_errno;
@ -579,12 +578,12 @@ void blr_master_response(ROUTER_INSTANCE* router, GWBUF* buf)
router->active_logs = 0;
if (router->reconnect_pending)
{
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
atomic_add(&router->handling_threads, -1);
blr_restart_master(router);
return;
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
atomic_add(&router->handling_threads, -1);
return;
}
@ -598,9 +597,9 @@ void blr_master_response(ROUTER_INSTANCE* router, GWBUF* buf)
blr_restart_master(router);
}
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
router->active_logs = 0;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
atomic_add(&router->handling_threads, -1);
}
@ -811,7 +810,7 @@ static bool verify_checksum(ROUTER_INSTANCE* router, size_t len, uint8_t* ptr)
*/
static void reset_errors(ROUTER_INSTANCE* router, REP_HEADER* hdr)
{
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
/* set mysql errno to 0 */
router->m_errno = 0;
@ -823,7 +822,7 @@ static void reset_errors(ROUTER_INSTANCE* router, REP_HEADER* hdr)
}
router->m_errmsg = NULL;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
#ifdef SHOW_EVENTS
printf("blr: event type 0x%02x, flags 0x%04x, "
"event size %d, event timestamp %" PRIu32 "\n",
@ -889,10 +888,10 @@ void blr_handle_binlog_record(ROUTER_INSTANCE* router, GWBUF* pkt)
if (router->master_event_state == BLR_EVENT_DONE)
{
/** This is the start of a new event */
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
router->stats.n_binlogs++;
router->stats.n_binlogs_ses++;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
/* Check for semi-sync in event with OK byte[4]:
* move pointer 2 bytes ahead and set check_packet_len accordingly
@ -1182,7 +1181,7 @@ int blr_rotate_event(ROUTER_INSTANCE* router, uint8_t* ptr, REP_HEADER* hdr)
blr_file_update_gtid(router);
}
}
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
router->rotating = 0;
/* remove current binlog encryption context */
@ -1191,7 +1190,7 @@ int blr_rotate_event(ROUTER_INSTANCE* router, uint8_t* ptr, REP_HEADER* hdr)
MXS_FREE(router->encryption_ctx);
router->encryption_ctx = NULL;
}
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
return rotated;
}
@ -1383,7 +1382,7 @@ void blr_stop_start_master(ROUTER_INSTANCE* router)
}
}
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
router->master_state = BLRM_SLAVE_STOPPED;
@ -1412,7 +1411,7 @@ void blr_stop_start_master(ROUTER_INSTANCE* router)
}
router->master_state = BLRM_UNCONNECTED;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
blr_master_reconnect(router);
}
@ -1432,13 +1431,13 @@ static bool blr_check_last_master_event(void* inst)
int master_state = BLRM_UNCONNECTED;
char task_name[BLRM_TASK_NAME_LEN + 1] = "";
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
master_check = blr_check_heartbeat(router);
master_state = router->master_state;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
if (!master_check)
{
@ -1812,7 +1811,7 @@ static void blr_terminate_master_replication(ROUTER_INSTANCE* router,
*(msg_err + msg_len) = '\0';
std::string s(msg_err);
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
char* old_errmsg = router->m_errmsg;
router->m_errmsg = msg_err;
@ -1820,7 +1819,7 @@ static void blr_terminate_master_replication(ROUTER_INSTANCE* router,
router->master_state = BLRM_SLAVE_STOPPED;
router->stats.n_binlog_errors++;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
MXS_FREE(old_errmsg);
// TODO: Would it or would it not be safe to access router->m_errmsg here?
@ -1947,7 +1946,7 @@ void blr_notify_all_slaves(ROUTER_INSTANCE* router)
ROUTER_SLAVE* slave;
int notified = 0;
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
slave = router->slaves;
while (slave)
{
@ -1960,7 +1959,7 @@ void blr_notify_all_slaves(ROUTER_INSTANCE* router)
slave = slave->next;
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
if (notified > 0)
{
@ -2938,7 +2937,7 @@ bool blr_handle_fake_rotate(ROUTER_INSTANCE* router,
return false;
}
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
/* Set writing pos to 4 if Master GTID */
if (router->mariadb10_master_gtid && pos == 4)
@ -2960,7 +2959,7 @@ bool blr_handle_fake_rotate(ROUTER_INSTANCE* router,
router->rotating = 1;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
return blr_rotate_event(router, ptr, hdr);
}
@ -3009,11 +3008,11 @@ void blr_handle_fake_gtid_list(ROUTER_INSTANCE* router,
hole_size);
/* Set the offet for the write routine */
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
router->last_written = binlog_file_eof;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
// Write One Hole
// TODO: write small holes
@ -3026,7 +3025,7 @@ void blr_handle_fake_gtid_list(ROUTER_INSTANCE* router,
else
{
// Increment the internal offsets
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
router->last_written = hdr->next_pos;
router->last_event_pos = router->current_pos;
@ -3034,7 +3033,7 @@ void blr_handle_fake_gtid_list(ROUTER_INSTANCE* router,
router->binlog_position = router->current_pos;
router->current_safe_event = router->current_pos;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
}
}
}

163
server/modules/routing/binlogrouter/blr_slave.cc
View File

@ -45,7 +45,6 @@
#include <maxscale/router.h>
#include <maxscale/server.h>
#include <maxscale/service.h>
#include <maxscale/spinlock.h>
#include <maxscale/utils.h>
#include <maxscale/version.h>
@ -1767,7 +1766,7 @@ static int blr_slave_send_slave_hosts(ROUTER_INSTANCE* router, ROUTER_SLAVE* sla
blr_slave_send_eof(router, slave, 7);
seqno = 8;
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
sptr = router->slaves;
while (sptr)
{
@ -1807,7 +1806,7 @@ static int blr_slave_send_slave_hosts(ROUTER_INSTANCE* router, ROUTER_SLAVE* sla
}
sptr = sptr->next;
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
return blr_slave_send_eof(router, slave, seqno);
}
@ -1916,9 +1915,9 @@ static int blr_slave_binlog_dump(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, G
}
/* Get the current router binlog file */
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
strcpy(slave->binlog_name, router->binlog_name);
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
/* Set the safe pos */
slave->binlog_pos = 4;
@ -2010,14 +2009,14 @@ static int blr_slave_binlog_dump(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, G
{
bool force_disconnect = false;
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
if (router->pending_transaction.state > BLRM_NO_TRANSACTION
&& blr_is_current_binlog(router, slave)
&& (slave->binlog_pos > router->binlog_position))
{
force_disconnect = true;
}
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
if (force_disconnect)
{
@ -2320,7 +2319,7 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
int do_return;
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
do_return = 0;
@ -2332,14 +2331,14 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
do_return = 1;
}
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
if (do_return)
{
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
slave->cstate &= ~CS_BUSY;
slave->cstate |= CS_EXPECTCB;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
poll_fake_write_event(slave->dcb);
return 0;
@ -2368,10 +2367,10 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
if (rotating)
{
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
slave->cstate |= CS_EXPECTCB;
slave->cstate &= ~CS_BUSY;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
poll_fake_write_event(slave->dcb);
return rval;
}
@ -2464,7 +2463,7 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
* Read it if slave->encryption_ctx is NULL and
* set the slave->encryption_ctx accordingly
*/
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
if (hdr.event_type == MARIADB10_START_ENCRYPTION_EVENT
&& !slave->encryption_ctx)
@ -2522,7 +2521,7 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
/* set next pos */
slave->binlog_pos = hdr.next_pos;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
gwbuf_free(record);
record = NULL;
@ -2563,10 +2562,10 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
err_msg[BINLOG_ERROR_MSG_LEN] = '\0';
if (rotating)
{
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
slave->cstate |= CS_EXPECTCB;
slave->cstate &= ~CS_BUSY;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
poll_fake_write_event(slave->dcb);
return rval;
}
@ -2740,11 +2739,11 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
slave->binlog_name,
read_errmsg);
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
slave->state = BLRS_ERRORED;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
/*
* Send an error that will stop slave replication
@ -2782,9 +2781,9 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
}
/* Remove BUSY state */
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
slave->cstate &= ~CS_BUSY;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
mxb_assert(hdr.ok == SLAVE_POS_READ_OK);
@ -2802,8 +2801,8 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
/**
* (1) Same name and pos as current router file: aka Up To Date
*/
spinlock_acquire(&router->binlog_lock);
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&router->binlog_lock);
pthread_mutex_lock(&slave->catch_lock);
/*
* Now check again since we hold the router->binlog_lock
@ -2813,8 +2812,8 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
|| !blr_is_current_binlog(router, slave))
{
slave->cstate |= CS_EXPECTCB;
spinlock_release(&slave->catch_lock);
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&slave->catch_lock);
pthread_mutex_unlock(&router->binlog_lock);
/* Force slave to read events via catchup routine */
poll_fake_write_event(slave->dcb);
@ -2830,8 +2829,8 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
*/
slave->cstate |= CS_WAIT_DATA;
spinlock_release(&slave->catch_lock);
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&slave->catch_lock);
pthread_mutex_unlock(&router->binlog_lock);
}
}
else
@ -2860,9 +2859,9 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
*/
if (f_tree)
{
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
blr_slave_info_save(&slave->f_info, &current_info, c_prefix);
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
}
/**
@ -2888,7 +2887,7 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
* registration is on:
* set CS_WAIT_DATA and return.
*/
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
if (f_tree)
{
@ -2910,7 +2909,7 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
*/
slave->cstate = CS_WAIT_DATA;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
#ifndef BLFILE_IN_SLAVE
/* Close file */
@ -2931,13 +2930,13 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
*/
if (router->master_state == BLRM_BINLOGDUMP)
{
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
/* Router state is BLRM_BINLOGDUMP (aka replicating) */
if (slave->stats.n_failed_read < MISSING_FILE_READ_RETRIES)
{
slave->stats.n_failed_read++;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
/* Log warning for missing file */
blr_slave_log_next_file_action(router,
@ -2954,7 +2953,7 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
*/
slave->state = BLRS_ERRORED;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
/* Log error for missing file */
blr_slave_log_next_file_action(router,
@ -2986,7 +2985,7 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
* We need to deal with current slave file:
* restore first the GTID info into slave->f_info
*/
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
if (f_tree)
{
memcpy(&slave->f_info,
@ -3003,7 +3002,7 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
*/
slave->cstate = CS_WAIT_DATA;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
}
else
{
@ -3043,9 +3042,9 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
next_file))
#endif
{
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
slave->cstate |= CS_EXPECTCB;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
/**
* Note:
* Fake rotate just written to client,
@ -3076,9 +3075,9 @@ int blr_slave_catchup(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave, bool large)
* (perhaps some ignorable / skipped events)
* just retry to read again.
*/
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
slave->cstate |= CS_EXPECTCB;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
poll_fake_write_event(slave->dcb);
}
}
@ -3122,15 +3121,15 @@ int blr_slave_callback(DCB* dcb, DCB_REASON reason, void* data)
{
if (slave->state == BLRS_DUMPING)
{
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
if (slave->cstate & CS_BUSY)
{
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
return 0;
}
slave->cstate &= ~(CS_EXPECTCB);
slave->cstate |= CS_BUSY;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
slave->stats.n_dcb++;
@ -3569,7 +3568,7 @@ static int blr_slave_disconnect_server(ROUTER_INSTANCE* router,
int n;
int server_found = 0;
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
sptr = router->slaves;
/* look for server_id among all registered slaves */
@ -3606,7 +3605,7 @@ static int blr_slave_disconnect_server(ROUTER_INSTANCE* router,
}
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
/**
* Server id was not found
@ -3667,7 +3666,7 @@ static int blr_slave_disconnect_all(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave
seqno++);
blr_slave_send_eof(router, slave, seqno++);
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
sptr = router->slaves;
while (sptr)
@ -3687,7 +3686,7 @@ static int blr_slave_disconnect_all(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave
sptr->dcb->remote,
sptr->serverid);
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
blr_slave_send_error(router,
slave,
@ -3723,7 +3722,7 @@ static int blr_slave_disconnect_all(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave
sptr = sptr->next;
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
blr_slave_send_eof(router, slave, seqno);
@ -3849,7 +3848,7 @@ static int blr_stop_slave(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave)
}
}
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
router->master_state = BLRM_SLAVE_STOPPED;
@ -3882,7 +3881,7 @@ static int blr_stop_slave(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave)
router->reconnect_pending = 0;
router->active_logs = 0;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
MXS_NOTICE("%s: STOP SLAVE executed by %s@%s. Disconnecting from master [%s]:%d, "
"read up to log %s, pos %lu, transaction safe pos %lu",
@ -3950,11 +3949,11 @@ static int blr_start_slave(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave)
return 1;
}
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
router->master_state = BLRM_UNCONNECTED;
router->retry_count = 0;
router->config_index = 0; // Always start from the default configuration.
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
/**
* Check whether to create the new binlog (router->binlog_name)
@ -4032,7 +4031,7 @@ static int blr_start_slave(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave)
router->binlog_name,
4);
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
router->pending_transaction.state = BLRM_NO_TRANSACTION;
router->last_safe_pos = 0;
@ -4041,7 +4040,7 @@ static int blr_start_slave(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave)
router->binlog_position = 4;
router->current_safe_event = 4;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
/* Send warning message to mysql command */
blr_slave_send_warning_message(router, slave, msg);
@ -4446,7 +4445,7 @@ static int blr_apply_change_master(ROUTER_INSTANCE* router,
{
int rc = 0;
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
if (index == static_cast<int>(router->configs.size()))
{
@ -4463,7 +4462,7 @@ static int blr_apply_change_master(ROUTER_INSTANCE* router,
rc = blr_apply_change_master_0(router, config, error);
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
return rc;
}
@ -6180,7 +6179,7 @@ static bool blr_send_slave_heartbeat(void* inst)
ROUTER_INSTANCE* router = (ROUTER_INSTANCE*) inst;
time_t t_now = time(0);
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
sptr = router->slaves;
@ -6210,7 +6209,7 @@ static bool blr_send_slave_heartbeat(void* inst)
sptr = sptr->next;
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
return true;
}
@ -6449,7 +6448,7 @@ static int blr_set_master_ssl(ROUTER_INSTANCE* router,
bool blr_notify_waiting_slave(ROUTER_SLAVE* slave)
{
bool ret = false;
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
if (slave->cstate & CS_WAIT_DATA)
{
ret = true;
@ -6457,7 +6456,7 @@ bool blr_notify_waiting_slave(ROUTER_SLAVE* slave)
poll_fake_write_event(slave->dcb);
slave->cstate &= ~CS_WAIT_DATA;
}
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
return ret;
}
@ -6547,13 +6546,13 @@ static int blr_slave_read_ste(ROUTER_INSTANCE* router,
/* Set the pos of first encrypted event */
new_encryption_ctx->first_enc_event_pos = fde_end_pos + hdr.event_size;
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
SLAVE_ENCRYPTION_CTX* old_encryption_ctx = slave->encryption_ctx;
/* Set the new encryption ctx into slave */
slave->encryption_ctx = new_encryption_ctx;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
/* Free previous encryption ctx */
MXS_FREE(old_encryption_ctx);
@ -6800,9 +6799,9 @@ static bool blr_handle_simple_select_stmt(ROUTER_INSTANCE* router,
if (router->mariadb10_compat
&& router->mariadb10_gtid)
{
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
strcpy(mariadb_gtid, router->last_mariadb_gtid);
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
}
/* Return empty gtid_slave_pos if master GTID registration is off */
@ -7035,7 +7034,7 @@ static bool blr_slave_gtid_request(ROUTER_INSTANCE* router,
memset(&f_gtid, 0, sizeof(f_gtid));
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
// Set gtid as current router gtid
strcpy(last_gtid, router->last_mariadb_gtid);
// Set file as router current file
@ -7048,7 +7047,7 @@ static bool blr_slave_gtid_request(ROUTER_INSTANCE* router,
f_gtid.gtid_elms.domain_id = router->mariadb10_gtid_domain;
f_gtid.gtid_elms.server_id = router->orig_masterid;
}
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
MXS_INFO("Slave %lu is registering with MariaDB GTID '%s'",
(unsigned long)slave->serverid,
@ -8127,7 +8126,7 @@ static bool blr_handle_admin_stmt(ROUTER_INSTANCE* router,
MXS_ERROR("%s: %s", router->service->name, error_string);
}
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
/* Set the BLRM_UNCONFIGURED state */
router->master_state = BLRM_UNCONFIGURED;
@ -8138,7 +8137,7 @@ static bool blr_handle_admin_stmt(ROUTER_INSTANCE* router,
router->m_errmsg = NULL;
router->m_errno = 0;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
if (removed_cfg == -1)
{
@ -8253,11 +8252,11 @@ static bool blr_handle_admin_stmt(ROUTER_INSTANCE* router,
if (ret != 0)
{
/* file operation failure: restore config */
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
blr_master_apply_config(router, current_master);
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
snprintf(error_string,
sizeof(error_string),
@ -8277,13 +8276,13 @@ static bool blr_handle_admin_stmt(ROUTER_INSTANCE* router,
}
/* Mark as active the master server struct */
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
if (!router->service->dbref->server->is_active)
{
router->service->dbref->server->is_active = true;
router->service->dbref->active = true;
}
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
/**
* check if router is BLRM_UNCONFIGURED
@ -8291,11 +8290,11 @@ static bool blr_handle_admin_stmt(ROUTER_INSTANCE* router,
*/
if (rc == 1 || router->master_state == BLRM_UNCONFIGURED)
{
spinlock_acquire(&router->lock);
pthread_mutex_lock(&router->lock);
router->master_state = BLRM_SLAVE_STOPPED;
spinlock_release(&router->lock);
pthread_mutex_unlock(&router->lock);
/*
* The binlog server has just been configured
@ -8414,9 +8413,9 @@ static void blr_slave_skip_empty_files(ROUTER_INSTANCE* router,
char next_file[BINLOG_FNAMELEN + 1] = "";
// Save the current router binlog filename
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
strcpy(router_curr_file, router->binlog_name);
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
// Set the starting filename
strcpy(binlog_file, slave->binlog_name);
@ -8538,12 +8537,12 @@ static int blr_show_binary_logs(ROUTER_INSTANCE* router,
BINARY_LOG_DATA_RESULT result = {};
/* Get current binlog finename and position */
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
strcpy(current_file, router->binlog_name);
current_pos = router->current_pos;
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
/**
* First part of result set:
@ -9336,9 +9335,9 @@ static bool blr_check_connecting_slave(const ROUTER_INSTANCE* router,
if (!rv)
{
/* Force BLRS_ERRORED state */
spinlock_acquire(&slave->catch_lock);
pthread_mutex_lock(&slave->catch_lock);
slave->state = BLRS_ERRORED;
spinlock_release(&slave->catch_lock);
pthread_mutex_unlock(&slave->catch_lock);
/* Send error that stops slave replication */
blr_send_custom_error(slave->dcb,
@ -9775,7 +9774,7 @@ static void blr_slave_log_next_file_action(const ROUTER_INSTANCE* router,
bool have_heartbeat = router->send_slave_heartbeat
&& (slave->heartbeat > 0);
spinlock_acquire(&router->binlog_lock);
pthread_mutex_lock(&router->binlog_lock);
if (s_tree)
{
/* Get master file prefix */
@ -9789,7 +9788,7 @@ static void blr_slave_log_next_file_action(const ROUTER_INSTANCE* router,
slave->f_info.gtid_elms.domain_id,
slave->f_info.gtid_elms.server_id);
}
spinlock_release(&router->binlog_lock);
pthread_mutex_unlock(&router->binlog_lock);
switch (log_action)
{

2
server/modules/routing/binlogrouter/test/testbinlog.cc
View File

@ -32,9 +32,7 @@
#include <maxscale/server.h>
#include <maxscale/router.h>
#include <maxbase/atomic.h>
#include <maxscale/spinlock.h>
#include <maxscale/dcb.h>
#include <maxscale/spinlock.h>
#include <maxscale/housekeeper.h>
#include <time.h>
#include <maxscale/log.h>

11
server/modules/routing/cli/cli.cc
View File

@ -35,7 +35,6 @@
#include <maxscale/router.h>
#include <maxscale/modinfo.h>
#include <maxbase/atomic.h>
#include <maxscale/spinlock.h>
#include <maxscale/dcb.h>
#include <maxscale/alloc.h>
#include <maxscale/poll.h>
@ -121,7 +120,7 @@ static MXS_ROUTER* createInstance(SERVICE* service, MXS_CONFIG_PARAMETER* params
}
inst->service = service;
spinlock_init(&inst->lock);
pthread_mutex_init(&inst->lock, NULL);
inst->sessions = NULL;
return (MXS_ROUTER*)inst;
@ -147,10 +146,10 @@ static MXS_ROUTER_SESSION* newSession(MXS_ROUTER* instance, MXS_SESSION* session
memset(client->cmdbuf, 0, 80);
spinlock_acquire(&inst->lock);
pthread_mutex_lock(&inst->lock);
client->next = inst->sessions;
inst->sessions = client;
spinlock_release(&inst->lock);
pthread_mutex_unlock(&inst->lock);
session->state = SESSION_STATE_READY;
@ -170,7 +169,7 @@ static void closeSession(MXS_ROUTER* instance, MXS_ROUTER_SESSION* router_sessio
CLI_SESSION* session = (CLI_SESSION*)router_session;
spinlock_acquire(&inst->lock);
pthread_mutex_lock(&inst->lock);
if (inst->sessions == session)
{
inst->sessions = session->next;
@ -187,7 +186,7 @@ static void closeSession(MXS_ROUTER* instance, MXS_ROUTER_SESSION* router_sessio
ptr->next = session->next;
}
}
spinlock_release(&inst->lock);
pthread_mutex_unlock(&inst->lock);
/**
* Router session is freed in session.c:session_close, when session who
* owns it, is freed.

19
server/modules/routing/debugcli/debugcli.cc
View File

@ -34,7 +34,6 @@
#include <maxscale/router.h>
#include <maxscale/modinfo.h>
#include <maxbase/atomic.h>
#include <maxscale/spinlock.h>
#include <maxscale/dcb.h>
#include <maxscale/alloc.h>
#include <maxscale/poll.h>
@ -53,7 +52,7 @@ static uint64_t getCapabilities(MXS_ROUTER* instance);
extern int execute_cmd(CLI_SESSION* cli);
static SPINLOCK instlock;
static pthread_mutex_t instlock;
static CLI_INSTANCE* instances;
/**
@ -67,7 +66,7 @@ static CLI_INSTANCE* instances;
extern "C" MXS_MODULE* MXS_CREATE_MODULE()
{
MXS_NOTICE("Initialise debug CLI router module.");
spinlock_init(&instlock);
pthread_mutex_init(&instlock, NULL);
instances = NULL;
static MXS_ROUTER_OBJECT MyObject =
@ -125,7 +124,7 @@ static MXS_ROUTER* createInstance(SERVICE* service, MXS_CONFIG_PARAMETER* params
}
inst->service = service;
spinlock_init(&inst->lock);
pthread_mutex_init(&inst->lock, NULL);
inst->sessions = NULL;
/*
@ -133,10 +132,10 @@ static MXS_ROUTER* createInstance(SERVICE* service, MXS_CONFIG_PARAMETER* params
* insert this router instance into the linked list of routers
* that have been created with this module.
*/
spinlock_acquire(&instlock);
pthread_mutex_lock(&instlock);
inst->next = instances;
instances = inst;
spinlock_release(&instlock);
pthread_mutex_unlock(&instlock);
return (MXS_ROUTER*)inst;
}
@ -161,10 +160,10 @@ static MXS_ROUTER_SESSION* newSession(MXS_ROUTER* instance, MXS_SESSION* session
memset(client->cmdbuf, 0, 80);
spinlock_acquire(&inst->lock);
pthread_mutex_lock(&inst->lock);
client->next = inst->sessions;
inst->sessions = client;
spinlock_release(&inst->lock);
pthread_mutex_unlock(&inst->lock);
session->state = SESSION_STATE_READY;
@ -187,7 +186,7 @@ static void closeSession(MXS_ROUTER* instance, MXS_ROUTER_SESSION* router_sessio
CLI_SESSION* session = (CLI_SESSION*)router_session;
spinlock_acquire(&inst->lock);
pthread_mutex_lock(&inst->lock);
if (inst->sessions == session)
{
inst->sessions = session->next;
@ -204,7 +203,7 @@ static void closeSession(MXS_ROUTER* instance, MXS_ROUTER_SESSION* router_sessio
ptr->next = session->next;
}
}
spinlock_release(&inst->lock);
pthread_mutex_unlock(&inst->lock);
/**
* Router session is freed in session.c:session_close, when session who
* owns it, is freed.

13
server/modules/routing/debugcli/debugcmd.cc
View File

@ -49,7 +49,6 @@
#include <maxscale/router.h>
#include <maxscale/server.hh>
#include <maxscale/service.h>
#include <maxscale/spinlock.h>
#include <maxscale/users.h>
#include <maxscale/utils.h>
#include <maxscale/version.h>
@ -1232,7 +1231,7 @@ struct subcommand flushoptions[] =
};
/** This is used to prevent concurrent creation or removal of servers */
static SPINLOCK server_mod_lock = SPINLOCK_INIT;
static pthread_mutex_t server_mod_lock = PTHREAD_MUTEX_INITIALIZER;
/**
* Create a new server
@ -1251,7 +1250,7 @@ static void createServer(DCB* dcb,
char* protocol,
char* authenticator)
{
spinlock_acquire(&server_mod_lock);
pthread_mutex_lock(&server_mod_lock);
if (server_find_by_unique_name(name) == NULL)
{
@ -1269,7 +1268,7 @@ static void createServer(DCB* dcb,
dcb_printf(dcb, "Server '%s' already exists.\n", name);
}
spinlock_release(&server_mod_lock);
pthread_mutex_unlock(&server_mod_lock);
}
static void createListener(DCB* dcb,
@ -2082,7 +2081,7 @@ static bool user_is_authorized(DCB* dcb)
return rval;
}
static SPINLOCK debugcmd_lock = SPINLOCK_INIT;
static pthread_mutex_t debugcmd_lock = PTHREAD_MUTEX_INITIALIZER;
static const char item_separator[] =
"----------------------------------------------------------------------------\n";
@ -2177,7 +2176,7 @@ int execute_cmd(CLI_SESSION* cli)
argc = i - 2; /* The number of extra arguments to commands */
spinlock_acquire(&debugcmd_lock);
pthread_mutex_lock(&debugcmd_lock);
if (!strcasecmp(args[0], "help"))
{
@ -2504,7 +2503,7 @@ int execute_cmd(CLI_SESSION* cli)
args[0]);
}
spinlock_release(&debugcmd_lock);
pthread_mutex_unlock(&debugcmd_lock);
memset(cli->cmdbuf, 0, CMDBUFLEN);

19
server/modules/routing/maxinfo/maxinfo.cc
View File

@ -41,7 +41,6 @@
#include <maxscale/modutil.h>
#include <maxscale/monitor.h>
#include <maxbase/atomic.h>
#include <maxscale/spinlock.h>
#include <maxscale/dcb.h>
#include <maxscale/maxscale.h>
#include <maxscale/log.h>
@ -81,7 +80,7 @@ static void handleError(MXS_ROUTER* instance,
mxs_error_action_t action,
bool* succp);
static SPINLOCK instlock;
static pthread_mutex_t instlock;
static INFO_INSTANCE* instances;
/**
@ -95,7 +94,7 @@ static INFO_INSTANCE* instances;
extern "C" MXS_MODULE* MXS_CREATE_MODULE()
{
MXS_NOTICE("Initialise MaxInfo router module.");
spinlock_init(&instlock);
pthread_mutex_init(&instlock, NULL);
instances = NULL;
static MXS_ROUTER_OBJECT MyObject =
@ -155,17 +154,17 @@ static MXS_ROUTER* createInstance(SERVICE* service, MXS_CONFIG_PARAMETER* params
inst->sessions = NULL;
inst->service = service;
spinlock_init(&inst->lock);
pthread_mutex_init(&inst->lock, NULL);
/*
* We have completed the creation of the instance data, so now
* insert this router instance into the linked list of routers
* that have been created with this module.
*/
spinlock_acquire(&instlock);
pthread_mutex_lock(&instlock);
inst->next = instances;
instances = inst;
spinlock_release(&instlock);
pthread_mutex_unlock(&instlock);
return (MXS_ROUTER*)inst;
}
@ -190,10 +189,10 @@ static MXS_ROUTER_SESSION* newSession(MXS_ROUTER* instance, MXS_SESSION* session
client->dcb = session->client_dcb;
client->queue = NULL;
spinlock_acquire(&inst->lock);
pthread_mutex_lock(&inst->lock);
client->next = inst->sessions;
inst->sessions = client;
spinlock_release(&inst->lock);
pthread_mutex_unlock(&inst->lock);
session->state = SESSION_STATE_READY;
@ -213,7 +212,7 @@ static void closeSession(MXS_ROUTER* instance, MXS_ROUTER_SESSION* router_sessio
INFO_SESSION* session = (INFO_SESSION*)router_session;
spinlock_acquire(&inst->lock);
pthread_mutex_lock(&inst->lock);
if (inst->sessions == session)
{
inst->sessions = session->next;
@ -230,7 +229,7 @@ static void closeSession(MXS_ROUTER* instance, MXS_ROUTER_SESSION* router_sessio
ptr->next = session->next;
}
}
spinlock_release(&inst->lock);
pthread_mutex_unlock(&inst->lock);
/**
* Router session is freed in session.c:session_close, when session who
* owns it, is freed.

5
server/modules/routing/maxinfo/maxinfo.h
View File

@ -31,7 +31,6 @@
#include <maxscale/cdefs.h>
#include <maxscale/service.h>
#include <maxscale/session.h>
#include <maxscale/spinlock.h>
#include <maxscale/resultset.hh>
struct maxinfo_session;
@ -42,8 +41,8 @@ struct maxinfo_session;
*/
typedef struct maxinfo_instance
{
SPINLOCK lock; /*< The instance spinlock */
SERVICE* service; /*< The debug cli service */
pthread_mutex_t lock; /*< The instance spinlock */
SERVICE* service;/*< The debug cli service */
struct maxinfo_session
* sessions; /*< Linked list of sessions within this instance */
struct maxinfo_instance

1
server/modules/routing/maxinfo/maxinfo_error.cc
View File

@ -35,7 +35,6 @@
#include <maxscale/modinfo.h>
#include <maxscale/modutil.h>
#include <maxbase/atomic.h>
#include <maxscale/spinlock.h>
#include <maxscale/dcb.h>
#include <maxscale/poll.h>
#include <maxscale/log.h>

1
server/modules/routing/maxinfo/maxinfo_exec.cc
View File

@ -43,7 +43,6 @@
#include <maxscale/router.h>
#include <maxscale/server.hh>
#include <maxscale/service.h>
#include <maxscale/spinlock.h>
#include <maxscale/version.h>
#include "../../../core/internal/maxscale.h"

1
server/modules/routing/maxinfo/maxinfo_parse.cc
View File

@ -37,7 +37,6 @@
#include <maxscale/modinfo.h>
#include <maxscale/modutil.h>
#include <maxbase/atomic.h>
#include <maxscale/spinlock.h>
#include <maxscale/dcb.h>
#include <maxscale/poll.h>
#include <maxscale/log.h>

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

@ -82,7 +82,6 @@
#include <maxscale/server.hh>
#include <maxscale/router.h>
#include <maxbase/atomic.hh>
#include <maxscale/spinlock.h>
#include <maxscale/dcb.h>
#include <maxscale/modinfo.h>
#include <maxscale/log.h>

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

@ -30,7 +30,6 @@
#include <maxscale/modutil.h>
#include <maxscale/query_classifier.h>
#include <maxscale/router.h>
#include <maxscale/spinlock.h>
#include <maxscale/mysql_utils.h>
#include <maxscale/routingworker.h>

1
server/modules/routing/readwritesplit/rwsplit_mysql.cc
View File

@ -23,7 +23,6 @@
#include <maxscale/log.h>
#include <maxscale/query_classifier.h>
#include <maxscale/dcb.h>
#include <maxscale/spinlock.h>
#include <maxscale/modinfo.h>
#include <maxscale/modutil.h>
#include <maxscale/protocol/mysql.h>

1
server/modules/routing/schemarouter/schemarouterinstance.cc
View File

@ -29,7 +29,6 @@
#include <maxscale/query_classifier.h>
#include <maxscale/router.h>
#include <maxscale/secrets.h>
#include <maxscale/spinlock.h>
using std::string;