1<?xml version="1.0" encoding="UTF-8" standalone="no"?> 2<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd"> 3<html xmlns="http://www.w3.org/1999/xhtml"> 4 <head> 5 <meta http-equiv="Content-Type" content="text/html; charset=UTF-8" /> 6 <title>Transaction Example</title> 7 <link rel="stylesheet" href="gettingStarted.css" type="text/css" /> 8 <meta name="generator" content="DocBook XSL Stylesheets V1.62.4" /> 9 <link rel="home" href="index.html" title="Getting Started with Berkeley DB Transaction Processing" /> 10 <link rel="up" href="wrapup.html" title="Chapter 6. Summary and Examples" /> 11 <link rel="previous" href="wrapup.html" title="Chapter 6. Summary and Examples" /> 12 <link rel="next" href="inmem_txnexample_c.html" title="In-Memory Transaction Example" /> 13 </head> 14 <body> 15 <div class="navheader"> 16 <table width="100%" summary="Navigation header"> 17 <tr> 18 <th colspan="3" align="center">Transaction Example</th> 19 </tr> 20 <tr> 21 <td width="20%" align="left"><a accesskey="p" href="wrapup.html">Prev</a> </td> 22 <th width="60%" align="center">Chapter 6. Summary and Examples</th> 23 <td width="20%" align="right"> <a accesskey="n" href="inmem_txnexample_c.html">Next</a></td> 24 </tr> 25 </table> 26 <hr /> 27 </div> 28 <div class="sect1" lang="en" xml:lang="en"> 29 <div class="titlepage"> 30 <div> 31 <div> 32 <h2 class="title" style="clear: both"><a id="txnexample_c"></a>Transaction Example</h2> 33 </div> 34 </div> 35 <div></div> 36 </div> 37 <p> 38 The following code provides a fully functional example of a 39 multi-threaded transactional DB application. For improved 40 portability across platforms, this examples uses pthreads to 41 provide threading support. 42 </p> 43 <p> 44 The example opens an environment and database and then creates 5 45 threads, each of which writes 500 records to the database. The keys 46 used for these writes are pre-determined strings, while the data is 47 a random value. This means that the actual data is arbitrary and 48 therefore uninteresting; we picked it only because it requires 49 minimum code to implement and therefore will stay out of the way of 50 the main points of this example. 51 </p> 52 <p> 53 Each thread writes 10 records under a single transaction 54 before committing and writing another 10 (this is repeated 50 55 times). At the end of each transaction, but before committing, each 56 thread calls a function that uses a cursor to read every record in 57 the database. We do this in order to make some points about 58 database reads in a transactional environment. 59 </p> 60 <p> 61 Of course, each writer thread performs deadlock detection as 62 described in this manual. In addition, normal recovery is performed 63 when the environment is opened. 64 </p> 65 <p> 66 We start with our normal <tt class="literal">include</tt> directives: 67 </p> 68 <pre class="programlisting">// File TxnGuide.cpp 69 70// We assume an ANSI-compatible compiler 71#include <db_cxx.h> 72#include <pthread.h> 73#include <iostream> 74 75#ifdef _WIN32 76extern int getopt(int, char * const *, const char *); 77#else 78#include <unistd.h> 79#endif </pre> 80 <p> 81 We also need a directive that we use to identify how many threads we 82 want our program to create: 83</p> 84 <pre class="programlisting">// Run 5 writers threads at a time. 85#define NUMWRITERS 5 </pre> 86 <p> 87 Next we declare a couple of global 88 variables (used by our threads), and we provide our forward 89 declarations for the functions used by this example. 90</p> 91 <pre class="programlisting">// Printing of pthread_t is implementation-specific, so we 92// create our own thread IDs for reporting purposes. 93int global_thread_num; 94pthread_mutex_t thread_num_lock; 95 96// Forward declarations 97int countRecords(Db *, DbTxn *); 98int openDb(Db **, const char *, const char *, DbEnv *, u_int32_t); 99int usage(void); 100void *writerThread(void *); </pre> 101 <p> 102 We now implement our usage function, which identifies our only command line 103 parameter: 104</p> 105 <pre class="programlisting">// Usage function 106int 107usage() 108{ 109 std::cerr << " [-h <database_home_directory>]" << std::endl; 110 return (EXIT_FAILURE); 111} </pre> 112 <p> 113 With that, we have finished up our program's housekeeping, and we can 114 now move on to the main part of our program. As usual, we begin with 115 <tt class="function">main()</tt>. First we declare all our variables, and 116 then we initialize our DB handles. 117</p> 118 <pre class="programlisting">int 119main(int argc, char *argv[]) 120{ 121 // Initialize our handles 122 Db *dbp = NULL; 123 DbEnv *envp = NULL; 124 125 pthread_t writerThreads[NUMWRITERS]; 126 int ch, i; 127 u_int32_t envFlags; 128 char *dbHomeDir; 129 130 // Application name 131 const char *progName = "TxnGuide"; 132 133 // Database file name 134 const char *fileName = "mydb.db"; </pre> 135 <p> 136 Now we need to parse our command line. In this case, all we want is to 137 know where our environment directory is. If the <tt class="literal">-h</tt> 138 option is not provided when this example is run, the current working 139 directory is used instead. 140</p> 141 <pre class="programlisting"> // Parse the command line arguments 142#ifdef _WIN32 143 dbHomeDir = ".\\"; 144#else 145 dbHomeDir = "./"; 146#endif 147 while ((ch = getopt(argc, argv, "h:")) != EOF) 148 switch (ch) { 149 case 'h': 150 dbHomeDir = optarg; 151 break; 152 case '?': 153 default: 154 return (usage()); 155 } </pre> 156 <p> 157 Next we create our database handle, and we define our environment open flags. 158 There are a few things to notice here: 159</p> 160 <div class="itemizedlist"> 161 <ul type="disc"> 162 <li> 163 <p> 164 We specify <tt class="literal">DB_RECOVER</tt>, which means that normal 165 recovery is run every time we start the application. This is 166 highly desirable and recommended for most 167 applications. 168 </p> 169 </li> 170 <li> 171 <p> 172 We also specify <tt class="literal">DB_THREAD</tt>, which means our 173 environment handle will be free-threaded. This is very 174 important because we will be sharing the environment handle 175 across threads. 176 </p> 177 </li> 178 </ul> 179 </div> 180 <pre class="programlisting"> // Env open flags 181 envFlags = 182 DB_CREATE | // Create the environment if it does not exist 183 DB_RECOVER | // Run normal recovery. 184 DB_INIT_LOCK | // Initialize the locking subsystem 185 DB_INIT_LOG | // Initialize the logging subsystem 186 DB_INIT_TXN | // Initialize the transactional subsystem. This 187 // also turns on logging. 188 DB_INIT_MPOOL | // Initialize the memory pool (in-memory cache) 189 DB_THREAD; // Cause the environment to be free-threaded 190 191 try { 192 // Create and open the environment 193 envp = new DbEnv(0); </pre> 194 <p> 195 Now we configure how we want deadlock detection performed. In our case, we will cause DB to perform deadlock 196 detection by walking its internal lock tables looking for a block every time a lock is requested. Further, in the 197 event of a deadlock, the thread that holds the youngest lock will receive the deadlock notification. 198 </p> 199 <pre class="programlisting"> // Indicate that we want db to internally perform deadlock 200 // detection. Also indicate that the transaction with 201 // the fewest number of write locks will receive the 202 // deadlock notification in the event of a deadlock. 203 envp->set_lk_detect(DB_LOCK_MINWRITE); </pre> 204 <p> 205 Now we open our environment. 206</p> 207 <pre class="programlisting"> // If we had utility threads (for running checkpoints or 208 // deadlock detection, for example) we would spawn those 209 // here. However, for a simple example such as this, 210 // that is not required. 211 212 envp->open(dbHomeDir, envFlags, 0); </pre> 213 <p> 214 Now we call the function that will open our database for us. This is 215 not very interesting, except that you will notice that we are 216 specifying <tt class="literal">DB_DUPSORT</tt>. This is required purely by 217 the data that we are writing to the database, and it is only necessary 218 if you run the application more than once without first deleting the environment. 219</p> 220 <p> 221The implementation of <tt class="function">open_db()</tt> is described 222 later in this section. 223</p> 224 <pre class="programlisting"> // Open the database 225 openDb(&dbp, progName, fileName, envp, DB_DUPSORT); </pre> 226 <p> 227 Now we create our threads. In this example we are using pthreads 228 for our threading package. A description of threading (beyond how 229 it impacts DB usage) is beyond the scope of this manual. 230 However, the things that we are doing here should be familiar to 231 anyone who has prior experience with any threading package. We are 232 simply initializing a mutex, creating our threads, and then joining 233 our threads, which causes our program to wait until the joined 234 threads have completed before continuing operations in the main 235 thread. 236 </p> 237 <pre class="programlisting"> // Initialize a pthread mutex. Used to help provide thread ids. 238 (void)pthread_mutex_init(&thread_num_lock, NULL); 239 240 // Start the writer threads. 241 for (i = 0; i < NUMWRITERS; i++) 242 (void)pthread_create(&writerThreads[i], NULL, 243 writerThread, (void *)dbp); 244 245 // Join the writers 246 for (i = 0; i < NUMWRITERS; i++) 247 (void)pthread_join(writerThreads[i], NULL); 248 249 } catch(DbException &e) { 250 std::cerr << "Error opening database environment: " 251 << dbHomeDir << std::endl; 252 std::cerr << e.what() << std::endl; 253 return (EXIT_FAILURE); 254 } </pre> 255 <p> 256 Finally, to wrap up <tt class="function">main()</tt>, we close out our 257 database and environment handle, as is normal for any DB 258 application. Notice that this is where our <tt class="literal">err</tt> 259 label is placed in our application. If any database operation prior 260 to this point in the program returns an error status, the program 261 simply jumps to this point and closes our handles if necessary 262 before exiting the application completely. 263 </p> 264 <pre class="programlisting"> try { 265 // Close our database handle if it was opened. 266 if (dbp != NULL) 267 dbp->close(0); 268 269 // Close our environment if it was opened. 270 if (envp != NULL) 271 envp->close(0); 272 } catch(DbException &e) { 273 std::cerr << "Error closing database and environment." 274 << std::endl; 275 std::cerr << e.what() << std::endl; 276 return (EXIT_FAILURE); 277 } 278 279 // Final status message and return. 280 281 std::cout << "I'm all done." << std::endl; 282 return (EXIT_SUCCESS); 283} </pre> 284 <p> 285 Now that we have completed <tt class="function">main()</tt>, we need to 286 implement the function that our writer threads will actually run. This 287 is where the bulk of our transactional code resides. 288</p> 289 <p> 290 We start as usual with variable declarations and initialization. 291</p> 292 <pre class="programlisting">// A function that performs a series of writes to a 293// Berkeley DB database. The information written 294// to the database is largely nonsensical, but the 295// mechanisms of transactional commit/abort and 296// deadlock detection are illustrated here. 297void * 298writerThread(void *args) 299{ 300 int j, thread_num; 301 int max_retries = 20; // Max retry on a deadlock 302 char *key_strings[] = {"key 1", "key 2", "key 3", "key 4", 303 "key 5", "key 6", "key 7", "key 8", 304 "key 9", "key 10"}; 305 306 Db *dbp = (Db *)args; 307 DbEnv *envp = dbp->get_env(); </pre> 308 <p> 309 Now we want a thread number for reporting purposes. It is possible to 310 use the <tt class="literal">pthread_t</tt> value directly for this purpose, 311 but how that is done unfortunately differs depending 312 on the pthread implementation you are using. So instead we use a 313 mutex-protected global variable to obtain a simple integer for 314 our reporting purposes. 315</p> 316 <p> 317 Note that we are also use this thread id for initializing a random number generator, which we do here. 318 We use this random number generator for data generation. 319</p> 320 <pre class="programlisting"> // Get the thread number 321 (void)pthread_mutex_lock(&thread_num_lock); 322 global_thread_num++; 323 thread_num = global_thread_num; 324 (void)pthread_mutex_unlock(&thread_num_lock); 325 326 // Initialize the random number generator 327 srand((u_int)pthread_self()); </pre> 328 <p> 329 Now we begin the loop that we use to write data to the database. 330 331 <span> 332 Notice that in this top loop, we begin a new transaction. 333 </span> 334 335 We will actually use 50 transactions per writer 336 thread, although we will only ever have one active transaction per 337 thread at a time. Within each transaction, we will perform 10 338 database writes. 339 </p> 340 <p> 341 By combining multiple writes together under a single transaction, 342 we increase the likelihood that a deadlock will occur. Normally, 343 you want to reduce the potential for a deadlock and in this case 344 the way to do that is to perform a single write per transaction. 345 To avoid deadlocks, we could be using auto commit to 346 write to our database for this workload. 347 </p> 348 <p> 349 However, we want to show deadlock handling and by performing 350 multiple writes per transaction we can actually observe deadlocks 351 occurring. We also want to underscore the idea that you can 352 combing multiple database operations together in a single atomic 353 unit of work in order to improve the efficiency of your writes. 354 </p> 355 <pre class="programlisting"> // Perform 50 transactions 356 for (int i=0; i<50; i++) { 357 DbTxn *txn; 358 bool retry = true; 359 int retry_count = 0; 360 // while loop is used for deadlock retries 361 while (retry) { 362 // try block used for deadlock detection and 363 // general db exception handling 364 try { 365 366 // Begin our transaction. We group multiple writes in 367 // this thread under a single transaction so as to 368 // (1) show that you can atomically perform multiple 369 // writes at a time, and (2) to increase the chances 370 // of a deadlock occurring so that we can observe our 371 // deadlock detection at work. 372 373 // Normally we would want to avoid the potential for 374 // deadlocks, so for this workload the correct thing 375 // would be to perform our puts with auto commit. But 376 // that would excessively simplify our example, so we 377 // do the "wrong" thing here instead. 378 txn = NULL; 379 envp->txn_begin(NULL, &txn, 0); </pre> 380 <p> 381 Now we begin the inner loop that we use to actually 382 perform the write. 383 </p> 384 <pre class="programlisting"> // Perform the database write for this transaction. 385 for (j = 0; j < 10; j++) { 386 Dbt key, value; 387 key.set_data(key_strings[j]); 388 key.set_size((strlen(key_strings[j]) + 1) * 389 sizeof(char)); 390 391 int payload = rand() + i; 392 value.set_data(&payload); 393 value.set_size(sizeof(int)); 394 395 // Perform the database put 396 dbp->put(txn, &key, &value, 0); 397 } </pre> 398 <p> 399 Having completed the inner database write loop, we could simply 400 commit the transaction and continue on to the next block of 10 401 writes. However, we want to first illustrate a few points about 402 transactional processing so instead we call our 403 404 <tt class="function">countRecords()</tt> 405 406 function before calling the transaction 407 commit. 408 409 <tt class="function">countRecords()</tt> 410 uses a cursor to read every 411 record in the database and return a count of the number of records 412 that it found. 413 </p> 414 <pre class="programlisting"> // countRecords runs a cursor over the entire database. 415 // We do this to illustrate issues of deadlocking 416 std::cout << thread_num << " : Found " 417 << countRecords(dbp, NULL) 418 << " records in the database." << std::endl; 419 420 std::cout << thread_num << " : committing txn : " << i 421 << std::endl; 422 423 // commit 424 try { 425 txn->commit(0); 426 retry = false; 427 txn = NULL; 428 } catch (DbException &e) { 429 std::cout << "Error on txn commit: " 430 << e.what() << std::endl; 431 } </pre> 432 <p> 433 Finally, we finish our try block. Notice how we examine the 434 exceptions to determine whether we need to 435 abort (or abort/retry in the case of a deadlock) our current 436 transaction. 437 </p> 438 <pre class="programlisting"> } catch (DbDeadlockException &de) { 439 // First thing we MUST do is abort the transaction. 440 if (txn != NULL) 441 (void)txn->abort(); 442 443 // Now we decide if we want to retry the operation. 444 // If we have retried less than max_retries, 445 // increment the retry count and goto retry. 446 if (retry_count < max_retries) { 447 std::cout << "############### Writer " << thread_num 448 << ": Got DB_LOCK_DEADLOCK.\n" 449 << "Retrying write operation." 450 << std::endl; 451 retry_count++; 452 retry = true; 453 } else { 454 // Otherwise, just give up. 455 std::cerr << "Writer " << thread_num 456 << ": Got DeadLockException and out of " 457 << "retries. Giving up." << std::endl; 458 retry = false; 459 } 460 } catch (DbException &e) { 461 std::cerr << "db put failed" << std::endl; 462 std::cerr << e.what() << std::endl; 463 if (txn != NULL) 464 txn->abort(); 465 retry = false; 466 } catch (std::exception &ee) { 467 std::cerr << "Unknown exception: " << ee.what() << std::endl; 468 return (0); 469 } 470 } 471 } 472 return (0); 473} </pre> 474 <p> 475 <span> 476 We want to back up for a moment and take a look at the call to <tt class="function">countRecords()</tt>. 477 </span> 478 If you look at the 479 480 <tt class="function">countRecords()</tt> 481 function prototype at the beginning of this example, you will see that the 482 function's second parameter takes a transaction handle. However, 483 our usage of the function here does not pass a transaction handle 484 through to the function. 485 </p> 486 <p> 487 488 Because 489 490 <tt class="function">countRecords()</tt> 491 reads every record in the database, if used incorrectly the thread 492 will self-deadlock. The writer thread has just written 500 records 493 to the database, but because the transaction used for that write 494 has not yet been committed, each of those 500 records are still 495 locked by the thread's transaction. If we then simply run a 496 non-transactional cursor over the database from within the same 497 thread that has locked those 500 records, the cursor will 498 block when it tries to read one of those transactional 499 protected records. The thread immediately stops operation at that 500 point while the cursor waits for the read lock it has 501 requested. Because that read lock will never be released (the thread 502 can never make any forward progress), this represents a 503 self-deadlock for the the thread. 504 </p> 505 <p> 506 There are three ways to prevent this self-deadlock: 507 </p> 508 <div class="orderedlist"> 509 <ol type="1"> 510 <li> 511 <p> 512 We can move the call to 513 514 <tt class="function">countRecords()</tt> 515 to a point after the thread's transaction has committed. 516 </p> 517 </li> 518 <li> 519 <p> 520 We can allow 521 522 <tt class="function">countRecords()</tt> 523 to operate under the same transaction as all of the writes 524 were performed (this is what the transaction parameter for 525 the function is for). 526 </p> 527 </li> 528 <li> 529 <p> 530 We can reduce our isolation guarantee for the application 531 by allowing uncommitted reads. 532 </p> 533 </li> 534 </ol> 535 </div> 536 <p> 537 For this example, we choose to use option 3 (uncommitted reads) to avoid 538 the deadlock. This means that we have to open our database such 539 that it supports uncommitted reads, and we have to open our cursor handle 540 so that it knows to perform uncommitted reads. 541 </p> 542 <p> 543 Note that in <a href="inmem_txnexample_c.html">In-Memory Transaction Example</a>, 544 we simply perform the cursor operation using the same transaction 545 as is used for the thread's writes. 546 </p> 547 <p> 548 The following is the 549 550 <tt class="function">countRecords()</tt> 551 implementation. There is not anything particularly interesting 552 about this function other than specifying uncommitted reads when 553 we open the cursor handle, but we include the function here anyway 554 for the sake of completeness. 555 </p> 556 <pre class="programlisting">// This simply counts the number of records contained in the 557// database and returns the result. 558// 559// Note that this method exists only for illustrative purposes. 560// A more straight-forward way to count the number of records in 561// a database is to use the Database.getStats() method. 562int 563countRecords(Db *dbp, DbTxn *txn) 564{ 565 566 Dbc *cursorp = NULL; 567 int count = 0; 568 569 try { 570 // Get the cursor 571 dbp->cursor(txn, &cursorp, DB_READ_UNCOMMITTED); 572 573 Dbt key, value; 574 while (cursorp->get(&key, &value, DB_NEXT) == 0) { 575 count++; 576 } 577 } catch (DbDeadlockException &de) { 578 std::cerr << "countRecords: got deadlock" << std::endl; 579 cursorp->close(); 580 throw de; 581 } catch (DbException &e) { 582 std::cerr << "countRecords error:" << std::endl; 583 std::cerr << e.what() << std::endl; 584 } 585 586 if (cursorp != NULL) { 587 try { 588 cursorp->close(); 589 } catch (DbException &e) { 590 std::cerr << "countRecords: cursor close failed:" << std::endl; 591 std::cerr << e.what() << std::endl; 592 } 593 } 594 595 return (count); 596} </pre> 597 <p> 598 Finally, we provide the implementation of our 599 600 <tt class="function">openDb()</tt> 601 function. This function should hold 602 no surprises for you. Note, however, that we do specify uncommitted reads 603 when we open the database. If we did not do this, then our 604 605 <tt class="function">countRecords()</tt> 606 function would cause our 607 thread to self-deadlock because the cursor could not be opened to 608 support uncommitted reads (that flag on the cursor open would, in fact, 609 be silently ignored by DB). 610 </p> 611 <pre class="programlisting">// Open a Berkeley DB database 612int 613openDb(Db **dbpp, const char *progname, const char *fileName, 614 DbEnv *envp, u_int32_t extraFlags) 615{ 616 int ret; 617 u_int32_t openFlags; 618 619 try { 620 Db *dbp = new Db(envp, 0); 621 622 // Point to the new'd Db 623 *dbpp = dbp; 624 625 if (extraFlags != 0) 626 ret = dbp->set_flags(extraFlags); 627 628 // Now open the database 629 openFlags = DB_CREATE | // Allow database creation 630 DB_READ_UNCOMMITTED | // Allow uncommitted reads 631 DB_AUTO_COMMIT; // Allow auto commit 632 633 dbp->open(NULL, // Txn pointer 634 fileName, // File name 635 NULL, // Logical db name 636 DB_BTREE, // Database type (using btree) 637 openFlags, // Open flags 638 0); // File mode. Using defaults 639 } catch (DbException &e) { 640 std::cerr << progname << "open_db: db open failed:" << std::endl; 641 std::cerr << e.what() << std::endl; 642 return (EXIT_FAILURE); 643 } 644 645 return (EXIT_SUCCESS); 646} </pre> 647 <p> 648 This completes our transactional example. If you would like to 649 experiment with this code, you can find the example in the following 650 location in your DB distribution: 651</p> 652 <pre class="programlisting"><span class="emphasis"><em>DB_INSTALL</em></span>/examples_cxx/txn_guide</pre> 653 </div> 654 <div class="navfooter"> 655 <hr /> 656 <table width="100%" summary="Navigation footer"> 657 <tr> 658 <td width="40%" align="left"><a accesskey="p" href="wrapup.html">Prev</a> </td> 659 <td width="20%" align="center"> 660 <a accesskey="u" href="wrapup.html">Up</a> 661 </td> 662 <td width="40%" align="right"> <a accesskey="n" href="inmem_txnexample_c.html">Next</a></td> 663 </tr> 664 <tr> 665 <td width="40%" align="left" valign="top">Chapter 6. Summary and Examples </td> 666 <td width="20%" align="center"> 667 <a accesskey="h" href="index.html">Home</a> 668 </td> 669 <td width="40%" align="right" valign="top"> In-Memory Transaction Example</td> 670 </tr> 671 </table> 672 </div> 673 </body> 674</html> 675