Changeset 35045 for platform/mysql

Timestamp:

08/25/09 18:13:43 (4 years ago)

Author:

kazuho

Message:

add mycached_stop()

Location:

platform/mysql/mycached/trunk

Files:

• cac (added)
• cac/cac_mutex.h (added)
• cac/cac_rwlock.h (added)
• cac/test_mutex.cc (added)
• cac/test_rwlock.cc (added)
• mycached.cc (modified) (5 diffs)
• mycached.hh (modified) (4 diffs)
• mycached_as_udf.cc (modified) (1 diff)
• picoev.h (modified) (1 diff)

platform/mysql/mycached/trunk/mycached.cc

@@ -21 +21 @@
 #include <sys/uio.h>
 }
+#include "cac/cac_mutex.h"
 #include "mycached.hh"
 
@@ -30 +31 @@
 #define SEND_CLIENT_ERROR(sock, es) send_error(sock, "CLIENT_ERROR ", es)
 #define SEND_SERVER_ERROR(sock, es) send_error(sock, "SERVER_ERROR ", es)
+
+struct mycached_server_info {
+  int listen_sock_;
+  pthread_t worker_threads_[NUM_THREADS];
+  mycached_server_info() : listen_sock_(-1) {}
+};
+
+static cac_mutex_t<mycached_server_info> server_info_(NULL);
 
 inline void setup_thd(THD** thd) {
@@ -310 +319 @@
 }
 
-int conn_t::listen_sock_ = -1;
+thread_t::thread_t(THD* thd, int listen_sock)
+  : thd_(thd), listen_sock_(listen_sock), loop_(NULL)
+{
+  loop_ = picoev_create_loop(TIMEOUT_SECS);
+  picoev_add(loop_, listen_sock_, PICOEV_READ, 0, accept_conn, this);
+  memset(buf_pool_, 0, sizeof(buf_pool_));
+}
+
+thread_t::~thread_t()
+{
+  picoev_del(loop_, listen_sock_);
+  close(listen_sock_);
+  picoev_destroy_loop(loop_);
+  for (int i = 0; i < NUM_POOLED_BUFS_PER_THREAD; ++i) {
+    free((void*)((long)buf_pool_[i] & ~(PICOEV_PAGE_SIZE - 1)));
+  }
+  delete thd_;
+}
+
+void
+thread_t::accept_conn()
+{
+  int sock = accept(listen_sock_, NULL, NULL);
+  if (sock != -1) {
+    if (sock >= MAX_FDS) {
+      SEND_SERVER_ERROR(sock, "too many connections");
+      close(sock);
+      return;
+    }
+    setup_sock(sock);
+    new conn_t(this, sock);
+  }
+}
+
+void
+thread_t::event_loop()
+{
+  setup_thd(&thd_);
+
+  // main loop
+  while (cac_mutex_t<mycached_server_info>::lockref(server_info_)->listen_sock_
+         != -1) {
+    picoev_loop_once(loop_, 1);
+  }
+  
+  // close all sockets
+  for (int fd = -1;
+       (fd = picoev_next_fd(loop_, fd)) != -1;
+       ) {
+    if (fd != listen_sock_) {
+      delete reinterpret_cast<conn_t*>(picoev.fds[fd].cb_arg);
+    }
+  }
+}
+
+void
+thread_t::accept_conn(picoev_loop* loop, int fd, int, void* cb_arg)
+{
+  thread_t* thd = reinterpret_cast<thread_t*>(cb_arg);
+  thd->accept_conn();
+}
+
+void*
+thread_t::event_loop(void* _thd)
+{
+  thread_t* thd = (thread_t*)_thd;
+  thd->event_loop();
+  delete thd;
+  return NULL;
+}
+
+bool
+thread_t::start_server(unsigned host, unsigned short port)
+{
+  cac_mutex_t<mycached_server_info>::lockref server_info(server_info_);
+  
+  if (server_info->listen_sock_ != -1) {
+    sql_print_error("mycached is already running");
+    return false;
+  }
+  
+  server_info->listen_sock_ = socket(AF_INET, SOCK_STREAM, 0);
+  assert(server_info->listen_sock_ != -1);
+  int on = 1;
+  setsockopt(server_info->listen_sock_, SOL_SOCKET, SO_REUSEADDR, &on,
+             sizeof(on));
+  sockaddr_in listen_addr;
+  listen_addr.sin_family = AF_INET;
+  listen_addr.sin_port = htons(port);
+  listen_addr.sin_addr.s_addr = htonl(host);
+  if (bind(server_info->listen_sock_, (sockaddr*)&listen_addr,
+           sizeof(listen_addr))
+      != 0
+      || listen(server_info->listen_sock_, 128) != 0) {
+    close(server_info->listen_sock_);
+    server_info->listen_sock_ = -1;
+    sql_print_error("could not listen to port\n");
+    return false;
+  }
+  
+  if (picoev.max_fd == 0) {
+    picoev_init(MAX_FDS);
+  }
+  
+  for (int i = 0; i < NUM_THREADS; ++i) {
+    // dup socket for per-thread accept(2)
+    int sock = dup(server_info->listen_sock_);
+    assert(sock != -1);
+    setup_sock(sock);
+    pthread_create(server_info->worker_threads_ + i, NULL, event_loop,
+                   new thread_t(new THD(), sock));
+  }
+  
+  return true;
+}
+
+bool
+thread_t::stop_server()
+{
+  { // close global listen sock
+    cac_mutex_t<mycached_server_info>::lockref server_info(server_info_);
+    if (server_info->listen_sock_ == -1) {
+      return false;
+    }
+    close(server_info->listen_sock_);
+    server_info->listen_sock_ = -1;
+  }
+  
+  // wait for all threads to terminate
+  for (int i = 0; i < NUM_THREADS; ++i) {
+    pthread_t tid
+      = cac_mutex_t<mycached_server_info>::lockref(server_info_)
+      ->worker_threads_[i];
+    pthread_join(tid, NULL);
+  }
+  
+  return true;
+}
 
 size_t
