root/platform/mysql/mycached/trunk/mycached.hh @ 35004

/*
 * mycached - memcached protocol handler for mysqld
 * 
 * Copyright (C) 2009 Cybozu Labs, Inc.
 * 
 * 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 FreeSoftware
 * Foundation; either 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
 */

#ifndef mycached_hh
#define mycached_hh

#include <cassert>
#include <stdexcept>
#include <string>
#include <vector>
extern "C" {
#include "picoev.h"
}

#define MYSQL_SERVER 1

#include "mysql_priv.h"
#include "sql_select.h"

#define NUM_THREADS 4
#define MAX_FDS 1024
#define TIMEOUT_SECS 3600
#define REQBUF_SZ 1048576

namespace mycached {
  
  struct error_t : public std::domain_error {
    error_t(const std::string& what) : std::domain_error(what) {}
  };
  
  struct range_string {
    const char* first_, * last_;
    range_string() : first_(NULL), last_(NULL) {}
    range_string(const char* first, const char* last)
      : first_(first), last_(last) {}
    size_t size() const { return last_ - first_; }
    void clear() { first_ = last_ = NULL; }
    bool to_int(long long& v) const {
      v = 0;
      bool is_negative = false;
      const char* p = first_;
      if (p != last_ && *p == '-') {
        is_negative = true;
        ++p;
      }
      if (p == last_) {
        return false;
      }
      do {
        if (! ('0' <= *p && *p <= '9')) {
          return false;
        }
        v = v * 10 + *p - '0';
      } while (++p != last_);
      if (is_negative) {
        v *= -1;
      }
      return v;
    }
  };

  inline
  bool
  operator==(const range_string& x, const range_string& y)
  {
    if (x.size() != y.size()) {
      return false;
    }
    if (x.first_ == y.first_) {
      return true;
    }
    if (x.first_ == NULL || y.first_ == NULL) {
      return true;
    }
    for (const char* xp = x.first_, * yp = y.first_;
         xp != x.last_;
         ++xp, ++yp) {
      if (*xp != *yp) {
        return false;
      }
    }
    return true;
  }
  
  inline
  bool
  operator!=(const range_string& x, const range_string& y)
  {
    return ! (x == y);
  }

  struct key_t {
    range_string db_;
    range_string table_;
    range_string key_str_;
    uchar key_value_[8];
    range_string entire_key() const {
      return range_string(db_.first_, key_str_.last_);
    }
  };

  struct fetcher_t {
    THD* thd_;
    Query_tables_list tables_;
    size_t num_tables_;
    const std::vector<key_t>& keys_;
    enum {
      state_not_locked,
      state_locked,
      state_index_is_open
    } state_;
    fetcher_t(THD* thd, const std::vector<key_t>& keys);
    ~fetcher_t();
    void fetch(std::string& response);
    void register_table(const range_string& db, const range_string& table);
    TABLE_LIST* get_table(const range_string& db, const range_string& table,
                          size_t& idx) {
      idx = 0;
      for (TABLE_LIST* tl = tables_.query_tables;
           tl != NULL;
           tl = tl->next_global, ++idx) {
        if (table.size() == tl->table_name_length
            && strncmp(table.first_, tl->table_name, table.size()) == 0
            && db.size() == tl->db_length
            && strncmp(db.first_, tl->db, db.size()) == 0) {
          return tl;
        }
      }
      return NULL;
    }
    static void build_response(const key_t& key, TABLE* table,
                               std::string& response);
  };
  
  struct conn_t {
    enum {
      rr_invalid_request = 0,
      rr_too_long,
      rr_closed
    };
    THD* thd_;
    int sock_;
    size_t res_off_;
    std::string res_buf_;
    size_t req_off_;
    char req_buf_[REQBUF_SZ]; // TODO shrinkable
    static int listen_sock_;
    conn_t(THD* thd, picoev_loop* loop, int sock)
      : thd_(thd), sock_(sock), res_off_(0), res_buf_(), req_off_(0) {
      picoev_add(loop, sock_, PICOEV_READ, TIMEOUT_SECS, handle_conn, this);
    }
    size_t parse_request(std::vector<key_t>& keys, bool& is_partial) const;
    bool build_send_response(const std::vector<key_t>& keys);
    bool handle_request();
    void handle_conn(picoev_loop* loop);
    static const char* get_token(const char* s, const char* s_end,
                                 range_string& token) {
      const char* first = s;
      for (; s != s_end; ++s) {
        switch (*s) {
        case '\r': case '\n': case ' ': case '.':
          token = range_string(first, s);
          return s;
        default:
          break;
        }
      }
      token.clear();
      return NULL;
    }
    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);
  };

}

#endif