trunk/src/ha_tritonn.cc

#ifdef USE_PRAGMA_IMPLEMENTATION
#pragma implementation        // gcc: Class implementation
#endif

#include "mysql_priv.h"
#undef PACKAGE
#undef VERSION
#undef HAVE_DTRACE
#undef _DTRACE_VERSION

#include "tritonn_config.h"

/* We define DTRACE after mysql_priv.h in case it disabled dtrace in the main server */
#ifdef HAVE_DTRACE
#define _DTRACE_VERSION 1
#else
#endif

#include "ha_tritonn.h"
#include <mysql/plugin.h>

/* create ha_tritonn instance  */
static handler *tritonn_create_handler(handlerton *hton,
                                       TABLE_SHARE *table,
                                       MEM_ROOT *mem_root);

static TRITONN_DB *tdb;
static sen_db *get_sen_db(const char *db_name);

/* delete database files */
static void tritonn_drop_database(handlerton *hton, char* path);

/* flush logfile */
static bool tritonn_flush_logs(handlerton *hton);

/* show engine status */
static bool tritonn_show_status(handlerton *hton, THD *thd, stat_print_fn *print,
                                enum ha_stat_type stat);

static void dump_create(const char *name, TABLE *table_arg,
                        HA_CREATE_INFO *create_info);

static void call_senna_ql(sen_ctx *ctx, const char *str, bool disp=true);

static char *mysqltype2sennatype_string(enum_field_types types);

/* handlerton capability flags. see handler.h for more information */
static int tritonn_hton_flags = HTON_ALTER_NOT_SUPPORTED | HTON_CAN_RECREATE |
    HTON_TEMPORARY_NOT_SUPPORTED | HTON_NO_PARTITION;

/* Variables for tritonn share methods */
static HASH tritonn_open_tables; ///< Hash used to track the number of open tables; variable for tritonn share methods
pthread_mutex_t tritonn_mutex;   ///< This is the mutex used to init the hash; variable for tritonn share methods

static uchar* get_key(TRITONN_SHARE *share, size_t *length,
                      my_bool not_used __attribute__((unused)))
{
  *length=share->table_name_length;
  return (uchar*) share->table_name;
}

static int tritonn_init_hton(void *p)
{
  DBUG_ENTER("tritonn_init_hton");
  DBTN;

  handlerton *hton;

  hton= (handlerton *)p;
  VOID(pthread_mutex_init(&tritonn_mutex, MY_MUTEX_INIT_FAST));
  (void) hash_init(&tritonn_open_tables, system_charset_info, 32, 0, 0,
                   (hash_get_key) get_key, 0, 0);

  hton->state         = SHOW_OPTION_YES;
  hton->create        = tritonn_create_handler;
  hton->drop_database = tritonn_drop_database;
  hton->flush_logs    = tritonn_flush_logs;
  hton->show_status   = tritonn_show_status;
  hton->flags         = tritonn_hton_flags;

  sen_init();
  DBUG_RETURN(0);
}

static int tritonn_deinit_hton(void *p)
{
  int error= 0;
  DBUG_ENTER("tritonn_deinit_hton");
  DBTN;

  if (tritonn_open_tables.records)
    error= 1;
  hash_free(&tritonn_open_tables);
  while (tdb != NULL) {
    sen_db_close(tdb->db);
    tdb = tdb->next;
  }
  pthread_mutex_destroy(&tritonn_mutex);

  sen_fin();
  DBUG_RETURN(0);
}

/** @brief
  Skeleton of simple lock controls. The "share" it creates is a structure we will
  pass to each tritonn handler. Do you have to have one of these? Well, you have
  pieces that are used for locking, and they are needed to function.
*/
static TRITONN_SHARE *get_share(const char *table_name, TABLE *table)
{
  TRITONN_SHARE *share;
  uint length;
  char *tmp_name;

  DBTN;

  pthread_mutex_lock(&tritonn_mutex);
  length=(uint) strlen(table_name);

  if (!(share=(TRITONN_SHARE*) hash_search(&tritonn_open_tables,
                                           (uchar*) table_name,
                                           length)))
  {
    if (!(share=(TRITONN_SHARE *)
          my_multi_malloc(MYF(MY_WME | MY_ZEROFILL),
                          &share, sizeof(*share),
                          &tmp_name, length+1,
                          NullS)))
    {
      pthread_mutex_unlock(&tritonn_mutex);
      return NULL;
    }

    share->use_count=0;
    share->table_name_length=length;
    share->table_name=tmp_name;
    strmov(share->table_name,table_name);
    if (my_hash_insert(&tritonn_open_tables, (uchar*) share))
      goto error;
    thr_lock_init(&share->lock);
    pthread_mutex_init(&share->mutex,MY_MUTEX_INIT_FAST);
  }
  share->use_count++;
  pthread_mutex_unlock(&tritonn_mutex);

  return share;

error:
  pthread_mutex_unlock(&tritonn_mutex);
  pthread_mutex_destroy(&share->mutex);
  my_free((uchar *) share, MYF(0));

  return NULL;
}

