forks/erlang-sqlite3
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merged with http://github.com/dreid/erlang-sqlite3

Browse Source
This commit is contained in:
Alexey Romanov
2010-10-14 10:24:18 +04:00
parent 85a9c312fb cfe57ab918
commit 91439ec043
22 changed files with 108 additions and 6075 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
priv/Makefile.in
View File

@@ -1,16 +0,0 @@
ERL_INTERFACE=@ERLANG_LIB_DIR_erl_interface@
ERL_ROOT=@ERLANG_ROOT_DIR@
GCC=gcc
LDFLAGS=-shared -L/usr/lib -lsqlite3 -undefined -fPIC -lerl_interface -L$(ERL_INTERFACE)/lib -lei
SRCS=sqlite3_drv.c sqlite3_drv.h
OUTPUT=sqlite3_drv.so
CFLAGS=-o ../ebin/${OUTPUT} -I$(ERL_ROOT)/usr/include/ -I$(ERL_INTERFACE)/include/ -DNULL_ATOM=@NULL_ATOM@
all: ../ebin/${OUTPUT}
../ebin/${OUTPUT}: sqlite3_drv.c sqlite3_drv.h
${GCC} sqlite3_drv.c ${LDFLAGS} ${CFLAGS}
clean:
rm -rf ../ebin/${OUTPUT}

62
priv/erl_comm.c
View File

@@ -1,62 +0,0 @@
/* erl_comm.c */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include "erl_comm.h"
int read_exact(byte *buf, int len);
int write_exact(byte *buf, int len);
int
read_cmd(byte *buf)
{
int len;
if (read_exact(buf, 2) != 2)
return(-1);
len = (buf[0] << 8) | buf[1];
return read_exact(buf, len);
}
int
write_cmd(byte *buf, int len)
{
byte li;
li = (len >> 8) & 0xff;
write_exact(&li, 1);
li = len & 0xff;
write_exact(&li, 1);
return write_exact(buf, len);
}
int
read_exact(byte *buf, int len)
{
int i, got=0;
do {
if ((i = read(0, buf+got, len-got)) <= 0)
return(i);
got += i;
} while (got<len);
return(len);
}
int
write_exact(byte *buf, int len)
{
int i, wrote = 0;
do {
if ((i = write(1, buf+wrote, len-wrote)) <= 0)
return (i);
wrote += i;
} while (wrote<len);
return (len);
}

8
priv/erl_comm.h
View File

@@ -1,8 +0,0 @@
#ifndef _ERL_COMM_H
#define _ERL_COMM_H
typedef unsigned char byte;
int read_cmd(byte *buf);
int write_cmd(byte *buf, int len);
#endif

383
priv/sqlite3_drv.c
View File

