cli.cpp

FastDb是高效的内存数据库系统

//-< CLI.CPP >-------------------------------------------------------*--------*
// FastDB                    Version 1.0         (c) 1999  GARRET    *     ?  *
// (Main Memory Database Management System)                          *   /\|  *
//                                                                   *  /  \  *
//                          Created:     13-Jan-2000  K.A. Knizhnik  * / [] \ *
//                          Last update: 13-Jan-2000  K.A. Knizhnik  * GARRET *
//-------------------------------------------------------------------*--------*
// Call level interface client part implementation
//-------------------------------------------------------------------*--------*

#define INSIDE_FASTDB

#include <ctype.h>
#include "stdtp.h"
#include "sockio.h"
#include "repsock.h"
#include "sync.h"
#include "cli.h"
#include "cliproto.h"

struct parameter_binding
{
  parameter_binding* next;
  char* name;
  int   var_type;
  int   var_len;
  void* var_ptr;

  ~parameter_binding()
  {
    delete[] name;
  }
};

struct column_binding
{
  column_binding* next;
  char* name;
  int   var_type;
  int*  var_len;
  void* var_ptr;
  void* arr_ptr;
  int   arr_len;
  cli_column_get_ex get_fnc;
  cli_column_set_ex set_fnc;

  ~column_binding()
  {
    delete[] name;
  }
};

struct session_desc;

struct statement_desc
{
  int                id;
  statement_desc*    next;
  char*              stmt;
  column_binding*    columns;
  parameter_binding* params;
  session_desc*      session;
  bool               for_update;
  bool               updated;
  bool               prepared;
  bool               autoincrement;
  cli_oid_t          oid;
  int                stmt_len;
  int                n_params;
  int                n_columns;
  int                columns_len;

  void deallocate()
  {
    delete[] stmt;
    column_binding *cb, *next_cb;

    for (cb = columns; cb != NULL; cb = next_cb)
    {
      next_cb = cb->next;
      delete cb;
    }

    parameter_binding *pb, *next_pb;

    for (pb = params; pb != NULL; pb = next_pb)
    {
      next_pb = pb->next;
      delete pb;
    }
  }

  statement_desc(int id, statement_desc* next)
  {
    this->id = id;
    this->next = next;
  }

  statement_desc()
  {}

}

;

struct session_desc
{
  int             id;
  session_desc*   next;
  socket_t*       sock;
  statement_desc* stmts;

  session_desc(int id, session_desc* next)
  {
    this->id = id;
    this->next = next;
  }

  session_desc()
  {}

}

;

template<class T>

class descriptor_table
{

protected:
  T**         table;
  T*          free_desc;
  int         descriptor_table_size;
  dbMutex     mutex;

public:
  descriptor_table()
  {
    int i;
    descriptor_table_size = 16;
    table = new T*[descriptor_table_size];
    T* next = NULL;

    for (i = 0; i < descriptor_table_size; i++)
    {
      table[i] = next = new T(i, next);
    }

    free_desc = next;
  }

  T* get
    (int desc)
  {
    dbCriticalSection cs(mutex);
    return (desc >= descriptor_table_size) ? (T*)0 : table[desc];
  }

  T* allocate()
  {
    dbCriticalSection cs(mutex);

    if (free_desc == NULL)
    {
      int i, n;
      T** desc = new T*[descriptor_table_size * 2];
      memcpy(desc, table, descriptor_table_size*sizeof(T*));
      delete[] table;
      table = desc;
      T* next = NULL;

      for (i = descriptor_table_size, n = i*2; i < n; i++)
      {
        table[i] = next = new T(i, next);
      }

      free_desc = next;
      descriptor_table_size = n;
    }

    T* desc = free_desc;
    free_desc = desc->next;
    return desc;
  }

  void deallocate(T* desc)
  {
    dbCriticalSection cs(mutex);
    desc->next = free_desc;
    free_desc = desc;
  }
};

static descriptor_table<session_desc>   sessions;

static descriptor_table<statement_desc> statements;