@@ -523 +669 @@
   
  CLOSE_CONN:
-  // all cleanups should be done here
-  if (req_buf_ != NULL) {
-    thd_->return_buf(req_buf_);
-    req_buf_ = NULL;
-  }
-  picoev_del(loop, sock_);
-  close(sock_);
   delete this;
 }
@@ -539 +678 @@
   self->handle_conn(loop);
 }
-
-void
-conn_t::accept_conn(picoev_loop* loop, int fd, int, void* cb_arg)
-{
-  thread_t* thd = reinterpret_cast<thread_t*>(cb_arg);
-  int sock = accept(fd, NULL, NULL);
-  if (sock == -1) {
-    return;
-  }
-  if (sock >= MAX_FDS) {
-    SEND_SERVER_ERROR(sock, "too many connections");
-    close(sock);
-    return;
-  }
-  setup_sock(sock);
-  new conn_t(thd, loop, sock);
-}
-
-void*
-conn_t::event_loop(void* _thd)
-{
-  thread_t thd((THD*)_thd);
-  setup_thd(&thd.thd_);
-
-  picoev_loop* loop = picoev_create_loop(TIMEOUT_SECS);
-  // use a dup socket instead since it is not possible to share a single
-  // file descriptor with multiple threads using picoev
-  int sock = dup(listen_sock_);
-  assert(sock != -1);
-  setup_sock(sock);
-  picoev_add(loop, sock, PICOEV_READ, 0, accept_conn, &thd);
-  while (1) {
-    picoev_loop_once(loop, 60);
-  }
-  picoev_destroy_loop(loop);
-  
-  delete thd.thd_;
-  return NULL;
-}
-
-bool
-conn_t::start_server(unsigned host, unsigned short port)
-{
-  if (listen_sock_ != -1) {
-    sql_print_error("mycached is already running");
-    return false;
-  }
-  
-  listen_sock_ = socket(AF_INET, SOCK_STREAM, 0);
-  assert(listen_sock_ != -1);
-  int on = 1;
-  setsockopt(listen_sock_, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
-  sockaddr_in listen_addr;
-  listen_addr.sin_family = AF_INET;
-  listen_addr.sin_port = htons(port);
-  listen_addr.sin_addr.s_addr = htonl(host);
-  if (bind(listen_sock_, (sockaddr*)&listen_addr, sizeof(listen_addr)) != 0
-      || listen(listen_sock_, 128) != 0) {
-    close(listen_sock_);
-    listen_sock_ = -1;
-    sql_print_error("could not listen to port\n");
-    return false;
-  }
-  
-  if (picoev.max_fd == 0) {
-    picoev_init(MAX_FDS);
-  }
-  for (int i = 0; i < NUM_THREADS; ++i) {
-    pthread_t tid;
-    pthread_create(&tid, NULL, event_loop, new THD());
-    pthread_detach(tid);
-  }
-  
-  return true;
-}

platform/mysql/mycached/trunk/mycached.hh

@@ -156 +156 @@
   struct thread_t {
     THD* thd_;
+    int listen_sock_; // per thread socket
+    picoev_loop* loop_;
     char* buf_pool_[NUM_POOLED_BUFS_PER_THREAD];
-    thread_t(THD* thd) : thd_(thd) {
-      memset(buf_pool_, 0, sizeof(buf_pool_));
-    }
-    ~thread_t() {
-      for (int i = 0; i < NUM_POOLED_BUFS_PER_THREAD; ++i) {
-        free((void*)((long)buf_pool_[i] & ~(PICOEV_PAGE_SIZE - 1)));
-      }
-    }
+    thread_t(THD* thd, int listen_sock);
+    ~thread_t();
+    void accept_conn();
+    void event_loop();
     char* steal_buf() {
       char* r = buf_pool_[0];
@@ -185 +183 @@
       }
     }
+    static void accept_conn(picoev_loop* loop, int fd, int, void* cb_arg);
+    static void* event_loop(void* _thd);
+    static bool start_server(unsigned host, unsigned short port);
+    static bool stop_server();
   };
   
