greg/bdberl
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

Merge branch 'master' of ecommons@wax.hive:bdberl

Browse source 
Conflicts:

	c_src/bdberl_drv.c
	src/bdberl_port.erl
This commit is contained in:
Dave Smith 2008-12-11 07:30:14 -07:00
parent aa3928b2c3 71f64d9666
commit b4c6aae90c
4 changed files with 341 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -1,3 +1,6 @@
*.o
*.so
*.beam
*.tmproj
c_src/system
test/logs

84
c_src/bdberl_drv.c
View file

@ -19,8 +19,8 @@
/**
* Function prototypes
*/
static int open_database(const char* name, DBTYPE type, PortData* data, int* errno);
static int close_database(int dbref, PortData* data);
static int open_database(const char* name, DBTYPE type, unsigned flags, PortData* data, int* errno);
static int close_database(int dbref, unsigned flags, PortData* data);
static void do_async_put(void* arg);
static void do_async_get(void* arg);
@ -94,6 +94,11 @@ static TPool* G_TPOOL_TXNS;
#define WRITE_LOCK(L) erl_drv_rwlock_rwlock(L)
#define WRITE_UNLOCK(L) erl_drv_rwlock_rwunlock(L)
#define UNPACK_BYTE(_buf, _off) (_buf[_off])
#define UNPACK_INT(_buf, _off) *((int*)(_buf+(_off)))
#define UNPACK_STRING(_buf, _off) (char*)(_buf+(_off))
#define UNPACK_BLOB(_buf, _off) (void*)(_buf+(_off))
#define RETURN_BH(bh, outbuf) *outbuf = (char*)bh.bin; return bh.bin->orig_size;
#define RETURN_INT(val, outbuf) { \
@ -222,7 +227,7 @@ static void bdberl_drv_stop(ErlDrvData handle)
// Close all the databases we previously opened
while (d->dbrefs)
{
close_database(d->dbrefs->dbref, d);
close_database(d->dbrefs->dbref, 0, d);
}
@ -265,12 +270,13 @@ static int bdberl_drv_control(ErlDrvData handle, unsigned int cmd,
case CMD_OPEN_DB:
{
// Extract the type code and filename from the inbuf
// Inbuf is: <<Type:8, Name/bytes, 0:8>>
DBTYPE type = (DBTYPE)((char)*inbuf);
char* name = (char*)(inbuf+1);
// Inbuf is: <<Flags:32/unsigned, Type:8, Name/bytes, 0:8>>
unsigned flags = UNPACK_INT(inbuf, 0);
DBTYPE type = (DBTYPE) UNPACK_BYTE(inbuf, 4);
char* name = UNPACK_STRING(inbuf, 5);
int dbref;
int status;
int rc = open_database(name, type, d, &dbref);
int rc = open_database(name, type, flags, d, &dbref);
if (rc == 0)
{
status = STATUS_OK;
@ -294,9 +300,12 @@ static int bdberl_drv_control(ErlDrvData handle, unsigned int cmd,
// TODO: If data is inflight, fail. Abort any open txns.
// Take the provided dbref and attempt to close it
int dbref = *((int*)inbuf);
int rc = close_database(dbref, d);
// Inbuf is: <<DbRef:32, Flags:32/unsigned>>
int dbref = UNPACK_INT(inbuf, 0);
unsigned flags = (unsigned) UNPACK_INT(inbuf, 4);
int rc = close_database(dbref, flags, d);
// Outbuf is: <<Rc:32>>
RETURN_INT(rc, outbuf);
}
@ -314,8 +323,11 @@ static int bdberl_drv_control(ErlDrvData handle, unsigned int cmd,
RETURN_INT(ERROR_TXN_OPEN, outbuf);
}
// Inbuf is <<Flags:32/unsigned>>
unsigned flags = UNPACK_INT(inbuf, 0);
// Outbuf is <<Rc:32>>
int rc = G_DB_ENV->txn_begin(G_DB_ENV, 0, &(d->txn), 0);
int rc = G_DB_ENV->txn_begin(G_DB_ENV, 0, &(d->txn), flags);
RETURN_INT(rc, outbuf);
}
case CMD_TXN_COMMIT:
@ -337,6 +349,11 @@ static int bdberl_drv_control(ErlDrvData handle, unsigned int cmd,
AsyncData* adata = zalloc(sizeof(AsyncData));
adata->port = d;
if (cmd == CMD_TXN_COMMIT)
{
adata->payload = (void*) UNPACK_INT(inbuf, 0);
}
// Update port data to indicate we have an operation in progress
d->async_op = cmd;
@ -362,7 +379,7 @@ static int bdberl_drv_control(ErlDrvData handle, unsigned int cmd,
}
// Inbuf is: << DbRef:32, Rest/binary>>
int dbref = *((int*)inbuf);
int dbref = UNPACK_INT(inbuf, 0);
// Make sure this port currently has dbref open -- if it doesn't, error out. Of note,
// if it's in our list, we don't need to grab the RWLOCK, as we don't have to worry about
@ -510,7 +527,7 @@ static void bdberl_drv_ready_input(ErlDrvData handle, ErlDrvEvent event)
}
}
static int open_database(const char* name, DBTYPE type, PortData* data, int* dbref_res)
static int open_database(const char* name, DBTYPE type, unsigned int flags, PortData* data, int* dbref_res)
{
*dbref_res = -1;
@ -592,9 +609,9 @@ static int open_database(const char* name, DBTYPE type, PortData* data, int* dbr
WRITE_UNLOCK(G_DATABASES_RWLOCK);
return rc;
}
// Attempt to open our database
rc = db->open(db, 0, name, 0, type, DB_CREATE | DB_AUTO_COMMIT | DB_THREAD, 0);
rc = db->open(db, 0, name, 0, type, flags, 0);
if (rc != 0)
{
// Failure while opening the database -- cleanup the handle, drop the lock
@ -623,7 +640,7 @@ static int open_database(const char* name, DBTYPE type, PortData* data, int* dbr
}
}
static int close_database(int dbref, PortData* data)
static int close_database(int dbref, unsigned flags, PortData* data)
{
// printf("Closing %d for port %p\n", dbref, data->port);
@ -647,7 +664,7 @@ static int close_database(int dbref, PortData* data)
{
printf("Closing actual database for dbref %d\n", dbref);
// Close out the BDB handle
database->db->close(database->db, 0);
database->db->close(database->db, flags);
// Remove the entry from the names map
hive_hash_remove(G_DATABASES_NAMES, database->name);
@ -666,9 +683,10 @@ static int close_database(int dbref, PortData* data)
static void do_async_put(void* arg)
{
// Payload is: <<DbRef:32, Flags:32, KeyLen:32, Key:KeyLen, ValLen:32, Val:ValLen>>
AsyncData* adata = (AsyncData*)arg;
// printf("%p: do_async_put\n", adata->port->port);
unsigned flags = UNPACK_INT(adata->payload, 4);
// Setup DBTs
DBT key;
DBT value;
@ -676,16 +694,15 @@ static void do_async_put(void* arg)
memset(&value, '\0', sizeof(DBT));
// Parse payload into DBTs
// Payload is: << DbRef:32, KeyLen:32, Key:KeyLen, ValLen:32, Val:ValLen>>
key.size = *((int*)(adata->payload + 4));
key.data = (void*)(adata->payload + 8);
value.size = *((int*)(adata->payload + 8 + key.size));
value.data = (void*)(adata->payload + 8 + key.size + 4);
key.size = UNPACK_INT(adata->payload, 8);
key.data = UNPACK_BLOB(adata->payload, 12);
value.size = UNPACK_INT(adata->payload, 12 + key.size);
value.data = UNPACK_BLOB(adata->payload, 12 + key.size + 4);
// Execute the actual put -- we'll process the result back in the driver_async_ready function
// All databases are opened with AUTO_COMMIT, so if msg->port->txn is NULL, the put will still
// be atomic
adata->rc = adata->db->put(adata->db, adata->port->txn, &key, &value, 0);
adata->rc = adata->db->put(adata->db, adata->port->txn, &key, &value, flags);
// If any error occurs while we have a txn action, abort it
if (adata->port->txn && adata->rc)
@ -699,9 +716,10 @@ static void do_async_put(void* arg)
static void do_async_get(void* arg)
{
// Payload is: << DbRef:32, Flags:32, KeyLen:32, Key:KeyLen >>
AsyncData* adata = (AsyncData*)arg;
// printf("%p: do_async_get\n", adata->port->port);
unsigned flags = UNPACK_INT(adata->payload, 4);
// Setup DBTs
DBT key;
DBT value;
@ -709,9 +727,8 @@ static void do_async_get(void* arg)
memset(&value, '\0', sizeof(DBT));
// Parse payload into DBT
// Payload is: << DbRef:32, KeyLen:32, Key:KeyLen >>
key.size = *((int*)(adata->payload + 4));
key.data = (void*)(adata->payload + 8);
key.size = UNPACK_INT(adata->payload, 8);
key.data = UNPACK_BLOB(adata->payload, 12);
// Allocate memory to hold the value -- hard code initial size to 4k
// TODO: Make this smarter!
@ -719,13 +736,13 @@ static void do_async_get(void* arg)
value.ulen = 4096;
value.flags = DB_DBT_USERMEM;
int rc = adata->db->get(adata->db, adata->port->txn, &key, &value, 0);
int rc = adata->db->get(adata->db, adata->port->txn, &key, &value, flags);
while (rc == DB_BUFFER_SMALL)
{
// Grow our value buffer and try again
value.data = driver_realloc(value.data, value.size);
value.ulen = value.size;
rc = adata->db->get(adata->db, adata->port->txn, &key, &value, 0);
rc = adata->db->get(adata->db, adata->port->txn, &key, &value, flags);
}
adata->payload = value.data;
@ -749,7 +766,8 @@ static void do_async_txnop(void* arg)
// Execute the actual commit/abort
if (adata->port->async_op == CMD_TXN_COMMIT)
{
adata->rc = adata->port->txn->commit(adata->port->txn, 0);
unsigned flags = (unsigned) adata->payload;
adata->rc = adata->port->txn->commit(adata->port->txn, flags);
}
else
{

299
src/bdberl_port.erl
View file

@ -7,10 +7,11 @@
-module(bdberl_port).
-export([new/0,
open_database/3,
open_database/3, open_database/4,
close_database/2,
txn_begin/1, txn_commit/1, txn_abort/1,
put/4,
txn_begin/1, txn_begin/2,
txn_commit/1, txn_commit/2, txn_abort/1,
put/4, put/5,
get/3]).
-define(CMD_NONE, 0).
@ -26,6 +27,217 @@
-define(DB_TYPE_BTREE, 1).
-define(DB_TYPE_HASH, 2).
%% DB API flags
-define(DB_AGGRESSIVE, 16#00000001).
-define(DB_ARCH_ABS, 16#00000001).
-define(DB_ARCH_DATA, 16#00000002).
-define(DB_ARCH_LOG, 16#00000004).
-define(DB_ARCH_REMOVE, 16#00000008).
-define(DB_AUTO_COMMIT, 16#00000100).
-define(DB_CDB_ALLDB, 16#00000004).
-define(DB_CHKSUM, 16#00000004).
-define(DB_CKP_INTERNAL, 16#00000002).
-define(DB_CREATE, 16#00000001).
-define(DB_CXX_NO_EXCEPTIONS, 16#00000002).
-define(DB_DIRECT, 16#00000002).
-define(DB_DIRECT_DB, 16#00000040).
-define(DB_DSYNC_DB, 16#00000080).
-define(DB_DUP, 16#00000008).
-define(DB_DUPSORT, 16#00000002).
-define(DB_DURABLE_UNKNOWN, 16#00000020).
-define(DB_ENCRYPT, 16#00000001).
-define(DB_ENCRYPT_AES, 16#00000001).
-define(DB_EXCL, 16#00000400).
-define(DB_EXTENT, 16#00000004).
-define(DB_FAST_STAT, 16#00000001).
-define(DB_FCNTL_LOCKING, 16#00001000).
-define(DB_FLUSH, 16#00000001).
-define(DB_FORCE, 16#00000001).
-define(DB_FOREIGN_ABORT, 16#00000001).
-define(DB_FOREIGN_CASCADE, 16#00000002).
-define(DB_FOREIGN_NULLIFY, 16#00000004).
-define(DB_FREELIST_ONLY, 16#00000001).
-define(DB_FREE_SPACE, 16#00000002).
-define(DB_IGNORE_LEASE, 16#00001000).
-define(DB_IMMUTABLE_KEY, 16#00000002).
-define(DB_INIT_CDB, 16#00000020).
-define(DB_INIT_LOCK, 16#00000040).
-define(DB_INIT_LOG, 16#00000080).
-define(DB_INIT_MPOOL, 16#00000100).
-define(DB_INIT_REP, 16#00000200).
-define(DB_INIT_TXN, 16#00000400).
-define(DB_INORDER, 16#00000010).
-define(DB_JOIN_NOSORT, 16#00000001).
-define(DB_LOCKDOWN, 16#00000800).
-define(DB_LOCK_NOWAIT, 16#00000001).
-define(DB_LOCK_RECORD, 16#00000002).
-define(DB_LOCK_SET_TIMEOUT, 16#00000004).
-define(DB_LOCK_SWITCH, 16#00000008).
-define(DB_LOCK_UPGRADE, 16#00000010).
-define(DB_LOG_AUTO_REMOVE, 16#00000004).
-define(DB_LOG_CHKPNT, 16#00000002).
-define(DB_LOG_COMMIT, 16#00000004).
-define(DB_LOG_DIRECT, 16#00000001).
-define(DB_LOG_DSYNC, 16#00000002).
-define(DB_LOG_IN_MEMORY, 16#00000008).
-define(DB_LOG_NOCOPY, 16#00000008).
-define(DB_LOG_NOT_DURABLE, 16#00000010).
-define(DB_LOG_WRNOSYNC, 16#00000020).
-define(DB_LOG_ZERO, 16#00000010).
-define(DB_MPOOL_CREATE, 16#00000001).
-define(DB_MPOOL_DIRTY, 16#00000002).
-define(DB_MPOOL_DISCARD, 16#00000001).
-define(DB_MPOOL_EDIT, 16#00000004).
-define(DB_MPOOL_FREE, 16#00000008).
-define(DB_MPOOL_LAST, 16#00000010).
-define(DB_MPOOL_NEW, 16#00000020).
-define(DB_MPOOL_NOFILE, 16#00000001).
-define(DB_MPOOL_NOLOCK, 16#00000002).
-define(DB_MPOOL_UNLINK, 16#00000002).
-define(DB_MULTIPLE, 16#00002000).
-define(DB_MULTIPLE_KEY, 16#00000100).
-define(DB_MULTIVERSION, 16#00000008).
-define(DB_MUTEX_ALLOCATED, 16#00000001).
-define(DB_MUTEX_LOCKED, 16#00000002).
-define(DB_MUTEX_LOGICAL_LOCK, 16#00000004).
-define(DB_MUTEX_PROCESS_ONLY, 16#00000008).
-define(DB_MUTEX_SELF_BLOCK, 16#00000010).
-define(DB_NOLOCKING, 16#00000200).
-define(DB_NOMMAP, 16#00000010).
-define(DB_NOORDERCHK, 16#00000002).
-define(DB_NOPANIC, 16#00000400).
-define(DB_NO_AUTO_COMMIT, 16#00002000).
-define(DB_ODDFILESIZE, 16#00000040).
-define(DB_ORDERCHKONLY, 16#00000004).
-define(DB_OVERWRITE, 16#00001000).
-define(DB_PANIC_ENVIRONMENT, 16#00002000).
-define(DB_PRINTABLE, 16#00000008).
-define(DB_PRIVATE, 16#00001000).
-define(DB_PR_PAGE, 16#00000010).
-define(DB_PR_RECOVERYTEST, 16#00000020).
-define(DB_RDONLY, 16#00000080).
-define(DB_RDWRMASTER, 16#00004000).
-define(DB_READ_COMMITTED, 16#00000400).
-define(DB_READ_UNCOMMITTED, 16#00000200).
-define(DB_RECNUM, 16#00000020).
-define(DB_RECOVER, 16#00000010).
-define(DB_RECOVER_FATAL, 16#00002000).
-define(DB_REGION_INIT, 16#00004000).
-define(DB_REGISTER, 16#00004000).
-define(DB_RENUMBER, 16#00000040).
-define(DB_REPMGR_CONF_2SITE_STRICT, 16#00000001).
-define(DB_REPMGR_PEER, 16#00000001).
-define(DB_REP_ANYWHERE, 16#00000001).
-define(DB_REP_CLIENT, 16#00000001).
-define(DB_REP_CONF_BULK, 16#00000002).
-define(DB_REP_CONF_DELAYCLIENT, 16#00000004).
-define(DB_REP_CONF_LEASE, 16#00000008).
-define(DB_REP_CONF_NOAUTOINIT, 16#00000010).
-define(DB_REP_CONF_NOWAIT, 16#00000020).
-define(DB_REP_ELECTION, 16#00000004).
-define(DB_REP_MASTER, 16#00000002).
-define(DB_REP_NOBUFFER, 16#00000002).
-define(DB_REP_PERMANENT, 16#00000004).
-define(DB_REP_REREQUEST, 16#00000008).
-define(DB_REVSPLITOFF, 16#00000080).
-define(DB_RMW, 16#00000800).
-define(DB_RPCCLIENT, 16#00000001).
-define(DB_SALVAGE, 16#00000040).
-define(DB_SA_SKIPFIRSTKEY, 16#00010000).
-define(DB_SEQ_DEC, 16#00000001).
-define(DB_SEQ_INC, 16#00000002).
-define(DB_SEQ_RANGE_SET, 16#00000004).
-define(DB_SEQ_WRAP, 16#00000008).
-define(DB_SEQ_WRAPPED, 16#00000010).
-define(DB_SET_LOCK_TIMEOUT, 16#00000002).
-define(DB_SET_TXN_NOW, 16#00000004).
-define(DB_SET_TXN_TIMEOUT, 16#00000001).
-define(DB_SNAPSHOT, 16#00000100).
-define(DB_STAT_ALL, 16#00000002).
-define(DB_STAT_CLEAR, 16#00000001).
-define(DB_STAT_LOCK_CONF, 16#00000004).
-define(DB_STAT_LOCK_LOCKERS, 16#00000008).
-define(DB_STAT_LOCK_OBJECTS, 16#00000010).
-define(DB_STAT_LOCK_PARAMS, 16#00000020).
-define(DB_STAT_MEMP_HASH, 16#00000004).
-define(DB_STAT_MEMP_NOERROR, 16#00000008).
-define(DB_STAT_SUBSYSTEM, 16#00000004).
-define(DB_ST_DUPOK, 16#00000100).
-define(DB_ST_DUPSET, 16#00000200).
-define(DB_ST_DUPSORT, 16#00000400).
-define(DB_ST_IS_RECNO, 16#00000800).
-define(DB_ST_OVFL_LEAF, 16#00001000).
-define(DB_ST_RECNUM, 16#00002000).
-define(DB_ST_RELEN, 16#00004000).
-define(DB_ST_TOPLEVEL, 16#00008000).
-define(DB_SYSTEM_MEM, 16#00008000).
-define(DB_THREAD, 16#00000004).
-define(DB_TIME_NOTGRANTED, 16#00008000).
-define(DB_TRUNCATE, 16#00008000).
-define(DB_TXN_NOSYNC, 16#00000001).
-define(DB_TXN_NOT_DURABLE, 16#00000200).
-define(DB_TXN_NOWAIT, 16#00000002).
-define(DB_TXN_SNAPSHOT, 16#00000800).
-define(DB_TXN_SYNC, 16#00000004).
-define(DB_TXN_WAIT, 16#00000008).
-define(DB_TXN_WRITE_NOSYNC, 16#00000020).
-define(DB_UNREF, 16#00000080).
-define(DB_UPGRADE, 16#00000001).
-define(DB_USE_ENVIRON, 16#00000002).
-define(DB_USE_ENVIRON_ROOT, 16#00000008).
-define(DB_VERB_DEADLOCK, 16#00000001).
-define(DB_VERB_FILEOPS, 16#00000002).
-define(DB_VERB_FILEOPS_ALL, 16#00000004).
-define(DB_VERB_RECOVERY, 16#00000008).
-define(DB_VERB_REGISTER, 16#00000010).
-define(DB_VERB_REPLICATION, 16#00000020).
-define(DB_VERB_REPMGR_CONNFAIL, 16#00000040).
-define(DB_VERB_REPMGR_MISC, 16#00000080).
-define(DB_VERB_REP_ELECT, 16#00000100).
-define(DB_VERB_REP_LEASE, 16#00000200).
-define(DB_VERB_REP_MISC, 16#00000400).
-define(DB_VERB_REP_MSGS, 16#00000800).
-define(DB_VERB_REP_SYNC, 16#00001000).
-define(DB_VERB_WAITSFOR, 16#00002000).
-define(DB_VERIFY, 16#00000002).
-define(DB_WRITEOPEN, 16#00010000).
-define(DB_XA_CREATE, 16#00000800).
-define(DB_YIELDCPU, 16#00010000).
%% DB access method and cursor operation values. Each value is an operation
%% code to which additional bit flags are added.
-define(DB_AFTER, 1).
-define(DB_APPEND, 2).
-define(DB_BEFORE, 3).
-define(DB_CONSUME, 4).
-define(DB_CONSUME_WAIT, 5).
-define(DB_CURRENT, 6).
-define(DB_FIRST, 7).
-define(DB_GET_BOTH, 8).
-define(DB_GET_BOTHC, 9).
-define(DB_GET_BOTH_RANGE, 10).
-define(DB_GET_RECNO, 11).
-define(DB_JOIN_ITEM, 12).
-define(DB_KEYFIRST, 13).
-define(DB_KEYLAST, 14).
-define(DB_LAST, 15).
-define(DB_NEXT, 16).
-define(DB_NEXT_DUP, 17).
-define(DB_NEXT_NODUP, 18).
-define(DB_NODUPDATA, 19).
-define(DB_NOOVERWRITE, 20).
-define(DB_NOSYNC, 21).
-define(DB_POSITION, 22).
-define(DB_PREV, 23).
-define(DB_PREV_DUP, 24).
-define(DB_PREV_NODUP, 25).
-define(DB_SET, 26).
-define(DB_SET_RANGE, 27).
-define(DB_SET_RECNO, 28).
-define(DB_UPDATE_SECONDARY, 29).
-define(DB_WRITECURSOR, 30).
-define(DB_WRITELOCK, 31).
-define(STATUS_OK, 0).
-define(STATUS_ERROR, 1).
@ -47,14 +259,17 @@ new() ->
Port = open_port({spawn, bdberl_drv}, [binary]),
{ok, Port}.
open_database(Port, Name, Type) ->
open_database(Port, Name, Type, [create, auto_commit, threaded]).
open_database(Port, Name, Type, Opts) ->
%% Map database type into an integer code
case Type of
btree -> TypeCode = ?DB_TYPE_BTREE;
hash -> TypeCode = ?DB_TYPE_HASH
end,
Cmd = <<TypeCode:8/native-integer, (list_to_binary(Name))/bytes, 0:8/native-integer>>,
end,
Flags = process_flags(Opts),
Cmd = <<Flags:32/unsigned-native-integer, TypeCode:8/native-integer, (list_to_binary(Name))/bytes, 0:8/native-integer>>,
case erlang:port_control(Port, ?CMD_OPEN_DB, Cmd) of
<<?STATUS_OK:8, DbRef:32/native>> ->
{ok, DbRef};
@ -63,7 +278,11 @@ open_database(Port, Name, Type) ->
end.
close_database(Port, DbRef) ->
Cmd = <<DbRef:32/native-integer>>,
close_database(Port, DbRef, []).
close_database(Port, DbRef, Opts) ->
Flags = process_flags(Opts),
Cmd = <<DbRef:32/native-integer, Flags:32/unsigned-native-integer>>,
case erlang:port_control(Port, ?CMD_CLOSE_DB, Cmd) of
<<0:32/native-integer>> ->
{error, invalid_dbref};
@ -72,15 +291,24 @@ close_database(Port, DbRef) ->
end.
txn_begin(Port) ->
<<Result:32/native>> = erlang:port_control(Port, ?CMD_TXN_BEGIN, <<>>),
txn_begin(Port, []).
txn_begin(Port, Opts) ->
Flags = process_flags(Opts),
Cmd = <<Flags:32/unsigned-native>>,
<<Result:32/native>> = erlang:port_control(Port, ?CMD_TXN_BEGIN, Cmd),
case decode_rc(Result) of
ok -> ok;
Error -> {error, {txn_begin, Error}}
end.
txn_commit(Port) ->
<<Result:32/native>> = erlang:port_control(Port, ?CMD_TXN_COMMIT, <<>>),
txn_commit(Port, []).
txn_commit(Port, Opts) ->
Flags = process_flags(Opts),
Cmd = <<Flags:32/unsigned-native>>,
<<Result:32/native>> = erlang:port_control(Port, ?CMD_TXN_COMMIT, Cmd),
case decode_rc(Result) of
ok ->
receive
@ -103,11 +331,14 @@ txn_abort(Port) ->
{error, {txn_abort, Error}}
end.
put(Port, DbRef, Key, Value) ->
put(Port, DbRef, Key, Value, []).
put(Port, DbRef, Key, Value, Opts) ->
{KeyLen, KeyBin} = to_binary(Key),
{ValLen, ValBin} = to_binary(Value),
Cmd = <<DbRef:32/native, KeyLen:32/native, KeyBin/bytes, ValLen:32/native, ValBin/bytes>>,
Flags = process_flags(Opts),
Cmd = <<DbRef:32/native, Flags:32/unsigned-native, KeyLen:32/native, KeyBin/bytes, ValLen:32/native, ValBin/bytes>>,
<<Result:32/native>> = erlang:port_control(Port, ?CMD_PUT, Cmd),
case decode_rc(Result) of
ok ->
@ -119,10 +350,13 @@ put(Port, DbRef, Key, Value) ->
{error, {put, decode_rc(Error)}}
end.
get(Port, DbRef, Key) ->
get(Port, DbRef, Key, []).
get(Port, DbRef, Key, Opts) ->
{KeyLen, KeyBin} = to_binary(Key),
Cmd = <<DbRef:32/native, KeyLen:32/native, KeyBin/bytes>>,
Flags = process_flags(Opts),
Cmd = <<DbRef:32/native, Flags:32/unsigned-native, KeyLen:32/native, KeyBin/bytes>>,
<<Result:32/native>> = erlang:port_control(Port, ?CMD_GET, Cmd),
case decode_rc(Result) of
ok ->
@ -151,7 +385,6 @@ decode_rc(?ERROR_DB_LOCK_NOTGRANTED) -> lock_not_granted;
decode_rc(?ERROR_DB_LOCK_DEADLOCK) -> deadlock;
decode_rc(Rc) -> {unknown, Rc}.
%%
%% Convert a term into a binary, returning a tuple with the binary and the length of the binary
%%
@ -160,3 +393,39 @@ to_binary(Term) ->
{size(Bin), Bin}.
process_flags([]) ->
0;
process_flags([Flag|Flags]) ->
flag_value(Flag) bor process_flags(Flags).
flag_value(Flag) ->
case Flag of
append -> ?DB_APPEND;
auto_commit -> ?DB_AUTO_COMMIT;
consume -> ?DB_CONSUME;
consume_wait -> ?DB_CONSUME_WAIT;
create -> ?DB_CREATE;
exclusive -> ?DB_EXCL;
get_both -> ?DB_GET_BOTH;
ignore_lease -> ?DB_IGNORE_LEASE;
multiple -> ?DB_MULTIPLE;
multiversion -> ?DB_MULTIVERSION;
no_duplicate -> ?DB_NODUPDATA;
no_mmap -> ?DB_NOMMAP;
no_overwrite -> ?DB_NOOVERWRITE;
no_sync -> ?DB_NOSYNC;
read_committed -> ?DB_READ_COMMITTED;
read_uncommitted -> ?DB_READ_UNCOMMITTED;
readonly -> ?DB_RDONLY;
rmw -> ?DB_RMW;
set_recno -> ?DB_SET_RECNO;
threaded -> ?DB_THREAD;
truncate -> ?DB_TRUNCATE;
txn_no_sync -> ?DB_TXN_NOSYNC;
txn_no_wait -> ?DB_TXN_NOWAIT;
txn_snapshot -> ?DB_TXN_SNAPSHOT;
txn_sync -> ?DB_TXN_SYNC;
txn_wait -> ?DB_TXN_WAIT;
txn_write_nosync -> ?DB_TXN_WRITE_NOSYNC
end.

12
test/port_SUITE.erl
View file

@ -12,12 +12,12 @@ all() ->
% [test_db].
[test_put].% test_txn].
init_per_testcase(TestCase, Config) ->
init_per_testcase(_TestCase, Config) ->
Config.
end_per_testcase(TestCase, _Config) ->
end_per_testcase(_TestCase, _Config) ->
ok.
test_db(_Config) ->
{ok, P} = bdberl_port:new(),
@ -71,9 +71,3 @@ test_txn(_Config) ->
ok = bdberl_port:txn_begin(P),
not_found = bdberl_port:get(P, 0, akey),
ok = bdberl_port:txn_commit(P).