int cli_open(char const* server_url,
             int         max_connect_attempts,
             int         reconnect_timeout_sec)
{
  socket_t* sock;
  int n_addresses = 1;
  char const* start = server_url;
  char const* end;

  while ((end = strchr(start, ',')) != NULL)
  {
    start = end + 1;
    n_addresses += 1;
  }

  if (n_addresses == 1)
  {
    sock = socket_t::connect(server_url,
                             socket_t::sock_any_domain,
                             max_connect_attempts,
                             reconnect_timeout_sec);
  }
  else
  {
    char** addresses = new char*[n_addresses];
    start = server_url;

    for (int i = 0; i < n_addresses; i++)
    {
      end = strchr(start, ',');

      if (end == NULL)
      {
        end = start + strlen(start);
      }

      int len = end - start;
      char* addr = new char[len+1];
      memcpy(addr, start, len);
      addr[len] = '\0';
      start = end + 1;
      addresses[i] = addr;
    }

    sock = replication_socket_t::connect((char const**)addresses,
                                         n_addresses,
                                         max_connect_attempts,
                                         reconnect_timeout_sec);

    while (--n_addresses >= 0)
    {
      delete[] addresses[n_addresses];
    }

    delete[] addresses;
  }

  if (!sock->is_ok())
  {
    delete sock;
    return cli_connection_refused;
  }

  session_desc* session = sessions.allocate();
  session->sock = sock;
  session->stmts = NULL;
  return session->id;
}


int cli_close(int session)
{
  statement_desc *stmt, *next;
  session_desc* s = sessions.get(session);

  if (s == NULL)
  {
    return cli_bad_descriptor;
  }

  cli_request req;
  req.length = sizeof(req);
  req.cmd = cli_cmd_close_session;
  req.pack();
  int result = cli_ok;

  if (!s->sock->write(&req, sizeof req))
  {
    result = cli_network_error;
  }

  delete s->sock;
  s->sock = NULL;

  for (stmt = s->stmts; stmt != NULL; stmt = next)
  {
    next = stmt->next;
    stmt->deallocate();
    statements.deallocate(stmt);
  }

  sessions.deallocate(s);
  return result;
}

int cli_statement(int session, char const* stmt_str)
{
  session_desc* s = sessions.get(session);

  if (s == NULL)
  {
    return cli_bad_descriptor;
  }

  statement_desc* stmt = statements.allocate();
  stmt->stmt = new char[strlen(stmt_str)+1];
  stmt->columns = NULL;
  stmt->params = NULL;
  stmt->session = s;
  stmt->for_update = 0;
  stmt->prepared = false;
  stmt->n_params = 0;
  stmt->n_columns = 0;
  stmt->columns_len = 0;
  stmt->oid = 0;
  stmt->next = s->stmts;
  stmt->updated = false;
  s->stmts = stmt;
  char const* p = stmt_str;
  char* dst = stmt->stmt;
  parameter_binding** last = &stmt->params;

  while (*p != '\0')
  {
    if (*p == '\'')
    {
      do
      {
        do
        {
          *dst++ = *p++;
        }
        while (*p != '\0' && *p != '\'');

        *dst++ = *p;

        if (*p == '\0')
        {
          *last = NULL;
          stmt->deallocate();
          statements.deallocate(stmt);
          return cli_bad_statement;
        }
      }
      while (*++p == '\'');
    }
    else if (*p == '%')
    {
      stmt->n_params += 1;
      char const* q = p++;

      while (isalnum((unsigned char)*p) || *p == '_')
        p += 1;

      if (*p == '%')
      {
        *last = NULL;
        stmt->deallocate();
        statements.deallocate(stmt);
        return cli_bad_statement;
      }

      parameter_binding* pb = new parameter_binding;
      int len = p - q;
      pb->name = new char[len+1];
      memcpy(pb->name, q, len);
      pb->name[len] = '\0';
      *last = pb;
      last = &pb->next;
      pb->var_ptr = NULL;
      *dst++ = '\0';
    }
    else
    {
      *dst++ = *p++;
    }
  }

  if (dst == stmt->stmt || *(dst-1) != '\0')
  {
    *dst++ = '\0';
  }

  stmt->stmt_len = dst - stmt->stmt;
  *last = NULL;
  return stmt->id;
}

int cli_parameter(int         statement,
                  char const* param_name,
                  int         var_type,
                  void*       var_ptr)
{
  if ((unsigned)var_type > cli_pasciiz)
  {
    return cli_unsupported_type;
  }

  statement_desc* s = statements.get(statement);

  if (s == NULL)
  {
    return cli_bad_descriptor;
  }

  s->prepared = false;

  for (parameter_binding* pb = s->params; pb != NULL; pb = pb->next)
  {
    if (strcmp(pb->name, param_name) == 0)
    {
      pb->var_ptr  = var_ptr;
      pb->var_type = var_type;
      return cli_ok;
    }
  }

  return cli_parameter_not_found;
}