/** @brief
  Free lock controls. We call this whenever we close a table. If the table had
  the last reference to the share, then we free memory associated with it.
*/
static int free_share(TRITONN_SHARE *share)
{
  DBTN;
  pthread_mutex_lock(&tritonn_mutex);
  if (!--share->use_count)
  {
    hash_delete(&tritonn_open_tables, (uchar*) share);
    thr_lock_delete(&share->lock);
    pthread_mutex_destroy(&share->mutex);
    my_free((uchar *) share, MYF(0));
  }
  pthread_mutex_unlock(&tritonn_mutex);

  return 0;
}

/* create ha_tritonn instance  */
static handler* tritonn_create_handler(handlerton *hton,
                                       TABLE_SHARE *table,
                                       MEM_ROOT *mem_root)
{
  DBTN;
  return new (mem_root) ha_tritonn(hton, table);
}

/* delete database files */
static void tritonn_drop_database(handlerton *hton, char* path)
{
  DBTN;
}

/* flush logfile */
static bool tritonn_flush_logs(handlerton *hton)
{
  DBTN;
  return true;
}

/* show engine status */
static bool tritonn_show_status(handlerton *hton, THD *thd, stat_print_fn *print,
                                enum ha_stat_type stat)
{
  DBTN;
  return true;
}

ha_tritonn::ha_tritonn(handlerton *hton, TABLE_SHARE *table_arg)
  :handler(hton, table_arg)
{
  DBTN;
}

ha_tritonn::~ha_tritonn()
{
  DBTN;
}

const char *ha_tritonn::table_type() const
{
  DBTN;
  return "TRITONN";
}

const char *ha_tritonn::index_type(uint inx) const
{
  DBTN;
  return "SENNA";
}

/* TODO: examin this feature */
static const char *ha_tritonn_exts[] = {
  NullS
};

const char **ha_tritonn::bas_ext() const
{
  DBTN;
  return ha_tritonn_exts;
}

// see $MYSQL_SRC/sql/handler.h "bits in table_flags"
ulonglong ha_tritonn::table_flags() const {
  DBTN;
  return (HA_NO_TRANSACTIONS | HA_STATS_RECORDS_IS_EXACT |
          HA_NO_PREFIX_CHAR_KEYS | HA_CAN_FULLTEXT | HA_NO_AUTO_INCREMENT);
}

// see $MYSQL_SRC/sql/handler.h, "bits in index_flags"
ulong ha_tritonn::index_flags(uint inx, uint part, bool all_parts) const {
  DBTN;
  return (HA_ONLY_WHOLE_INDEX);
}

int ha_tritonn::add_index(TABLE *table_arg, KEY *key_info, uint num_of_keys) {
  DBTN;
  return HA_ERR_WRONG_COMMAND;
}

int ha_tritonn::prepare_drop_index(TABLE *table_arg, uint *key_num, uint num_of_keys)
{
  DBTN;
  return HA_ERR_WRONG_COMMAND;
}

int ha_tritonn::final_drop_index(TABLE *table_arg)
{
  DBTN;
  return HA_ERR_WRONG_COMMAND;
}


