cli.cpp

实现内存数据库的源代码

//-< 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 "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;
    }
};    

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;
    }
};

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 = socket_t::connect(server_url, 
				       socket_t::sock_any_domain,
				       max_connect_attempts,
				       reconnect_timeout_sec);
    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 += 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_real8) { 
	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');
	    continue;
	  case cli_pasciiz:
	    s = *(char**)pb->var_ptr;
	    while ((*p++ = *s++) != '\0');
	    continue;
	  default:
	    switch (sizeof_type[pb->var_type]) { 
	      case 1:
		*p++ = *(char*)pb->var_ptr;
		continue;
	      case 2:
		p = pack2(p, *(int2*)pb->var_ptr);
		continue;
	      case 4:
		p = pack4(p, *(int4*)pb->var_ptr);
		continue;
	      case 8:
		p = pack8(p, *(int8*)pb->var_ptr);
		continue;
	    }
	}
    }