@@ -199 +201 @@
     size_t req_off_;
     char* req_buf_;
-    static int listen_sock_;
-    conn_t(thread_t* thd, picoev_loop* loop, int sock)
+    conn_t(thread_t* thd, int sock)
       : thd_(thd), sock_(sock), res_off_(0), res_buf_(), req_off_(0),
         req_buf_(NULL) {
-      picoev_add(loop, sock_, PICOEV_READ, TIMEOUT_SECS, handle_conn, this);
+      picoev_add(thd_->loop_, sock_, PICOEV_READ, TIMEOUT_SECS, handle_conn,
+                 this);
+    }
+    ~conn_t() {
+      if (req_buf_ != NULL) {
+        thd_->return_buf(req_buf_);
+      }
+      picoev_del(thd_->loop_, sock_);
+      close(sock_);
     }
     // all cleanups done in handle_conn
@@ -226 +235 @@
     }
     static void handle_conn(picoev_loop* loop, int, int, void* cb_arg);
-    static void accept_conn(picoev_loop* loop, int fd, int, void* cb_arg);
-    static void* event_loop(void* _thd);
-    static bool start_server(unsigned host, unsigned short port);
   };
 

platform/mysql/mycached/trunk/mycached_as_udf.cc

@@ -67 +67 @@
 mycached_start(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
 {
-  return mycached::conn_t::start_server(*(long long*)args->args[0],
-                                        *(long long*)args->args[1]);
+  return mycached::thread_t::start_server(*(long long*)args->args[0],
+                                          *(long long*)args->args[1]);
 }
+
+extern "C"
+my_bool
+mycached_stop_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
+{
+  if (args->arg_count != 0) {
+    strcpy(message, "mycached_stop(): invalid arguments");
+    return 1;
+  }
+  initid->maybe_null = 0;
+  return 0;
+}
+
+extern "C"
+void
+mycached_stop_deinit(UDF_INIT* initid)
+{
+}
+
+extern "C"
+long long
+mycached_stop(UDF_INIT* initid, UDF_ARGS* args, char* is_null, char* error)
+{
+  return mycached::thread_t::stop_server();
+}

platform/mysql/mycached/trunk/picoev.h

@@ -274 +274 @@
     }
     return 0;
+  }
+  
+  /* function to iterate registered information. To start iteration, set curfd
+     to -1 and call the function until -1 is returned */
+  PICOEV_INLINE
+  int picoev_next_fd(picoev_loop* loop, int curfd) {
+    if (curfd != -1) {
+      assert(PICOEV_IS_INITED_AND_FD_IN_RANGE(curfd));
+    }
+    while (++curfd < picoev.max_fd) {
+      if (loop->loop_id == picoev.fds[curfd].loop_id) {
+        return curfd;
+      }
+    }
+    return -1;
   }