int ha_tritonn::open(const char *name, int mode, uint test_if_locked)
{
  DBUG_ENTER("ha_tritonn::open");
  DBTN;

  if (!(share = get_share(name, table)))
    DBUG_RETURN(HA_ERR_OUT_OF_MEM);
  thr_lock_data_init(&share->lock,&lock,NULL);
  share->short_name = table_share->table_name.str;
  ctx = sen_ctx_open(get_sen_db(table_share->db.str),SEN_CTX_USEQL);
  call_senna_ql(ctx, "(define (select recs expr) (reverse (letrec "
                "((func (lambda (r x) (if x (let ((y (expr x))) "
                "(func (if y (cons y r) r) (recs ::+ x))) r)))) "
                "(func () (recs ::)))))",false);
  call_senna_ql(ctx, "(define (slot-exps class) (select <db> "
                "(lambda (x) (let ((s (x ::schema))) (and (pair? s) "
                "(eq? class (car s)) (list () (car (cdr (cdr s)))))))))",false);
  call_senna_ql(ctx, "(define (dump-table class) (if (< 0 class.:nrecords) "
                "(begin (disp `(: ,class (.:key ,@(slot-exps class)) 0 0) "
                ":tsv))))",false);

  DBUG_RETURN(0);
}

int ha_tritonn::close(void)
{
  DBUG_ENTER("ha_tritonn::close");
  DBTN;
  sen_ctx_close(ctx);
  DBUG_RETURN(free_share(share));
}

int ha_tritonn::rnd_init(bool scan)
{
  DBUG_ENTER("ha_tritonn::rnd_init");
  DBTN;
  char buf[1024];
  my_snprintf(buf,1024,"(dump-table <%s>)", share->short_name);
  call_senna_ql(ctx,buf, false);
  DBUG_RETURN(0);
}

int ha_tritonn::rnd_next(uchar *buf)
{
  DBUG_ENTER("ha_tritonn::rnd_next");
  DBTN;
  char *res;
  unsigned int res_len;
  int res_flags;
  char *row;
  sen_ctx_recv(ctx,&res,&res_len,&res_flags);
  if (res_flags & SEN_CTX_MORE) {
    char token[1024];
    int offset=0;
    int i;
    for (Field **field=table->field; *field ; field++) {
      for (i=0; offset < res_len && *(res+offset) != '\t'; offset++,i++) {
        token[i] = *(res+offset);
      }
      offset++;
      switch((*field)->result_type()) {
      case STRING_RESULT:
        (*field)->store(token,i,system_charset_info);
        break;
      case REAL_RESULT:
        token[i] = '\0';
        (*field)->store(atof(token));
        break;
      case INT_RESULT:
        token[i] = '\0';
        (*field)->store(atoi(token),false);
        break;
      case ROW_RESULT:
        (*field)->store("row_result",10,system_charset_info);
        break;
      case DECIMAL_RESULT:
        (*field)->store("decimal",7,system_charset_info);
        break;
      }
    }
    memset(buf, 0, table->s->null_bytes);
    DBUG_RETURN(0);
  } else {
    DBUG_RETURN(HA_ERR_END_OF_FILE);
  }
}

void ha_tritonn::position(const uchar *record)
{
  DBUG_ENTER("ha_tritonn::position");
  DBTN;
  DBUG_VOID_RETURN;
}

int ha_tritonn::rnd_pos(uchar * buf, uchar *pos)
{
  DBUG_ENTER("ha_tritonn::rnd_pos");
  DBTN;
  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
}

int ha_tritonn::info(uint flag)
{
  DBUG_ENTER("ha_tritonn::info");
  DBTN;
  char buf[1024];
  char *res;
  unsigned int res_len;
  int res_flag;
  int i;
  my_snprintf(buf,1024,"(<%s> ::nrecords)", share->short_name);
  call_senna_ql(ctx,buf,false);
  sen_ctx_recv(ctx,&res,&res_len,&res_flag);
  char *res2 = (char*) sql_alloc(res_len+1);
  for (i=0; i<res_len; i++) {
    *(res2+i) = *(res+i);
  }
  *(res2+res_len) = '\0';
  stats.records=atoi(res2);
  //  printf("[SennaQL] returned nrecords = %d\n",stats.records);
  DBUG_RETURN(0);
}

THR_LOCK_DATA **ha_tritonn::store_lock(THD *thd,
                                       THR_LOCK_DATA **to,
                                       enum thr_lock_type lock_type)
{
  DBTN;
  if (lock_type != TL_IGNORE && lock.type == TL_UNLOCK)
    lock.type=lock_type;
  *to++= &lock;
  return to;
}

