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"

BEGIN_FASTDB_NAMESPACE

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;   
    void* user_data;

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

struct session_desc;

#define DEFAULT_BUF_SIZE 256

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;
    char*              buf;
    size_t             buf_size;

    void deallocate() { 
        delete[] stmt;
        column_binding *cb, *next_cb;
        for (cb = columns; cb != NULL; cb = next_cb) { 
            next_cb = cb->next;
            delete cb;
        }
        if (buf != NULL && buf_size > DEFAULT_BUF_SIZE) { 
            delete[] buf;
            buf_size = 0;
            buf = NULL;
        }
        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;
        buf = NULL;
        buf_size = 0;
    }
    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;
    }

    ~descriptor_table() {
        for (int i = 0; i < descriptor_table_size; i++) {
            delete table[i];
        }
        delete[] table;
    }

    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;

END_FASTDB_NAMESPACE
USE_FASTDB_NAMESPACE

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 (var_type != cli_rectangle && (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, NULL);
}

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,
                        void*             user_data)
{
    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;
    cb->user_data = user_data;
    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) {