greg/stasis-aries-wal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Starting work on the actual graph-generation portion of the benchmarks. :)

Browse source
This commit is contained in:
Sears Russell 2004-10-20 06:21:21 +00:00
parent 903ba4b97b
commit af9e7adee0
17 changed files with 2227 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
benchmarks/Makefile.am
View file

@ -1,5 +1,7 @@
LDADD=$(top_builddir)/src/2pc/lib2pc.a $(top_builddir)/src/libdfa/libdfa.a \
$(top_builddir)/src/lladd/liblladd.a $(top_builddir)/src/pbl/libpbl.a \
$(top_builddir)/src/libdfa/librw.a
bin_PROGRAMS=naiveHash logicalHash readLogicalHash
bin_PROGRAMS=naiveHash logicalHash readLogicalHash naiveMultiThreaded logicalMultThreaded rawSet arrayListSet
AM_CFLAGS= -g -Wall -pedantic -std=gnu99
SUBDIRS=berkeleyDB

44
benchmarks/arrayListSet.c Normal file
View file

@ -0,0 +1,44 @@
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <lladd/transactional.h>
#include <unistd.h>
int main(int argc, char** argv) {
assert(argc == 3);
int xact_count = atoi(argv[1]);
int count = atoi(argv[2]);
/* unlink("storefile.txt");
unlink("logfile.txt");
unlink("blob0_file.txt");
unlink("blob1_file.txt");*/
Tinit();
int xid = Tbegin();
Tcommit(xid);
recordid arrayList = TarrayListAlloc(xid, 512, 2, sizeof(int));
recordid rid = arrayList;
rid.slot = 0;
int i = 0;
int k;
for(k = 0; k < xact_count; k++) {
xid = Tbegin();
for(; i < (count*(k+1)) ; i++) {
TarrayListExtend(xid, arrayList, 1);
Tset(xid, rid, &i);
rid.slot++;
}
Tcommit(xid);
}
Tdeinit();
}

595
benchmarks/berkeleyDB/bdbHash.c Normal file
View file