int ha_tritonn::create(const char *name, TABLE *table_arg,
                       HA_CREATE_INFO *create_info)
{
  DBUG_ENTER("ha_tritonn::create");
  DBTN;
  char buf[FN_REFLEN];
  //dump_create(name,table_arg,create_info);
  ctx = sen_ctx_open(get_sen_db(table_share->db.str),SEN_CTX_USEQL);
  my_snprintf(buf,1024,"(ptable '<%s>)",table_arg->s->table_name);
  call_senna_ql(ctx,buf);
  int i;
  /* col1 is assigned to pkey so i=1 here */
  for (i=1; i < table_arg->s->fields; i++) {
    Field *field = table_arg->s->field[i];
    my_snprintf(buf,1024,"(<%s> ::def :%s <%s>)",
                create_info->alias,
                field->field_name,
                mysqltype2sennatype_string(field->type()));
    call_senna_ql(ctx,buf);
  }
  sen_ctx_close(ctx);
  DBUG_RETURN(0); 
}

uint ha_tritonn::max_supported_keys() const
{
  DBTN;
  return 1024;
}

uint ha_tritonn::max_supported_key_length() const
{
  DBTN;
  return 1024;
}

uint ha_tritonn::max_supported_key_part_length() const
{
  DBTN;
  return 1024;
}


struct st_mysql_storage_engine storage_engine_structure=
{ MYSQL_HANDLERTON_INTERFACE_VERSION };

mysql_declare_plugin(tritonn)
{
  MYSQL_STORAGE_ENGINE_PLUGIN,
  &storage_engine_structure,
  "TRITONN",
  "Tritonn Project",
  "Tritonn storage engine",
  PLUGIN_LICENSE_BSD,
  tritonn_init_hton, /* Plugin Init */
  tritonn_deinit_hton, /* Plugin Deinit */
  0x0001 /* 0.1 */,
  NULL,                       /* status variables                */
  NULL,                       /* system variables                */
  NULL                        /* config options                  */
}
mysql_declare_plugin_end;

char *a_type[] = {"FIELD_ITEM","FUNC_ITEM", "SUM_FUNC_ITEM", "STRING_ITEM",
                  "INT_ITEM", "REAL_ITEM", "NULL_ITEM", "VARBIN_ITEM",
                  "COPY_STR_ITEM", "FIELD_AVG_ITEM", "DEFAULT_VALUE_ITEM",
                  "PROC_ITEM","COND_ITEM", "REF_ITEM", "FIELD_STD_ITEM",
                  "FIELD_VARIANCE_ITEM", "INSERT_VALUE_ITEM",
                  "SUBSELECT_ITEM", "ROW_ITEM", "CACHE_ITEM", "TYPE_HOLDER",
                  "PARAM_ITEM", "TRIGGER_FIELD_ITEM", "DECIMAL_ITEM",
                  "XPATH_NODESET", "XPATH_NODESET_CMP",
                  "VIEW_FIXER_ITEM"};

char *f_type[] ={"UNKNOWN_FUNC","EQ_FUNC","EQUAL_FUNC","NE_FUNC","LT_FUNC","LE_FUNC",
                 "GE_FUNC","GT_FUNC","FT_FUNC",
                 "LIKE_FUNC","ISNULL_FUNC","ISNOTNULL_FUNC",
                 "COND_AND_FUNC", "COND_OR_FUNC", "COND_XOR_FUNC",
                 "BETWEEN", "IN_FUNC", "MULT_EQUAL_FUNC",
                 "INTERVAL_FUNC", "ISNOTNULLTEST_FUNC",
                 "SP_EQUALS_FUNC", "SP_DISJOINT_FUNC","SP_INTERSECTS_FUNC",
                 "SP_TOUCHES_FUNC","SP_CROSSES_FUNC","SP_WITHIN_FUNC",
                 "SP_CONTAINS_FUNC","SP_OVERLAPS_FUNC",
                 "SP_STARTPOINT","SP_ENDPOINT","SP_EXTERIORRING",
                 "SP_POINTN","SP_GEOMETRYN","SP_INTERIORRINGN",
                 "NOT_FUNC", "NOT_ALL_FUNC",
                 "NOW_FUNC", "TRIG_COND_FUNC",
                 "SUSERVAR_FUNC", "GUSERVAR_FUNC", "COLLATE_FUNC",
                 "EXTRACT_FUNC", "CHAR_TYPECAST_FUNC", "FUNC_SP", "UDF_FUNC",
                 "NEG_FUNC" };

