network-mysqld.c

Mysql Proxy本身是个很好的mysql负载均衡工具,但是其本身有bug:当多个mysql 做slave的时候,如果一个slave死掉,会影响别的slave也死掉!这个文件修复了这个bug!

/* Copyright (C) 2007 MySQL AB

   This program 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 of the License.

   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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */ 

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

/** 
 * @page protocol MySQL Protocol
 *
 * The MySQL Protocol is spilt up into the four phases
 *
 * -# connect
 * -# auth
 * -# query
 * -# and the close state
 *
 * @dot
 * digraph states {
 *   graph [rankdir=LR];
 *   node [fontname=Helvetica, fontsize=10];
 *
 *   connect [ shape=record ];
 *   close [ shape=record ];
 *
 *   subgraph { 
 *     label = "client";
 *     color = black;
 *     rank = same;
 *     node [ style=filled, fillcolor=lightblue ];
 *     connect; 
 *     auth; 
 *     oldauth; 
 *     query; 
 *     local;
 *   }
 *
 *   subgraph { 
 *     label = "server";
 *     rank = same; 
 *     node [ style=filled, fillcolor=orange ];
 *     handshake; 
 *     authres; 
 *     result; 
 *     infile;
 *   }
 *   
 *   subgraph { 
 *     edge [ fontcolor=blue, color=blue, fontsize=10, fontname=Helvetica ];
 *
 *     connect->handshake [ label = "connecting server" ];
 *     auth->authres [ label = "capabilities, password, default-db", URL="http://forge.mysql.com/wiki/MySQL_Internals_ClientServer_Protocol#Client_Authentication_Packet" ]; 
 *     oldauth->authres [ label = "scrambled password" ] ; 
 *     query->result [ label = "command (COM_*)", URL="http://forge.mysql.com/wiki/MySQL_Internals_ClientServer_Protocol#Command_Packet" ] ;
 *     query->infile [ label = "LOAD DATA INFILE LOCAL" ];
 *     local->result [ label = "file content"];
 *   }
 *
 *   subgraph {
 *     edge [ fontcolor=red, color=red, fontsize=10, fontname=Helvetica ];
 *     handshake->close [ label = "ERR: host denied", URL="http://forge.mysql.com/wiki/MySQL_Internals_ClientServer_Protocol#Error_Packet" ];
 *     handshake->auth [ label = "0x10: handshake", URL="http://forge.mysql.com/wiki/MySQL_Internals_ClientServer_Protocol#Handshake_Initialization_Packet" ];
 *     authres->oldauth [ label = "EOF: old password reauth" ];
 *     authres->query [ label = "OK: auth done", URL="http://forge.mysql.com/wiki/MySQL_Internals_ClientServer_Protocol#OK_Packet" ];
 *     authres->close [ label = "ERR: auth failed", URL="http://forge.mysql.com/wiki/MySQL_Internals_ClientServer_Protocol#Error_Packet" ];
 *     result->query [ label = "result for COM_*", URL="http://forge.mysql.com/wiki/MySQL_Internals_ClientServer_Protocol#Result_Set_Header_Packet" ] ;
 *     result->close [ label = "COM_QUIT" ];
 *     result->result [ label = "COM_BINLOG_DUMP" ];
 *     infile->local [ label = "EOF: filename" ];
 *   }
 * }
 * @enddot
 * 
 * Unfolded the sequence diagrams of the different use-cases:
 *
 * -# the client connects to the server and waits for data to return @msc
 *   client, backend;
 *   --- [ label = "connect to backend" ];
 *   client->backend  [ label = "INIT" ];
 * @endmsc
 * -# the auth-phase handles the new SHA1-style passwords and the old scramble() passwords 
 *   -# 4.1+ passwords @msc
 *   client, backend;
 *   --- [ label = "authenticate" ];
 *   backend->client [ label = "HANDSHAKE" ];
 *   client->backend [ label = "AUTH" ];
 *   backend->client [ label = "AUTH_RESULT" ];
 * @endmsc
 *   -# pre-4.1 passwords @msc
 *   client, backend;
 *   --- [ label = "authenticate" ];
 *   backend->client [ label = "HANDSHAKE" ];
 *   client->backend [ label = "AUTH" ];
 *   backend->client [ label = "OLD_PASSWORD_SCRAMBLE" ];
 *   client->backend [ label = "OLD_PASSWORD_AUTH" ];
 *   backend->client [ label = "AUTH_RESULT" ];
 * @endmsc
 * -# the query-phase repeats 
 *   -# COM_QUERY and friends @msc
 *   client, backend;
 *   --- [ label = "query result phase" ];
 *   client->backend [ label = "QUERY" ];
 *   backend->client [ label = "QUERY_RESULT" ];
 * @endmsc
 *   -# COM_QUIT @msc
 *   client, backend;
 *   --- [ label = "query result phase" ];
 *   client->backend [ label = "QUERY" ];
 *   backend->client [ label = "connection close" ];
 * @endmsc
 *   -# COM_BINLOG_DUMP @msc
 *   client, backend;
 *   --- [ label = "query result phase" ];
 *   client->backend [ label = "QUERY" ];
 *   backend->client [ label = "QUERY_RESULT" ];
 *   ... [ label = "more binlog entries" ];
 *   backend->client [ label = "QUERY_RESULT"];
 * @endmsc
 */