@@ -1,383 +0,0 @@
#include "sqlite3_drv.h"
// Callback Array
static ErlDrvEntry basic_driver_entry = {
NULL, /* init */
start, /* startup (defined below) */
stop, /* shutdown (defined below) */
NULL, /* output */
NULL, /* ready_input */
NULL, /* ready_output */
"sqlite3_drv", /* the name of the driver */
NULL, /* finish */
NULL, /* handle */
control, /* control */
NULL, /* timeout */
NULL, /* outputv (defined below) */
ready_async, /* ready_async */
NULL, /* flush */
NULL, /* call */
NULL, /* event */
ERL_DRV_EXTENDED_MARKER, /* ERL_DRV_EXTENDED_MARKER */
ERL_DRV_EXTENDED_MAJOR_VERSION, /* ERL_DRV_EXTENDED_MAJOR_VERSION */
ERL_DRV_EXTENDED_MAJOR_VERSION, /* ERL_DRV_EXTENDED_MINOR_VERSION */
ERL_DRV_FLAG_USE_PORT_LOCKING /* ERL_DRV_FLAGs */
};
DRIVER_INIT(basic_driver) {
return &basic_driver_entry;
}
// Driver Start
static ErlDrvData start(ErlDrvPort port, char* cmd) {
sqlite3_drv_t* retval = (sqlite3_drv_t*) driver_alloc(sizeof(sqlite3_drv_t));
struct sqlite3 *db = 0;
int status = 0;
retval->log = fopen ("/tmp/erlang-sqlite3-drv.log", "a+");
if (!retval->log) {
fprintf (stderr, "Can't create log file\n");
}
fprintf (retval->log, "--- Start erlang-sqlite3 driver\nCommand line: [%s]\n", cmd);
const char *db_name = strstr (cmd, " ");
if (!db_name) {
fprintf (retval->log, "ERROR: DB name should be passed at command line\n");
db_name = DB_PATH;
} else {
++db_name;
}
// Create and open the database
sqlite3_open(db_name, &db);
status = sqlite3_errcode(db);
if(status != SQLITE_OK) {
fprintf(retval->log, "ERROR: Unabled to open file: %s because %s\n\n", DB_PATH, sqlite3_errmsg(db));
} else {
fprintf(retval->log, "Opened file %s\n", db_name);
}
// Set the state for the driver
retval->port = port;
retval->db = db;
retval->key = 42; //FIXME: Just a magic number, make real key
#define STR_(ARG) #ARG
#define STR(ARG) STR_(ARG)
retval->atom_error = driver_mk_atom ("error");
retval->atom_columns = driver_mk_atom ("columns");
retval->atom_rows = driver_mk_atom ("rows");
retval->atom_null = driver_mk_atom (STR (NULL_ATOM));
retval->atom_id = driver_mk_atom ("id");
retval->atom_ok = driver_mk_atom ("ok");
retval->atom_unknown_cmd = driver_mk_atom ("uknown_command");
fflush (retval->log);
return (ErlDrvData) retval;
}
// Driver Stop
static void stop(ErlDrvData handle) {
sqlite3_drv_t* driver_data = (sqlite3_drv_t*) handle;
sqlite3_close(driver_data->db);
fclose (driver_data->log);
driver_data->log = 0;
driver_free(driver_data);
}
// Handle input from Erlang VM
static int control(ErlDrvData drv_data, unsigned int command, char *buf,
int len, char **rbuf, int rlen) {
sqlite3_drv_t* driver_data = (sqlite3_drv_t*) drv_data;
switch(command) {
case CMD_SQL_EXEC:
sql_exec(driver_data, buf, len);
break;
default:
unknown(driver_data, buf, len);
}
return 0;
}
static inline int return_error(sqlite3_drv_t *drv, const char *error, ErlDrvTermData **spec, int *terms_count) {
*spec = (ErlDrvTermData *)calloc(7, sizeof(ErlDrvTermData));
(*spec)[0] = ERL_DRV_ATOM;
(*spec)[1] = drv->atom_error;
(*spec)[2] = ERL_DRV_STRING;
(*spec)[3] = (ErlDrvTermData)error;
(*spec)[4] = strlen(error);
(*spec)[5] = ERL_DRV_TUPLE;
(*spec)[6] = 2;
*terms_count = 7;
return 0;
}
static int sql_exec(sqlite3_drv_t *drv, char *command, int command_size) {
int result, next_row;
char *rest = NULL;
sqlite3_stmt *statement;
// fprintf(drv->log, "Preexec: %.*s\n", command_size, command);
// fflush (drv->log);
result = sqlite3_prepare_v2(drv->db, command, command_size, &statement, (const char **)&rest);
if(result != SQLITE_OK) {
ErlDrvTermData *dataset;
int term_count;
return_error(drv, sqlite3_errmsg(drv->db), &dataset, &term_count);
driver_output_term(drv->port, dataset, term_count);
return 0;
}
async_sqlite3_command *async_command = (async_sqlite3_command *)calloc(1, sizeof(async_sqlite3_command));
async_command->driver_data = drv;
async_command->statement = statement;
// fprintf(drv->log, "Driver async: %d %p\n", SQLITE_VERSION_NUMBER, async_command->statement);
// fflush (drv->log);
if (sqlite3_threadsafe()) {
drv->async_handle = driver_async(drv->port, &drv->key, sql_exec_async, async_command, sql_free_async);
} else {
sql_exec_async(async_command);
ready_async((ErlDrvData)drv, (ErlDrvThreadData)async_command);
}
return 0;
}
static void sql_free_async(void *_async_command)
{
int i;
async_sqlite3_command *async_command = (async_sqlite3_command *)_async_command;
free(async_command->dataset);
async_command->driver_data->async_handle = 0;
if (async_command->floats) {
free(async_command->floats);
}
for (i = 0; i < async_command->binaries_count; i++) {
driver_free_binary(async_command->binaries[i]);
}
if(async_command->binaries) {
free(async_command->binaries);
}
if (async_command->statement) {
sqlite3_finalize(async_command->statement);
}
free(async_command);
}
static void sql_exec_async(void *_async_command) {
async_sqlite3_command *async_command = (async_sqlite3_command *)_async_command;
ErlDrvTermData *dataset = async_command->dataset;
int term_count = async_command->term_count;
int row_count = async_command->row_count;
sqlite3_drv_t *drv = async_command->driver_data;
int result, next_row, column_count;
char *error = NULL;
char *rest = NULL;
sqlite3_stmt *statement = async_command->statement;
double *floats = NULL;
int float_count = 0;
ErlDrvBinary **binaries = NULL;
int binaries_count = 0;
int i;
column_count = sqlite3_column_count(statement);
dataset = NULL;
term_count += 2;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[term_count - 2] = ERL_DRV_PORT;
dataset[term_count - 1] = driver_mk_port(drv->port);
if (column_count > 0) {
int base = term_count;
term_count += 2 + column_count*3 + 1 + 2 + 2 + 2;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[base] = ERL_DRV_ATOM;
dataset[base + 1] = drv->atom_columns;
for (i = 0; i < column_count; i++) {
char *column_name = (char *)sqlite3_column_name(statement, i);
// fprintf(drv->log, "Column: %s\n", column_name);
// fflush (drv->log);
dataset[base + 2 + (i*3)] = ERL_DRV_STRING;
dataset[base + 2 + (i*3) + 1] = (ErlDrvTermData) column_name;
dataset[base + 2 + (i*3) + 2] = strlen (column_name);
}
dataset[base + 2 + column_count*3 + 0] = ERL_DRV_NIL;
dataset[base + 2 + column_count*3 + 1] = ERL_DRV_LIST;
dataset[base + 2 + column_count*3 + 2] = column_count + 1;
dataset[base + 2 + column_count*3 + 3] = ERL_DRV_TUPLE;
dataset[base + 2 + column_count*3 + 4] = 2;
dataset[base + 2 + column_count*3 + 5] = ERL_DRV_ATOM;
dataset[base + 2 + column_count*3 + 6] = drv->atom_rows;
}
// fprintf(drv->log, "Exec: %s\n", sqlite3_sql(statement));
// fflush (drv->log);
while ((next_row = sqlite3_step(statement)) == SQLITE_ROW) {
for (i = 0; i < column_count; i++) {
// fprintf(drv->log, "Column %d type: %d\n", i, sqlite3_column_type(statement, i));
// fflush (drv->log);
switch (sqlite3_column_type(statement, i)) {
case SQLITE_INTEGER: {
term_count += 2;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[term_count - 2] = ERL_DRV_INT;
dataset[term_count - 1] = sqlite3_column_int(statement, i);
break;
}
case SQLITE_FLOAT: {
float_count++;
floats = realloc(floats, sizeof(double) * float_count);
floats[float_count - 1] = sqlite3_column_double(statement, i);
term_count += 2;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[term_count - 2] = ERL_DRV_FLOAT;
dataset[term_count - 1] = (ErlDrvTermData)&floats[float_count - 1];
break;
}
case SQLITE_BLOB:
case SQLITE_TEXT: {
int bytes = sqlite3_column_bytes(statement, i);
binaries_count++;
binaries = realloc(binaries, sizeof(*binaries) * binaries_count);
binaries[binaries_count - 1] = driver_alloc_binary(bytes);
binaries[binaries_count - 1]->orig_size = bytes;
memcpy(binaries[binaries_count - 1]->orig_bytes, sqlite3_column_blob(statement, i), bytes);
term_count += 4;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[term_count - 4] = ERL_DRV_BINARY;
dataset[term_count - 3] = (ErlDrvTermData)binaries[binaries_count - 1];
dataset[term_count - 2] = bytes;
dataset[term_count - 1] = 0;
break;
}
case SQLITE_NULL: {
term_count += 2;
dataset = realloc (dataset, sizeof (*dataset) * term_count);
dataset[term_count - 2] = ERL_DRV_ATOM;
dataset[term_count - 1] = drv->atom_null;
break;
}
}
}
term_count += 2;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[term_count - 2] = ERL_DRV_TUPLE;
dataset[term_count - 1] = column_count;
row_count++;
}
async_command->row_count = row_count;
async_command->floats = floats;
async_command->binaries = binaries;
async_command->binaries_count = binaries_count;
if (next_row == SQLITE_BUSY) {
return_error(drv, "SQLite3 database is busy", &async_command->dataset, &async_command->term_count);
return;
}
if (next_row != SQLITE_DONE) {
return_error(drv, sqlite3_errmsg(drv->db), &async_command->dataset, &async_command->term_count);
return;
}
if (column_count > 0) {
term_count += 3;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[term_count - 3] = ERL_DRV_NIL;
dataset[term_count - 2] = ERL_DRV_LIST;
dataset[term_count - 1] = row_count + 1;
term_count += 2;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[term_count - 2] = ERL_DRV_TUPLE;
dataset[term_count - 1] = 2;
term_count += 3;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[term_count - 3] = ERL_DRV_NIL;
dataset[term_count - 2] = ERL_DRV_LIST;
dataset[term_count - 1] = 3;
} else if (strcasestr(sqlite3_sql(statement), "INSERT")) {
long long rowid = sqlite3_last_insert_rowid(drv->db);
term_count += 6;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[term_count - 6] = ERL_DRV_ATOM;
dataset[term_count - 5] = drv->atom_id;
dataset[term_count - 4] = ERL_DRV_INT;
dataset[term_count - 3] = rowid;
dataset[term_count - 2] = ERL_DRV_TUPLE;
dataset[term_count - 1] = 2;
} else {
term_count += 6;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[term_count - 6] = ERL_DRV_ATOM;
dataset[term_count - 5] = drv->atom_ok;
dataset[term_count - 4] = ERL_DRV_INT;
dataset[term_count - 3] = next_row;
dataset[term_count - 2] = ERL_DRV_TUPLE;
dataset[term_count - 1] = 2;
}
term_count += 2;
dataset = realloc(dataset, sizeof(*dataset) * term_count);
dataset[term_count - 2] = ERL_DRV_TUPLE;
dataset[term_count - 1] = 2;
async_command->dataset = dataset;
async_command->term_count = term_count;
// fprintf(drv->log, "Total term count: %p %d, rows count: %dx%d\n", statement, term_count, column_count, row_count);
// fflush (drv->log);
}
static void ready_async(ErlDrvData drv_data, ErlDrvThreadData thread_data)
{
async_sqlite3_command *async_command = (async_sqlite3_command *)thread_data;
sqlite3_drv_t *drv = async_command->driver_data;
int res = driver_output_term(drv->port, async_command->dataset, async_command->term_count);
// fprintf(drv->log, "Total term count: %p %d, rows count: %d (%d)\n", async_command->statement, async_command->term_count, async_command->row_count, res);
// fflush (drv->log);
sql_free_async(async_command);
}
// Unkown Command
static int unknown(sqlite3_drv_t *drv, char *command, int command_size) {
// Return {error, unkown_command}
ErlDrvTermData spec[] = {ERL_DRV_ATOM, drv->atom_error,
ERL_DRV_ATOM, drv->atom_unknown_cmd,
ERL_DRV_TUPLE, 2};
return driver_output_term(drv->port, spec, sizeof(spec) / sizeof(spec[0]));
}