extern int string2my_decimal(uint mask, const String *str, my_decimal *d);
#include "item_func.h"
#include "my_decimal.h"

void dump_condition(Item *cond)
{
  char buff[256];
  String str(buff,(uint32) sizeof(buff), system_charset_info);
  str.length(0);
  cond->print(&str, QT_ORDINARY);
  str.append("\0");
  printf("COND* name=%s, type()=%s", cond->name, a_type[cond->type()]);
  if (cond->type() == Item::FUNC_ITEM) {
    Item_func *item = (Item_func*) cond;
    printf(", functype=%s", f_type[item->functype()]);
  } else if (cond->type() == Item::INT_ITEM) {
    Item_int *item = (Item_int*) cond;
    printf(", value=%d", item->value);
  } else if (cond->type() == Item::FIELD_ITEM) {
  }
  printf(", str=%s",str.ptr());
  printf("\n");
  if (cond->next != NULL) dump_condition(cond->next);
  return;
}


const COND* ha_tritonn::cond_push(const COND *cond)
{
  printf("===ha_tritonn::cond_push is called===\n");
  Item *item = (Item*) cond;
  dump_condition(item);
  return cond;
}

static void dump_create(const char *name, TABLE *table_arg,
                        HA_CREATE_INFO *create_info)
{
  int i;
  printf("===dump_create is called===\n");
  printf("argments:: name=%s, table_arg=%p, create_info=%p\n",name,table_arg,create_info);
  printf("table_arg->s->table_name   = %s\n",table_arg->s->table_name);
  printf("table_arg->s->db           = %s\n", table_arg->s->db);
  printf("table_arg->s->path         = %s\n", table_arg->s->path);
  printf("table_arg->s->fields       = %d\n", table_arg->s->fields);
  printf("table_arg->s->keys         = %d\n", table_arg->s->keys);
  printf("table_arg->s->primary_key  = %d\n", table_arg->s->primary_key);
  printf("table_arg->s->row_type     = %d\n", table_arg->s->row_type);
  printf("create_info->alias         = %s\n", create_info->alias);
  printf("create_info->table_charset = %p\n", create_info->table_charset);
  printf("create_info->default_table_charset = %p\n", create_info->default_table_charset);
  printf("create_info->connect_string.str = %s\n",create_info->connect_string.str);
  printf("create_info->comment.str   = %s\n", create_info->comment.str);

  for (i=0; i < table_arg->s->fieldnames.count; i++) {
    printf("  table_arg->s->fieldnames.type_names=%s\n", table_arg->s->fieldnames.type_names[i]);
  }
  for (i=0; i < table_arg->s->keynames.count; i++) {
    printf("  table_arg->s->keynames.type_names=%s\n", table_arg->s->keynames.type_names[i]);
  }

  for (i=0; i < table_arg->s->fields; i++) {
    Field *field = table_arg->s->field[i];
    printf("  field=%p\n",field);
    printf("    field->field_name = %s\n", field->field_name);
    printf("    field->comment.str = %s\n", field->comment.str);
    printf("    field->field_length = %d\n", field->field_length);
    printf("    field->unireg_chekc = %d\n", field->unireg_check);
    printf("    field->flags = %d\n", field->flags);
    printf("    field->field_index = %d\n", field->field_index);
    printf("    field->type() = %d\n", field->type());
    printf("    field->result_type() = %d\n", field->result_type());
  }

  int keys = table_arg->s->keys;
  for (i=0; i < keys; i++) {
    KEY *key = table_arg->key_info+i;
    printf("  key=%p\n",key);
    printf("  key->name      = %s\n", key->name);
    printf("  key->algorithm = %d\n", key->algorithm);
    printf("  key->key_parts = %d\n", key->key_parts);
    int j;
    for (j=0; j < key->key_parts; j++) {
      KEY_PART_INFO *part = key->key_part+j;
      printf("    part=%p\n",part);
      printf("    part->fieldnr=%d\n",part->fieldnr);
      printf("    part->field->field_name=%s\n",part->field->field_name);
      printf("    part->field->comment.str=%s\n",part->field->comment.str);
    }
  }
}

