<commit_before>
// vim:sw=2:ai

/*
 * Copyright (C) 2010 DeNA Co.,Ltd.. All rights reserved.
 * See COPYRIGHT.txt for details.
 */

#include <stdlib.h>
#include <stdio.h>
#include <string.h>

#include "database.hpp"
#include "string_util.hpp"
#include "escape.hpp"
#include "mysql_incl.hpp"

#define DBG_KEY(x)
#define DBG_SHUT(x)
#define DBG_LOCK(x)
#define DBG_THR(x)
#define DBG_CMP(x)
#define DBG_FLD(x)
#define DBG_FILTER(x)
#define DBG_REFCNT(x)
#define DBG_KEYLEN(x)
#define DBG_DELETED

/* status variables */
unsigned long long int open_tables_count;
unsigned long long int close_tables_count;
unsigned long long int lock_tables_count;
unsigned long long int unlock_tables_count;
unsigned long long int index_exec_count;

namespace dena {

prep_stmt::prep_stmt()
  : dbctx(0), table_id(static_cast<size_t>(-1)),
    idxnum(static_cast<size_t>(-1))
{
}
prep_stmt::prep_stmt(dbcontext_i *c, size_t tbl, size_t idx,
  const fields_type& rf, const fields_type& ff)
  : dbctx(c), table_id(tbl), idxnum(idx), ret_fields(rf), filter_fields(ff)
{
  if (dbctx) {
    dbctx->table_addref(table_id);
  }
}
prep_stmt::~prep_stmt()
{
  if (dbctx) {
    dbctx->table_release(table_id);
  }
}

prep_stmt::prep_stmt(const prep_stmt& x)
  : dbctx(x.dbctx), table_id(x.table_id), idxnum(x.idxnum),
  ret_fields(x.ret_fields), filter_fields(x.filter_fields)
{
  if (dbctx) {
    dbctx->table_addref(table_id);
  }
}

prep_stmt&
prep_stmt::operator =(const prep_stmt& x)
{
  if (this != &x) {
    if (dbctx) {
      dbctx->table_release(table_id);
    }
    dbctx = x.dbctx;
    table_id = x.table_id;
    idxnum = x.idxnum;
    ret_fields = x.ret_fields;
    filter_fields = x.filter_fields;
    if (dbctx) {
      dbctx->table_addref(table_id);
    }
  }
  return *this;
}

struct database : public database_i, private noncopyable {
  database(const config& c);
  virtual ~database();
  virtual dbcontext_ptr create_context(bool for_write) volatile;
  virtual void stop() volatile;
  virtual const config& get_conf() const volatile;
 public:
  int child_running;
 private:
  config conf;
};

struct tablevec_entry {
  TABLE *table;
  size_t refcount;
  bool modified;
  tablevec_entry() : table(0), refcount(0), modified(false) { }
};

struct expr_user_lock : private noncopyable {
  expr_user_lock(THD *thd, int timeout)
    : lck_key("handlersocket_wr", 16, &my_charset_latin1),
      lck_timeout(timeout),
      lck_func_get_lock(&lck_key, &lck_timeout),
      lck_func_release_lock(&lck_key)
  {
    lck_key.fix_fields(thd, 0);
    lck_timeout.fix_fields(thd, 0);
    lck_func_get_lock.fix_fields(thd, 0);
    lck_func_release_lock.fix_fields(thd, 0);
  }
  long long get_lock() {
    return lck_func_get_lock.val_int();
  }
  long long release_lock() {
    return lck_func_release_lock.val_int();
  }
 private:
  Item_string lck_key;
  Item_int lck_timeout;
  Item_func_get_lock lck_func_get_lock;
  Item_func_release_lock lck_func_release_lock;
};

struct dbcontext : public dbcontext_i, private noncopyable {
  dbcontext(volatile database *d, bool for_write);
  virtual ~dbcontext();
  virtual void init_thread(const void *stack_botton,
    volatile int& shutdown_flag);
  virtual void term_thread();
  virtual bool check_alive();
  virtual void lock_tables_if();
  virtual void unlock_tables_if();
  virtual bool get_commit_error();
  virtual void clear_error();
  virtual void close_tables_if();
  virtual void table_addref(size_t tbl_id);
  virtual void table_release(size_t tbl_id);
  virtual void cmd_open_index(dbcallback_i& cb, size_t pst_id, const char *dbn,
    const char *tbl, const char *idx, const char *retflds,
    const char *filflds);
  virtual void cmd_exec_on_index(dbcallback_i& cb, const cmd_exec_args& args);
  virtual void set_statistics(size_t num_conns, size_t num_active);
 private:
  int set_thread_message(const char *fmt, ...)
    __attribute__((format (printf, 2, 3)));
  bool parse_fields(TABLE *const table, const char *str,
    prep_stmt::fields_type& flds);
  void cmd_insert_internal(dbcallback_i& cb, const prep_stmt& pst,
    const string_ref *fvals, size_t fvalslen);
  void cmd_sql_internal(dbcallback_i& cb, const prep_stmt& pst,
    const string_ref *fvals, size_t fvalslen);
  void cmd_find_internal(dbcallback_i& cb, const prep_stmt& pst,
    ha_rkey_function find_flag, const cmd_exec_args& args);
  size_t calc_filter_buf_size(TABLE *table, const prep_stmt& pst,
    const record_filter *filters);
  bool fill_filter_buf(TABLE *table, const prep_stmt& pst,
    const record_filter *filters, uchar *filter_buf, size_t len);
  int check_filter(dbcallback_i& cb, TABLE *table, const prep_stmt& pst,
    const record_filter *filters, const uchar *filter_buf);
  void resp_record(dbcallback_i& cb, TABLE *const table, const prep_stmt& pst);
  void dump_record(dbcallback_i& cb, TABLE *const table, const prep_stmt& pst);
  int modify_record(dbcallback_i& cb, TABLE *const table,
    const prep_stmt& pst, const cmd_exec_args& args, char mod_op,
    size_t& modified_count);
 private:
  typedef std::vector<tablevec_entry> table_vec_type;
  typedef std::pair<std::string, std::string> table_name_type;
  typedef std::map<table_name_type, size_t> table_map_type;
 private:
  volatile database *const dbref;
  bool for_write_flag;
  THD *thd;
  MYSQL_LOCK *lock;
  bool lock_failed;
  std::auto_ptr<expr_user_lock> user_lock;
  int user_level_lock_timeout;
  bool user_level_lock_locked;
  bool commit_error;
  std::vector<char> info_message_buf;
  table_vec_type table_vec;
  table_map_type table_map;
  #if MYSQL_VERSION_ID >= 50505
  MDL_request *mdl_request;
  #else
  void *mdl_request;
  #endif
};

database::database(const config& c)
  : child_running(1), conf(c)
{
}

database::~database()
{
}

dbcontext_ptr
database::create_context(bool for_write) volatile
{
  return dbcontext_ptr(new dbcontext(this, for_write));
}

void
database::stop() volatile
{
  child_running = false;
}

const config&
database::get_conf() const volatile
{
  return const_cast<const config&>(conf);
}

database_ptr
database_i::create(const config& conf)
{
  return database_ptr(new database(conf));
}

dbcontext::dbcontext(volatile database *d, bool for_write)
  : dbref(d), for_write_flag(for_write), thd(0), lock(0), lock_failed(false),
    user_level_lock_timeout(0), user_level_lock_locked(false),
    commit_error(false), mdl_request(0)
{
  info_message_buf.resize(8192);
  user_level_lock_timeout = d->get_conf().get_int("wrlock_timeout", 12);
}

dbcontext::~dbcontext()
{
}

namespace {

int
wait_server_to_start(THD *thd, volatile int& shutdown_flag)
{
  int r = 0;
  DBG_SHUT(fprintf(stderr, "HNDSOCK wsts\n"));
  pthread_mutex_lock(&LOCK_server_started);
  while (!mysqld_server_started) {
    timespec abstime = { };
    set_timespec(abstime, 1);
    pthread_cond_timedwait(&COND_server_started, &LOCK_server_started,
      &abstime);
    pthread_mutex_unlock(&LOCK_server_started);
    pthread_mutex_lock(&thd->mysys_var->mutex);
    THD::killed_state st = thd->killed;
    pthread_mutex_unlock(&thd->mysys_var->mutex);
    DBG_SHUT(fprintf(stderr, "HNDSOCK wsts kst %d\n", (int)st));
    pthread_mutex_lock(&LOCK_server_started);
    if (st != THD::NOT_KILLED) {
      DBG_SHUT(fprintf(stderr, "HNDSOCK wsts kst %d break\n", (int)st));
      r = -1;
      break;
    }
    if (shutdown_flag) {
      DBG_SHUT(fprintf(stderr, "HNDSOCK wsts kst shut break\n"));
      r = -1;
      break;
    }
  }
  pthread_mutex_unlock(&LOCK_server_started);
  DBG_SHUT(fprintf(stderr, "HNDSOCK wsts done\n"));
  return r;
}

}; // namespace

void
dbcontext::init_thread(const void *stack_bottom, volatile int& shutdown_flag)
{
  DBG_THR(fprintf(stderr, "HNDSOCK init thread\n"));
  {
    my_thread_init();
    thd = new THD;
    thd->thread_stack = (char *)stack_bottom;
    DBG_THR(const size_t of = (char *)(&thd->thread_stack) - (char *)thd);
    DBG_THR(fprintf(stderr, "thread_stack = %p sz=%zu of=%zu\n",
      thd->thread_stack, sizeof(THD), of));
    thd->store_globals();
    thd->system_thread = static_cast<enum_thread_type>(1<<30UL);
    const NET v = { 0 };
    thd->net = v;
    if (for_write_flag) {
      #if MYSQL_VERSION_ID >= 50505
      thd->variables.option_bits |= OPTION_BIN_LOG;
      #else
      thd->options |= OPTION_BIN_LOG;
      #endif
      safeFree(thd->db);
      thd->db = 0;
      thd->db = my_strdup("handlersocket", MYF(0));
    }
    my_pthread_setspecific_ptr(THR_THD, thd);
    DBG_THR(fprintf(stderr, "HNDSOCK x0 %p\n", thd));
  }
  {
    pthread_mutex_lock(&LOCK_thread_count);
    thd->thread_id = thread_id++;
    threads.append(thd);
    ++thread_count;
    pthread_mutex_unlock(&LOCK_thread_count);
  }

  DBG_THR(fprintf(stderr, "HNDSOCK init thread wsts\n"));
  wait_server_to_start(thd, shutdown_flag);
  DBG_THR(fprintf(stderr, "HNDSOCK init thread done\n"));

  thd_proc_info(thd, &info_message_buf[0]);
  set_thread_message("hs:listening");
  DBG_THR(fprintf(stderr, "HNDSOCK x1 %p\n", thd));

  #if MYSQL_VERSION_ID >= 50508
  mdl_request = new(thd->mem_root) MDL_request;
  mdl_request->init(MDL_key::TABLE, "", "",
    for_write_flag ?  MDL_SHARED_WRITE : MDL_SHARED_READ, MDL_STATEMENT);
  #elif MYSQL_VERSION_ID >= 50505
  mdl_request = MDL_request::create(MDL_key::TABLE, "", "",
    for_write_flag ?  MDL_SHARED_WRITE : MDL_SHARED_READ, thd->mem_root);
  #endif

  lex_start(thd);

  user_lock.reset(new expr_user_lock(thd, user_level_lock_timeout));
}

int
dbcontext::set_thread_message(const char *fmt, ...)
{
  va_list ap;
  va_start(ap, fmt);
  const int n = vsnprintf(&info_message_buf[0], info_message_buf.size(),
    fmt, ap);
  va_end(ap);
  return n;
}

void
dbcontext::term_thread()
{
  DBG_THR(fprintf(stderr, "HNDSOCK thread end %p\n", thd));
  unlock_tables_if();
  my_pthread_setspecific_ptr(THR_THD, 0);
  {
    pthread_mutex_lock(&LOCK_thread_count);
    delete thd;
    thd = 0;
    --thread_count;
    pthread_mutex_unlock(&LOCK_thread_count);
    my_thread_end();
  }
}

bool
dbcontext::check_alive()
{
  pthread_mutex_lock(&thd->mysys_var->mutex);
  THD::killed_state st = thd->killed;
  pthread_mutex_unlock(&thd->mysys_var->mutex);
  DBG_SHUT(fprintf(stderr, "chk HNDSOCK kst %p %p %d %zu\n", thd, &thd->killed,
    (int)st, sizeof(*thd)));
  if (st != THD::NOT_KILLED) {
    DBG_SHUT(fprintf(stderr, "chk HNDSOCK kst %d break\n", (int)st));
    return false;
  }
  return true;
}

void
dbcontext::lock_tables_if()
{
  if (lock_failed) {
    return;
  }
  if (for_write_flag && !user_level_lock_locked) {
    if (user_lock->get_lock()) {
      user_level_lock_locked = true;
    } else {
      lock_failed = true;
      return;
    }
  }
  if (lock == 0) {
    const size_t num_max = table_vec.size();
    TABLE *tables[num_max ? num_max : 1]; /* GNU */
    size_t num_open = 0;
    for (size_t i = 0; i < num_max; ++i) {
      if (table_vec[i].refcount > 0) {
	tables[num_open++] = table_vec[i].table;
      }
      table_vec[i].modified = false;
    }
    #if MYSQL_VERSION_ID >= 50505
    lock = thd->lock = mysql_lock_tables(thd, &tables[0], num_open, 0);
    #else
    bool need_reopen= false;
    lock = thd->lock = mysql_lock_tables(thd, &tables[0], num_open,
      MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN, &need_reopen);
    #endif
    statistic_increment(lock_tables_count, &LOCK_status);
    thd_proc_info(thd, &info_message_buf[0]);
    DENA_VERBOSE(100, fprintf(stderr, "HNDSOCK lock tables %p %p %zu %zu\n",
      thd, lock, num_max, num_open));
    if (lock == 0) {
      lock_failed = true;
      DENA_VERBOSE(10, fprintf(stderr, "HNDSOCK failed to lock tables %p\n",
	thd));
    }
    if (for_write_flag) {
      #if MYSQL_VERSION_ID >= 50505
      thd->set_current_stmt_binlog_format_row();
      #else
      thd->current_stmt_binlog_row_based = 1;
      #endif
    }
  }
  DBG_LOCK(fprintf(stderr, "HNDSOCK tblnum=%d\n", (int)tblnum));
}

void
dbcontext::unlock_tables_if()
{
  if (lock != 0) {
    DENA_VERBOSE(100, fprintf(stderr, "HNDSOCK unlock tables\n"));
    if (for_write_flag) {
      for (size_t i = 0; i < table_vec.size(); ++i) {
	if (table_vec[i].modified) {
	  query_cache_invalidate3(thd, table_vec[i].table, 0);
	    /* invalidate immediately */
	}
      }
      bool suc = true;
      #if MYSQL_VERSION_ID >= 50505
      suc = (trans_commit_stmt(thd) == 0);
      #else
      suc = (ha_autocommit_or_rollback(thd, 0) == 0);
      #endif<commit_msg>temp commit: query cache<commit_after>
// vim:sw=2:ai

/*
 * Copyright (C) 2010 DeNA Co.,Ltd.. All rights reserved.
 * See COPYRIGHT.txt for details.
 */

#include <stdlib.h>
#include <stdio.h>
#include <string.h>

#include "database.hpp"
#include "string_util.hpp"
#include "escape.hpp"
#include "mysql_incl.hpp"

#define DBG_KEY(x)
#define DBG_SHUT(x)
#define DBG_LOCK(x)
#define DBG_THR(x)
#define DBG_CMP(x)
#define DBG_FLD(x)
#define DBG_FILTER(x)
#define DBG_REFCNT(x)
#define DBG_KEYLEN(x)
#define DBG_DELETED

/* status variables */
unsigned long long int open_tables_count;
unsigned long long int close_tables_count;
unsigned long long int lock_tables_count;
unsigned long long int unlock_tables_count;
unsigned long long int index_exec_count;

namespace dena {

prep_stmt::prep_stmt()
  : dbctx(0), table_id(static_cast<size_t>(-1)),
    idxnum(static_cast<size_t>(-1))
{
}
prep_stmt::prep_stmt(dbcontext_i *c, size_t tbl, size_t idx,
  const fields_type& rf, const fields_type& ff)
  : dbctx(c), table_id(tbl), idxnum(idx), ret_fields(rf), filter_fields(ff)
{
  if (dbctx) {
    dbctx->table_addref(table_id);
  }
}
prep_stmt::~prep_stmt()
{
  if (dbctx) {
    dbctx->table_release(table_id);
  }
}

prep_stmt::prep_stmt(const prep_stmt& x)
  : dbctx(x.dbctx), table_id(x.table_id), idxnum(x.idxnum),
  ret_fields(x.ret_fields), filter_fields(x.filter_fields)
{
  if (dbctx) {
    dbctx->table_addref(table_id);
  }
}

prep_stmt&
prep_stmt::operator =(const prep_stmt& x)
{
  if (this != &x) {
    if (dbctx) {
      dbctx->table_release(table_id);
    }
    dbctx = x.dbctx;
    table_id = x.table_id;
    idxnum = x.idxnum;
    ret_fields = x.ret_fields;
    filter_fields = x.filter_fields;
    if (dbctx) {
      dbctx->table_addref(table_id);
    }
  }
  return *this;
}

struct database : public database_i, private noncopyable {
  database(const config& c);
  virtual ~database();
  virtual dbcontext_ptr create_context(bool for_write) volatile;
  virtual void stop() volatile;
  virtual const config& get_conf() const volatile;
 public:
  int child_running;
 private:
  config conf;
};

struct tablevec_entry {
  TABLE *table;
  size_t refcount;
  bool modified;
  tablevec_entry() : table(0), refcount(0), modified(false) { }
};

struct expr_user_lock : private noncopyable {
  expr_user_lock(THD *thd, int timeout)
    : lck_key("handlersocket_wr", 16, &my_charset_latin1),
      lck_timeout(timeout),
      lck_func_get_lock(&lck_key, &lck_timeout),
      lck_func_release_lock(&lck_key)
  {
    lck_key.fix_fields(thd, 0);
    lck_timeout.fix_fields(thd, 0);
    lck_func_get_lock.fix_fields(thd, 0);
    lck_func_release_lock.fix_fields(thd, 0);
  }
  long long get_lock() {
    return lck_func_get_lock.val_int();
  }
  long long release_lock() {
    return lck_func_release_lock.val_int();
  }
 private:
  Item_string lck_key;
  Item_int lck_timeout;
  Item_func_get_lock lck_func_get_lock;
  Item_func_release_lock lck_func_release_lock;
};

struct dbcontext : public dbcontext_i, private noncopyable {
  dbcontext(volatile database *d, bool for_write);
  virtual ~dbcontext();
  virtual void init_thread(const void *stack_botton,
    volatile int& shutdown_flag);
  virtual void term_thread();
  virtual bool check_alive();
  virtual void lock_tables_if();
  virtual void unlock_tables_if();
  virtual bool get_commit_error();
  virtual void clear_error();
  virtual void close_tables_if();
  virtual void table_addref(size_t tbl_id);
  virtual void table_release(size_t tbl_id);
  virtual void cmd_open_index(dbcallback_i& cb, size_t pst_id, const char *dbn,
    const char *tbl, const char *idx, const char *retflds,
    const char *filflds);
  virtual void cmd_exec_on_index(dbcallback_i& cb, const cmd_exec_args& args);
  virtual void set_statistics(size_t num_conns, size_t num_active);
 private:
  int set_thread_message(const char *fmt, ...)
    __attribute__((format (printf, 2, 3)));
  bool parse_fields(TABLE *const table, const char *str,
    prep_stmt::fields_type& flds);
  void cmd_insert_internal(dbcallback_i& cb, const prep_stmt& pst,
    const string_ref *fvals, size_t fvalslen);
  void cmd_sql_internal(dbcallback_i& cb, const prep_stmt& pst,
    const string_ref *fvals, size_t fvalslen);
  void cmd_find_internal(dbcallback_i& cb, const prep_stmt& pst,
    ha_rkey_function find_flag, const cmd_exec_args& args);
  size_t calc_filter_buf_size(TABLE *table, const prep_stmt& pst,
    const record_filter *filters);
  bool fill_filter_buf(TABLE *table, const prep_stmt& pst,
    const record_filter *filters, uchar *filter_buf, size_t len);
  int check_filter(dbcallback_i& cb, TABLE *table, const prep_stmt& pst,
    const record_filter *filters, const uchar *filter_buf);
  void resp_record(dbcallback_i& cb, TABLE *const table, const prep_stmt& pst);
  void dump_record(dbcallback_i& cb, TABLE *const table, const prep_stmt& pst);
  int modify_record(dbcallback_i& cb, TABLE *const table,
    const prep_stmt& pst, const cmd_exec_args& args, char mod_op,
    size_t& modified_count);
 private:
  typedef std::vector<tablevec_entry> table_vec_type;
  typedef std::pair<std::string, std::string> table_name_type;
  typedef std::map<table_name_type, size_t> table_map_type;
 private:
  volatile database *const dbref;
  bool for_write_flag;
  THD *thd;
  MYSQL_LOCK *lock;
  bool lock_failed;
  std::auto_ptr<expr_user_lock> user_lock;
  int user_level_lock_timeout;
  bool user_level_lock_locked;
  bool commit_error;
  std::vector<char> info_message_buf;
  table_vec_type table_vec;
  table_map_type table_map;
  #if MYSQL_VERSION_ID >= 50505
  MDL_request *mdl_request;
  #else
  void *mdl_request;
  #endif
};

database::database(const config& c)
  : child_running(1), conf(c)
{
}

database::~database()
{
}

dbcontext_ptr
database::create_context(bool for_write) volatile
{
  return dbcontext_ptr(new dbcontext(this, for_write));
}

void
database::stop() volatile
{
  child_running = false;
}

const config&
database::get_conf() const volatile
{
  return const_cast<const config&>(conf);
}

database_ptr
database_i::create(const config& conf)
{
  return database_ptr(new database(conf));
}

dbcontext::dbcontext(volatile database *d, bool for_write)
  : dbref(d), for_write_flag(for_write), thd(0), lock(0), lock_failed(false),
    user_level_lock_timeout(0), user_level_lock_locked(false),
    commit_error(false), mdl_request(0)
{
  info_message_buf.resize(8192);
  user_level_lock_timeout = d->get_conf().get_int("wrlock_timeout", 12);
}

dbcontext::~dbcontext()
{
}

namespace {

int
wait_server_to_start(THD *thd, volatile int& shutdown_flag)
{
  int r = 0;
  DBG_SHUT(fprintf(stderr, "HNDSOCK wsts\n"));
  pthread_mutex_lock(&LOCK_server_started);
  while (!mysqld_server_started) {
    timespec abstime = { };
    set_timespec(abstime, 1);
    pthread_cond_timedwait(&COND_server_started, &LOCK_server_started,
      &abstime);
    pthread_mutex_unlock(&LOCK_server_started);
    pthread_mutex_lock(&thd->mysys_var->mutex);
    THD::killed_state st = thd->killed;
    pthread_mutex_unlock(&thd->mysys_var->mutex);
    DBG_SHUT(fprintf(stderr, "HNDSOCK wsts kst %d\n", (int)st));
    pthread_mutex_lock(&LOCK_server_started);
    if (st != THD::NOT_KILLED) {
      DBG_SHUT(fprintf(stderr, "HNDSOCK wsts kst %d break\n", (int)st));
      r = -1;
      break;
    }
    if (shutdown_flag) {
      DBG_SHUT(fprintf(stderr, "HNDSOCK wsts kst shut break\n"));
      r = -1;
      break;
    }
  }
  pthread_mutex_unlock(&LOCK_server_started);
  DBG_SHUT(fprintf(stderr, "HNDSOCK wsts done\n"));
  return r;
}

}; // namespace

void
dbcontext::init_thread(const void *stack_bottom, volatile int& shutdown_flag)
{
  DBG_THR(fprintf(stderr, "HNDSOCK init thread\n"));
  {
    my_thread_init();
    thd = new THD;
    thd->thread_stack = (char *)stack_bottom;
    DBG_THR(const size_t of = (char *)(&thd->thread_stack) - (char *)thd);
    DBG_THR(fprintf(stderr, "thread_stack = %p sz=%zu of=%zu\n",
      thd->thread_stack, sizeof(THD), of));
    thd->store_globals();
    thd->system_thread = static_cast<enum_thread_type>(1<<30UL);
    const NET v = { 0 };
    thd->net = v;
    if (for_write_flag) {
      #if MYSQL_VERSION_ID >= 50505
      thd->variables.option_bits |= OPTION_BIN_LOG;
      #else
      thd->options |= OPTION_BIN_LOG;
      #endif
      safeFree(thd->db);
      thd->db = 0;
      thd->db = my_strdup("handlersocket", MYF(0));
    }
    my_pthread_setspecific_ptr(THR_THD, thd);
    DBG_THR(fprintf(stderr, "HNDSOCK x0 %p\n", thd));
  }
  {
    pthread_mutex_lock(&LOCK_thread_count);
    thd->thread_id = thread_id++;
    threads.append(thd);
    ++thread_count;
    pthread_mutex_unlock(&LOCK_thread_count);
  }

  DBG_THR(fprintf(stderr, "HNDSOCK init thread wsts\n"));
  wait_server_to_start(thd, shutdown_flag);
  DBG_THR(fprintf(stderr, "HNDSOCK init thread done\n"));

  thd_proc_info(thd, &info_message_buf[0]);
  set_thread_message("hs:listening");
  DBG_THR(fprintf(stderr, "HNDSOCK x1 %p\n", thd));

  #if MYSQL_VERSION_ID >= 50508
  mdl_request = new(thd->mem_root) MDL_request;
  mdl_request->init(MDL_key::TABLE, "", "",
    for_write_flag ?  MDL_SHARED_WRITE : MDL_SHARED_READ, MDL_STATEMENT);
  #elif MYSQL_VERSION_ID >= 50505
  mdl_request = MDL_request::create(MDL_key::TABLE, "", "",
    for_write_flag ?  MDL_SHARED_WRITE : MDL_SHARED_READ, thd->mem_root);
  #endif

  lex_start(thd);

  user_lock.reset(new expr_user_lock(thd, user_level_lock_timeout));
}

int
dbcontext::set_thread_message(const char *fmt, ...)
{
  va_list ap;
  va_start(ap, fmt);
  const int n = vsnprintf(&info_message_buf[0], info_message_buf.size(),
    fmt, ap);
  va_end(ap);
  return n;
}

void
dbcontext::term_thread()
{
  DBG_THR(fprintf(stderr, "HNDSOCK thread end %p\n", thd));
  unlock_tables_if();
  my_pthread_setspecific_ptr(THR_THD, 0);
  {
    pthread_mutex_lock(&LOCK_thread_count);
    delete thd;
    thd = 0;
    --thread_count;
    pthread_mutex_unlock(&LOCK_thread_count);
    my_thread_end();
  }
}

bool
dbcontext::check_alive()
{
  pthread_mutex_lock(&thd->mysys_var->mutex);
  THD::killed_state st = thd->killed;
  pthread_mutex_unlock(&thd->mysys_var->mutex);
  DBG_SHUT(fprintf(stderr, "chk HNDSOCK kst %p %p %d %zu\n", thd, &thd->killed,
    (int)st, sizeof(*thd)));
  if (st != THD::NOT_KILLED) {
    DBG_SHUT(fprintf(stderr, "chk HNDSOCK kst %d break\n", (int)st));
    return false;
  }
  return true;
}

void
dbcontext::lock_tables_if()
{
  if (lock_failed) {
    return;
  }
  if (for_write_flag && !user_level_lock_locked) {
    if (user_lock->get_lock()) {
      user_level_lock_locked = true;
    } else {
      lock_failed = true;
      return;
    }
  }
  if (lock == 0) {
    const size_t num_max = table_vec.size();
    TABLE *tables[num_max ? num_max : 1]; /* GNU */
    size_t num_open = 0;
    for (size_t i = 0; i < num_max; ++i) {
      if (table_vec[i].refcount > 0) {
	tables[num_open++] = table_vec[i].table;
      }
      table_vec[i].modified = false;
    }
    #if MYSQL_VERSION_ID >= 50505
    lock = thd->lock = mysql_lock_tables(thd, &tables[0], num_open, 0);
    #else
    bool need_reopen= false;
    lock = thd->lock = mysql_lock_tables(thd, &tables[0], num_open,
      MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN, &need_reopen);
    #endif
    statistic_increment(lock_tables_count, &LOCK_status);
    thd_proc_info(thd, &info_message_buf[0]);
    DENA_VERBOSE(100, fprintf(stderr, "HNDSOCK lock tables %p %p %zu %zu\n",
      thd, lock, num_max, num_open));
    if (lock == 0) {
      lock_failed = true;
      DENA_VERBOSE(10, fprintf(stderr, "HNDSOCK failed to lock tables %p\n",
	thd));
    }
    if (for_write_flag) {
      #if MYSQL_VERSION_ID >= 50505
      thd->set_current_stmt_binlog_format_row();
      #else
      thd->current_stmt_binlog_row_based = 1;
      #endif
    }
  }
  DBG_LOCK(fprintf(stderr, "HNDSOCK tblnum=%d\n", (int)tblnum));
}

void
dbcontext::unlock_tables_if()
{
  if (lock != 0) {
    DENA_VERBOSE(100, fprintf(stderr, "HNDSOCK unlock tables\n"));
    if (for_write_flag) {
      for (size_t i = 0; i < table_vec.size(); ++i) {
	if (table_vec[i].modified) {
	  query_cache_invalidate3(thd, table_vec[i].table, 1);
	}
      }
      bool suc = true;
      #if MYSQL_VERSION_ID >= 50505
      suc = (trans_commit_stmt(thd) == 0);
      #else
      suc = (ha_autocommit_or_rollback(thd, 0) == 0);
      #endif
{'fraction_non_alphanumeric': 0.12905965621840243, 'fraction_numerical': 0.00865520728008089, 'mean_word_length': 4.406953859610759, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 117, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4522714', 'n_tokens_mistral': 9757, 'n_tokens_neox': 9523, 'n_words': 1882}