开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 数据库系统 > MySQL数据库 > 查看源码
ex_thread.c
资源名称：mysql-3.23.35.tar.gz　[点击查看]
上传用户：tsgydb
上传日期：2007-04-14
资源大小：10674k
文件大小：12k
源码类别：

MySQL数据库

开发平台：

Visual C++

ex_thread.c：源码内容
  1. /*-
  2.  * See the file LICENSE for redistribution information.
  3.  *
  4.  * Copyright (c) 1997, 1998, 1999, 2000
  5.  * Sleepycat Software.  All rights reserved.
  6.  *
  7.  * $Id: ex_thread.c,v 11.9 2000/05/31 15:10:04 bostic Exp $
  8.  */
  10. #include "db_config.h"
  12. #ifndef NO_SYSTEM_INCLUDES
  13. #include <sys/types.h>
  15. #if TIME_WITH_SYS_TIME
  16. #include <sys/time.h>
  17. #include <time.h>
  18. #else
  19. #if HAVE_SYS_TIME_H
  20. #include <sys/time.h>
  21. #else
  22. #include <time.h>
  23. #endif
  24. #endif
  26. #include <errno.h>
  27. #include <pthread.h>
  28. #include <stdio.h>
  29. #include <stdlib.h>
  30. #include <string.h>
  31. #include <unistd.h>
  32. #endif
  34. #include <db.h>
  36. /*
  37.  * NB: This application is written using POSIX 1003.1b-1993 pthreads
  38.  * interfaces, which may not be portable to your system.
  39.  */
  40. extern int sched_yield __P((void)); /* Pthread yield function. */
  42. DB_ENV *db_init __P((char *));
  43. void   *deadlock __P((void *));
  44. void fatal __P((char *, int, int));
  45. int main __P((int, char *[]));
  46. int reader __P((int));
  47. void stats __P((void));
  48. void   *trickle __P((void *));
  49. void   *tstart __P((void *));
  50. void usage __P((void));
  51. void word __P((void));
  52. int writer __P((int));
  54. struct _statistics {
  55. int aborted; /* Write. */
  56. int aborts; /* Read/write. */
  57. int adds; /* Write. */
  58. int deletes; /* Write. */
  59. int txns; /* Write. */
  60. int found; /* Read. */
  61. int notfound; /* Read. */
  62. } *perf;
  64. const char
  65. *progname = "ex_thread"; /* Program name. */
  67. #define DATABASE "access.db" /* Database name. */
  68. #define WORDLIST "../test/wordlist" /* Dictionary. */
  70. /*
  71.  * We can seriously increase the number of collisions and transaction
  72.  * aborts by yielding the scheduler after every DB call.  Specify the
  73.  * -p option to do this.
  74.  */
  75. int punish; /* -p */
  76. int nlist; /* -n */
  77. int nreaders; /* -r */
  78. int verbose; /* -v */
  79. int nwriters; /* -w */
  81. DB     *dbp; /* Database handle. */
  82. DB_ENV *dbenv; /* Database environment. */
  83. int nthreads; /* Total threads. */
  84. char  **list; /* Word list. */
  86. /*
  87.  * ex_thread --
  88.  * Run a simple threaded application of some numbers of readers and
  89.  * writers competing for a set of words.
  90.  *
  91.  * Example UNIX shell script to run this program:
  92.  * % rm -rf TESTDIR
  93.  * % mkdir TESTDIR
  94.  * % ex_thread -h TESTDIR
  95.  */
  96. int
  97. main(argc, argv)
  98. int argc;
  99. char *argv[];
  100. {
  101. extern char *optarg;
  102. extern int errno, optind;
  103. pthread_t *tids;
  104. int ch, i, ret;
  105. char *home;
  106. void *retp;
  108. nlist = 1000;
  109. nreaders = nwriters = 4;
  110. home = "TESTDIR";
  111. while ((ch = getopt(argc, argv, "h:pn:r:vw:")) != EOF)
  112. switch (ch) {
  113. case 'h':
  114. home = optarg;
  115. break;
  116. case 'p':
  117. punish = 1;
  118. break;
  119. case 'n':
  120. nlist = atoi(optarg);
  121. break;
  122. case 'r':
  123. nreaders = atoi(optarg);
  124. break;
  125. case 'v':
  126. verbose = 1;
  127. break;
  128. case 'w':
  129. nwriters = atoi(optarg);
  130. break;
  131. case '?':
  132. default:
  133. usage();
  134. }
  135. argc -= optind;
  136. argv += optind;
  138. /* Initialize the random number generator. */
  139. srand(getpid() | time(NULL));
  141. /* Build the key list. */
  142. word();
  144. /* Remove the previous database. */
  145. (void)unlink(DATABASE);
  147. /* Initialize the database environment. */
  148. dbenv = db_init(home);
  150. /* Initialize the database. */
  151. if ((ret = db_create(&dbp, dbenv, 0)) != 0) {
  152. dbenv->err(dbenv, ret, "db_create");
  153. (void)dbenv->close(dbenv, 0);
  154. return (1);
  155. }
  156. if ((ret = dbp->set_pagesize(dbp, 1024)) != 0) {
  157. dbp->err(dbp, ret, "set_pagesize");
  158. goto err;
  159. }
  160. if ((ret = dbp->open(dbp,
  161.      DATABASE, NULL, DB_BTREE, DB_CREATE | DB_THREAD, 0664)) != 0) {
  162. dbp->err(dbp, ret, "%s: open", DATABASE);
  163. goto err;
  164. }
  166. nthreads = nreaders + nwriters + 2;
  167. printf("Running: readers %d, writers %dn", nreaders, nwriters);
  168. fflush(stdout);
  170. /* Create statistics structures, offset by 1. */
  171. if ((perf = calloc(nreaders + nwriters + 1, sizeof(*perf))) == NULL)
  172. fatal(NULL, errno, 1);
  174. /* Create thread ID structures. */
  175. if ((tids = malloc(nthreads * sizeof(pthread_t))) == NULL)
  176. fatal(NULL, errno, 1);
  178. /* Create reader/writer threads. */
  179. for (i = 0; i < nreaders + nwriters; ++i)
  180. if (pthread_create(&tids[i], NULL, tstart, (void *)i))
  181. fatal("pthread_create", errno, 1);
  183. /* Create buffer pool trickle thread. */
  184. if (pthread_create(&tids[i], NULL, trickle, &i))
  185. fatal("pthread_create", errno, 1);
  186. ++i;
  188. /* Create deadlock detector thread. */
  189. if (pthread_create(&tids[i], NULL, deadlock, &i))
  190. fatal("pthread_create", errno, 1);
  192. /* Wait for the threads. */
  193. for (i = 0; i < nthreads; ++i)
  194. (void)pthread_join(tids[i], &retp);
  196. err: (void)dbp->close(dbp, 0);
  197. (void)dbenv->close(dbenv, 0);
  199. return (0);
  200. }
  202. int
  203. reader(id)
  204. int id;
  205. {
  206. DBT key, data;
  207. int n, ret;
  208. char buf[64];
  210. /*
  211.  * DBT's must use local memory or malloc'd memory if the DB handle
  212.  * is accessed in a threaded fashion.
  213.  */
  214. memset(&key, 0, sizeof(DBT));
  215. memset(&data, 0, sizeof(DBT));
  216. data.flags = DB_DBT_MALLOC;
  218. /*
  219.  * Read-only threads do not require transaction protection, unless
  220.  * there's a need for repeatable reads.
  221.  */
  222. for (;;) {
  223. /* Pick a key at random, and look it up. */
  224. n = rand() % nlist;
  225. key.data = list[n];
  226. key.size = strlen(key.data);
  228. if (verbose) {
  229. sprintf(buf, "reader: %d: list entry %dn", id, n);
  230. write(STDOUT_FILENO, buf, strlen(buf));
  231. }
  233. switch (ret = dbp->get(dbp, NULL, &key, &data, 0)) {
  234. case DB_LOCK_DEADLOCK: /* Deadlock. */
  235. ++perf[id].aborts;
  236. break;
  237. case 0: /* Success. */
  238. ++perf[id].found;
  239. free(data.data);
  240. break;
  241. case DB_NOTFOUND: /* Not found. */
  242. ++perf[id].notfound;
  243. break;
  244. default:
  245. sprintf(buf,
  246.     "reader %d: dbp->get: %s", id, (char *)key.data);
  247. fatal(buf, ret, 0);
  248. }
  249. }
  250. return (0);
  251. }
  253. int
  254. writer(id)
  255. int id;
  256. {
  257. DBT key, data;
  258. DB_TXN *tid;
  259. time_t now, then;
  260. int n, ret;
  261. char buf[256], dbuf[10000];
  263. time(&now);
  264. then = now;
  266. /*
  267.  * DBT's must use local memory or malloc'd memory if the DB handle
  268.  * is accessed in a threaded fashion.
  269.  */
  270. memset(&key, 0, sizeof(DBT));
  271. memset(&data, 0, sizeof(DBT));
  272. data.data = dbuf;
  273. data.ulen = sizeof(dbuf);
  274. data.flags = DB_DBT_USERMEM;
  276. for (;;) {
  277. /* Pick a random key. */
  278. n = rand() % nlist;
  279. key.data = list[n];
  280. key.size = strlen(key.data);
  282. if (verbose) {
  283. sprintf(buf, "writer: %d: list entry %dn", id, n);
  284. write(STDOUT_FILENO, buf, strlen(buf));
  285. }
  287. /* Abort and retry. */
  288. if (0) {
  289. retry: if ((ret = txn_abort(tid)) != 0)
  290. fatal("txn_abort", ret, 1);
  291. ++perf[id].aborts;
  292. ++perf[id].aborted;
  293. }
  295. /* Thread #1 prints out the stats every 20 seconds. */
  296. if (id == 1) {
  297. time(&now);
  298. if (now - then >= 20) {
  299. stats();
  300. then = now;
  301. }
  302. }
  304. /* Begin the transaction. */
  305. if ((ret = txn_begin(dbenv, NULL, &tid, 0)) != 0)
  306. fatal("txn_begin", ret, 1);
  308. /*
  309.  * Get the key.  If it doesn't exist, add it.  If it does
  310.  * exist, delete it.
  311.  */
  312. switch (ret = dbp->get(dbp, tid, &key, &data, 0)) {
  313. case DB_LOCK_DEADLOCK:
  314. goto retry;
  315. case 0:
  316. goto delete;
  317. case DB_NOTFOUND:
  318. goto add;
  319. }
  321. sprintf(buf, "writer: %d: dbp->get", id);
  322. fatal(buf, ret, 1);
  323. /* NOTREACHED */
  325. delete: /* Delete the key. */
  326. switch (ret = dbp->del(dbp, tid, &key, 0)) {
  327. case DB_LOCK_DEADLOCK:
  328. goto retry;
  329. case 0:
  330. ++perf[id].deletes;
  331. goto commit;
  332. }
  334. sprintf(buf, "writer: %d: dbp->del", id);
  335. fatal(buf, ret, 1);
  336. /* NOTREACHED */
  338. add: /* Add the key.  1 data item in 30 is an overflow item. */
  339. data.size = 20 + rand() % 128;
  340. if (rand() % 30 == 0)
  341. data.size += 8192;
  343. switch (ret = dbp->put(dbp, tid, &key, &data, 0)) {
  344. case DB_LOCK_DEADLOCK:
  345. goto retry;
  346. case 0:
  347. ++perf[id].adds;
  348. goto commit;
  349. default:
  350. sprintf(buf, "writer: %d: dbp->put", id);
  351. fatal(buf, ret, 1);
  352. }
  354. commit: /* The transaction finished, commit it. */
  355. if ((ret = txn_commit(tid, 0)) != 0)
  356. fatal("txn_commit", ret, 1);
  358. /*
  359.  * Every time the thread completes 20 transactions, show
  360.  * our progress.
  361.  */
  362. if (++perf[id].txns % 20 == 0) {
  363. sprintf(buf,
  364. "writer: %2d: adds: %4d: deletes: %4d: aborts: %4d: txns: %4dn",
  365.     id, perf[id].adds, perf[id].deletes,
  366.     perf[id].aborts, perf[id].txns);
  367. write(STDOUT_FILENO, buf, strlen(buf));
  368. }
  370. /*
  371.  * If this thread was aborted more than 5 times before
  372.  * the transaction finished, complain.
  373.  */
  374. if (perf[id].aborted > 5) {
  375. sprintf(buf,
  376. "writer: %2d: adds: %4d: deletes: %4d: aborts: %4d: txns: %4d: ABORTED: %2dn",
  377.     id, perf[id].adds, perf[id].deletes,
  378.     perf[id].aborts, perf[id].txns, perf[id].aborted);
  379. write(STDOUT_FILENO, buf, strlen(buf));
  380. }
  381. perf[id].aborted = 0;
  382. }
  383. return (0);
  384. }
  386. /*
  387.  * stats --
  388.  * Display reader/writer thread statistics.  To display the statistics
  389.  * for the mpool trickle or deadlock threads, use db_stat(1).
  390.  */
  391. void
  392. stats()
  393. {
  394. int id;
  395. char *p, buf[8192];
  397. p = buf + sprintf(buf, "-------------n");
  398. for (id = 0; id < nreaders + nwriters;)
  399. if (id++ < nwriters)
  400. p += sprintf(p,
  401. "writer: %2d: adds: %4d: deletes: %4d: aborts: %4d: txns: %4dn",
  402.     id, perf[id].adds,
  403.     perf[id].deletes, perf[id].aborts, perf[id].txns);
  404. else
  405. p += sprintf(p,
  406. "reader: %2d: found: %5d: notfound: %5d: aborts: %4dn",
  407.     id, perf[id].found,
  408.     perf[id].notfound, perf[id].aborts);
  409. p += sprintf(p, "-------------n");
  411. write(STDOUT_FILENO, buf, p - buf);
  412. }
  414. /*
  415.  * db_init --
  416.  * Initialize the environment.
  417.  */
  418. DB_ENV *
  419. db_init(home)
  420. char *home;
  421. {
  422. DB_ENV *dbenv;
  423. int ret;
  425. if (punish) {
  426. (void)db_env_set_pageyield(1);
  427. (void)db_env_set_func_yield(sched_yield);
  428. }
  430. if ((ret = db_env_create(&dbenv, 0)) != 0) {
  431. fprintf(stderr,
  432.     "%s: db_env_create: %sn", progname, db_strerror(ret));
  433. exit (1);
  434. }
  435. dbenv->set_errfile(dbenv, stderr);
  436. dbenv->set_errpfx(dbenv, progname);
  437. (void)dbenv->set_cachesize(dbenv, 0, 100 * 1024, 0);
  438. (void)dbenv->set_lg_max(dbenv, 200000);
  440. if ((ret = dbenv->open(dbenv, home,
  441.     DB_CREATE | DB_INIT_LOCK | DB_INIT_LOG |
  442.     DB_INIT_MPOOL | DB_INIT_TXN | DB_THREAD, 0)) != 0) {
  443. dbenv->err(dbenv, ret, NULL);
  444. (void)dbenv->close(dbenv, 0);
  445. exit (1);
  446. }
  447. return (dbenv);
  448. }
  450. /*
  451.  * tstart --
  452.  * Thread start function for readers and writers.
  453.  */
  454. void *
  455. tstart(arg)
  456. void *arg;
  457. {
  458. pthread_t tid;
  459. u_int id;
  461. id = (u_int)arg + 1;
  463. tid = pthread_self();
  465. if (id <= (u_int)nwriters) {
  466. printf("write thread %d starting: tid: %lun", id, (u_long)tid);
  467. fflush(stdout);
  468. writer(id);
  469. } else {
  470. printf("read thread %d starting: tid: %lun", id, (u_long)tid);
  471. fflush(stdout);
  472. reader(id);
  473. }
  475. /* NOTREACHED */
  476. return (NULL);
  477. }
  479. /*
  480.  * deadlock --
  481.  * Thread start function for lock_detect().
  482.  */
  483. void *
  484. deadlock(arg)
  485. void *arg;
  486. {
  487. struct timeval t;
  488. pthread_t tid;
  490. arg = arg; /* XXX: shut the compiler up. */
  491. tid = pthread_self();
  493. printf("deadlock thread starting: tid: %lun", (u_long)tid);
  494. fflush(stdout);
  496. t.tv_sec = 0;
  497. t.tv_usec = 100000;
  498. for (;;) {
  499. (void)lock_detect(dbenv,
  500.     DB_LOCK_CONFLICT, DB_LOCK_YOUNGEST, NULL);
  502. /* Check every 100ms. */
  503. (void)select(0, NULL, NULL, NULL, &t);
  504. }
  506. /* NOTREACHED */
  507. return (NULL);
  508. }
  510. /*
  511.  * trickle --
  512.  * Thread start function for memp_trickle().
  513.  */
  514. void *
  515. trickle(arg)
  516. void *arg;
  517. {
  518. pthread_t tid;
  519. int wrote;
  520. char buf[64];
  522. arg = arg; /* XXX: shut the compiler up. */
  523. tid = pthread_self();
  525. printf("trickle thread starting: tid: %lun", (u_long)tid);
  526. fflush(stdout);
  528. for (;;) {
  529. (void)memp_trickle(dbenv, 10, &wrote);
  530. if (verbose) {
  531. sprintf(buf, "trickle: wrote %dn", wrote);
  532. write(STDOUT_FILENO, buf, strlen(buf));
  533. }
  534. if (wrote == 0) {
  535. sleep(1);
  536. sched_yield();
  537. }
  538. }
  540. /* NOTREACHED */
  541. return (NULL);
  542. }
  544. /*
  545.  * word --
  546.  * Build the dictionary word list.
  547.  */
  548. void
  549. word()
  550. {
  551. FILE *fp;
  552. int cnt;
  553. char buf[256];
  555. if ((fp = fopen(WORDLIST, "r")) == NULL)
  556. fatal(WORDLIST, errno, 1);
  558. if ((list = malloc(nlist * sizeof(char *))) == NULL)
  559. fatal(NULL, errno, 1);
  561. for (cnt = 0; cnt < nlist; ++cnt) {
  562. if (fgets(buf, sizeof(buf), fp) == NULL)
  563. break;
  564. if ((list[cnt] = strdup(buf)) == NULL)
  565. fatal(NULL, errno, 1);
  566. }
  567. nlist = cnt; /* In case nlist was larger than possible. */
  568. }
  570. /*
  571.  * fatal --
  572.  * Report a fatal error and quit.
  573.  */
  574. void
  575. fatal(msg, err, syserr)
  576. char *msg;
  577. int err, syserr;
  578. {
  579. fprintf(stderr, "%s: ", progname);
  580. if (msg != NULL) {
  581. fprintf(stderr, "%s", msg);
  582. if (syserr)
  583. fprintf(stderr, ": ");
  584. }
  585. if (syserr)
  586. fprintf(stderr, "%s", strerror(err));
  587. fprintf(stderr, "n");
  588. exit (1);
  590. /* NOTREACHED */
  591. }
  593. /*
  594.  * usage --
  595.  * Usage message.
  596.  */
  597. void
  598. usage()
  599. {
  600. (void)fprintf(stderr,
  601.     "usage: %s [-pv] [-h home] [-n words] [-r readers] [-w writers]n",
  602.     progname);
  603. exit(1);
  604. }