62
priv/sqlite3_drv.h
View File

@@ -1,62 +0,0 @@
#include <erl_driver.h>
#include <ei.h>
#include <stdio.h>
#include <string.h>
#include <sqlite3.h>
#include <erl_interface.h>
#if SQLITE_VERSION_NUMBER < 3006001
#error "SQLite3 of version 3.6.1 minumum required"
#endif
// Path to file where data will be stored.
// It will be created if it doesn't exist
#define DB_PATH "./store.db"
// Binary commands between Erlang VM and Driver
#define CMD_SQL_EXEC 2
// #define CMD_DEL 3
// Number of bytes for each key
// (160 bits for SHA1 hash)
#define KEY_SIZE 20
// Define struct to hold state across calls
typedef struct sqlite3_drv_t {
ErlDrvPort port;
unsigned int key;
struct sqlite3 *db;
long async_handle;
FILE *log;
ErlDrvTermData atom_error;
ErlDrvTermData atom_columns;
ErlDrvTermData atom_rows;
ErlDrvTermData atom_null;
ErlDrvTermData atom_id;
ErlDrvTermData atom_ok;
ErlDrvTermData atom_unknown_cmd;
} sqlite3_drv_t;
typedef struct async_sqlite3_command {
sqlite3_drv_t *driver_data;
sqlite3_stmt *statement;
ErlDrvTermData *dataset;
int term_count;
int row_count;
double *floats;
int binaries_count;
ErlDrvBinary **binaries;
} async_sqlite3_command;
static ErlDrvData start(ErlDrvPort port, char* cmd);
static void stop(ErlDrvData handle);
static int control(ErlDrvData drv_data, unsigned int command, char *buf,
int len, char **rbuf, int rlen);
static int sql_exec(sqlite3_drv_t *drv, char *buf, int len);
static void sql_exec_async(void *async_command);
static void sql_free_async(void *async_command);
static void ready_async(ErlDrvData drv_data, ErlDrvThreadData thread_data);
static int unknown(sqlite3_drv_t *bdb_drv, char *buf, int len);