#include <sys/types.h>

#ifdef HAVE_SYS_FILIO_H
#include <sys/filio.h> /* required for FIONREAD on solaris */
#endif

#ifndef _WIN32
#include <sys/ioctl.h>
#include <sys/socket.h>

#include <arpa/inet.h> /** inet_ntoa */
#include <netinet/in.h>
#include <netinet/tcp.h>

#include <netdb.h>
#include <unistd.h>
#else
#include <winsock2.h>
#include <io.h>
#define ioctl ioctlsocket
#endif

#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <fcntl.h>
#include <errno.h>
#ifdef HAVE_SIGNAL_H
#include <signal.h>
#endif

#include <glib.h>

#include <mysql.h>
#include <mysqld_error.h>

#include "network-mysqld.h"
#include "network-mysqld-proto.h"
#include "network-conn-pool.h"

#ifdef _WIN32
extern volatile int agent_shutdown;
#else
extern volatile sig_atomic_t agent_shutdown;
#endif

#ifdef _WIN32
#define E_NET_CONNRESET WSAECONNRESET
#define E_NET_CONNABORTED WSAECONNABORTED
#define E_NET_WOULDBLOCK WSAEWOULDBLOCK
#define E_NET_INPROGRESS WSAEINPROGRESS
#else
#define E_NET_CONNRESET ECONNRESET
#define E_NET_CONNABORTED ECONNABORTED
#define E_NET_INPROGRESS EINPROGRESS
#if EWOULDBLOCK == EAGAIN
/**
 * some system make EAGAIN == EWOULDBLOCK which would lead to a 
 * error in the case handling
 *
 * set it to -1 as this error should never happen
 */
#define E_NET_WOULDBLOCK -1
#else
#define E_NET_WOULDBLOCK EWOULDBLOCK
#endif
#endif

/**
 * a handy marco for constant strings 
 */
#define C(x) x, sizeof(x) - 1

/**
 * call the cleanup callback for the current connection
 *
 * @param srv    global context
 * @param con    connection context
 *
 * @return       RET_SUCCESS on success
 */
retval_t plugin_call_cleanup(network_mysqld *srv, network_mysqld_con *con) {
	NETWORK_MYSQLD_PLUGIN_FUNC(func) = NULL;

	func = con->plugins.con_cleanup;
	
	if (!func) return RET_SUCCESS;

	return (*func)(srv, con);
}


/**
 * create a connection 
 *
 * @param srv    global context
 * @return       a connection context
 */
network_mysqld_con *network_mysqld_con_init(network_mysqld *srv) {
	network_mysqld_con *con;

	con = g_new0(network_mysqld_con, 1);

	con->srv = srv;

	g_ptr_array_add(srv->cons, con);

	return con;
}

/**
 * free a connection 
 *
 * closes the client and server sockets 
 *
 * @param con    connection context
 */
void network_mysqld_con_free(network_mysqld_con *con) {
	if (!con) return;

	if (con->server) network_socket_free(con->server);
	if (con->client) network_socket_free(con->client);

	/* we are still in the conns-array */

	g_ptr_array_remove_fast(con->srv->cons, con);

	g_free(con);
}


/**
 * type-safe free()-ing of the internal SQL tables
 *
 * @param s    a network_mysqld_tables 
 * @see g_hash_table_new_full()
 */
static void network_mysqld_tables_free_void(void *s) {
	network_mysqld_table_free(s);
}

/**
 * create a global context
 *
 * @see network_mysqld_tables_free_void()
 */
network_mysqld *network_mysqld_init() {
	network_mysqld *m;

	m = g_new0(network_mysqld, 1);

	m->event_base  = NULL;

	m->tables      = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, network_mysqld_tables_free_void);
	
	m->cons        = g_ptr_array_new();
	
	return m;
}

/**
 * init libevent
 *
 * kqueue has to be called after the fork() of daemonize
 *
 * @param m      global context
 */
void network_mysqld_init_libevent(network_mysqld *m) {
	m->event_base  = event_init();
}

/**
 * free the global scope
 *
 * closes all open connections, cleans up all plugins
 *
 * @param m      global context
 */
