listener.cc

/* Copyright (C) 2003 MySQL AB & MySQL Finland AB & TCX DataKonsult 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; either version 2 of the License, or
   (at your option) any later version.

   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 __GNUC__
#pragma implementation
#endif

#include "listener.h"

#include <m_string.h>
#include <mysql.h>
#include <violite.h>
#include <sys/un.h>
#include <sys/stat.h>

#include "thread_registry.h"
#include "options.h"
#include "instance_map.h"
#include "log.h"
#include "mysql_connection.h"


/*
  Listener_thread - incapsulates listening functionality
*/

class Listener_thread: public Listener_thread_args
{
public:
  Listener_thread(const Listener_thread_args &args);
  ~Listener_thread();
  void run();
private:
  ulong total_connection_count;
  Thread_info thread_info;
private:
  void handle_new_mysql_connection(Vio *vio);
};


Listener_thread::Listener_thread(const Listener_thread_args &args) :
  Listener_thread_args(args.thread_registry, args.options, args.user_map,
                       args.instance_map)
  ,total_connection_count(0)
  ,thread_info(pthread_self())
{
}


Listener_thread::~Listener_thread()
{
}


/*
  Listener_thread::run() - listen all supported sockets and spawn a thread
  to handle incoming connection.
  Using 'die' in case of syscall failure is OK now - we don't hold any
  resources and 'die' kills the signal thread automatically. To be rewritten
  one day.
  See also comments in mysqlmanager.cc to picture general Instance Manager
  architecture.
*/

void Listener_thread::run()
{
  enum { LISTEN_BACK_LOG_SIZE = 5 };            // standard backlog size
  int flags;
  int arg= 1;                             /* value to be set by setsockopt */
  int unix_socket;
  uint im_port;

  thread_registry.register_thread(&thread_info);

  my_thread_init();

  /* I. prepare 'listen' sockets */

  int ip_socket= socket(AF_INET, SOCK_STREAM, 0);
  if (ip_socket == INVALID_SOCKET)
  {
    log_error("Listener_thead::run(): socket(AF_INET) failed, %s",
              strerror(errno));
    goto err;
  }

  struct sockaddr_in ip_socket_address;
  bzero(&ip_socket_address, sizeof(ip_socket_address));

  ulong im_bind_addr;
  if (options.bind_address != 0)
  {
    if ((im_bind_addr= (ulong) inet_addr(options.bind_address)) == INADDR_NONE)
      im_bind_addr= htonl(INADDR_ANY);
  }
  else
    im_bind_addr= htonl(INADDR_ANY);
  im_port= options.port_number;

  ip_socket_address.sin_family= AF_INET;
  ip_socket_address.sin_addr.s_addr = im_bind_addr;


  ip_socket_address.sin_port= (unsigned short)
                              htons((unsigned short) im_port);

  setsockopt(ip_socket, SOL_SOCKET, SO_REUSEADDR, (char*) &arg, sizeof(arg));
  if (bind(ip_socket, (struct sockaddr *) &ip_socket_address,
           sizeof(ip_socket_address)))
  {
    log_error("Listener_thread::run(): bind(ip socket) failed, '%s'",
              strerror(errno));
   goto err;
  }

  if (listen(ip_socket, LISTEN_BACK_LOG_SIZE))
  {
    log_error("Listener_thread::run(): listen(ip socket) failed, %s",
              strerror(errno));
    goto err;
  }
      /* set the socket nonblocking */
  flags= fcntl(ip_socket, F_GETFL, 0);
  fcntl(ip_socket, F_SETFL, flags | O_NONBLOCK);
    /* make sure that instances won't be listening our sockets */
  flags= fcntl(ip_socket, F_GETFD, 0);
  fcntl(ip_socket, F_SETFD, flags | FD_CLOEXEC);

  log_info("accepting connections on ip socket");

  /*--------------------------------------------------------------*/
  unix_socket= socket(AF_UNIX, SOCK_STREAM, 0);
  if (unix_socket == INVALID_SOCKET)
  {
    log_error("Listener_thead::run(): socket(AF_UNIX) failed, %s",
              strerror(errno));
    goto err;
  }

  struct sockaddr_un unix_socket_address;
  bzero(&unix_socket_address, sizeof(unix_socket_address));

  unix_socket_address.sun_family= AF_UNIX;
  strmake(unix_socket_address.sun_path, options.socket_file_name,
          sizeof(unix_socket_address.sun_path));
  unlink(unix_socket_address.sun_path); // in case we have stale socket file

  {
    /*
      POSIX specifies default permissions for a pathname created by bind
      to be 0777. We need everybody to have access to the socket.
    */
    mode_t old_mask= umask(0);
    if (bind(unix_socket, (struct sockaddr *) &unix_socket_address,
             sizeof(unix_socket_address)))
    {
      log_error("Listener_thread::run(): bind(unix socket) failed, "
                "socket file name is '%s', error '%s'",
                unix_socket_address.sun_path, strerror(errno));
      goto err;
    }
    umask(old_mask);

    if (listen(unix_socket, LISTEN_BACK_LOG_SIZE))
    {
      log_error("Listener_thread::run(): listen(unix socket) failed, %s",
                strerror(errno));
      goto err;
    }

      /* set the socket nonblocking */
    flags= fcntl(unix_socket, F_GETFL, 0);
    fcntl(unix_socket, F_SETFL, flags | O_NONBLOCK);
      /* make sure that instances won't be listening our sockets */
    flags= fcntl(unix_socket, F_GETFD, 0);
    fcntl(unix_socket, F_SETFD, flags | FD_CLOEXEC);
  }
  log_info("accepting connections on unix socket %s",
           unix_socket_address.sun_path);

  /* II. Listen sockets and spawn childs */

  {
    int n= max(unix_socket, ip_socket) + 1;
    fd_set read_fds;

    FD_ZERO(&read_fds);
    FD_SET(unix_socket, &read_fds);
    FD_SET(ip_socket, &read_fds);

    while (thread_registry.is_shutdown() == false)
    {
      fd_set read_fds_arg= read_fds;

      /*
        When using valgrind 2.0 this syscall doesn't get kicked off by a
        signal during shutdown. This results in failing assert
        (Thread_registry::~Thread_registry). Valgrind 2.2 works fine.
      */
      int rc= select(n, &read_fds_arg, 0, 0, 0);


      if (rc == -1 && errno != EINTR)
        log_error("Listener_thread::run(): select() failed, %s",
                  strerror(errno));
      else
      {
        /* Assuming that rc > 0 as we asked to wait forever */
        if (FD_ISSET(unix_socket, &read_fds_arg))
        {
          int client_fd= accept(unix_socket, 0, 0);
          /* accept may return -1 (failure or spurious wakeup) */
          if (client_fd >= 0)                    // connection established
          {
            if (Vio *vio= vio_new(client_fd, VIO_TYPE_SOCKET, 1))
              handle_new_mysql_connection(vio);
            else
            {
              shutdown(client_fd, SHUT_RDWR);
              close(client_fd);
            }
          }
        }
        else
          if (FD_ISSET(ip_socket, &read_fds_arg))
          {
            int client_fd= accept(ip_socket, 0, 0);
            /* accept may return -1 (failure or spurious wakeup) */
            if (client_fd >= 0)                    // connection established
            {
              if (Vio *vio= vio_new(client_fd, VIO_TYPE_TCPIP, 0))
              {
                handle_new_mysql_connection(vio);
              }
              else
              {
                shutdown(client_fd, SHUT_RDWR);
                close(client_fd);
              }
            }
        }
      }
    }
  }

  /* III. Release all resources and exit */

  log_info("Listener_thread::run(): shutdown requested, exiting...");

  close(unix_socket);
  close(ip_socket);
  unlink(unix_socket_address.sun_path);

  thread_registry.unregister_thread(&thread_info);
  my_thread_end();
  return;

err:
  thread_registry.unregister_thread(&thread_info);
  thread_registry.request_shutdown();
  my_thread_end();
  return;
}