194
priv/sqlite_port.c
View File

@@ -1,194 +0,0 @@
#include <stdio.h>
#include <stdlib.h>
#include <sqlite3.h>
#include <string.h>
#include "erl_comm.h"
#include "erl_interface.h"
#include "ei.h"
#define MASTER_QUERY "select * from sqlite_master where type='table';"
static FILE *log;
static ETERM *result;
void respond(ETERM *r);
void send_error(char *err_msg);
void send_result();
void send_ok();
// 4 sql = CREATE TABLE t1 (t1key INTEGER PRIMARY KEY, data TEXT, num double, timeEnter DATE)
static int list_tables(void *notUsed, int argc, char **argv, char **azColName)
{
if (result == 0) {
result = erl_mk_empty_list();
}
fprintf(log, "%d %s = %s\n", 2, azColName[2], argv[2]);
fprintf(log, "\n");
fflush(log);
result = erl_cons(erl_mk_atom(argv[2]), result);
return 0;
}
static int callback(void *notUsed, int argc, char **argv, char **azColName)
{
ETERM **record_list;
int i;
record_list = malloc(argc * sizeof(ETERM *));
fprintf(log, "runs %d\n", argc);
for (i = 0; i < argc; i++) {
fprintf(log, "%s = %s\n", azColName[i], argv[i] ? argv[i] : "NULL");
if (argv[i]) {
record_list[i] = erl_mk_string(argv[i]);
}
else {
record_list[i] = erl_mk_empty_list();
}
}
fprintf(log, "\n");
fflush(log);
result = erl_cons(erl_mk_tuple(record_list, argc), result);
free(record_list);
return 0;
}
int main(int argc, char **argv)
{
ETERM *tuplep;
ETERM *fnp, *argp;
byte buf[1024];
sqlite3 *db;
char *zErrMsg = 0;
int rc;
log = fopen("/tmp/sqlite_port.log", "a+");
fprintf(log, "******start log (%s)******\n", argv[1]);
fflush(log);
rc = sqlite3_open(argv[1], &db);
if (rc) {
sqlite3_close(db);
exit(1);
}
erl_init(NULL, 0);
while (read_cmd(buf) > 0) {
tuplep = erl_decode(buf);
fnp = erl_element(1, tuplep);
argp = erl_element(2, tuplep);
if (strncmp((const char *)ERL_ATOM_PTR(fnp), "close", 5) == 0) {
fprintf(log, "closing sqlite3_close\n");
fflush(log);
sqlite3_close(db);
break;
}
else if (strncmp((const char *)ERL_ATOM_PTR(fnp), "list_tables", 11) == 0) {
fprintf(log, "calling list_tables\n");
result = 0;
rc = sqlite3_exec(db, MASTER_QUERY, list_tables, 0, &zErrMsg);
if (rc != SQLITE_OK) {
send_error(zErrMsg);
sqlite3_free(zErrMsg);
}
else if (result != 0) {
send_result();
}
else {
// not an error and no results. still need to return something
send_ok();
}
fflush(log);
}
else if (strncmp((const char *)ERL_ATOM_PTR(fnp), "sql_exec", 8) == 0) {
fprintf(log, "calling sqlite3_exec %s\n", erl_iolist_to_string(argp));
result = 0;
rc = sqlite3_exec(db, erl_iolist_to_string(argp), callback, 0, &zErrMsg);
if (rc != SQLITE_OK) {
send_error(zErrMsg);
sqlite3_free(zErrMsg);
}
else if (result != 0) {
send_result();
}
else {
// not an error and no results. still need to return something
send_ok();
}
fflush(log);
}
erl_free_compound(tuplep);
erl_free_term(fnp);
erl_free_term(argp);
}
fprintf(log, "******end log******\n");
fclose(log);
return 0;
}
void send_error(char *err_msg)
{
ETERM *tup_list[2];
ETERM *to_send;
tup_list[0] = erl_mk_atom("sql_error");
tup_list[1] = erl_mk_string(err_msg);
to_send = erl_mk_tuple(tup_list, 2);
fprintf(log, "SQL Error: %s\n", err_msg);
respond(to_send);
erl_free_term(tup_list[0]);
erl_free_term(tup_list[1]);
erl_free_compound(to_send);
}
void send_result()
{
fprintf(log, "returning at len %d\n", erl_term_len(result));
respond(result);
erl_free_compound(result);
result = 0;
}
void send_ok()
{
ETERM *to_send;
to_send = erl_mk_atom("ok");
fprintf(log, "returning ok at len %d\n", erl_term_len(to_send));
respond(to_send);
erl_free_term(to_send);
}
void respond(ETERM *r)
{
byte buf[2048];
bzero(buf, 2048);
erl_encode(r, buf);
write_cmd(buf, erl_term_len(r));
fprintf(log, "sending response back\n");
}