int cli_column(int         statement,
               char const* column_name,
               int         var_type,
               int*        var_len,
               void*       var_ptr)
{
  statement_desc* s = statements.get(statement);

  if (s == NULL)
  {
    return cli_bad_descriptor;
  }

  s->prepared = false;
  column_binding* cb = new column_binding;
  int len = strlen(column_name) + 1;
  cb->name = new char[len];
  s->columns_len += len;
  cb->next = s->columns;
  s->columns = cb;
  s->n_columns += 1;
  strcpy(cb->name, column_name);
  cb->var_type = var_type;
  cb->var_len = var_len;
  cb->var_ptr = var_ptr;
  cb->set_fnc = NULL;
  cb->get_fnc = NULL;
  return cli_ok;
}

int cli_array_column(int            statement,
                     char const*    column_name,
                     int            var_type,
                     void*          var_ptr,

                     cli_column_set set
                       ,
                       cli_column_get get
                        )
  {
    return cli_array_column_ex(statement, column_name, var_type, var_ptr,

                               (cli_column_set_ex)set
                               , (cli_column_get_ex)get
                              );
  }

int cli_array_column_ex(int               statement,
                        char const*       column_name,
                        int               var_type,
                        void*             var_ptr,

                        cli_column_set_ex set
                          ,
                          cli_column_get_ex get
                           )
  {
    statement_desc* s = statements.get(statement);

    if (s == NULL)
    {
      return cli_bad_descriptor;
    }

    if (var_type < cli_asciiz || var_type > cli_array_of_string)
    {
      return cli_unsupported_type;
    }

    s->prepared = false;
    column_binding* cb = new column_binding;
    int len = strlen(column_name) + 1;
    cb->name = new char[len];
    s->columns_len += len;
    cb->next = s->columns;
    s->columns = cb;
    s->n_columns += 1;
    strcpy(cb->name, column_name);
    cb->var_type = var_type;
    cb->var_len = NULL;
    cb->var_ptr = var_ptr;

    cb->set_fnc = set

                    ;
    cb->get_fnc = get

                    ;
    return cli_ok;
  }

int cli_fetch(int statement, int for_update)
{
  parameter_binding* pb;
  column_binding*    cb;
  statement_desc* stmt = statements.get(statement);
  char *p, *s;

  if (stmt == NULL)
  {
    return cli_bad_descriptor;
  }

  stmt->for_update = for_update;
  int msg_size = sizeof(cli_request) + 1;

  for (pb = stmt->params; pb != NULL; pb = pb->next)
  {
    if (pb->var_ptr == NULL)
    {
      return cli_unbound_parameter;
    }

    if (pb->var_type == cli_asciiz)
    {
      msg_size += strlen((char*)pb->var_ptr) + 1;
    }
    else if (pb->var_type == cli_pasciiz)
    {
      msg_size += strlen(*(char**)pb->var_ptr) + 1;
    }
    else
    {
      msg_size += sizeof_type[pb->var_type];
    }
  }

  stmt->oid = 0;

  if (!stmt->prepared)
  {
    msg_size += 4 + stmt->stmt_len + stmt->n_params;
    msg_size += stmt->columns_len + stmt->n_columns;
  }

  dbSmallBuffer buf(msg_size);
  p = buf;
  cli_request* req = (cli_request*)p;
  req->length  = msg_size;
  req->cmd     = stmt->prepared
                 ? cli_cmd_execute : cli_cmd_prepare_and_execute;
  req->stmt_id = statement;
  req->pack();
  p += sizeof(cli_request);

  if (!stmt->prepared)
  {
    *p++ = stmt->n_params;
    *p++ = stmt->n_columns;
    p = pack2(p, stmt->stmt_len + stmt->n_params);
    pb = stmt->params;
    char* end = p + stmt->stmt_len + stmt->n_params;
    char* src = stmt->stmt;

    while (p < end)
    {
      while ((*p++ = *src++) != '\0')

        ;
      if (pb != NULL)
      {
        *p++ = pb->var_type == cli_pasciiz ? cli_asciiz : pb->var_type;
        pb = pb->next;
      }
    }

    for (cb = stmt->columns; cb != NULL; cb = cb->next)
    {
      *p++ = cb->var_type;
      s = cb->name;

      while ((*p++ = *s++) != '\0')

        ;
    }
  }

  *p++ = for_update;

  for (pb = stmt->params; pb != NULL; pb = pb->next)
  {
    switch (pb->var_type)
    {

    case cli_asciiz:
      s = (char*)pb->var_ptr;

      while ((*p++ = *s++) != '\0')