/*
  Create new mysql connection. Created thread is responsible for deletion of
  the Mysql_connection_thread_args and Vio instances passed to it.
  SYNOPSYS
    handle_new_mysql_connection()
*/

void Listener_thread::handle_new_mysql_connection(Vio *vio)
{
  if (Mysql_connection_thread_args *mysql_thread_args=
      new Mysql_connection_thread_args(vio, thread_registry, user_map,
                                       ++total_connection_count,
                                       instance_map)
      )
  {
    /*
      Initialize thread attributes to create detached thread; it seems
      easier to do it ad-hoc than have a global variable for attributes.
    */
    pthread_t mysql_thd_id;
    pthread_attr_t mysql_thd_attr;
    pthread_attr_init(&mysql_thd_attr);
    pthread_attr_setdetachstate(&mysql_thd_attr, PTHREAD_CREATE_DETACHED);
    if (pthread_create(&mysql_thd_id, &mysql_thd_attr, mysql_connection,
                       mysql_thread_args))
    {
      delete mysql_thread_args;
      vio_delete(vio);
      log_error("handle_one_mysql_connection(): pthread_create(mysql) failed");
    }
    pthread_attr_destroy(&mysql_thd_attr);
  }
  else
    vio_delete(vio);
}


C_MODE_START


pthread_handler_decl(listener, arg)
{
  Listener_thread_args *args= (Listener_thread_args *) arg;
  Listener_thread listener(*args);
  listener.run();
  /*
    args is a stack variable because listener thread lives as long as the
    manager process itself
  */
  return 0;
}


C_MODE_END