Take new sharding implementation into use
The sharding implementation now uses a class to abstract the details of the shard. This allows for different design where each session makes a copy of the global shard map which is then used for the duration of the session. In addition to making the desing a bit clearer to understand, it also removes lock competition between threads. Due to the change to C++, the main entry points need to be wrapped in the exception-safety macros. The next step in the refactoring will be to use the router template. This will remove the need to manually define them.
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@ -28,6 +28,10 @@
|
|||||||
#define MXS_MODULE_NAME "schemarouter"
|
#define MXS_MODULE_NAME "schemarouter"
|
||||||
|
|
||||||
#include <maxscale/cdefs.h>
|
#include <maxscale/cdefs.h>
|
||||||
|
|
||||||
|
#include <set>
|
||||||
|
#include <string>
|
||||||
|
|
||||||
#include <maxscale/dcb.h>
|
#include <maxscale/dcb.h>
|
||||||
#include <maxscale/hashtable.h>
|
#include <maxscale/hashtable.h>
|
||||||
#include <maxscale/protocol/mysql.h>
|
#include <maxscale/protocol/mysql.h>
|
||||||
@ -250,7 +254,7 @@ struct schemarouter_session
|
|||||||
bool rses_transaction_active; /*< Is a transaction active */
|
bool rses_transaction_active; /*< Is a transaction active */
|
||||||
struct schemarouter_instance *router; /*< The router instance */
|
struct schemarouter_instance *router; /*< The router instance */
|
||||||
struct schemarouter_session* next; /*< List of router sessions */
|
struct schemarouter_session* next; /*< List of router sessions */
|
||||||
shard_map_t *shardmap; /*< Database hash containing names of the databases
|
Shard shardmap; /**< Database hash containing names of the databases
|
||||||
* mapped to the servers that contain them */
|
* mapped to the servers that contain them */
|
||||||
char connect_db[MYSQL_DATABASE_MAXLEN + 1]; /*< Database the user was trying to connect to */
|
char connect_db[MYSQL_DATABASE_MAXLEN + 1]; /*< Database the user was trying to connect to */
|
||||||
char current_db[MYSQL_DATABASE_MAXLEN + 1]; /*< Current active database */
|
char current_db[MYSQL_DATABASE_MAXLEN + 1]; /*< Current active database */
|
||||||
@ -275,7 +279,7 @@ struct schemarouter_session
|
|||||||
*/
|
*/
|
||||||
typedef struct schemarouter_instance
|
typedef struct schemarouter_instance
|
||||||
{
|
{
|
||||||
HASHTABLE* shard_maps; /*< Shard maps hashed by user name */
|
ShardManager shard_manager; /*< Shard maps hashed by user name */
|
||||||
SERVICE* service; /*< Pointer to service */
|
SERVICE* service; /*< Pointer to service */
|
||||||
SCHEMAROUTER_SESSION* connections; /*< List of client connections */
|
SCHEMAROUTER_SESSION* connections; /*< List of client connections */
|
||||||
SPINLOCK lock; /*< Lock for the instance data */
|
SPINLOCK lock; /*< Lock for the instance data */
|
||||||
@ -286,9 +290,9 @@ typedef struct schemarouter_instance
|
|||||||
ROUTER_STATS stats; /*< Statistics for this router */
|
ROUTER_STATS stats; /*< Statistics for this router */
|
||||||
struct schemarouter_instance* next; /*< Next router on the list */
|
struct schemarouter_instance* next; /*< Next router on the list */
|
||||||
bool available_slaves; /*< The router has some slaves available */
|
bool available_slaves; /*< The router has some slaves available */
|
||||||
HASHTABLE* ignored_dbs; /*< List of databases to ignore when the
|
std::set<std::string> ignored_dbs; /*< List of databases to ignore when the
|
||||||
* database mapping finds multiple servers
|
* database mapping finds multiple servers
|
||||||
* with the same database */
|
* with the same database */
|
||||||
pcre2_code* ignore_regex; /*< Databases matching this regex will
|
pcre2_code* ignore_regex; /*< Databases matching this regex will
|
||||||
* not cause the session to be terminated
|
* not cause the session to be terminated
|
||||||
* if they are found on more than one server. */
|
* if they are found on more than one server. */
|
||||||
|
@ -15,109 +15,96 @@
|
|||||||
|
|
||||||
#include <maxscale/alloc.h>
|
#include <maxscale/alloc.h>
|
||||||
|
|
||||||
int hashkeyfun(const void* key)
|
Shard::Shard():
|
||||||
|
m_last_updated(time(NULL))
|
||||||
{
|
{
|
||||||
if (key == NULL)
|
|
||||||
{
|
|
||||||
return 0;
|
|
||||||
}
|
|
||||||
int hash = 0, c = 0;
|
|
||||||
const char* ptr = (const char*)key;
|
|
||||||
while ((c = *ptr++))
|
|
||||||
{
|
|
||||||
hash = c + (hash << 6) + (hash << 16) - hash;
|
|
||||||
}
|
|
||||||
|
|
||||||
return hash;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
int hashcmpfun(const void* v1, const void* v2)
|
Shard::~Shard()
|
||||||
{
|
{
|
||||||
const char* i1 = (const char*) v1;
|
|
||||||
const char* i2 = (const char*) v2;
|
|
||||||
|
|
||||||
return strcmp(i1, i2);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
void keyfreefun(void* data)
|
bool Shard::add_location(string db, SERVER* target)
|
||||||
{
|
{
|
||||||
MXS_FREE(data);
|
return m_map.insert(make_pair(db, target)).second;
|
||||||
}
|
}
|
||||||
|
|
||||||
shard_map_t* shard_map_alloc()
|
SERVER* Shard::get_location(string db)
|
||||||
{
|
{
|
||||||
shard_map_t *rval = (shard_map_t*) MXS_MALLOC(sizeof(shard_map_t));
|
SERVER* rval = NULL;
|
||||||
|
ServerMap::iterator iter = m_map.find(db);
|
||||||
|
|
||||||
if (rval)
|
if (iter != m_map.end())
|
||||||
{
|
{
|
||||||
if ((rval->hash = hashtable_alloc(SCHEMAROUTER_HASHSIZE, hashkeyfun, hashcmpfun)))
|
rval = iter->second;
|
||||||
{
|
|
||||||
HASHCOPYFN kcopy = (HASHCOPYFN)strdup;
|
|
||||||
hashtable_memory_fns(rval->hash, kcopy, kcopy, keyfreefun, keyfreefun);
|
|
||||||
spinlock_init(&rval->lock);
|
|
||||||
rval->last_updated = 0;
|
|
||||||
rval->state = SHMAP_UNINIT;
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
MXS_FREE(rval);
|
|
||||||
rval = NULL;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return rval;
|
return rval;
|
||||||
}
|
}
|
||||||
|
|
||||||
enum shard_map_state shard_map_update_state(shard_map_t *self, double refresh_min_interval)
|
bool Shard::stale(double max_interval) const
|
||||||
{
|
{
|
||||||
spinlock_acquire(&self->lock);
|
time_t now = time(NULL);
|
||||||
double tdiff = difftime(time(NULL), self->last_updated);
|
|
||||||
if (tdiff > refresh_min_interval)
|
return difftime(now, m_last_updated) > max_interval;
|
||||||
{
|
|
||||||
self->state = SHMAP_STALE;
|
|
||||||
}
|
|
||||||
enum shard_map_state state = self->state;
|
|
||||||
spinlock_release(&self->lock);
|
|
||||||
return state;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
void replace_shard_map(shard_map_t **target, shard_map_t **source)
|
bool Shard::empty() const
|
||||||
{
|
{
|
||||||
shard_map_t *tgt = *target;
|
return m_map.size() == 0;
|
||||||
shard_map_t *src = *source;
|
|
||||||
tgt->last_updated = src->last_updated;
|
|
||||||
tgt->state = src->state;
|
|
||||||
hashtable_free(tgt->hash);
|
|
||||||
tgt->hash = src->hash;
|
|
||||||
MXS_FREE(src);
|
|
||||||
*source = NULL;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
shard_map_t* get_latest_shard_map(shard_map_t *stored, shard_map_t *current)
|
void Shard::get_content(ServerMap& dest)
|
||||||
{
|
{
|
||||||
shard_map_t *map = stored;
|
for (ServerMap::iterator it = m_map.begin(); it != m_map.end(); it++)
|
||||||
|
|
||||||
spinlock_acquire(&map->lock);
|
|
||||||
|
|
||||||
if (map->state == SHMAP_STALE)
|
|
||||||
{
|
{
|
||||||
replace_shard_map(&map, ¤t);
|
dest.insert(*it);
|
||||||
}
|
}
|
||||||
else if (map->state != SHMAP_READY)
|
}
|
||||||
|
|
||||||
|
bool Shard::newer_than(const Shard& shard) const
|
||||||
|
{
|
||||||
|
return m_last_updated > shard.m_last_updated;
|
||||||
|
}
|
||||||
|
|
||||||
|
ShardManager::ShardManager()
|
||||||
|
{
|
||||||
|
spinlock_init(&m_lock);
|
||||||
|
}
|
||||||
|
|
||||||
|
ShardManager::~ShardManager()
|
||||||
|
{
|
||||||
|
}
|
||||||
|
|
||||||
|
Shard ShardManager::get_shard(string user, double max_interval)
|
||||||
|
{
|
||||||
|
SpinLockGuard guard(m_lock);
|
||||||
|
|
||||||
|
ShardMap::iterator iter = m_maps.find(user);
|
||||||
|
|
||||||
|
if (iter == m_maps.end() || iter->second.stale(max_interval))
|
||||||
{
|
{
|
||||||
MXS_WARNING("Shard map state is not ready but"
|
// No previous shard or a stale shard, construct a new one
|
||||||
"it is in use. Replacing it with a newer one.");
|
|
||||||
replace_shard_map(&map, ¤t);
|
if (iter != m_maps.end())
|
||||||
}
|
{
|
||||||
else
|
m_maps.erase(iter);
|
||||||
{
|
}
|
||||||
/**
|
|
||||||
* Another thread has already updated the shard map for this user
|
return Shard();
|
||||||
*/
|
|
||||||
hashtable_free(current->hash);
|
|
||||||
MXS_FREE(current);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
spinlock_release(&map->lock);
|
// Found valid shard
|
||||||
|
return iter->second;
|
||||||
|
}
|
||||||
|
|
||||||
return map;
|
void ShardManager::update_shard(Shard& shard, string user)
|
||||||
}
|
{
|
||||||
|
SpinLockGuard guard(m_lock);
|
||||||
|
ShardMap::iterator iter = m_maps.find(user);
|
||||||
|
|
||||||
|
if (iter == m_maps.end() || shard.newer_than(iter->second))
|
||||||
|
{
|
||||||
|
m_maps[user] = shard;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
@ -15,65 +15,114 @@
|
|||||||
|
|
||||||
#include <maxscale/cppdefs.hh>
|
#include <maxscale/cppdefs.hh>
|
||||||
|
|
||||||
|
#include <map>
|
||||||
|
#include <string>
|
||||||
|
#include <list>
|
||||||
|
|
||||||
#include <maxscale/service.h>
|
#include <maxscale/service.h>
|
||||||
#include <maxscale/hashtable.h>
|
#include <maxscale/hashtable.h>
|
||||||
#include <maxscale/spinlock.hh>
|
#include <maxscale/spinlock.hh>
|
||||||
|
|
||||||
enum shard_map_state
|
using namespace maxscale;
|
||||||
|
using std::map;
|
||||||
|
using std::string;
|
||||||
|
|
||||||
|
/** This contains the database to server mapping */
|
||||||
|
typedef map<string, SERVER*> ServerMap;
|
||||||
|
|
||||||
|
class Shard
|
||||||
{
|
{
|
||||||
SHMAP_UNINIT, /*< No databases have been added to this shard map */
|
public:
|
||||||
SHMAP_READY, /*< All available databases have been added */
|
Shard();
|
||||||
SHMAP_STALE /*< The shard map has old data or has not been updated recently */
|
~Shard();
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @brief Add a database location
|
||||||
|
*
|
||||||
|
* @param db Database to add
|
||||||
|
* @param target Target where database is located
|
||||||
|
*
|
||||||
|
* @return True if location was added
|
||||||
|
*/
|
||||||
|
bool add_location(string db, SERVER* target);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @brief Retrieve the location of a database
|
||||||
|
*
|
||||||
|
* @param db Database to locate
|
||||||
|
*
|
||||||
|
* @return The database or NULL if no server contains the database
|
||||||
|
*/
|
||||||
|
SERVER* get_location(string db);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @brief Check if shard contains stale information
|
||||||
|
*
|
||||||
|
* @param max_interval The maximum lifetime of the shard
|
||||||
|
*
|
||||||
|
* @return True if the shard is stale
|
||||||
|
*/
|
||||||
|
bool stale(double max_interval) const;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @brief Check if shard is empty
|
||||||
|
*
|
||||||
|
* @return True if shard contains no locations
|
||||||
|
*/
|
||||||
|
bool empty() const;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @brief Retrieve all database to server mappings
|
||||||
|
*
|
||||||
|
* @param keys A map where the database to server mappings are added
|
||||||
|
*/
|
||||||
|
void get_content(ServerMap& dest);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @brief Check if this shard is newer than the other shard
|
||||||
|
*
|
||||||
|
* @param shard The other shard to check
|
||||||
|
*
|
||||||
|
* @return True if this shard is newer
|
||||||
|
*/
|
||||||
|
bool newer_than(const Shard& shard) const;
|
||||||
|
|
||||||
|
private:
|
||||||
|
ServerMap m_map;
|
||||||
|
time_t m_last_updated;
|
||||||
};
|
};
|
||||||
|
|
||||||
/**
|
typedef map<string, Shard> ShardMap;
|
||||||
* A map of the shards tied to a single user.
|
|
||||||
*/
|
class ShardManager
|
||||||
typedef struct shard_map
|
|
||||||
{
|
{
|
||||||
HASHTABLE *hash; /*< A hashtable of database names and the servers which
|
public:
|
||||||
* have these databases. */
|
ShardManager();
|
||||||
SPINLOCK lock;
|
~ShardManager();
|
||||||
time_t last_updated;
|
|
||||||
enum shard_map_state state; /*< State of the shard map */
|
|
||||||
} shard_map_t;
|
|
||||||
|
|
||||||
/** TODO: Replace these */
|
/**
|
||||||
int hashkeyfun(const void* key);
|
* @brief Retrieve or create a shard
|
||||||
int hashcmpfun(const void *, const void *);
|
*
|
||||||
void keyfreefun(void* data);
|
* @param user User whose shard to retrieve
|
||||||
|
* @param max_lifetime The maximum lifetime of a shard
|
||||||
|
*
|
||||||
|
* @return The latest version of the shard or a newly created shard if no
|
||||||
|
* old version is available
|
||||||
|
*/
|
||||||
|
Shard get_shard(string user, double max_lifetime);
|
||||||
|
|
||||||
/** TODO: Don't use this everywhere */
|
/**
|
||||||
/** Size of the hashtable used to store ignored databases */
|
* @brief Update the shard information
|
||||||
#define SCHEMAROUTER_HASHSIZE 100
|
*
|
||||||
|
* The shard information is updated if the new shard contains more up to date
|
||||||
|
* information than the one stored in the shard manager.
|
||||||
|
*
|
||||||
|
* @param shard New version of the shard
|
||||||
|
* @param user The user whose shard this is
|
||||||
|
*/
|
||||||
|
void update_shard(Shard& shard, string user);
|
||||||
|
|
||||||
/**
|
private:
|
||||||
* Allocate a shard map and initialize it.
|
SPINLOCK m_lock;
|
||||||
* @return Pointer to new shard_map_t or NULL if memory allocation failed
|
ShardMap m_maps;
|
||||||
*/
|
};
|
||||||
shard_map_t* shard_map_alloc();
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Check if the shard map is out of date and update its state if necessary.
|
|
||||||
* @param router Router instance
|
|
||||||
* @param map Shard map to update
|
|
||||||
* @return Current state of the shard map
|
|
||||||
*/
|
|
||||||
enum shard_map_state shard_map_update_state(shard_map_t *self, double refresh_min_interval);
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Replace a shard map with another one. This function copies the contents of
|
|
||||||
* the source shard map to the target and frees the source memory.
|
|
||||||
* @param target Target shard map to replace
|
|
||||||
* @param source Source shard map to use
|
|
||||||
*/
|
|
||||||
void replace_shard_map(shard_map_t **target, shard_map_t **source);
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Return the newer of two shard maps
|
|
||||||
*
|
|
||||||
* @param stored The currently stored shard map
|
|
||||||
* @param current The replacement map the current client is using
|
|
||||||
* @return The newer of the two shard maps
|
|
||||||
*/
|
|
||||||
shard_map_t* get_latest_shard_map(shard_map_t *stored, shard_map_t *current);
|
|
||||||
|
Reference in New Issue
Block a user