1/* File: txn_guide.c */ 2 3/* We assume an ANSI-compatible compiler */ 4#include <stdio.h> 5#include <stdlib.h> 6#include <string.h> 7#include <db.h> 8 9#ifdef _WIN32 10#include <windows.h> 11#define PATHD '\\' 12extern int getopt(int, char * const *, const char *); 13extern char *optarg; 14 15typedef HANDLE thread_t; 16#define thread_create(thrp, attr, func, arg) \ 17 (((*(thrp) = CreateThread(NULL, 0, \ 18 (LPTHREAD_START_ROUTINE)(func), (arg), 0, NULL)) == NULL) ? -1 : 0) 19#define thread_join(thr, statusp) \ 20 ((WaitForSingleObject((thr), INFINITE) == WAIT_OBJECT_0) && \ 21 GetExitCodeThread((thr), (LPDWORD)(statusp)) ? 0 : -1) 22 23typedef HANDLE mutex_t; 24#define mutex_init(m, attr) \ 25 (((*(m) = CreateMutex(NULL, FALSE, NULL)) != NULL) ? 0 : -1) 26#define mutex_lock(m) \ 27 ((WaitForSingleObject(*(m), INFINITE) == WAIT_OBJECT_0) ? 0 : -1) 28#define mutex_unlock(m) (ReleaseMutex(*(m)) ? 0 : -1) 29#else 30#include <pthread.h> 31#include <unistd.h> 32#define PATHD '/' 33 34typedef pthread_t thread_t; 35#define thread_create(thrp, attr, func, arg) \ 36 pthread_create((thrp), (attr), (func), (arg)) 37#define thread_join(thr, statusp) pthread_join((thr), (statusp)) 38 39typedef pthread_mutex_t mutex_t; 40#define mutex_init(m, attr) pthread_mutex_init((m), (attr)) 41#define mutex_lock(m) pthread_mutex_lock(m) 42#define mutex_unlock(m) pthread_mutex_unlock(m) 43#endif 44 45/* Run 5 writers threads at a time. */ 46#define NUMWRITERS 5 47 48/* 49 * Printing of a thread_t is implementation-specific, so we 50 * create our own thread IDs for reporting purposes. 51 */ 52int global_thread_num; 53mutex_t thread_num_lock; 54 55/* Forward declarations */ 56int count_records(DB *, DB_TXN *); 57int open_db(DB **, const char *, const char *, DB_ENV *, u_int32_t); 58int usage(void); 59void *writer_thread(void *); 60 61/* Usage function */ 62int 63usage() 64{ 65 fprintf(stderr, " [-h <database_home_directory>]\n"); 66 return (EXIT_FAILURE); 67} 68 69int 70main(int argc, char *argv[]) 71{ 72 /* Initialize our handles */ 73 DB *dbp = NULL; 74 DB_ENV *envp = NULL; 75 76 thread_t writer_threads[NUMWRITERS]; 77 int ch, i, ret, ret_t; 78 u_int32_t env_flags; 79 char *db_home_dir; 80 /* Application name */ 81 const char *prog_name = "txn_guide"; 82 /* Database file name */ 83 const char *file_name = "mydb.db"; 84 85 /* Parse the command line arguments */ 86#ifdef _WIN32 87 db_home_dir = ".\\"; 88#else 89 db_home_dir = "./"; 90#endif 91 while ((ch = getopt(argc, argv, "h:")) != EOF) 92 switch (ch) { 93 case 'h': 94 db_home_dir = optarg; 95 break; 96 case '?': 97 default: 98 return (usage()); 99 } 100 101 /* Create the environment */ 102 ret = db_env_create(&envp, 0); 103 if (ret != 0) { 104 fprintf(stderr, "Error creating environment handle: %s\n", 105 db_strerror(ret)); 106 goto err; 107 } 108 109 /* 110 * Indicate that we want db to perform lock detection internally. 111 * Also indicate that the transaction with the fewest number of 112 * write locks will receive the deadlock notification in 113 * the event of a deadlock. 114 */ 115 ret = envp->set_lk_detect(envp, DB_LOCK_MINWRITE); 116 if (ret != 0) { 117 fprintf(stderr, "Error setting lock detect: %s\n", 118 db_strerror(ret)); 119 goto err; 120 } 121 122 env_flags = 123 DB_CREATE | /* Create the environment if it does not exist */ 124 DB_RECOVER | /* Run normal recovery. */ 125 DB_INIT_LOCK | /* Initialize the locking subsystem */ 126 DB_INIT_LOG | /* Initialize the logging subsystem */ 127 DB_INIT_TXN | /* Initialize the transactional subsystem. This 128 * also turns on logging. */ 129 DB_INIT_MPOOL | /* Initialize the memory pool (in-memory cache) */ 130 DB_THREAD; /* Cause the environment to be free-threaded */ 131 132 /* Now actually open the environment */ 133 ret = envp->open(envp, db_home_dir, env_flags, 0); 134 if (ret != 0) { 135 fprintf(stderr, "Error opening environment: %s\n", 136 db_strerror(ret)); 137 goto err; 138 } 139 140 /* 141 * If we had utility threads (for running checkpoints or 142 * deadlock detection, for example) we would spawn those 143 * here. However, for a simple example such as this, 144 * that is not required. 145 */ 146 147 /* Open the database */ 148 ret = open_db(&dbp, prog_name, file_name, 149 envp, DB_DUPSORT); 150 if (ret != 0) 151 goto err; 152 153 /* Initialize a mutex. Used to help provide thread ids. */ 154 (void)mutex_init(&thread_num_lock, NULL); 155 156 /* Start the writer threads. */ 157 for (i = 0; i < NUMWRITERS; i++) 158 (void)thread_create( 159 &writer_threads[i], NULL, writer_thread, (void *)dbp); 160 161 /* Join the writers */ 162 for (i = 0; i < NUMWRITERS; i++) 163 (void)thread_join(writer_threads[i], NULL); 164 165err: 166 /* Close our database handle, if it was opened. */ 167 if (dbp != NULL) { 168 ret_t = dbp->close(dbp, 0); 169 if (ret_t != 0) { 170 fprintf(stderr, "%s database close failed: %s\n", 171 file_name, db_strerror(ret_t)); 172 ret = ret_t; 173 } 174 } 175 176 /* Close our environment, if it was opened. */ 177 if (envp != NULL) { 178 ret_t = envp->close(envp, 0); 179 if (ret_t != 0) { 180 fprintf(stderr, "environment close failed: %s\n", 181 db_strerror(ret_t)); 182 ret = ret_t; 183 } 184 } 185 186 /* Final status message and return. */ 187 printf("I'm all done.\n"); 188 return (ret == 0 ? EXIT_SUCCESS : EXIT_FAILURE); 189} 190 191/* 192 * A function that performs a series of writes to a 193 * Berkeley DB database. The information written 194 * to the database is largely nonsensical, but the 195 * mechanism of transactional commit/abort and 196 * deadlock detection is illustrated here. 197 */ 198void * 199writer_thread(void *args) 200{ 201 static char *key_strings[] = { 202 "key 1", "key 2", "key 3", "key 4", "key 5", 203 "key 6", "key 7", "key 8", "key 9", "key 10" 204 }; 205 DB *dbp; 206 DB_ENV *envp; 207 DBT key, value; 208 DB_TXN *txn; 209 int i, j, payload, ret, thread_num; 210 int retry_count, max_retries = 20; /* Max retry on a deadlock */ 211 212 dbp = (DB *)args; 213 envp = dbp->get_env(dbp); 214 215 /* Get the thread number */ 216 (void)mutex_lock(&thread_num_lock); 217 global_thread_num++; 218 thread_num = global_thread_num; 219 (void)mutex_unlock(&thread_num_lock); 220 221 /* Initialize the random number generator */ 222 srand(thread_num); 223 224 /* Write 50 times and then quit */ 225 for (i = 0; i < 50; i++) { 226 retry_count = 0; /* Used for deadlock retries */ 227 228 /* 229 * Some think it is bad form to loop with a goto statement, but 230 * we do it anyway because it is the simplest and clearest way 231 * to achieve our abort/retry operation. 232 */ 233retry: 234 /* Begin our transaction. We group multiple writes in 235 * this thread under a single transaction so as to 236 * (1) show that you can atomically perform multiple writes 237 * at a time, and (2) to increase the chances of a 238 * deadlock occurring so that we can observe our 239 * deadlock detection at work. 240 * 241 * Normally we would want to avoid the potential for deadlocks, 242 * so for this workload the correct thing would be to perform our 243 * puts with autocommit. But that would excessively simplify our 244 * example, so we do the "wrong" thing here instead. 245 */ 246 ret = envp->txn_begin(envp, NULL, &txn, 0); 247 if (ret != 0) { 248 envp->err(envp, ret, "txn_begin failed"); 249 return ((void *)EXIT_FAILURE); 250 } 251 for (j = 0; j < 10; j++) { 252 /* Set up our key and values DBTs */ 253 memset(&key, 0, sizeof(DBT)); 254 key.data = key_strings[j]; 255 key.size = (u_int32_t)strlen(key_strings[j]) + 1; 256 257 memset(&value, 0, sizeof(DBT)); 258 payload = rand() + i; 259 value.data = &payload; 260 value.size = sizeof(int); 261 262 /* Perform the database put. */ 263 switch (ret = dbp->put(dbp, txn, &key, &value, 0)) { 264 case 0: 265 break; 266 /* 267 * Our database is configured for sorted duplicates, 268 * so there is a potential for a KEYEXIST error return. 269 * If we get one, simply ignore it and continue on. 270 * 271 * Note that you will see KEYEXIST errors only after you 272 * have run this program at least once. 273 */ 274 case DB_KEYEXIST: 275 printf("Got keyexists.\n"); 276 break; 277 /* 278 * Here's where we perform deadlock detection. If 279 * DB_LOCK_DEADLOCK is returned by the put operation, 280 * then this thread has been chosen to break a deadlock. 281 * It must abort its operation, and optionally retry the 282 * put. 283 */ 284 case DB_LOCK_DEADLOCK: 285 /* 286 * First thing that we MUST do is abort the 287 * transaction. 288 */ 289 (void)txn->abort(txn); 290 /* 291 * Now we decide if we want to retry the operation. 292 * If we have retried less than max_retries, 293 * increment the retry count and goto retry. 294 */ 295 if (retry_count < max_retries) { 296 printf("Writer %i: Got DB_LOCK_DEADLOCK.\n", 297 thread_num); 298 printf("Writer %i: Retrying write operation.\n", 299 thread_num); 300 retry_count++; 301 goto retry; 302 } 303 /* 304 * Otherwise, just give up. 305 */ 306 printf("Writer %i: ", thread_num); 307 printf("Got DB_LOCK_DEADLOCK and out of retries.\n"); 308 printf("Writer %i: Giving up.\n", thread_num); 309 return ((void *)EXIT_FAILURE); 310 /* 311 * If a generic error occurs, we simply abort the 312 * transaction and exit the thread completely. 313 */ 314 default: 315 envp->err(envp, ret, "db put failed"); 316 ret = txn->abort(txn); 317 if (ret != 0) 318 envp->err(envp, ret, 319 "txn abort failed"); 320 return ((void *)EXIT_FAILURE); 321 } /** End case statement **/ 322 323 } /** End for loop **/ 324 325 /* 326 * print the number of records found in the database. 327 * See count_records() for usage information. 328 */ 329 printf("Thread %i. Record count: %i\n", thread_num, 330 count_records(dbp, NULL)); 331 332 /* 333 * If all goes well, we can commit the transaction and 334 * exit the thread. 335 */ 336 ret = txn->commit(txn, 0); 337 if (ret != 0) { 338 envp->err(envp, ret, "txn commit failed"); 339 return ((void *)EXIT_FAILURE); 340 } 341 } 342 return ((void *)EXIT_SUCCESS); 343} 344 345/* 346 * This simply counts the number of records contained in the 347 * database and returns the result. You can use this function 348 * in three ways: 349 * 350 * First call it with an active txn handle. 351 * Secondly, configure the cursor for uncommitted reads (this 352 * is what the example currently does). 353 * Third, call count_records AFTER the writer has committed 354 * its transaction. 355 * 356 * If you do none of these things, the writer thread will 357 * self-deadlock. 358 * 359 * Note that this function exists only for illustrative purposes. 360 * A more straight-forward way to count the number of records in 361 * a database is to use DB->stat() or DB->stat_print(). 362 */ 363 364int 365count_records(DB *dbp, DB_TXN *txn) 366{ 367 368 DBT key, value; 369 DBC *cursorp; 370 int count, ret; 371 372 cursorp = NULL; 373 count = 0; 374 375 /* Get the cursor */ 376 ret = dbp->cursor(dbp, txn, &cursorp, 377 DB_READ_UNCOMMITTED); 378 if (ret != 0) { 379 dbp->err(dbp, ret, 380 "count_records: cursor open failed."); 381 goto cursor_err; 382 } 383 384 /* Get the key DBT used for the database read */ 385 memset(&key, 0, sizeof(DBT)); 386 memset(&value, 0, sizeof(DBT)); 387 do { 388 ret = cursorp->get(cursorp, &key, &value, DB_NEXT); 389 switch (ret) { 390 case 0: 391 count++; 392 break; 393 case DB_NOTFOUND: 394 break; 395 default: 396 dbp->err(dbp, ret, 397 "Count records unspecified error"); 398 goto cursor_err; 399 } 400 } while (ret == 0); 401 402cursor_err: 403 if (cursorp != NULL) { 404 ret = cursorp->close(cursorp); 405 if (ret != 0) { 406 dbp->err(dbp, ret, 407 "count_records: cursor close failed."); 408 } 409 } 410 411 return (count); 412} 413 414/* Open a Berkeley DB database */ 415int 416open_db(DB **dbpp, const char *progname, const char *file_name, 417 DB_ENV *envp, u_int32_t extra_flags) 418{ 419 int ret; 420 u_int32_t open_flags; 421 DB *dbp; 422 423 /* Initialize the DB handle */ 424 ret = db_create(&dbp, envp, 0); 425 if (ret != 0) { 426 fprintf(stderr, "%s: %s\n", progname, 427 db_strerror(ret)); 428 return (EXIT_FAILURE); 429 } 430 431 /* Point to the memory malloc'd by db_create() */ 432 *dbpp = dbp; 433 434 if (extra_flags != 0) { 435 ret = dbp->set_flags(dbp, extra_flags); 436 if (ret != 0) { 437 dbp->err(dbp, ret, 438 "open_db: Attempt to set extra flags failed."); 439 return (EXIT_FAILURE); 440 } 441 } 442 443 /* Now open the database */ 444 open_flags = DB_CREATE | /* Allow database creation */ 445 DB_READ_UNCOMMITTED | /* Allow dirty reads */ 446 DB_AUTO_COMMIT; /* Allow autocommit */ 447 448 ret = dbp->open(dbp, /* Pointer to the database */ 449 NULL, /* Txn pointer */ 450 file_name, /* File name */ 451 NULL, /* Logical db name */ 452 DB_BTREE, /* Database type (using btree) */ 453 open_flags, /* Open flags */ 454 0); /* File mode. Using defaults */ 455 if (ret != 0) { 456 dbp->err(dbp, ret, "Database '%s' open failed", 457 file_name); 458 return (EXIT_FAILURE); 459 } 460 return (EXIT_SUCCESS); 461} 462