release/src/router/mysql/server-tools/instance-manager/listener.cc

   1 /*
   2    Copyright (c) 2003-2007 MySQL AB, 2009 Sun Microsystems, Inc.
   3    Use is subject to license terms.
   4
   5    This program is free software; you can redistribute it and/or modify
   6    it under the terms of the GNU General Public License as published by
   7    the Free Software Foundation; version 2 of the License.
   8
   9    This program is distributed in the hope that it will be useful,
  10    but WITHOUT ANY WARRANTY; without even the implied warranty of
  11    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12    GNU General Public License for more details.
  13
  14    You should have received a copy of the GNU General Public License
  15    along with this program; if not, write to the Free Software
  16    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
  17 */
  18
  19 #if defined(__GNUC__) && defined(USE_PRAGMA_IMPLEMENTATION)
  20 #pragma implementation
  21 #endif
  22
  23 #include "listener.h"
  24
  25 #include <my_global.h>
  26 #include <mysql.h>
  27 #include <violite.h>
  28
  29 #include <sys/stat.h>
  30 #ifndef __WIN__
  31 #include <sys/un.h>
  32 #endif
  33
  34 #include "log.h"
  35 #include "mysql_connection.h"
  36 #include "options.h"
  37 #include "portability.h"
  38 #include "priv.h"
  39 #include "thread_registry.h"
  40
  41
  42 static void set_non_blocking(int socket)
  43 {
  44 #ifndef __WIN__
  45   int flags= fcntl(socket, F_GETFL, 0);
  46   fcntl(socket, F_SETFL, flags | O_NONBLOCK);
  47 #else
  48   u_long arg= 1;
  49   ioctlsocket(socket, FIONBIO, &arg);
  50 #endif
  51 }
  52
  53
  54 static void set_no_inherit(int socket)
  55 {
  56 #ifndef __WIN__
  57   int flags= fcntl(socket, F_GETFD, 0);
  58   fcntl(socket, F_SETFD, flags | FD_CLOEXEC);
  59 #endif
  60 }
  61
  62 const int Listener::LISTEN_BACK_LOG_SIZE= 5;     /* standard backlog size */
  63
  64 Listener::Listener(Thread_registry *thread_registry_arg,
  65                    User_map *user_map_arg)
  66   :thread_registry(thread_registry_arg),
  67   user_map(user_map_arg),
  68   total_connection_count(0),
  69   num_sockets(0)
  70 {
  71 }
  72
  73
  74 /*
  75   Listener::run() - listen all supported sockets and spawn a thread
  76   to handle incoming connection.
  77   Using 'die' in case of syscall failure is OK now - we don't hold any
  78   resources and 'die' kills the signal thread automatically. To be rewritten
  79   one day.
  80   See also comments in mysqlmanager.cc to picture general Instance Manager
  81   architecture.
  82 */
  83
  84 void Listener::run()
  85 {
  86   int i, n= 0;
  87
  88 #ifndef __WIN__
  89   struct sockaddr_un unix_socket_address;
  90 #endif
  91
  92   log_info("Listener: started.");
  93
  94   thread_registry->register_thread(&thread_info);
  95
  96   FD_ZERO(&read_fds);
  97
  98   /* I. prepare 'listen' sockets */
  99   if (create_tcp_socket())
 100     goto err;
 101
 102 #ifndef __WIN__
 103   if (create_unix_socket(unix_socket_address))
 104     goto err;
 105 #endif
 106
 107   /* II. Listen sockets and spawn childs */
 108   for (i= 0; i < num_sockets; i++)
 109     n= max(n, sockets[i]);
 110   n++;
 111
 112   timeval tv;
 113   while (!thread_registry->is_shutdown())
 114   {
 115     fd_set read_fds_arg= read_fds;
 116     /*
 117       We should reintialize timer as on linux it is modified
 118       to reflect amount of time not slept.
 119     */
 120     tv.tv_sec= 0;
 121     tv.tv_usec= 100000;
 122
 123     /*
 124       When using valgrind 2.0 this syscall doesn't get kicked off by a
 125       signal during shutdown. This results in failing assert
 126       (Thread_registry::~Thread_registry). Valgrind 2.2 works fine.
 127     */
 128     int rc= select(n, &read_fds_arg, 0, 0, &tv);
 129
 130     if (rc == 0 || rc == -1)
 131     {
 132       if (rc == -1 && errno != EINTR)
 133         log_error("Listener: select() failed: %s.",
 134                   (const char *) strerror(errno));
 135       continue;
 136     }
 137
 138
 139     for (int socket_index= 0; socket_index < num_sockets; socket_index++)
 140     {
 141       /* Assuming that rc > 0 as we asked to wait forever */
 142       if (FD_ISSET(sockets[socket_index], &read_fds_arg))
 143       {
 144         int client_fd= accept(sockets[socket_index], 0, 0);
 145         /* accept may return -1 (failure or spurious wakeup) */
 146         if (client_fd >= 0)                    // connection established
 147         {
 148           set_no_inherit(client_fd);
 149
 150           struct st_vio *vio=
 151             vio_new(client_fd,
 152                     socket_index == 0 ?  VIO_TYPE_SOCKET : VIO_TYPE_TCPIP,
 153                     socket_index == 0 ? 1 : 0);
 154
 155           if (vio != NULL)
 156             handle_new_mysql_connection(vio);
 157           else
 158           {
 159             shutdown(client_fd, SHUT_RDWR);
 160             closesocket(client_fd);
 161           }
 162         }
 163       }
 164     }
 165   }
 166
 167   /* III. Release all resources and exit */
 168
 169   log_info("Listener: shutdown requested, exiting...");
 170
 171   for (i= 0; i < num_sockets; i++)
 172     closesocket(sockets[i]);
 173
 174 #ifndef __WIN__
 175   unlink(unix_socket_address.sun_path);
 176 #endif
 177
 178   thread_registry->unregister_thread(&thread_info);
 179
 180   log_info("Listener: finished.");
 181   return;
 182
 183 err:
 184   log_error("Listener: failed to initialize. Initiate shutdown...");
 185
 186   // we have to close the ip sockets in case of error
 187   for (i= 0; i < num_sockets; i++)
 188     closesocket(sockets[i]);
 189
 190   thread_registry->set_error_status();
 191   thread_registry->unregister_thread(&thread_info);
 192   thread_registry->request_shutdown();
 193   return;
 194 }
 195
 196 int Listener::create_tcp_socket()
 197 {
 198   /* value to be set by setsockopt */
 199   int arg= 1;
 200
 201   int ip_socket= socket(AF_INET, SOCK_STREAM, 0);
 202   if (ip_socket == INVALID_SOCKET)
 203   {
 204     log_error("Listener: socket(AF_INET) failed: %s.",
 205               (const char *) strerror(errno));
 206     return -1;
 207   }
 208
 209   struct sockaddr_in ip_socket_address;
 210   bzero(&ip_socket_address, sizeof(ip_socket_address));
 211
 212   ulong im_bind_addr;
 213   if (Options::Main::bind_address != 0)
 214   {
 215     im_bind_addr= (ulong) inet_addr(Options::Main::bind_address);
 216
 217     if (im_bind_addr == (ulong) INADDR_NONE)
 218       im_bind_addr= htonl(INADDR_ANY);
 219   }
 220   else
 221     im_bind_addr= htonl(INADDR_ANY);
 222   uint im_port= Options::Main::port_number;
 223
 224   ip_socket_address.sin_family= AF_INET;
 225   ip_socket_address.sin_addr.s_addr= im_bind_addr;
 226
 227
 228   ip_socket_address.sin_port= (unsigned short)
 229     htons((unsigned short) im_port);
 230
 231   setsockopt(ip_socket, SOL_SOCKET, SO_REUSEADDR, (char*) &arg, sizeof(arg));
 232   if (bind(ip_socket, (struct sockaddr *) &ip_socket_address,
 233            sizeof(ip_socket_address)))
 234   {
 235     log_error("Listener: bind(ip socket) failed: %s.",
 236               (const char *) strerror(errno));
 237     closesocket(ip_socket);
 238     return -1;
 239   }
 240
 241   if (listen(ip_socket, LISTEN_BACK_LOG_SIZE))
 242   {
 243     log_error("Listener: listen(ip socket) failed: %s.",
 244               (const char *) strerror(errno));
 245     closesocket(ip_socket);
 246     return -1;
 247   }
 248
 249   /* set the socket nonblocking */
 250   set_non_blocking(ip_socket);
 251
 252   /* make sure that instances won't be listening our sockets */
 253   set_no_inherit(ip_socket);
 254
 255   FD_SET(ip_socket, &read_fds);
 256   sockets[num_sockets++]= ip_socket;
 257   log_info("Listener: accepting connections on ip socket (port: %d)...",
 258            (int) im_port);
 259   return 0;
 260 }
 261
 262 #ifndef __WIN__
 263 int Listener::
 264 create_unix_socket(struct sockaddr_un &unix_socket_address)
 265 {
 266   int unix_socket= socket(AF_UNIX, SOCK_STREAM, 0);
 267   if (unix_socket == INVALID_SOCKET)
 268   {
 269     log_error("Listener: socket(AF_UNIX) failed: %s.",
 270               (const char *) strerror(errno));
 271     return -1;
 272   }
 273
 274   bzero(&unix_socket_address, sizeof(unix_socket_address));
 275
 276   unix_socket_address.sun_family= AF_UNIX;
 277   strmake(unix_socket_address.sun_path, Options::Main::socket_file_name,
 278           sizeof(unix_socket_address.sun_path) - 1);
 279   unlink(unix_socket_address.sun_path); // in case we have stale socket file
 280
 281   /*
 282     POSIX specifies default permissions for a pathname created by bind
 283     to be 0777. We need everybody to have access to the socket.
 284   */
 285   mode_t old_mask= umask(0);
 286   if (bind(unix_socket, (struct sockaddr *) &unix_socket_address,
 287            sizeof(unix_socket_address)))
 288   {
 289     log_error("Listener: bind(unix socket) failed for '%s': %s.",
 290               (const char *) unix_socket_address.sun_path,
 291               (const char *) strerror(errno));
 292     close(unix_socket);
 293     return -1;
 294   }
 295
 296   umask(old_mask);
 297
 298   if (listen(unix_socket, LISTEN_BACK_LOG_SIZE))
 299   {
 300     log_error("Listener: listen(unix socket) failed: %s.",
 301               (const char *) strerror(errno));
 302     close(unix_socket);
 303     return -1;
 304   }
 305
 306   /* set the socket nonblocking */
 307   set_non_blocking(unix_socket);
 308
 309   /* make sure that instances won't be listening our sockets */
 310   set_no_inherit(unix_socket);
 311
 312   log_info("Listener: accepting connections on unix socket '%s'...",
 313            (const char *) unix_socket_address.sun_path);
 314   sockets[num_sockets++]= unix_socket;
 315   FD_SET(unix_socket, &read_fds);
 316   return 0;
 317 }
 318 #endif
 319
 320
 321 /*
 322   Create new mysql connection. Created thread is responsible for deletion of
 323   the Mysql_connection and Vio instances passed to it.
 324   SYNOPSIS
 325     handle_new_mysql_connection()
 326 */
 327
 328 void Listener::handle_new_mysql_connection(struct st_vio *vio)
 329 {
 330   Mysql_connection *mysql_connection=
 331     new Mysql_connection(thread_registry, user_map,
 332                          vio, ++total_connection_count);
 333   if (mysql_connection == NULL || mysql_connection->start(Thread::DETACHED))
 334   {
 335     log_error("Listener: can not start connection handler.");
 336     delete mysql_connection;
 337     vio_delete(vio);
 338   }
 339   /* The connection will delete itself when the thread is finished */
 340 }