void call_senna_ql(sen_ctx *ctx, const char *str, bool disp)
{
  char *ql;
  char *res;
  unsigned int res_len;
  int res_flags;
  char *buf;
  unsigned int ql_len = strlen(str);
  ql = (char*) malloc(ql_len);
  strncpy(ql,str,ql_len);
  sen_ctx_send(ctx,ql,ql_len,0);
  if (disp == true) {
    do {
      sen_ctx_recv(ctx,&res,&res_len,&res_flags);
      buf = (char*) calloc(1,1024);
      printf("[SennaQL] %s => %s\n",str,strncat(buf,res,res_len));
      free(buf);
    } while (res_flags & SEN_CTX_MORE);
  }
  free(ql);
}

static char *mysqltype2sennatype_string(enum_field_types types)
{
  switch(types) {
  case MYSQL_TYPE_TINY:
  case MYSQL_TYPE_SHORT:
  case MYSQL_TYPE_INT24:
  case MYSQL_TYPE_LONG:
    return "int";
  case MYSQL_TYPE_FLOAT:
  case MYSQL_TYPE_DOUBLE:
    return "float";
  case MYSQL_TYPE_STRING:
  case MYSQL_TYPE_VAR_STRING:
  case MYSQL_TYPE_VARCHAR:
  case MYSQL_TYPE_BLOB:
    return "text";
  default:
    return "null";
  }
}

const char* tritonn_escape_string(const char* str){
  int i,len,cnt,offset;
  for (i=0,len=strlen(str),cnt=0; i<len; i++) {
    if (*(str+i) == '\\' || *(str+i) == '"') cnt++;
  }
  if (cnt == 0) return str;
  char *res = (char*) sql_alloc(len+cnt);
  for (i=0,offset=0; i<len; i++) {
    if (*(str+i) == '\\') {
      *(res+i+offset) = '\\';
      offset++;
      *(res+i+offset) = '\\';
    } else if (*(str+i) == '"') {
      *(res+i+offset) = '\\';
      offset++;
      *(res+i+offset) = '"';
    } else {
      *(res+i+offset) = *(str+i);
    }
  }
  return res;
}

int ha_tritonn::write_row(uchar* buf)
{
  DBTN;
  int i=0;
  char ql[1024];
  char* pkey;
  for (Field **field=table->field; *field; field++) {
    String str;
    (*field)->val_str(&str);
    str.chop();
    if (i==0) {
      pkey = (char*) sql_alloc(strlen(str.ptr()));
      strcpy(pkey,str.ptr());
      my_snprintf(ql,1024,"(<%s> ::new \"%s\")",share->short_name,
                  tritonn_escape_string(str.ptr()));
    } else {
      my_snprintf(ql,1024,"((<%s> : \"%s\") :%s \"%s\")",
                  share->short_name,
                  tritonn_escape_string(pkey), (*field)->field_name,
                  tritonn_escape_string(str.ptr()));
    }
    call_senna_ql(ctx,ql);
    i++;
  }
  return 0;
}

TRITONN_DB *open_or_create_db(const char *db_name)
{
  char file_name[FN_REFLEN];
  TRITONN_DB *d = (TRITONN_DB*) my_malloc(sizeof(TRITONN_DB), MYF(0));
  d->db_name_length = strlen(db_name);
  d->db_name = (char*) my_malloc(d->db_name_length,MYF(0));
  memcpy(d->db_name,db_name,d->db_name_length);
  my_snprintf(file_name,FN_REFLEN,"%s/tritonn.db",db_name);
  d->db = sen_db_open(file_name);
  if (d->db == NULL) {
    d->db = sen_db_create(file_name,0,sen_enc_utf8);
  }
  d->next = NULL;
  pthread_mutex_unlock(&tritonn_mutex);
  return d;
}

sen_db *get_sen_db(const char *db_name)
{
  pthread_mutex_lock(&tritonn_mutex);
  /* case1: no instance created yet */
  if (tdb == NULL) {
    tdb = open_or_create_db(db_name);
    pthread_mutex_unlock(&tritonn_mutex);
    return tdb->db;
  }
  /* case2: instance hit */
  TRITONN_DB *cur = tdb;
  while (true) {
    if (strcmp(cur->db_name,db_name) == 0) {
        pthread_mutex_unlock(&tritonn_mutex);
        return cur->db;
    } else if (cur->next != NULL) {
      cur = cur->next;
    } else {
      break;
    }
  }
  /* case3: instance not hit */
  cur->next = open_or_create_db(db_name);
  pthread_mutex_unlock(&tritonn_mutex);
  return cur->next->db;
}
