/* * This file is distributed as part of the SkySQL Gateway. It is free * software: you can redistribute it and/or modify it under the terms of the * GNU General Public License as published by the Free Software Foundation, * version 2. * * This program is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more * details. * * You should have received a copy of the GNU General Public License along with * this program; if not, write to the Free Software Foundation, Inc., 51 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. * * Copyright SkySQL Ab 2013 */ #include #include #include #include #include #include #include #include #include #include #include #include /** * @file readwritesplit.c The entry points for the read/write query splitting * router module. * * This file contains the entry points that comprise the API to the read write * query splitting router. * @verbatim * Revision History * * Date Who Description * 01/07/2013 Vilho Raatikka Initial implementation * 15/07/2013 Massimiliano Pinto Added clientReply * from master only in case of session change * 17/07/2013 Massimiliano Pinto clientReply is now used by mysql_backend * for all reply situations * 18/07/2013 Massimiliano Pinto routeQuery now handles COM_QUIT * as QUERY_TYPE_SESSION_WRITE * * @endverbatim */ static char *version_str = "V1.0.2"; static ROUTER* createInstance(SERVICE *service, char **options); static void* newSession(ROUTER *instance, SESSION *session); static void closeSession(ROUTER *instance, void *session); static int routeQuery(ROUTER *instance, void *session, GWBUF *queue); static void diagnostic(ROUTER *instance, DCB *dcb); static void clientReply(ROUTER* instance, void* router_session, GWBUF* queue, DCB *backend_dcb); static ROUTER_OBJECT MyObject = { createInstance, newSession, closeSession, routeQuery, diagnostic, clientReply }; static SPINLOCK instlock; static INSTANCE* instances; /** * Implementation of the mandatory version entry point * * @return version string of the module */ char * version() { return version_str; } /** * The module initialisation routine, called when the module * is first loaded. */ void ModuleInit() { skygw_log_write_flush(LOGFILE_MESSAGE, "Initialize read/write split router module.\n"); spinlock_init(&instlock); instances = NULL; } /** * The module entry point routine. It is this routine that * must populate the structure that is referred to as the * "module object", this is a structure with the set of * external entry points for this module. * * @return The module object */ ROUTER_OBJECT* GetModuleObject() { skygw_log_write(LOGFILE_TRACE, "Returning readwritesplit router module object."); return &MyObject; } /** * Create an instance of the router for a particular service * within the gateway. * * The job of ths entry point is to create the service wide data needed * for the query router. This is information needed to route queries that * is not related to any individual client session, exmaples of data that * might be stored in the ROUTER object for a particular query router are * connections counts, last used connection etc so that balancing may * take place. * * @param service The service this router is being create for * @param options The options for this query router * * @return The instance data for this new instance */ static ROUTER* createInstance( SERVICE* service, char** options) { INSTANCE* inst; SERVER* server; int n; int i; if ((inst = calloc(1, sizeof(INSTANCE))) == NULL) { return NULL; } inst->service = service; spinlock_init(&inst->lock); inst->connections = NULL; /** Calculate number of servers */ for (server = service->databases, n = 0; server; server = server->nextdb) { n++; } inst->servers = (BACKEND **)calloc(n + 1, sizeof(BACKEND *)); if (!inst->servers) { free(inst); return NULL; } if (options) { skygw_log_write_flush(LOGFILE_MESSAGE, "Router options supplied to read/write split router module but none are supported. The options will be ignored.\n"); } /** * We need an array of the backend servers in the instance structure so * that we can maintain a count of the number of connections to each * backend server. */ for (server = service->databases, n = 0; server; server = server->nextdb) { if ((inst->servers[n] = malloc(sizeof(BACKEND))) == NULL) { for (i = 0; i < n; i++) { free(inst->servers[i]); } free(inst->servers); free(inst); return NULL; } inst->servers[n]->server = server; inst->servers[n]->count = 0; n++; } inst->servers[n] = 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); inst->next = instances; instances = inst; spinlock_release(&instlock); return (ROUTER *)inst; } /** * Associate a new session with this instance of the router. * * The session is used to store all the data required for a particular * client connection. * * @param instance The router instance data * @param session The session itself * @return Session specific data for this session */ static void* newSession( ROUTER* instance, SESSION* session) { BACKEND* candidate = NULL; CLIENT_SESSION* client; INSTANCE* inst = (INSTANCE *)instance; int i; if ((client = (CLIENT_SESSION *)malloc(sizeof(CLIENT_SESSION))) == NULL) { return NULL; } /** * Find a backend server to connect to. This is the extent of the * load balancing algorithm we need to implement for this simple * connection router. */ for (i = 0; inst->servers[i]; i++) { if (inst->servers[i] && SERVER_IS_SLAVE(inst->servers[i]->server)) { candidate = inst->servers[i]; break; } } /** * Loop over all the servers and find any that have fewer connections * than our candidate server. * * If a server has less connections than the current candidate we mark this * as the new candidate to connect to. * * If a server has the same number of connections currently as the candidate * and has had less connections over time than the candidate it will also * become the new candidate. This has the effect of spreading the connections * over different servers during periods of very low load. */ for (i = 0; inst->servers[i]; i++) { if (inst->servers[i] && SERVER_IS_RUNNING(inst->servers[i]->server)) { if (SERVER_IS_SLAVE(inst->servers[i]->server)) { if (inst->servers[i]->count < candidate->count) { candidate = inst->servers[i]; } else if (inst->servers[i]->count == candidate->count && inst->servers[i]->server->stats.n_connections < candidate->server->stats.n_connections) { candidate = inst->servers[i]; } } else if (SERVER_IS_MASTER(inst->servers[i]->server)) { /** master is found */ inst->master = inst->servers[i]; } } } /* for */ if (candidate == NULL) { skygw_log_write_flush(LOGFILE_MESSAGE, "No suitable servers found for connection."); free(client); return NULL; } if (inst->master == NULL) { inst->master = inst->servers[i-1]; } /** * We now have a master and a slave server with the least connections. * Bump the connection counts for these servers. */ atomic_add(&candidate->count, 1); client->slave = candidate; atomic_add(&inst->master->count, 1); client->master = inst->master; ss_dassert(client->master->server != candidate->server); /** * Open the slave connection. */ if ((client->slaveconn = dcb_connect(candidate->server, session, candidate->server->protocol)) == NULL) { atomic_add(&candidate->count, -1); free(client); return NULL; } /** * Open the master connection. */ if ((client->masterconn = dcb_connect(client->master->server, session, client->master->server->protocol)) == NULL) { atomic_add(&client->master->count, -1); free(client); return NULL; } inst->stats.n_sessions += 1; /* Add this session to end of the list of active sessions */ spinlock_acquire(&inst->lock); client->next = inst->connections; inst->connections = client; spinlock_release(&inst->lock); return (void *)client; } /** * Close a session with the router, this is the mechanism * by which a router may cleanup data structure etc. * * @param instance The router instance data * @param session The session being closed */ static void closeSession( ROUTER* instance, void* router_session) { INSTANCE* inst = (INSTANCE *)instance; CLIENT_SESSION* session = (CLIENT_SESSION *)router_session; /** * Close the connection to the backend servers */ session->slaveconn->func.close(session->slaveconn); session->masterconn->func.close(session->masterconn); atomic_add(&session->slave->count, -1); atomic_add(&session->master->count, -1); atomic_add(&session->slave->server->stats.n_current, -1); atomic_add(&session->master->server->stats.n_current, -1); spinlock_acquire(&inst->lock); if (inst->connections == session) { inst->connections = session->next; } else { CLIENT_SESSION* ptr = inst->connections; while (ptr && ptr->next != session) { ptr = ptr->next; } if (ptr) { ptr->next = session->next; } } spinlock_release(&inst->lock); /* * We are no longer in the linked list, free * all the memory and other resources associated * to the client session. */ free(session); } /** * The main routing entry, this is called with every packet that is * received and has to be forwarded to the backend database. * * The routeQuery will make the routing decision based on the contents * of the instance, session and the query itself in the queue. The * data in the queue may not represent a complete query, it represents * the data that has been received. The query router itself is responsible * for buffering the partial query, a later call to the query router will * contain the remainder, or part thereof of the query. * * @param instance The query router instance * @param session The session associated with the client * @param queue Gateway buffer queue with the packets received * * @return The number of queries forwarded */ static int routeQuery( ROUTER* instance, void* router_session, GWBUF* queue) { skygw_query_type_t qtype = QUERY_TYPE_UNKNOWN; char* querystr = NULL; char* startpos; size_t len; unsigned char packet_type, *packet; int ret = 0; GWBUF *cq = NULL; INSTANCE* inst = (INSTANCE *)instance; CLIENT_SESSION* session = (CLIENT_SESSION *)router_session; inst->stats.n_queries++; packet = GWBUF_DATA(queue); packet_type = packet[4]; startpos = (char *)&packet[5]; len = packet[0]; len += 255*packet[1]; len += 255*255*packet[2]; switch(packet_type) { case COM_QUIT: /**< 1 QUIT will close all sessions */ case COM_INIT_DB: /**< 2 DDL must go to the master */ case COM_REFRESH: /**< 7 - I guess this is session but not sure */ case COM_DEBUG: /**< 0d all servers dump debug info to stdout */ case COM_PING: /**< 0e all servers are pinged */ case COM_CHANGE_USER: /**< 11 all servers change it accordingly */ qtype = QUERY_TYPE_SESSION_WRITE; break; case COM_CREATE_DB: /**< 5 DDL must go to the master */ case COM_DROP_DB: /**< 6 DDL must go to the master */ qtype = QUERY_TYPE_WRITE; break; case COM_QUERY: querystr = (char *)malloc(len); memcpy(querystr, startpos, len-1); memset(&querystr[len-1], 0, 1); qtype = skygw_query_classifier_get_type(querystr, 0); break; default: case COM_SHUTDOWN: /**< 8 where should shutdown be routed ? */ case COM_STATISTICS: /**< 9 ? */ case COM_PROCESS_INFO: /**< 0a ? */ case COM_CONNECT: /**< 0b ? */ case COM_PROCESS_KILL: /**< 0c ? */ case COM_TIME: /**< 0f should this be run in gateway ? */ case COM_DELAYED_INSERT: /**< 10 ? */ case COM_DAEMON: /**< 1d ? */ break; } #if defined(SS_DEBUG_) skygw_log_write(NULL, LOGFILE_TRACE, "String\t\"%s\"", querystr); skygw_log_write(NULL, LOGFILE_TRACE, "Packet type\t%s", STRPACKETTYPE(packet_type)); #endif switch (qtype) { case QUERY_TYPE_WRITE: #if defined(SS_DEBUG_) skygw_log_write(NULL, LOGFILE_TRACE, "Query type\t%s, routing to Master.", STRQTYPE(qtype)); #endif ret = session->masterconn->func.write(session->masterconn, queue); atomic_add(&inst->stats.n_master, 1); goto return_ret; break; case QUERY_TYPE_READ: #if defined(SS_DEBUG_) skygw_log_write(NULL, LOGFILE_TRACE, "Query type\t%s, routing to Slave.", STRQTYPE(qtype)); #endif ret = session->slaveconn->func.write(session->slaveconn, queue); atomic_add(&inst->stats.n_slave, 1); goto return_ret; break; case QUERY_TYPE_SESSION_WRITE: #if defined(SS_DEBUG_) skygw_log_write(NULL, LOGFILE_TRACE, "Query type\t%s, routing to All servers.", STRQTYPE(qtype)); #endif /** * TODO! Connection to all servers must be established, and * the command must be executed in them. */ cq = gwbuf_clone(queue); switch(packet_type) { case COM_QUIT: ret = session->masterconn->func.write(session->masterconn, queue); session->slaveconn->func.write(session->slaveconn, cq); break; case COM_CHANGE_USER: session->masterconn->func.auth(session->masterconn, NULL, session->masterconn->session, queue); session->slaveconn->func.auth(session->slaveconn, NULL, session->masterconn->session, cq); break; default: ret = session->masterconn->func.session(session->masterconn, (void *)queue); session->slaveconn->func.session(session->slaveconn, (void *)cq); break; } atomic_add(&inst->stats.n_all, 1); goto return_ret; break; default: #if defined(SS_DEBUG_) skygw_log_write(NULL, LOGFILE_TRACE, "Query type\t%s, routing to Master by default.", STRQTYPE(qtype)); #endif /** Is this really ok? */ ret = session->masterconn->func.write(session->masterconn, queue); atomic_add(&inst->stats.n_master, 1); goto return_ret; break; } return_ret: free(querystr); return ret; } /** * Diagnostics routine * * Print query router statistics to the DCB passed in * * @param instance The router instance * @param dcb The DCB for diagnostic output */ static void diagnostic(ROUTER *instance, DCB *dcb) { CLIENT_SESSION *session; INSTANCE *inst = (INSTANCE *)instance; int i = 0; spinlock_acquire(&inst->lock); session = inst->connections; while (session) { i++; session = session->next; } spinlock_release(&inst->lock); dcb_printf(dcb, "\tNumber of router sessions: %d\n", inst->stats.n_sessions); dcb_printf(dcb, "\tCurrent no. of router sessions: %d\n", i); dcb_printf(dcb, "\tNumber of queries forwarded: %d\n", inst->stats.n_queries); dcb_printf(dcb, "\tNumber of queries forwarded to master: %d\n", inst->stats.n_master); dcb_printf(dcb, "\tNumber of queries forwarded to slave: %d\n", inst->stats.n_slave); dcb_printf(dcb, "\tNumber of queries forwarded to all: %d\n", inst->stats.n_all); } /** * Client Reply routine * * The routine will reply to client for session change with master server data * * @param instance The router instance * @param router_session The router session * @param backend_dcb The backend DCB * @param queue The GWBUF with reply data */ static void clientReply(ROUTER* instance, void* router_session, GWBUF* queue, DCB *backend_dcb) { INSTANCE* inst = NULL; DCB *master = NULL; DCB *client = NULL; CLIENT_SESSION* session = NULL; inst = (INSTANCE *)instance; session = (CLIENT_SESSION *)router_session; master = session->masterconn; client = backend_dcb->session->client; if (backend_dcb->command == ROUTER_CHANGE_SESSION) { /* if backend_dcb is the master we can reply to the client */ if (backend_dcb == master) { master->session->client->func.write(master->session->client, queue); } else { /* just consume the gwbuf without writing to the client */ gwbuf_consume(queue, gwbuf_length(queue)); } } else { /* normal flow */ client->func.write(client, queue); } } ///