void network_mysqld_free(network_mysqld *m) {
	guint i;

	if (!m) return;

	for (i = 0; i < m->cons->len; i++) {
		network_mysqld_con *con = m->cons->pdata[i];

		plugin_call_cleanup(m, con);
		network_mysqld_con_free(con);
	}

	g_ptr_array_free(m->cons, TRUE);

	g_hash_table_destroy(m->tables);

	if (m->config.proxy.backend_addresses) {
		for (i = 0; m->config.proxy.backend_addresses[i]; i++) {
			g_free(m->config.proxy.backend_addresses[i]);
		}
		g_free(m->config.proxy.backend_addresses);
	}

	if (m->config.proxy.address) {
		network_mysqld_proxy_free(NULL);

		g_free(m->config.proxy.address);
	}
	if (m->config.admin.address) {
		g_free(m->config.admin.address);
	}
#ifdef HAVE_EVENT_BASE_FREE
	/* only recent versions have this call */
	event_base_free(m->event_base);
#endif
	
	g_free(m);
}



/**
 * translate a address-string into a network_address structure
 *
 * - if the address contains a colon we assume IPv4, 
 *   - ":3306" -> (tcp) "0.0.0.0:3306"
 * - if it starts with a / it is a unix-domain socket 
 *   - "/tmp/socket" -> (unix) "/tmp/socket"
 *
 * @param addr     the address-struct
 * @param address  the address string
 * @return 0 on success, -1 otherwise
 */
int network_mysqld_con_set_address(network_address *addr, gchar *address) {
	gchar *s;
	guint port;

	/* split the address:port */
	if (NULL != (s = strchr(address, ':'))) {
		port = strtoul(s + 1, NULL, 10);

		if (port == 0) {
			g_critical("<ip>:<port>, port is invalid or 0, has to be > 0, got '%s'", address);
			return -1;
		}
		if (port > 65535) {
			g_critical("<ip>:<port>, port is too large, has to be < 65536, got '%s'", address);

			return -1;
		}

		memset(&addr->addr.ipv4, 0, sizeof(struct sockaddr_in));

		if (address == s || 
		    0 == strcmp("0.0.0.0", address)) {
			/* no ip */
			addr->addr.ipv4.sin_addr.s_addr = htonl(INADDR_ANY);
		} else {
			struct hostent *he;

			*s = '\0';
			he = gethostbyname(address);
			*s = ':';

			if (NULL == he)  {
				g_error("resolving proxy-address '%s' failed: ", address);
			}

			g_assert(he->h_addrtype == AF_INET);
			g_assert(he->h_length == sizeof(struct in_addr));

			memcpy(&(addr->addr.ipv4.sin_addr.s_addr), he->h_addr_list[0], he->h_length);
		}

		addr->addr.ipv4.sin_family = AF_INET;
		addr->addr.ipv4.sin_port = htons(port);
		addr->len = sizeof(struct sockaddr_in);

		addr->str = g_strdup(address);
#ifdef HAVE_SYS_UN_H
	} else if (address[0] == '/') {
		if (strlen(address) >= sizeof(addr->addr.un.sun_path) - 1) {
			g_critical("unix-path is too long: %s", address);
			return -1;
		}

		addr->addr.un.sun_family = AF_UNIX;
		strcpy(addr->addr.un.sun_path, address);
		addr->len = sizeof(struct sockaddr_un);
		addr->str = g_strdup(address);
#endif
	} else {
		/* might be a unix socket */
		g_critical("%s.%d: network_mysqld_con_set_address(%s) failed: address has to be <ip>:<port> for TCP or a absolute path starting with / for Unix sockets", 
				__FILE__, __LINE__,
				address);
		return -1;
	}

	return 0;
}

/**
 * portable 'set non-blocking io'
 *
 * @param fd      socket-fd
 * @return        0
 */
int network_mysqld_con_set_non_blocking(int fd) {
	int ioctlvar;

#ifdef _WIN32
	ioctlvar = 1;
	if (0 != ioctlsocket(fd, FIONBIO, &ioctlvar)) {
		errno = WSAGetLastError();
		g_critical("%s.%d: ioctlsocket() failed: %s (%d)", 
				__FILE__, __LINE__,
				strerror(errno), errno);
        }
#else
	fcntl(fd, F_SETFL, O_NONBLOCK | O_RDWR);
#endif
	return 0;
}


/**
 * connect a socket
 *
 * the con->addr has to be set before 
 * 
 * @param con    a socket 
 * @return       0 on connected, -1 on error, -2 for try again
 * @see network_mysqld_set_address()
 */
int network_mysqld_con_connect(network_socket * con) {
	int val = 1;

	g_assert(con->addr.len);

	/**
	 * con->addr.addr.ipv4.sin_family is always mapped to the same field 
	 * even if it is not a IPv4 address as we use a union
	 */
	if (-1 == (con->fd = socket(con->addr.addr.ipv4.sin_family, SOCK_STREAM, 0))) {
#ifdef _WIN32
		errno = WSAGetLastError();
#endif
		g_critical("%s.%d: socket(%s) failed: %s", 
				__FILE__, __LINE__,
				con->addr.str, strerror(errno));