@ -0,0 +1,595 @@
#include <sys/types.h>
#include <sys/stat.h>
#include <errno.h>
#include <pthread.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <db.h>
#define ENV_DIRECTORY "TXNAPP"
void add_cat(DB_ENV *, DB *, char *, ...);
void run_xact(DB_ENV *, DB *, int, int);
/*void add_color(DB_ENV *, DB *, char *, int);
void add_fruit(DB_ENV *, DB *, char *, char *); */
void *checkpoint_thread(void *);
void log_archlist(DB_ENV *);
void *logfile_thread(void *);
void db_open(DB_ENV *, DB **, char *, int);
void env_dir_create(void);
void env_open(DB_ENV **);
void usage(void);
int
main(int argc, char *argv[])
{
extern int optind;
DB *db_cats; /*, *db_color, *db_fruit; */
DB_ENV *dbenv;
pthread_t ptid;
int ch, ret;
/* while ((ch = getopt(argc, argv, "")) != EOF)
switch (ch) {
case '?':
default:
usage();
}
argc -= optind;
argv += optind; */
assert(argc == 3);
env_dir_create();
env_open(&dbenv);
/* Start a checkpoint thread. */
if ((ret = pthread_create(
&ptid, NULL, checkpoint_thread, (void *)dbenv)) != 0) {
fprintf(stderr,
"txnapp: failed spawning checkpoint thread: %s\n",
strerror(ret));
exit (1);
}
/* Start a logfile removal thread. */
if ((ret = pthread_create(
&ptid, NULL, logfile_thread, (void *)dbenv)) != 0) {
fprintf(stderr,
"txnapp: failed spawning log file removal thread: %s\n",
strerror(ret));
exit (1);
}
/* Open database: Key is fruit class; Data is specific type. */
/* db_open(dbenv, &db_fruit, "fruit", 0); */
/* Open database: Key is a color; Data is an integer. */
/*db_open(dbenv, &db_color, "color", 0); */
/*
* Open database:
* Key is a name; Data is: company name, address, cat breeds.
*/
db_open(dbenv, &db_cats, "cats", 1);
/* add_fruit(dbenv, db_fruit, "apple", "yellow delicious");
add_color(dbenv, db_color, "blue", 0);
add_color(dbenv, db_color, "blue", 3); */
/* add_cat(dbenv, db_cats,
"Amy Adams",
"Sleepycat Software",
"394 E. Riding Dr., Carlisle, MA 01741, USA",
"abyssinian",
"bengal",
"chartreaux",
NULL);*/
int r;
int num_xact = atoi(argv[1]); //100;
int insert_per_xact = atoi(argv[2]); //1000;
for(r = 0; r < num_xact; r ++) {
run_xact(dbenv, db_cats, 1+r*insert_per_xact, insert_per_xact);
}
return (0);
}
void
env_dir_create()
{
struct stat sb;
/*
* If the directory exists, we're done. We do not further check
* the type of the file, DB will fail appropriately if it's the
* wrong type.
*/
if (stat(ENV_DIRECTORY, &sb) == 0)
return;
/* Create the directory, read/write/access owner only. */
if (mkdir(ENV_DIRECTORY, S_IRWXU) != 0) {
fprintf(stderr,
"txnapp: mkdir: %s: %s\n", ENV_DIRECTORY, strerror(errno));
exit (1);
}
}
void
env_open(DB_ENV **dbenvp)
{
DB_ENV *dbenv;
int ret;
/* Create the environment handle. */
if ((ret = db_env_create(&dbenv, 0)) != 0) {
fprintf(stderr,
"txnapp: db_env_create: %s\n", db_strerror(ret));
exit (1);
}
/* Set up error handling. */
dbenv->set_errpfx(dbenv, "txnapp");
dbenv->set_errfile(dbenv, stderr);
/* Do deadlock detection internally. */
if ((ret = dbenv->set_lk_detect(dbenv, DB_LOCK_DEFAULT)) != 0) {
dbenv->err(dbenv, ret, "set_lk_detect: DB_LOCK_DEFAULT");
exit (1);
}
/*
* Open a transactional environment:
* create if it doesn't exist
* free-threaded handle
* run recovery
* read/write owner only
*/
if ((ret = dbenv->open(dbenv, ENV_DIRECTORY,
DB_CREATE | DB_INIT_LOCK | DB_INIT_LOG |
DB_INIT_MPOOL | DB_INIT_TXN | DB_RECOVER | DB_THREAD,
S_IRUSR | S_IWUSR)) != 0) {
dbenv->err(dbenv, ret, "dbenv->open: %s", ENV_DIRECTORY);
exit (1);
}
*dbenvp = dbenv;
}
void *
checkpoint_thread(void *arg)
{
DB_ENV *dbenv;
int ret;
dbenv = arg;
dbenv->errx(dbenv, "Checkpoint thread: %lu", (u_long)pthread_self());
/* Checkpoint once a minute. */
for (;; sleep(60))
if ((ret = dbenv->txn_checkpoint(dbenv, 0, 0, 0)) != 0) {
dbenv->err(dbenv, ret, "checkpoint thread");
exit (1);
}
/* NOTREACHED */
}
void *
logfile_thread(void *arg)
{
DB_ENV *dbenv;
int ret;
char **begin, **list;
dbenv = arg;
dbenv->errx(dbenv,
"Log file removal thread: %lu", (u_long)pthread_self());
/* Check once every 5 minutes. */
for (;; sleep(300)) {
/* Get the list of log files. */
if ((ret =
dbenv->log_archive(dbenv, &list, DB_ARCH_ABS)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->log_archive");
exit (1);
}
/* Remove the log files. */
if (list != NULL) {
for (begin = list; *list != NULL; ++list)
if ((ret = remove(*list)) != 0) {
dbenv->err(dbenv,
ret, "remove %s", *list);
exit (1);
}
free (begin);
}
}
/* NOTREACHED */
}
void
log_archlist(DB_ENV *dbenv)
{
int ret;
char **begin, **list;
/* Get the list of database files. */
if ((ret = dbenv->log_archive(dbenv,
&list, DB_ARCH_ABS | DB_ARCH_DATA)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->log_archive: DB_ARCH_DATA");
exit (1);
}
if (list != NULL) {
for (begin = list; *list != NULL; ++list)
printf("database file: %s\n", *list);
free (begin);
}
/* Get the list of log files. */
if ((ret = dbenv->log_archive(dbenv,
&list, DB_ARCH_ABS | DB_ARCH_LOG)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->log_archive: DB_ARCH_LOG");
exit (1);
}
if (list != NULL) {
for (begin = list; *list != NULL; ++list)
printf("log file: %s\n", *list);
free (begin);
}
}
void
db_open(DB_ENV *dbenv, DB **dbp, char *name, int dups)
{
DB *db;
int ret;
/* Create the database handle. */
if ((ret = db_create(&db, dbenv, 0)) != 0) {
dbenv->err(dbenv, ret, "db_create");
exit (1);
}
/* Optionally, turn on duplicate data items. */
/* if (dups && (ret = db->set_flags(db, DB_DUP)) != 0) {
dbenv->err(dbenv, ret, "db->set_flags: DB_DUP");
exit (1);
} */
/*
* Open a database in the environment:
* create if it doesn't exist
* free-threaded handle
* read/write owner only
*/
if ((ret = db->open(db, NULL, name, NULL, /*DB_BTREE*//* DB_RECNO */DB_HASH,
DB_AUTO_COMMIT | DB_CREATE | DB_THREAD, S_IRUSR | S_IWUSR)) != 0) {
(void)db->close(db, 0);
dbenv->err(dbenv, ret, "db->open: %s", name);
exit (1);
}
*dbp = db;
}
void
add_fruit(DB_ENV *dbenv, DB *db, char *fruit, char *name)
{
DBT key, data;
DB_TXN *tid;
int ret;
/* Initialization. */
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
key.data = fruit;
key.size = strlen(fruit);
data.data = name;
data.size = strlen(name);
for (;;) {
/* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/* Store the value. */
switch (ret = db->put(db, tid, &key, &data, 0)) {
case 0:
/* Success: commit the change. */
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
return;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
break;
default:
/* Error: run recovery. */
dbenv->err(dbenv, ret, "dbc->put: %s/%s", fruit, name);
exit (1);
}
}
}
void
add_color(DB_ENV *dbenv, DB *dbp, char *color, int increment)
{
DBT key, data;
DB_TXN *tid;
int original, ret;
char buf[64];
/* Initialization. */
memset(&key, 0, sizeof(key));
key.data = color;
key.size = strlen(color);
memset(&data, 0, sizeof(data));
data.flags = DB_DBT_MALLOC;
for (;;) {
/* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/*
* Get the key. If it exists, we increment the value. If it
* doesn't exist, we create it.
*/
switch (ret = dbp->get(dbp, tid, &key, &data, 0)) {
case 0:
original = atoi(data.data);
break;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
continue;
case DB_NOTFOUND:
original = 0;
break;
default:
/* Error: run recovery. */
dbenv->err(
dbenv, ret, "dbc->get: %s/%d", color, increment);
exit (1);
}
if (data.data != NULL)
free(data.data);
/* Create the new data item. */
(void)snprintf(buf, sizeof(buf), "%d", original + increment);
data.data = buf;
data.size = strlen(buf) + 1;
/* Store the new value. */
switch (ret = dbp->put(dbp, tid, &key, &data, 0)) {
case 0:
/* Success: commit the change. */
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
return;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
break;
default:
/* Error: run recovery. */
dbenv->err(
dbenv, ret, "dbc->put: %s/%d", color, increment);
exit (1);
}
}
}
void
add_cat(DB_ENV *dbenv, DB *db, char *name, ...)
{
va_list ap;
DBC *dbc;
DBT key, data;
DB_TXN *tid;
int ret;
char *s;
/* Initialization. */
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
key.data = name;
key.size = strlen(name);
retry: /* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/* Delete any previously existing item. */
switch (ret = db->del(db, tid, &key, 0)) {
case 0:
case DB_NOTFOUND:
break;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
dbenv->err(dbenv, ret, "db->del: %s", name);
exit (1);
}
/* Create a cursor. */
if ((ret = db->cursor(db, tid, &dbc, 0)) != 0) {
dbenv->err(dbenv, ret, "db->cursor");
exit (1);
}
/* Append the items, in order. */
va_start(ap, name);
while ((s = va_arg(ap, char *)) != NULL) {
data.data = s;
data.size = strlen(s);
switch (ret = dbc->c_put(dbc, &key, &data, DB_KEYLAST)) {
case 0:
break;
case DB_LOCK_DEADLOCK:
va_end(ap);
/* Deadlock: retry the operation. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(
dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
/* Error: run recovery. */
dbenv->err(dbenv, ret, "dbc->put: %s/%s", name, s);
exit (1);
}
}
va_end(ap);
/* Success: commit the change. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
}
void
run_xact(DB_ENV *dbenv, DB *db, int offset, int count)
{
va_list ap;
DBC *dbc;
DBT key, data;
DB_TXN *tid;
int ret;
char *s;
/* Initialization. */
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
int keyPtr;
int valPtr;
key.data = &keyPtr;
key.size = sizeof(int);/*strlen(name);*/
data.data = &valPtr;
data.size = sizeof(int);
retry: /* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/* Delete any previously existing item. */
/* switch (ret = db->del(db, tid, &key, 0)) {
case 0:
case DB_NOTFOUND:
break;
case DB_LOCK_DEADLOCK:
/ * Deadlock: retry the operation. * /
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
dbenv->err(dbenv, ret, "db->del: %s", name);
exit (1);
} */
/* Create a cursor. */
if ((ret = db->cursor(db, tid, &dbc, 0)) != 0) {
dbenv->err(dbenv, ret, "db->cursor");
exit (1);
}
/* Append the items, in order. */
// va_start(ap, name);
// while ((s = va_arg(ap, char *)) != NULL) {
int q;
for(q = offset; q < offset + count; q++) {
keyPtr = q;
valPtr = q;
/* data.data = s;
data.size = strlen(s); */
// printf("A"); fflush(NULL);
switch (ret = dbc->c_put(dbc, &key, &data, DB_KEYLAST)) {
case 0:
// printf("B"); fflush(NULL);
break;
case DB_LOCK_DEADLOCK:
va_end(ap);
/* Deadlock: retry the operation. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(
dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
/* Error: run recovery. */
dbenv->err(dbenv, ret, "dbc->put: %d/%d", q, q);
exit (1);
}
}
va_end(ap);
/* Success: commit the change. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
}
void
usage()
{
(void)fprintf(stderr, "usage: txnapp\n");
exit(1);
}

595
benchmarks/berkeleyDB/bdbRaw.c Normal file
View file

@ -0,0 +1,595 @@
#include <sys/types.h>
#include <sys/stat.h>
#include <errno.h>
#include <pthread.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <db.h>
#define ENV_DIRECTORY "TXNAPP"
void add_cat(DB_ENV *, DB *, char *, ...);
void run_xact(DB_ENV *, DB *, int, int);
/*void add_color(DB_ENV *, DB *, char *, int);
void add_fruit(DB_ENV *, DB *, char *, char *); */
void *checkpoint_thread(void *);
void log_archlist(DB_ENV *);
void *logfile_thread(void *);
void db_open(DB_ENV *, DB **, char *, int);
void env_dir_create(void);
void env_open(DB_ENV **);
void usage(void);
int
main(int argc, char *argv[])
{
extern int optind;
DB *db_cats; /*, *db_color, *db_fruit; */
DB_ENV *dbenv;
pthread_t ptid;
int ch, ret;
/* while ((ch = getopt(argc, argv, "")) != EOF)
switch (ch) {
case '?':
default:
usage();
}
argc -= optind;
argv += optind; */
assert(argc == 3);
env_dir_create();
env_open(&dbenv);
/* Start a checkpoint thread. */
if ((ret = pthread_create(
&ptid, NULL, checkpoint_thread, (void *)dbenv)) != 0) {
fprintf(stderr,
"txnapp: failed spawning checkpoint thread: %s\n",
strerror(ret));
exit (1);
}
/* Start a logfile removal thread. */
if ((ret = pthread_create(
&ptid, NULL, logfile_thread, (void *)dbenv)) != 0) {
fprintf(stderr,
"txnapp: failed spawning log file removal thread: %s\n",
strerror(ret));
exit (1);
}
/* Open database: Key is fruit class; Data is specific type. */
/* db_open(dbenv, &db_fruit, "fruit", 0); */
/* Open database: Key is a color; Data is an integer. */
/*db_open(dbenv, &db_color, "color", 0); */
/*
* Open database:
* Key is a name; Data is: company name, address, cat breeds.
*/
db_open(dbenv, &db_cats, "cats", 1);
/* add_fruit(dbenv, db_fruit, "apple", "yellow delicious");
add_color(dbenv, db_color, "blue", 0);
add_color(dbenv, db_color, "blue", 3); */
/* add_cat(dbenv, db_cats,
"Amy Adams",
"Sleepycat Software",
"394 E. Riding Dr., Carlisle, MA 01741, USA",
"abyssinian",
"bengal",
"chartreaux",
NULL);*/
int r;
int num_xact = atoi(argv[1]); //100;
int insert_per_xact = atoi(argv[2]); //1000;
for(r = 0; r < num_xact; r ++) {
run_xact(dbenv, db_cats, 1+r*insert_per_xact, insert_per_xact);
}
return (0);
}
void
env_dir_create()
{
struct stat sb;
/*
* If the directory exists, we're done. We do not further check
* the type of the file, DB will fail appropriately if it's the
* wrong type.
*/
if (stat(ENV_DIRECTORY, &sb) == 0)
return;
/* Create the directory, read/write/access owner only. */
if (mkdir(ENV_DIRECTORY, S_IRWXU) != 0) {
fprintf(stderr,
"txnapp: mkdir: %s: %s\n", ENV_DIRECTORY, strerror(errno));
exit (1);
}
}
void
env_open(DB_ENV **dbenvp)
{
DB_ENV *dbenv;
int ret;
/* Create the environment handle. */
if ((ret = db_env_create(&dbenv, 0)) != 0) {
fprintf(stderr,
"txnapp: db_env_create: %s\n", db_strerror(ret));
exit (1);
}
/* Set up error handling. */
dbenv->set_errpfx(dbenv, "txnapp");
dbenv->set_errfile(dbenv, stderr);
/* Do deadlock detection internally. */
if ((ret = dbenv->set_lk_detect(dbenv, DB_LOCK_DEFAULT)) != 0) {
dbenv->err(dbenv, ret, "set_lk_detect: DB_LOCK_DEFAULT");
exit (1);
}
/*
* Open a transactional environment:
* create if it doesn't exist
* free-threaded handle
* run recovery
* read/write owner only
*/
if ((ret = dbenv->open(dbenv, ENV_DIRECTORY,
DB_CREATE | DB_INIT_LOCK | DB_INIT_LOG |
DB_INIT_MPOOL | DB_INIT_TXN | DB_RECOVER | DB_THREAD,
S_IRUSR | S_IWUSR)) != 0) {
dbenv->err(dbenv, ret, "dbenv->open: %s", ENV_DIRECTORY);
exit (1);
}
*dbenvp = dbenv;
}
void *
checkpoint_thread(void *arg)
{
DB_ENV *dbenv;
int ret;
dbenv = arg;
dbenv->errx(dbenv, "Checkpoint thread: %lu", (u_long)pthread_self());
/* Checkpoint once a minute. */
for (;; sleep(60))
if ((ret = dbenv->txn_checkpoint(dbenv, 0, 0, 0)) != 0) {
dbenv->err(dbenv, ret, "checkpoint thread");
exit (1);
}
/* NOTREACHED */
}
void *
logfile_thread(void *arg)
{
DB_ENV *dbenv;
int ret;
char **begin, **list;
dbenv = arg;
dbenv->errx(dbenv,
"Log file removal thread: %lu", (u_long)pthread_self());
/* Check once every 5 minutes. */
for (;; sleep(300)) {
/* Get the list of log files. */
if ((ret =
dbenv->log_archive(dbenv, &list, DB_ARCH_ABS)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->log_archive");
exit (1);
}
/* Remove the log files. */
if (list != NULL) {
for (begin = list; *list != NULL; ++list)
if ((ret = remove(*list)) != 0) {
dbenv->err(dbenv,
ret, "remove %s", *list);
exit (1);
}
free (begin);
}
}
/* NOTREACHED */
}
void
log_archlist(DB_ENV *dbenv)
{
int ret;
char **begin, **list;
/* Get the list of database files. */
if ((ret = dbenv->log_archive(dbenv,
&list, DB_ARCH_ABS | DB_ARCH_DATA)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->log_archive: DB_ARCH_DATA");
exit (1);
}
if (list != NULL) {
for (begin = list; *list != NULL; ++list)
printf("database file: %s\n", *list);
free (begin);
}
/* Get the list of log files. */
if ((ret = dbenv->log_archive(dbenv,
&list, DB_ARCH_ABS | DB_ARCH_LOG)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->log_archive: DB_ARCH_LOG");
exit (1);
}
if (list != NULL) {
for (begin = list; *list != NULL; ++list)
printf("log file: %s\n", *list);
free (begin);
}
}
void
db_open(DB_ENV *dbenv, DB **dbp, char *name, int dups)
{
DB *db;
int ret;
/* Create the database handle. */
if ((ret = db_create(&db, dbenv, 0)) != 0) {
dbenv->err(dbenv, ret, "db_create");
exit (1);
}
/* Optionally, turn on duplicate data items. */
/* if (dups && (ret = db->set_flags(db, DB_DUP)) != 0) {
dbenv->err(dbenv, ret, "db->set_flags: DB_DUP");
exit (1);
} */
/*
* Open a database in the environment:
* create if it doesn't exist
* free-threaded handle
* read/write owner only
*/
if ((ret = db->open(db, NULL, name, NULL, /*DB_BTREE*/ DB_RECNO /*DB_HASH*/,
DB_AUTO_COMMIT | DB_CREATE | DB_THREAD, S_IRUSR | S_IWUSR)) != 0) {
(void)db->close(db, 0);
dbenv->err(dbenv, ret, "db->open: %s", name);
exit (1);
}
*dbp = db;
}
void
add_fruit(DB_ENV *dbenv, DB *db, char *fruit, char *name)
{
DBT key, data;
DB_TXN *tid;
int ret;
/* Initialization. */
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
key.data = fruit;
key.size = strlen(fruit);
data.data = name;
data.size = strlen(name);
for (;;) {
/* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/* Store the value. */
switch (ret = db->put(db, tid, &key, &data, 0)) {
case 0:
/* Success: commit the change. */
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
return;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
break;
default:
/* Error: run recovery. */
dbenv->err(dbenv, ret, "dbc->put: %s/%s", fruit, name);
exit (1);
}
}
}
void
add_color(DB_ENV *dbenv, DB *dbp, char *color, int increment)
{
DBT key, data;
DB_TXN *tid;
int original, ret;
char buf[64];
/* Initialization. */
memset(&key, 0, sizeof(key));
key.data = color;
key.size = strlen(color);
memset(&data, 0, sizeof(data));
data.flags = DB_DBT_MALLOC;
for (;;) {
/* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/*
* Get the key. If it exists, we increment the value. If it
* doesn't exist, we create it.
*/
switch (ret = dbp->get(dbp, tid, &key, &data, 0)) {
case 0:
original = atoi(data.data);
break;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
continue;
case DB_NOTFOUND:
original = 0;
break;
default:
/* Error: run recovery. */
dbenv->err(
dbenv, ret, "dbc->get: %s/%d", color, increment);
exit (1);
}
if (data.data != NULL)
free(data.data);
/* Create the new data item. */
(void)snprintf(buf, sizeof(buf), "%d", original + increment);
data.data = buf;
data.size = strlen(buf) + 1;
/* Store the new value. */
switch (ret = dbp->put(dbp, tid, &key, &data, 0)) {
case 0:
/* Success: commit the change. */
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
return;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
break;
default:
/* Error: run recovery. */
dbenv->err(
dbenv, ret, "dbc->put: %s/%d", color, increment);
exit (1);
}
}
}
void
add_cat(DB_ENV *dbenv, DB *db, char *name, ...)
{
va_list ap;
DBC *dbc;
DBT key, data;
DB_TXN *tid;
int ret;
char *s;
/* Initialization. */
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
key.data = name;
key.size = strlen(name);
retry: /* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/* Delete any previously existing item. */
switch (ret = db->del(db, tid, &key, 0)) {
case 0:
case DB_NOTFOUND:
break;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
dbenv->err(dbenv, ret, "db->del: %s", name);
exit (1);
}
/* Create a cursor. */
if ((ret = db->cursor(db, tid, &dbc, 0)) != 0) {
dbenv->err(dbenv, ret, "db->cursor");
exit (1);
}
/* Append the items, in order. */
va_start(ap, name);
while ((s = va_arg(ap, char *)) != NULL) {
data.data = s;
data.size = strlen(s);
switch (ret = dbc->c_put(dbc, &key, &data, DB_KEYLAST)) {
case 0:
break;
case DB_LOCK_DEADLOCK:
va_end(ap);
/* Deadlock: retry the operation. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(
dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
/* Error: run recovery. */
dbenv->err(dbenv, ret, "dbc->put: %s/%s", name, s);
exit (1);
}
}
va_end(ap);
/* Success: commit the change. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
}
void
run_xact(DB_ENV *dbenv, DB *db, int offset, int count)
{
va_list ap;
DBC *dbc;
DBT key, data;
DB_TXN *tid;
int ret;
char *s;
/* Initialization. */
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
int keyPtr;
int valPtr;
key.data = &keyPtr;
key.size = sizeof(int);/*strlen(name);*/
data.data = &valPtr;
data.size = sizeof(int);
retry: /* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/* Delete any previously existing item. */
/* switch (ret = db->del(db, tid, &key, 0)) {
case 0:
case DB_NOTFOUND:
break;
case DB_LOCK_DEADLOCK:
/ * Deadlock: retry the operation. * /
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
dbenv->err(dbenv, ret, "db->del: %s", name);
exit (1);
} */
/* Create a cursor. */
if ((ret = db->cursor(db, tid, &dbc, 0)) != 0) {
dbenv->err(dbenv, ret, "db->cursor");
exit (1);
}
/* Append the items, in order. */
// va_start(ap, name);
// while ((s = va_arg(ap, char *)) != NULL) {
int q;
for(q = offset; q < offset + count; q++) {
keyPtr = q;
valPtr = q;
/* data.data = s;
data.size = strlen(s); */
// printf("A"); fflush(NULL);
switch (ret = dbc->c_put(dbc, &key, &data, DB_KEYLAST)) {
case 0:
// printf("B"); fflush(NULL);
break;
case DB_LOCK_DEADLOCK:
va_end(ap);
/* Deadlock: retry the operation. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(
dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
/* Error: run recovery. */
dbenv->err(dbenv, ret, "dbc->put: %d/%d", q, q);
exit (1);
}
}
va_end(ap);
/* Success: commit the change. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
}
void
usage()
{
(void)fprintf(stderr, "usage: txnapp\n");
exit(1);
}

595
benchmarks/berkeleyDB/transapp.c Normal file
View file

@ -0,0 +1,595 @@
#include <sys/types.h>
#include <sys/stat.h>
#include <errno.h>
#include <pthread.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <db.h>
#define ENV_DIRECTORY "TXNAPP"
void add_cat(DB_ENV *, DB *, char *, ...);
void run_xact(DB_ENV *, DB *, int, int);
/*void add_color(DB_ENV *, DB *, char *, int);
void add_fruit(DB_ENV *, DB *, char *, char *); */
void *checkpoint_thread(void *);
void log_archlist(DB_ENV *);
void *logfile_thread(void *);
void db_open(DB_ENV *, DB **, char *, int);
void env_dir_create(void);
void env_open(DB_ENV **);
void usage(void);
int
main(int argc, char *argv[])
{
extern int optind;
DB *db_cats; /*, *db_color, *db_fruit; */
DB_ENV *dbenv;
pthread_t ptid;
int ch, ret;
/* while ((ch = getopt(argc, argv, "")) != EOF)
switch (ch) {
case '?':
default:
usage();
}
argc -= optind;
argv += optind; */
assert(argc == 3);
env_dir_create();
env_open(&dbenv);
/* Start a checkpoint thread. */
if ((ret = pthread_create(
&ptid, NULL, checkpoint_thread, (void *)dbenv)) != 0) {
fprintf(stderr,
"txnapp: failed spawning checkpoint thread: %s\n",
strerror(ret));
exit (1);
}
/* Start a logfile removal thread. */
if ((ret = pthread_create(
&ptid, NULL, logfile_thread, (void *)dbenv)) != 0) {
fprintf(stderr,
"txnapp: failed spawning log file removal thread: %s\n",
strerror(ret));
exit (1);
}
/* Open database: Key is fruit class; Data is specific type. */
/* db_open(dbenv, &db_fruit, "fruit", 0); */
/* Open database: Key is a color; Data is an integer. */
/*db_open(dbenv, &db_color, "color", 0); */
/*
* Open database:
* Key is a name; Data is: company name, address, cat breeds.
*/
db_open(dbenv, &db_cats, "cats", 1);
/* add_fruit(dbenv, db_fruit, "apple", "yellow delicious");
add_color(dbenv, db_color, "blue", 0);
add_color(dbenv, db_color, "blue", 3); */
/* add_cat(dbenv, db_cats,
"Amy Adams",
"Sleepycat Software",
"394 E. Riding Dr., Carlisle, MA 01741, USA",
"abyssinian",
"bengal",
"chartreaux",
NULL);*/
int r;
int num_xact = atoi(argv[1]); //100;
int insert_per_xact = atoi(argv[2]); //1000;
for(r = 0; r < num_xact; r ++) {
run_xact(dbenv, db_cats, 1+r*insert_per_xact, insert_per_xact);
}
return (0);
}
void
env_dir_create()
{
struct stat sb;
/*
* If the directory exists, we're done. We do not further check
* the type of the file, DB will fail appropriately if it's the
* wrong type.
*/
if (stat(ENV_DIRECTORY, &sb) == 0)
return;
/* Create the directory, read/write/access owner only. */
if (mkdir(ENV_DIRECTORY, S_IRWXU) != 0) {
fprintf(stderr,
"txnapp: mkdir: %s: %s\n", ENV_DIRECTORY, strerror(errno));
exit (1);
}
}
void
env_open(DB_ENV **dbenvp)
{
DB_ENV *dbenv;
int ret;
/* Create the environment handle. */
if ((ret = db_env_create(&dbenv, 0)) != 0) {
fprintf(stderr,
"txnapp: db_env_create: %s\n", db_strerror(ret));
exit (1);
}
/* Set up error handling. */
dbenv->set_errpfx(dbenv, "txnapp");
dbenv->set_errfile(dbenv, stderr);
/* Do deadlock detection internally. */
if ((ret = dbenv->set_lk_detect(dbenv, DB_LOCK_DEFAULT)) != 0) {
dbenv->err(dbenv, ret, "set_lk_detect: DB_LOCK_DEFAULT");
exit (1);
}
/*
* Open a transactional environment:
* create if it doesn't exist
* free-threaded handle
* run recovery
* read/write owner only
*/
if ((ret = dbenv->open(dbenv, ENV_DIRECTORY,
DB_CREATE | DB_INIT_LOCK | DB_INIT_LOG |
DB_INIT_MPOOL | DB_INIT_TXN | DB_RECOVER | DB_THREAD,
S_IRUSR | S_IWUSR)) != 0) {
dbenv->err(dbenv, ret, "dbenv->open: %s", ENV_DIRECTORY);
exit (1);
}
*dbenvp = dbenv;
}
void *
checkpoint_thread(void *arg)
{
DB_ENV *dbenv;
int ret;
dbenv = arg;
dbenv->errx(dbenv, "Checkpoint thread: %lu", (u_long)pthread_self());
/* Checkpoint once a minute. */
for (;; sleep(60))
if ((ret = dbenv->txn_checkpoint(dbenv, 0, 0, 0)) != 0) {
dbenv->err(dbenv, ret, "checkpoint thread");
exit (1);
}
/* NOTREACHED */
}
void *
logfile_thread(void *arg)
{
DB_ENV *dbenv;
int ret;
char **begin, **list;
dbenv = arg;
dbenv->errx(dbenv,
"Log file removal thread: %lu", (u_long)pthread_self());
/* Check once every 5 minutes. */
for (;; sleep(300)) {
/* Get the list of log files. */
if ((ret =
dbenv->log_archive(dbenv, &list, DB_ARCH_ABS)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->log_archive");
exit (1);
}
/* Remove the log files. */
if (list != NULL) {
for (begin = list; *list != NULL; ++list)
if ((ret = remove(*list)) != 0) {
dbenv->err(dbenv,
ret, "remove %s", *list);
exit (1);
}
free (begin);
}
}
/* NOTREACHED */
}
void
log_archlist(DB_ENV *dbenv)
{
int ret;
char **begin, **list;
/* Get the list of database files. */
if ((ret = dbenv->log_archive(dbenv,
&list, DB_ARCH_ABS | DB_ARCH_DATA)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->log_archive: DB_ARCH_DATA");
exit (1);
}
if (list != NULL) {
for (begin = list; *list != NULL; ++list)
printf("database file: %s\n", *list);
free (begin);
}
/* Get the list of log files. */
if ((ret = dbenv->log_archive(dbenv,
&list, DB_ARCH_ABS | DB_ARCH_LOG)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->log_archive: DB_ARCH_LOG");
exit (1);
}
if (list != NULL) {
for (begin = list; *list != NULL; ++list)
printf("log file: %s\n", *list);
free (begin);
}
}
void
db_open(DB_ENV *dbenv, DB **dbp, char *name, int dups)
{
DB *db;
int ret;
/* Create the database handle. */
if ((ret = db_create(&db, dbenv, 0)) != 0) {
dbenv->err(dbenv, ret, "db_create");
exit (1);
}
/* Optionally, turn on duplicate data items. */
/* if (dups && (ret = db->set_flags(db, DB_DUP)) != 0) {
dbenv->err(dbenv, ret, "db->set_flags: DB_DUP");
exit (1);
} */
/*
* Open a database in the environment:
* create if it doesn't exist
* free-threaded handle
* read/write owner only
*/
if ((ret = db->open(db, NULL, name, NULL, /*DB_BTREE*/ DB_RECNO /*DB_HASH*/,
DB_AUTO_COMMIT | DB_CREATE | DB_THREAD, S_IRUSR | S_IWUSR)) != 0) {
(void)db->close(db, 0);
dbenv->err(dbenv, ret, "db->open: %s", name);
exit (1);
}
*dbp = db;
}
void
add_fruit(DB_ENV *dbenv, DB *db, char *fruit, char *name)
{
DBT key, data;
DB_TXN *tid;
int ret;
/* Initialization. */
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
key.data = fruit;
key.size = strlen(fruit);
data.data = name;
data.size = strlen(name);
for (;;) {
/* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/* Store the value. */
switch (ret = db->put(db, tid, &key, &data, 0)) {
case 0:
/* Success: commit the change. */
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
return;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
break;
default:
/* Error: run recovery. */
dbenv->err(dbenv, ret, "dbc->put: %s/%s", fruit, name);
exit (1);
}
}
}
void
add_color(DB_ENV *dbenv, DB *dbp, char *color, int increment)
{
DBT key, data;
DB_TXN *tid;
int original, ret;
char buf[64];
/* Initialization. */
memset(&key, 0, sizeof(key));
key.data = color;
key.size = strlen(color);
memset(&data, 0, sizeof(data));
data.flags = DB_DBT_MALLOC;
for (;;) {
/* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/*
* Get the key. If it exists, we increment the value. If it
* doesn't exist, we create it.
*/
switch (ret = dbp->get(dbp, tid, &key, &data, 0)) {
case 0:
original = atoi(data.data);
break;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
continue;
case DB_NOTFOUND:
original = 0;
break;
default:
/* Error: run recovery. */
dbenv->err(
dbenv, ret, "dbc->get: %s/%d", color, increment);
exit (1);
}
if (data.data != NULL)
free(data.data);
/* Create the new data item. */
(void)snprintf(buf, sizeof(buf), "%d", original + increment);
data.data = buf;
data.size = strlen(buf) + 1;
/* Store the new value. */
switch (ret = dbp->put(dbp, tid, &key, &data, 0)) {
case 0:
/* Success: commit the change. */
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
return;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
break;
default:
/* Error: run recovery. */
dbenv->err(
dbenv, ret, "dbc->put: %s/%d", color, increment);
exit (1);
}
}
}
void
add_cat(DB_ENV *dbenv, DB *db, char *name, ...)
{
va_list ap;
DBC *dbc;
DBT key, data;
DB_TXN *tid;
int ret;
char *s;
/* Initialization. */
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
key.data = name;
key.size = strlen(name);
retry: /* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/* Delete any previously existing item. */
switch (ret = db->del(db, tid, &key, 0)) {
case 0:
case DB_NOTFOUND:
break;
case DB_LOCK_DEADLOCK:
/* Deadlock: retry the operation. */
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
dbenv->err(dbenv, ret, "db->del: %s", name);
exit (1);
}
/* Create a cursor. */
if ((ret = db->cursor(db, tid, &dbc, 0)) != 0) {
dbenv->err(dbenv, ret, "db->cursor");
exit (1);
}
/* Append the items, in order. */
va_start(ap, name);
while ((s = va_arg(ap, char *)) != NULL) {
data.data = s;
data.size = strlen(s);
switch (ret = dbc->c_put(dbc, &key, &data, DB_KEYLAST)) {
case 0:
break;
case DB_LOCK_DEADLOCK:
va_end(ap);
/* Deadlock: retry the operation. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(
dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
/* Error: run recovery. */
dbenv->err(dbenv, ret, "dbc->put: %s/%s", name, s);
exit (1);
}
}
va_end(ap);
/* Success: commit the change. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
}
void
run_xact(DB_ENV *dbenv, DB *db, int offset, int count)
{
va_list ap;
DBC *dbc;
DBT key, data;
DB_TXN *tid;
int ret;
char *s;
/* Initialization. */
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
int keyPtr;
int valPtr;
key.data = &keyPtr;
key.size = sizeof(int);/*strlen(name);*/
data.data = &valPtr;
data.size = sizeof(int);
retry: /* Begin the transaction. */
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_ENV->txn_begin");
exit (1);
}
/* Delete any previously existing item. */
/* switch (ret = db->del(db, tid, &key, 0)) {
case 0:
case DB_NOTFOUND:
break;
case DB_LOCK_DEADLOCK:
/ * Deadlock: retry the operation. * /
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
dbenv->err(dbenv, ret, "db->del: %s", name);
exit (1);
} */
/* Create a cursor. */
if ((ret = db->cursor(db, tid, &dbc, 0)) != 0) {
dbenv->err(dbenv, ret, "db->cursor");
exit (1);
}
/* Append the items, in order. */
// va_start(ap, name);
// while ((s = va_arg(ap, char *)) != NULL) {
int q;
for(q = offset; q < offset + count; q++) {
keyPtr = q;
valPtr = q;
/* data.data = s;
data.size = strlen(s); */
// printf("A"); fflush(NULL);
switch (ret = dbc->c_put(dbc, &key, &data, DB_KEYLAST)) {
case 0:
// printf("B"); fflush(NULL);
break;
case DB_LOCK_DEADLOCK:
va_end(ap);
/* Deadlock: retry the operation. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(
dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->abort(tid)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->abort");
exit (1);
}
goto retry;
default:
/* Error: run recovery. */
dbenv->err(dbenv, ret, "dbc->put: %d/%d", q, q);
exit (1);
}
}
va_end(ap);
/* Success: commit the change. */
if ((ret = dbc->c_close(dbc)) != 0) {
dbenv->err(dbenv, ret, "dbc->c_close");
exit (1);
}
if ((ret = tid->commit(tid, 0)) != 0) {
dbenv->err(dbenv, ret, "DB_TXN->commit");
exit (1);
}
}
void
usage()
{
(void)fprintf(stderr, "usage: txnapp\n");
exit(1);
}

35
benchmarks/generateScripts.pl Executable file
View file

@ -0,0 +1,35 @@
#!/usr/bin/perl -w
use strict;
## This perl script generates the input to timer.pl, which in turn
## generates the input to plotting.pl, which generates performance
## graphs. :)
open(LLADD_THREADED, ">LLADD_THREADED.script" );
open(LLADD_LOGICAL, ">LLADD_LOGICAL.script" );
open(LLADD_PHYSICAL, ">LLADD_PHYSICAL.script" );
open(LLADD_RAW_PHYSICAL, ">LLADD_RAW_PHYSICAL.script");
open(BDB_RAW_INSERT, ">BDB_RAW_INSERT.script" );
open(BDB_HASH_INSERT, ">BDB_HASH_INSERT.script" );
for(my $i = 1; $i <= 10; $i += .5) {
my $insert_count = $i * 100000;
my $threaded_insert_count = $insert_count / 200;
print LLADD_THREADED "./logicalMultThreaded 200 $threaded_insert_count\n";
print LLADD_LOGICAL "./logicalHash 1 $insert_count\n";
print LLADD_PHYSICAL "./naiveHash 1 $insert_count\n";
print LLADD_RAW_PHYSICAL "./arrayListSet 1 $insert_count\n";
print BDB_RAW_INSERT "./berkeleyDB/bdbRaw 1 $insert_count\n";
print BDB_HASH_INSERT "./berkeleyDB/bdbHash 1 $insert_count\n";
}
close(LLADD_THREADED);
close(LLADD_LOGICAL);
close(LLADD_PHYSICAL);
close(LLADD_RAW_PHYSICAL);
close(BDB_RAW_INSERT);
close(BDB_HASH_INSERT);

6
benchmarks/logicalHash.c
View file

@ -13,10 +13,10 @@ int main(int argc, char** argv) {
int count = atoi(argv[2]);
int k;
unlink("storefile.txt");
/* unlink("storefile.txt");
unlink("logfile.txt");
unlink("blob0_file.txt");
unlink("blob1_file.txt");
unlink("blob1_file.txt"); */
Tinit();
int xid = Tbegin();
@ -41,6 +41,6 @@ int main(int argc, char** argv) {
}
/* Tdeinit(); */
Tdeinit();
}

73
benchmarks/logicalMultThreaded.c Normal file
View file

@ -0,0 +1,73 @@
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <lladd/transactional.h>
#include <unistd.h>
#include <pthread.h>
//static pthread_mutex_t hash_mutex = PTHREAD_MUTEX_INITIALIZER;
static int count;
static recordid hash;
static void * go (void * arg_ptr) {
// pthread_mutex_lock(&hash_mutex);
int k = *(int*)arg_ptr;
int j;
int xid = Tbegin();
for(j = k * count; j < (k+1) *(count) ; j++) {
TlogicalHashInsert(xid, hash, &j, sizeof(int), &j, sizeof(int));
// printf("(%d)", k);
}
Tcommit(xid);
/*
for(j = k * count; j < (k+1) *(count) ; j++) {
int tmp = -100;
TlogicalHashLookup(xid, hash, &j, sizeof(int), &tmp, sizeof(int));
assert(j == tmp);
} */
// pthread_mutex_unlock(&hash_mutex);
return NULL;
}
int main(int argc, char** argv) {
assert(argc == 3);
int thread_count = atoi(argv[1]);
count = atoi(argv[2]);
unlink("storefile.txt");
unlink("logfile.txt");
unlink("blob0_file.txt");
unlink("blob1_file.txt");
pthread_t * workers = malloc(sizeof(pthread_t) * thread_count);
Tinit();
int xid = Tbegin();
hash = ThashAlloc(xid, sizeof(int), sizeof(int));
Tcommit(xid);
int k;
for(k = 0; k < thread_count; k++) {
int * k_copy = malloc(sizeof(int));
*k_copy = k ;
pthread_create(&workers[k], NULL, go, k_copy);
}
for(k = 0; k < thread_count; k++) {
pthread_join(workers[k],NULL);
}
/* Tdeinit() */
}

32
benchmarks/naiveHash.c
View file

@ -4,17 +4,17 @@
#include <lladd/transactional.h>
#include <unistd.h>
int main(int argc, char** argv) {
assert(argc == 2);
assert(argc == 3);
int count = atoi(argv[1]);
int xact_count = atoi(argv[1]);
int count = atoi(argv[2]);
unlink("storefile.txt");
/* unlink("storefile.txt");
unlink("logfile.txt");
unlink("blob0_file.txt");
unlink("blob1_file.txt");
unlink("blob1_file.txt");*/
Tinit();
@ -22,16 +22,20 @@ int main(int argc, char** argv) {
recordid hash = ThashAlloc(xid, sizeof(int), sizeof(int));
int i;
for(i = 0; i < count ; i++) {
ThashInsert(xid, hash, &i, sizeof(int), &i, sizeof(int));
}
Tcommit(xid);
/* Tdeinit(); */
int i = 0;
int k;
for(k = 0; k < xact_count; k++) {
xid = Tbegin();
for(; i < (count*(k+1)) ; i++) {
ThashInsert(xid, hash, &i, sizeof(int), &i, sizeof(int));
}
Tcommit(xid);
}
Tdeinit();
}

64
benchmarks/naiveMultiThreaded.c Normal file
View file

@ -0,0 +1,64 @@
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <lladd/transactional.h>
#include <unistd.h>
#include <pthread.h>
static pthread_mutex_t hash_mutex = PTHREAD_MUTEX_INITIALIZER;
static int count;
static recordid hash;
static void * go (void * arg_ptr) {
pthread_mutex_lock(&hash_mutex);
int k = *(int*)arg_ptr;
int j;
int xid = Tbegin();
for(j = k * count; j < (k+1) *(count) ; j++) {
ThashInsert(xid, hash, &j, sizeof(int), &j, sizeof(int));
// printf("(%d)", k);
}
Tcommit(xid);
pthread_mutex_unlock(&hash_mutex);
return NULL;
}
int main(int argc, char** argv) {
assert(argc == 3);
int thread_count = atoi(argv[1]);
count = atoi(argv[2]);
unlink("storefile.txt");
unlink("logfile.txt");
unlink("blob0_file.txt");
unlink("blob1_file.txt");
pthread_t * workers = malloc(sizeof(pthread_t) * thread_count);
Tinit();
int xid = Tbegin();
hash = ThashAlloc(xid, sizeof(int), sizeof(int));
Tcommit(xid);
int k;
for(k = 0; k < thread_count; k++) {
int * k_copy = malloc(sizeof(int));
*k_copy = k ;
pthread_create(&workers[k], NULL, go, k_copy);
}
for(k = 0; k < thread_count; k++) {
pthread_join(workers[k],NULL);
}
/* Tdeinit() */
}

39
benchmarks/rawSet.c Normal file
View file

@ -0,0 +1,39 @@
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <lladd/transactional.h>
#include <unistd.h>
int main(int argc, char** argv) {
assert(argc == 3);
int xact_count = atoi(argv[1]);
int count = atoi(argv[2]);
/* unlink("storefile.txt");
unlink("logfile.txt");
unlink("blob0_file.txt");
unlink("blob1_file.txt");*/
Tinit();
int xid = Tbegin();
Tcommit(xid);
int i = 0;
int k;
for(k = 0; k < xact_count; k++) {
xid = Tbegin();
for(; i < (count*(k+1)) ; i++) {
Tset(xid, Talloc(xid, sizeof(int)), &i);
}
Tcommit(xid);
}
Tdeinit();
}

1
configure.in
View file

@ -89,5 +89,6 @@ AC_CONFIG_FILES([Makefile
test/monotree/Makefile
utilities/Makefile
benchmarks/Makefile
benchmarks/berkeleyDB/Makefile
])
AC_OUTPUT

1
lladd/operations/linearHash.h
View file

@ -24,6 +24,7 @@ void ThashInstantUpdate(int xid, recordid hashRid, const void * key, int keySize
void TlogicalHashUpdate(int xid, recordid hashRid, void * key, int keySize, void * val, int valSize);
void TlogicalHashInsert(int xid, recordid hashRid, void * key, int keySize, void * val, int valSize);
int TlogicalHashDelete(int xid, recordid hashRid, void * key, int keySize, void * val, int valSize);
int TlogicalHashLookup(int xid, recordid hashRid, void * key, int keySize, void * buf, int valSize);
Operation getLinearInsert();
Operation getLinearDelete();
Operation getUndoLinearInsert();

38
src/lladd/logger/logWriter.c
View file

@ -39,9 +39,18 @@ authors grant the U.S. Government and others acting in its behalf
permission to use and distribute the software in accordance with the
terms specified in this license.
---*/
#define _XOPEN_SOURCE 600
#include <sys/types.h>
#include <sys/stat.h>
#include <stdio.h>
/** For O_DIRECT. It's unclear that this is the correct thing to #define, but it works under linux. */
#define __USE_GNU
#include <fcntl.h>
#include <unistd.h>
#define _XOPEN_SOURCE 600
#include <stdlib.h>
#include <config.h>
#include <lladd/common.h>
@ -51,8 +60,7 @@ terms specified in this license.
#include "latches.h"
#include "io.h"
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <lladd/bufferManager.h>
@ -130,14 +138,22 @@ pthread_mutex_t truncateLog_mutex;
static int sought = 1;
int openLogWriter() {
#define BUFSIZE 1024*16
#define BUFSIZE 1024*96
char * buffer ;/*= malloc(BUFSIZE);*/
assert(!posix_memalign((void*)&(buffer), PAGE_SIZE, BUFSIZE));
log = fopen(LOG_FILE, "a+");
int logFD = open (LOG_FILE, O_CREAT | O_RDWR | O_APPEND | O_SYNC, S_IRWXU | S_IRWXG | S_IRWXO);
if(logFD == -1) {
perror("Couldn't open log file (A)");
abort();
}
log = fdopen(logFD, "a+");
// log = fopen(LOG_FILE, "a+");
if (log==NULL) {
assert(0);
perror("Couldn't open log file");
abort();
/*there was an error opening this file */
return FILE_WRITE_OPEN_ERROR;
}
@ -327,12 +343,13 @@ void syncLog() {
/* Wait to set the static variable until after the flush returns. */
fflush(log);
// Since we open the logfile with O_SYNC, fflush suffices.
#ifdef HAVE_FDATASYNC
/* Should be available in linux >= 2.4 */
fdatasync(fileno(log));
/* fdatasync(fileno(log)); */
#else
/* Slow - forces fs implementation to sync the file metadata to disk */
fsync(fileno(log));
/* fsync(fileno(log)); */
#endif
writelock(flushedLSN_lock, 0);
@ -371,7 +388,9 @@ void deleteLogWriter() {
static LogEntry * readLogEntry() {
LogEntry * ret = NULL;
long size, entrySize;
long size;
// assert(!posix_memalign((void*)&(size), 512, sizeof(long)));
long entrySize;
int nmemb;
if(feof(log)) {
@ -391,6 +410,7 @@ static LogEntry * readLogEntry() {
}
}
// assert(!posix_memalign(&ret, 512, (*size)));
ret = malloc(size);
nmemb = fread(ret, size, 1, log);

132
src/lladd/operations/linearHash.c
View file

@ -32,10 +32,12 @@ typedef struct {
void instant_expand (int xid, recordid hash, int next_split, int i, int keySize, int valSize);
pthread_mutex_t linearHashMutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t bucketUnlocked = PTHREAD_COND_INITIALIZER;
extern pblHashTable_t * openHashes ;
/*pblHashTable_t * openHashes = NULL; */
extern pblHashTable_t * lockedBuckets;
static int operateUndoInsert(int xid, Page * p, lsn_t lsn, recordid rid, const void * dat) {
int keySize = rid.size;
@ -124,9 +126,6 @@ void TlogicalHashInsert(int xid, recordid hashRid, void * key, int keySize, void
/* Write undo-only log entry. */
recordid * headerRidB = pblHtLookup(openHashes, &hashRid.page, sizeof(int));
assert(headerRidB);
hashRid.slot = valSize;
hashRid.size = keySize;
@ -134,8 +133,17 @@ void TlogicalHashInsert(int xid, recordid hashRid, void * key, int keySize, void
/* Perform redo-only insert. */
hashRid.size = sizeof(hashEntry) + keySize + valSize;
ThashInstantInsert(xid, hashRid, key, keySize, val, valSize);
pthread_mutex_lock(&linearHashMutex); /* @todo Finer grained locking for linear hash's expand? */
recordid * headerRidB = pblHtLookup(openHashes, &hashRid.page, sizeof(int));
assert(headerRidB);
instant_expand(xid, hashRid, headerNextSplit, headerHashBits, keySize, valSize);
pthread_mutex_unlock(&linearHashMutex);
}
int TlogicalHashDelete(int xid, recordid hashRid, void * key, int keySize, void * val, int valSize) {
@ -216,32 +224,79 @@ void instant_expand (int xid, recordid hash, int next_split, int i, int keySize,
}
}*/
/** you must hold linearHashMutex to call this function, which will release, and reaquire the mutex for you, as apprpriate. */
void instant_expand (int xid, recordid hash, int next_split, int i, int keySize, int valSize) {
/* Total hack; need to do this better, by storing stuff in the hash table headers.*/
static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t slow_mutex = PTHREAD_MUTEX_INITIALIZER;
static int count = 4096 * .25;
pthread_mutex_lock(&mutex);
count --;
int mycount = count;
pthread_mutex_unlock(&mutex);
#define AMORTIZE 1000
#define FF_AM 750
if(count <= 0 && !(count * -1) % FF_AM) {
recordid * headerRidB = pblHtLookup(openHashes, &(hash.page), sizeof(int));
if(mycount <= 0 && !(mycount * -1) % FF_AM) {
pthread_mutex_lock(&slow_mutex);
int j;
TarrayListInstantExtend(xid, hash, AMORTIZE);
// pthread_mutex_lock(&linearHashMutex);
recordid * headerRidB = pblHtLookup(openHashes, &(hash.page), sizeof(int));
for(j = 0; j < AMORTIZE; j++) {
if(next_split >= twoToThe(i-1)+2) {
i++;
next_split = 2;
}
while(pblHtLookup(lockedBuckets, &next_split, sizeof(int))) {
pthread_cond_wait(&bucketUnlocked, &linearHashMutex);
}
int other_bucket = next_split + twoToThe(i-1);
pblHtInsert(lockedBuckets, &next_split, sizeof(int), &other_bucket);
while(pblHtLookup(lockedBuckets, &other_bucket, sizeof(int))) {
pthread_cond_wait(&bucketUnlocked, &linearHashMutex);
}
pblHtInsert(lockedBuckets, &other_bucket, sizeof(int), &other_bucket);
pthread_mutex_unlock(&linearHashMutex);
instant_rehash(xid, hash, next_split, i, keySize, valSize);
pthread_mutex_lock(&linearHashMutex);
pblHtRemove(lockedBuckets, &next_split, sizeof(int));
pblHtRemove(lockedBuckets, &other_bucket, sizeof(int));
next_split++;
headerNextSplit = next_split;
headerHashBits = i;
}
instant_update_hash_header(xid, hash, i, next_split);
// pthread_mutex_unlock(&linearHashMutex);
pthread_mutex_unlock(&slow_mutex);
pthread_cond_broadcast(&bucketUnlocked);
}
}
void instant_update_hash_header(int xid, recordid hash, int i, int next_split) {
recordid * headerRidB = pblHtLookup(openHashes, &hash.page, sizeof(int));
@ -770,12 +825,26 @@ void ThashInstantInsert(int xid, recordid hashRid,
const void * key, int keySize,
const void * val, int valSize) {
pthread_mutex_lock(&linearHashMutex);
recordid * headerRidB = pblHtLookup(openHashes, &hashRid.page, sizeof(int));
assert(headerRidB);
int bucket = hash(key, keySize, headerHashBits, headerNextSplit - 2) + 2;
while(pblHtLookup(lockedBuckets, &bucket, sizeof(int))) {
pthread_cond_wait(&bucketUnlocked, &linearHashMutex);
bucket = hash(key, keySize, headerHashBits, headerNextSplit - 2) + 2;
}
int foo;
pblHtInsert(lockedBuckets, &bucket, sizeof(int), &foo );
headerRidB = NULL;
pthread_mutex_unlock(&linearHashMutex);
hashEntry * e = calloc(1,sizeof(hashEntry) + keySize + valSize);
memcpy(e+1, key, keySize);
memcpy(((byte*)(e+1)) + keySize, val, valSize);
@ -789,6 +858,11 @@ void ThashInstantInsert(int xid, recordid hashRid,
hashRid.slot = 0;
instant_insertIntoBucket(xid, hashRid, bucket, bucket_contents, e, keySize, valSize, 0);
pthread_mutex_lock(&linearHashMutex);
pblHtRemove(lockedBuckets, &bucket, sizeof(int));
pthread_cond_broadcast(&bucketUnlocked);
pthread_mutex_unlock(&linearHashMutex);
free(e);
}
@ -796,17 +870,38 @@ void ThashInstantInsert(int xid, recordid hashRid,
so that expand can be selectively called. */
void ThashInstantDelete(int xid, recordid hashRid,
const void * key, int keySize, int valSize) {
pthread_mutex_lock(&linearHashMutex);
recordid * headerRidB = pblHtLookup(openHashes, &hashRid.page, sizeof(int));
recordid tmp = hashRid;
tmp.slot = 1;
int bucket_number = hash(key, keySize, headerHashBits, headerNextSplit - 2) + 2;
while(pblHtLookup(lockedBuckets, &bucket_number, sizeof(int))) {
pthread_cond_wait(&bucketUnlocked, &linearHashMutex);
bucket_number = hash(key, keySize, headerHashBits, headerNextSplit - 2) + 2;
}
int foo;
pblHtInsert(lockedBuckets, &bucket_number, sizeof(int), &foo );
headerRidB = NULL;
pthread_mutex_unlock(&linearHashMutex);
recordid deleteMe;
hashRid.slot = bucket_number;
hashEntry * bucket_contents = malloc(sizeof(hashEntry) + keySize + valSize);
TreadUnlocked(xid, hashRid, bucket_contents);
hashRid.slot = 0;
if(instant_deleteFromBucket(xid, hashRid, bucket_number, bucket_contents, key, keySize, valSize, &deleteMe)) {
pthread_mutex_lock(&linearHashMutex);
pblHtRemove(lockedBuckets, &bucket_number, sizeof(int));
pthread_cond_broadcast(&bucketUnlocked);
pthread_mutex_unlock(&linearHashMutex);
/* Tdealloc(xid, deleteMe); */
}
}
@ -821,6 +916,7 @@ void ThashInstantDelete(int xid, recordid hashRid,
return 0;
}*/
/** @todo TlogicalHashUpdate is not threadsafe, is very slow, and is otherwise in need of help */
void TlogicalHashUpdate(int xid, recordid hashRid, void * key, int keySize, void * val, int valSize) {
void * dummy = malloc(valSize);
TlogicalHashDelete(xid, hashRid, key, keySize, dummy, valSize);
@ -848,3 +944,27 @@ int ThashLookup(int xid, recordid hashRid, void * key, int keySize, void * buf,
}
*/
int TlogicalHashLookup(int xid, recordid hashRid, void * key, int keySize, void * buf, int valSize) {
pthread_mutex_lock(&linearHashMutex);
recordid * headerRidB = pblHtLookup(openHashes, &(hashRid.page), sizeof(int));
/* printf("lookup header: %d %d\n", headerHashBits, headerNextSplit); */
recordid tmp = hashRid;
tmp.slot = 1;
int bucket_number = hash(key, keySize, headerHashBits, headerNextSplit - 2) + 2;
while(pblHtLookup(lockedBuckets, &bucket_number, sizeof(int))) {
pthread_cond_wait(&bucketUnlocked, &linearHashMutex);
bucket_number = hash(key, keySize, headerHashBits, headerNextSplit - 2) + 2;
}
pblHtInsert(lockedBuckets, &bucket_number, sizeof(int), &tmp);
pthread_mutex_unlock(&linearHashMutex);
int ret = findInBucket(xid, hashRid, bucket_number, key, keySize, buf, valSize);
pthread_mutex_lock(&linearHashMutex);
pblHtRemove(lockedBuckets, &bucket_number, sizeof(int));
pthread_mutex_unlock(&linearHashMutex);
pthread_cond_broadcast(&bucketUnlocked);
return ret;
}

3
src/lladd/operations/naiveLinearHash.c
View file

@ -28,6 +28,7 @@ typedef struct {
} hashEntry;
pblHashTable_t * openHashes = NULL;
pblHashTable_t * lockedBuckets = NULL;
@ -489,10 +490,12 @@ recordid ThashAlloc(int xid, int keySize, int valSize) {
void ThashInit() {
openHashes = pblHtCreate();
lockedBuckets = pblHtCreate();
}
void ThashDeinit() {
pblHtDelete(openHashes);
pblHtDelete(lockedBuckets);
}
void ThashInsert(int xid, recordid hashRid,

8
src/timing/timer.pl
View file

@ -23,7 +23,7 @@ my @ttbl;
sub parse_t_distribution {
## Takes the t-distribution.tbl file, and parses it into a nice, fat array.
open (T, "/home/morph/bin/t-distribution.tbl") || die "No T dist! $!";
open (T, "/home/sears/bin/t-distribution.tbl") || die "No T dist! $!";
my $i =0;
while(my $line = <T>) {
my @tok = split /\s+/, $line;
@ -41,17 +41,15 @@ sub parse_t_distribution {
sub runit {
my $cmd = shift;
`rm storefile.txt logfile.txt blob0_file.txt blob1_file.txt; sync`;
`rm -rf storefile.txt logfile.txt blob0_file.txt blob1_file.txt TXNAPP; sync; sleep 1`;
my $start_sec = `getTimeOfDay`;
chomp($start_sec);
system($cmd);
my $end_sec = `getTimeOfDay`;
chomp($start_sec);
chomp($end_sec);
my $time = ($end_sec - $start_sec) / 1000.0;