#include <sys/types.h>
#include <sys/time.h>
#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#ifdef _WIN32
extern int getopt(int, char * const *, const char *);
#else
#include <unistd.h>
#endif
#include <db.h>
extern int sched_yield __P((void));
int db_init __P((const char *));
void *deadlock __P((void *));
void fatal __P((const char *, int, int));
void onint __P((int));
int main __P((int, char *[]));
int reader __P((int));
void stats __P((void));
void *trickle __P((void *));
void *tstart __P((void *));
int usage __P((void));
void word __P((void));
int writer __P((int));
int quit;
struct _statistics {
int aborted;
int aborts;
int adds;
int deletes;
int txns;
int found;
int notfound;
} *perf;
const char
*progname = "ex_thread";
#define DATABASE "access.db"
#define WORDLIST "../test/wordlist"
int punish;
int nlist;
int nreaders;
int verbose;
int nwriters;
DB *dbp;
DB_ENV *dbenv;
int nthreads;
char **list;
int
main(argc, argv)
int argc;
char *argv[];
{
extern char *optarg;
extern int errno, optind;
DB_TXN *txnp;
pthread_t *tids;
int ch, i, ret;
const char *home;
void *retp;
txnp = NULL;
nlist = 1000;
nreaders = nwriters = 4;
home = "TESTDIR";
while ((ch = getopt(argc, argv, "h:pn:r:vw:")) != EOF)
switch (ch) {
case 'h':
home = optarg;
break;
case 'p':
punish = 1;
break;
case 'n':
nlist = atoi(optarg);
break;
case 'r':
nreaders = atoi(optarg);
break;
case 'v':
verbose = 1;
break;
case 'w':
nwriters = atoi(optarg);
break;
case '?':
default:
return (usage());
}
argc -= optind;
argv += optind;
srand(getpid() | time(NULL));
(void)signal(SIGINT, onint);
word();
(void)remove(DATABASE);
if ((ret = db_init(home)) != 0)
return (ret);
if ((ret = db_create(&dbp, dbenv, 0)) != 0) {
dbenv->err(dbenv, ret, "db_create");
(void)dbenv->close(dbenv, 0);
return (EXIT_FAILURE);
}
if ((ret = dbp->set_pagesize(dbp, 1024)) != 0) {
dbp->err(dbp, ret, "set_pagesize");
goto err;
}
if ((ret = dbenv->txn_begin(dbenv, NULL, &txnp, 0)) != 0)
fatal("txn_begin", ret, 1);
if ((ret = dbp->open(dbp, txnp,
DATABASE, NULL, DB_BTREE, DB_CREATE | DB_THREAD, 0664)) != 0) {
dbp->err(dbp, ret, "%s: open", DATABASE);
goto err;
} else {
ret = txnp->commit(txnp, 0);
txnp = NULL;
if (ret != 0)
goto err;
}
nthreads = nreaders + nwriters + 2;
printf("Running: readers %d, writers %d\n", nreaders, nwriters);
fflush(stdout);
if ((perf = calloc(nreaders + nwriters + 1, sizeof(*perf))) == NULL)
fatal(NULL, errno, 1);
if ((tids = malloc(nthreads * sizeof(pthread_t))) == NULL)
fatal(NULL, errno, 1);
for (i = 0; i < nreaders + nwriters; ++i)
if ((ret =
pthread_create(&tids[i], NULL, tstart, (void *)i)) != 0)
fatal("pthread_create", ret > 0 ? ret : errno, 1);
if (pthread_create(&tids[i], NULL, trickle, &i))
fatal("pthread_create", errno, 1);
++i;
if (pthread_create(&tids[i], NULL, deadlock, &i))
fatal("pthread_create", errno, 1);
for (i = 0; i < nthreads; ++i)
(void)pthread_join(tids[i], &retp);
printf("Exiting\n");
stats();
err: if (txnp != NULL)
(void)txnp->abort(txnp);
(void)dbp->close(dbp, 0);
(void)dbenv->close(dbenv, 0);
return (EXIT_SUCCESS);
}
int
reader(id)
int id;
{
DBT key, data;
int n, ret;
char buf[64];
memset(&key, 0, sizeof(DBT));
memset(&data, 0, sizeof(DBT));
data.flags = DB_DBT_MALLOC;
while (!quit) {
n = rand() % nlist;
key.data = list[n];
key.size = strlen(key.data);
if (verbose) {
sprintf(buf, "reader: %d: list entry %d\n", id, n);
write(STDOUT_FILENO, buf, strlen(buf));
}
switch (ret = dbp->get(dbp, NULL, &key, &data, 0)) {
case DB_LOCK_DEADLOCK:
++perf[id].aborts;
break;
case 0:
++perf[id].found;
free(data.data);
break;
case DB_NOTFOUND:
++perf[id].notfound;
break;
default:
sprintf(buf,
"reader %d: dbp->get: %s", id, (char *)key.data);
fatal(buf, ret, 0);
}
}
return (0);
}
int
writer(id)
int id;
{
DBT key, data;
DB_TXN *tid;
time_t now, then;
int n, ret;
char buf[256], dbuf[10000];
time(&now);
then = now;
memset(&key, 0, sizeof(DBT));
memset(&data, 0, sizeof(DBT));
data.data = dbuf;
data.ulen = sizeof(dbuf);
data.flags = DB_DBT_USERMEM;
while (!quit) {
n = rand() % nlist;
key.data = list[n];
key.size = strlen(key.data);
if (verbose) {
sprintf(buf, "writer: %d: list entry %d\n", id, n);
write(STDOUT_FILENO, buf, strlen(buf));
}
if (0) {
retry: if ((ret = tid->abort(tid)) != 0)
fatal("DB_TXN->abort", ret, 1);
++perf[id].aborts;
++perf[id].aborted;
}
if (id == 1) {
time(&now);
if (now - then >= 20) {
stats();
then = now;
}
}
if ((ret = dbenv->txn_begin(dbenv, NULL, &tid, 0)) != 0)
fatal("txn_begin", ret, 1);
switch (ret = dbp->get(dbp, tid, &key, &data, 0)) {
case DB_LOCK_DEADLOCK:
goto retry;
case 0:
goto delete;
case DB_NOTFOUND:
goto add;
}
sprintf(buf, "writer: %d: dbp->get", id);
fatal(buf, ret, 1);
delete:
switch (ret = dbp->del(dbp, tid, &key, 0)) {
case DB_LOCK_DEADLOCK:
goto retry;
case 0:
++perf[id].deletes;
goto commit;
}
sprintf(buf, "writer: %d: dbp->del", id);
fatal(buf, ret, 1);
add:
data.size = 20 + rand() % 128;
if (rand() % 30 == 0)
data.size += 8192;
switch (ret = dbp->put(dbp, tid, &key, &data, 0)) {
case DB_LOCK_DEADLOCK:
goto retry;
case 0:
++perf[id].adds;
goto commit;
default:
sprintf(buf, "writer: %d: dbp->put", id);
fatal(buf, ret, 1);
}
commit:
if ((ret = tid->commit(tid, 0)) != 0)
fatal("DB_TXN->commit", ret, 1);
if (++perf[id].txns % 20 == 0) {
sprintf(buf,
"writer: %2d: adds: %4d: deletes: %4d: aborts: %4d: txns: %4d\n",
id, perf[id].adds, perf[id].deletes,
perf[id].aborts, perf[id].txns);
write(STDOUT_FILENO, buf, strlen(buf));
}
if (perf[id].aborted > 5) {
sprintf(buf,
"writer: %2d: adds: %4d: deletes: %4d: aborts: %4d: txns: %4d: ABORTED: %2d\n",
id, perf[id].adds, perf[id].deletes,
perf[id].aborts, perf[id].txns, perf[id].aborted);
write(STDOUT_FILENO, buf, strlen(buf));
}
perf[id].aborted = 0;
}
return (0);
}
void
stats()
{
int id;
char *p, buf[8192];
p = buf + sprintf(buf, "-------------\n");
for (id = 0; id < nreaders + nwriters;)
if (id++ < nwriters)
p += sprintf(p,
"writer: %2d: adds: %4d: deletes: %4d: aborts: %4d: txns: %4d\n",
id, perf[id].adds,
perf[id].deletes, perf[id].aborts, perf[id].txns);
else
p += sprintf(p,
"reader: %2d: found: %5d: notfound: %5d: aborts: %4d\n",
id, perf[id].found,
perf[id].notfound, perf[id].aborts);
p += sprintf(p, "-------------\n");
write(STDOUT_FILENO, buf, p - buf);
}
int
db_init(home)
const char *home;
{
int ret;
if ((ret = db_env_create(&dbenv, 0)) != 0) {
fprintf(stderr,
"%s: db_env_create: %s\n", progname, db_strerror(ret));
return (EXIT_FAILURE);
}
if (punish) {
(void)dbenv->set_flags(dbenv, DB_YIELDCPU, 1);
(void)db_env_set_func_yield(sched_yield);
}
dbenv->set_errfile(dbenv, stderr);
dbenv->set_errpfx(dbenv, progname);
(void)dbenv->set_cachesize(dbenv, 0, 100 * 1024, 0);
(void)dbenv->set_lg_max(dbenv, 200000);
if ((ret = dbenv->open(dbenv, home,
DB_CREATE | DB_INIT_LOCK | DB_INIT_LOG |
DB_INIT_MPOOL | DB_INIT_TXN | DB_THREAD, 0)) != 0) {
dbenv->err(dbenv, ret, NULL);
(void)dbenv->close(dbenv, 0);
return (EXIT_FAILURE);
}
return (0);
}
void *
tstart(arg)
void *arg;
{
pthread_t tid;
u_int id;
id = (u_int)arg + 1;
tid = pthread_self();
if (id <= (u_int)nwriters) {
printf("write thread %d starting: tid: %lu\n", id, (u_long)tid);
fflush(stdout);
writer(id);
} else {
printf("read thread %d starting: tid: %lu\n", id, (u_long)tid);
fflush(stdout);
reader(id);
}
return (NULL);
}
void *
deadlock(arg)
void *arg;
{
struct timeval t;
pthread_t tid;
arg = arg;
tid = pthread_self();
printf("deadlock thread starting: tid: %lu\n", (u_long)tid);
fflush(stdout);
t.tv_sec = 0;
t.tv_usec = 100000;
while (!quit) {
(void)dbenv->lock_detect(dbenv, 0, DB_LOCK_YOUNGEST, NULL);
(void)select(0, NULL, NULL, NULL, &t);
}
return (NULL);
}
void *
trickle(arg)
void *arg;
{
pthread_t tid;
int wrote;
char buf[64];
arg = arg;
tid = pthread_self();
printf("trickle thread starting: tid: %lu\n", (u_long)tid);
fflush(stdout);
while (!quit) {
(void)dbenv->memp_trickle(dbenv, 10, &wrote);
if (verbose) {
sprintf(buf, "trickle: wrote %d\n", wrote);
write(STDOUT_FILENO, buf, strlen(buf));
}
if (wrote == 0) {
sleep(1);
sched_yield();
}
}
return (NULL);
}
void
word()
{
FILE *fp;
int cnt;
char buf[256];
if ((fp = fopen(WORDLIST, "r")) == NULL)
fatal(WORDLIST, errno, 1);
if ((list = malloc(nlist * sizeof(char *))) == NULL)
fatal(NULL, errno, 1);
for (cnt = 0; cnt < nlist; ++cnt) {
if (fgets(buf, sizeof(buf), fp) == NULL)
break;
if ((list[cnt] = strdup(buf)) == NULL)
fatal(NULL, errno, 1);
}
nlist = cnt;
}
void
fatal(msg, err, syserr)
const char *msg;
int err, syserr;
{
fprintf(stderr, "%s: ", progname);
if (msg != NULL) {
fprintf(stderr, "%s", msg);
if (syserr)
fprintf(stderr, ": ");
}
if (syserr)
fprintf(stderr, "%s", strerror(err));
fprintf(stderr, "\n");
exit(EXIT_FAILURE);
}
int
usage()
{
(void)fprintf(stderr,
"usage: %s [-pv] [-h home] [-n words] [-r readers] [-w writers]\n",
progname);
return (EXIT_FAILURE);
}
void
onint(signo)
int signo;
{
signo = 0;
quit = 1;
}