Edit File - File Manager

Edit File: txnexample_c.html

<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
  <head>
    <meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
    <title>Transaction Example</title>
    <link rel="stylesheet" href="gettingStarted.css" type="text/css" />
    <meta name="generator" content="DocBook XSL Stylesheets V1.62.4" />
    <link rel="home" href="index.html" title="Getting Started with Berkeley DB Transaction Processing" />
    <link rel="up" href="wrapup.html" title="Chapter 6. Summary and Examples" />
    <link rel="previous" href="wrapup.html" title="Chapter 6. Summary and Examples" />
    <link rel="next" href="inmem_txnexample_c.html" title="In-Memory Transaction Example" />
  </head>
  <body>
    <div class="navheader">
      <table width="100%" summary="Navigation header">
        <tr>
          <th colspan="3" align="center">Transaction Example</th>
        </tr>
        <tr>
          <td width="20%" align="left"><a accesskey="p" href="wrapup.html">Prev</a> </td>
          <th width="60%" align="center">Chapter 6. Summary and Examples</th>
          <td width="20%" align="right"> <a accesskey="n" href="inmem_txnexample_c.html">Next</a></td>
        </tr>
      </table>
      <hr />
    </div>
    <div class="sect1" lang="en" xml:lang="en">
      <div class="titlepage">
        <div>
          <div>
            <h2 class="title" style="clear: both"><a id="txnexample_c"></a>Transaction Example</h2>
          </div>
        </div>
        <div></div>
      </div>
      <p>
        The following code provides a fully functional example of a
        multi-threaded transactional DB application. For improved
        portability across platforms, this examples uses pthreads to
        provide threading support.
    </p>
      <p>
        The example opens an environment and database and then creates 5
        threads, each of which writes 500 records to the database. The keys
        used for these writes are pre-determined strings, while the data is
        a random value. This means that the actual data is arbitrary and
        therefore uninteresting; we picked it only because it requires
        minimum code to implement and therefore will stay out of the way of
        the main points of this example.
    </p>
      <p>
        Each thread writes 10 records under a single transaction
        before committing and writing another 10 (this is repeated 50
        times). At the end of each transaction, but before committing, each
        thread calls a function that uses a cursor to read every record in
        the database. We do this in order to make some points about
        database reads in a transactional environment.
    </p>
      <p>
        Of course, each writer thread performs deadlock detection as
        described in this manual. In addition, normal recovery is performed
        when the environment is opened.
    </p>
      <p>
        We start with our normal <tt class="literal">include</tt> directives:
    </p>
      <pre class="programlisting">// File TxnGuide.cpp

// We assume an ANSI-compatible compiler
#include &lt;db_cxx.h&gt;
#include &lt;pthread.h&gt;
#include &lt;iostream&gt;

#ifdef _WIN32
extern int getopt(int, char * const *, const char *);
#else
#include &lt;unistd.h&gt;
#endif  </pre>
      <p>
    We also need a directive that we use to identify how many threads we
    want our program to create:
</p>
      <pre class="programlisting">// Run 5 writers threads at a time.
#define NUMWRITERS 5 </pre>
      <p>
    Next we declare a couple of global
    variables (used by our threads), and we provide our forward
    declarations for the functions used by this example.
</p>
      <pre class="programlisting">// Printing of pthread_t is implementation-specific, so we
// create our own thread IDs for reporting purposes.
int global_thread_num;
pthread_mutex_t thread_num_lock;

// Forward declarations
int countRecords(Db *, DbTxn *);
int openDb(Db **, const char *, const char *, DbEnv *, u_int32_t);
int usage(void);
void *writerThread(void *);  </pre>
      <p>
    We now implement our usage function, which identifies our only command line
    parameter:
</p>
      <pre class="programlisting">// Usage function
int
usage()
{
    std::cerr &lt;&lt; " [-h &lt;database_home_directory&gt;]" &lt;&lt; std::endl;
    return (EXIT_FAILURE);
}  </pre>
      <p>
    With that, we have finished up our program's housekeeping, and we can
    now move on to the main part of our program. As usual, we begin with
    <tt class="function">main()</tt>. First we declare all our variables, and
    then we initialize our DB handles.
</p>
      <pre class="programlisting">int
main(int argc, char *argv[])
{
    // Initialize our handles
    Db *dbp = NULL;
    DbEnv *envp = NULL;

pthread_t writerThreads[NUMWRITERS];
    int ch, i;
    u_int32_t envFlags;
    char *dbHomeDir;

// Application name
    const char *progName = "TxnGuide";

// Database file name
    const char *fileName = "mydb.db";  </pre>
      <p>
    Now we need to parse our command line. In this case, all we want is to
    know where our environment directory is. If the <tt class="literal">-h</tt>
    option is not provided when this example is run, the current working
    directory is used instead.
</p>
      <pre class="programlisting">    // Parse the command line arguments
#ifdef _WIN32
    dbHomeDir = ".\\";
#else
    dbHomeDir = "./";
#endif
    while ((ch = getopt(argc, argv, "h:")) != EOF)
        switch (ch) {
        case 'h':
            dbHomeDir = optarg;
            break;
        case '?':
        default:
            return (usage());
        }  </pre>
      <p>
    Next we create our database handle, and we define our environment open flags.
    There are a few things to notice here:
</p>
      <div class="itemizedlist">
        <ul type="disc">
          <li>
            <p>
            We specify <tt class="literal">DB_RECOVER</tt>, which means that normal
            recovery is run every time we start the application. This is
            highly desirable and recommended for most
            applications.
        </p>
          </li>
          <li>
            <p>
            We also specify <tt class="literal">DB_THREAD</tt>, which means our
            environment handle will be free-threaded. This is very
            important because we will be sharing the environment handle
            across threads.
        </p>
          </li>
        </ul>
      </div>
      <pre class="programlisting">    // Env open flags
    envFlags =
      DB_CREATE     |  // Create the environment if it does not exist
      DB_RECOVER    |  // Run normal recovery.
      DB_INIT_LOCK  |  // Initialize the locking subsystem
      DB_INIT_LOG   |  // Initialize the logging subsystem
      DB_INIT_TXN   |  // Initialize the transactional subsystem. This
                       // also turns on logging.
      DB_INIT_MPOOL |  // Initialize the memory pool (in-memory cache)
      DB_THREAD;       // Cause the environment to be free-threaded

try {
        // Create and open the environment 
        envp = new DbEnv(0);  </pre>
      <p>
    Now we configure how we want deadlock detection performed. In our case, we will cause DB to perform deadlock
    detection by walking its internal lock tables looking for a block every time a lock is requested.  Further, in the
    event of a deadlock, the thread that holds the youngest lock will receive the deadlock notification.
 </p>
      <pre class="programlisting">        // Indicate that we want db to internally perform deadlock 
        // detection.  Also indicate that the transaction with 
        // the fewest number of write locks will receive the 
        // deadlock notification in the event of a deadlock.
        envp-&gt;set_lk_detect(DB_LOCK_MINWRITE);  </pre>
      <p>
    Now we open our environment.
</p>
      <pre class="programlisting">        // If we had utility threads (for running checkpoints or 
        // deadlock detection, for example) we would spawn those
        // here. However, for a simple example such as this,
        // that is not required.

envp-&gt;open(dbHomeDir, envFlags, 0); </pre>
      <p>
    Now we call the function that will open our database for us. This is
    not very interesting, except that you will notice that we are
    specifying <tt class="literal">DB_DUPSORT</tt>. This is required purely by
    the data that we are writing to the database, and it is only necessary 
    if you run the application more than once without first deleting the environment. 
</p>
      <p>
The implementation of <tt class="function">open_db()</tt> is described
    later in this section.
</p>
      <pre class="programlisting">        // Open the database
        openDb(&amp;dbp, progName, fileName, envp, DB_DUPSORT);  </pre>
      <p>
        Now we create our threads. In this example we are using pthreads
        for our threading package. A description of threading (beyond how
        it impacts DB usage) is beyond the scope of this manual. 
        However, the things that we are doing here should be familiar to
        anyone who has prior experience with any threading package. We are
        simply initializing a mutex, creating our threads, and then joining
        our threads, which causes our program to wait until the joined
        threads have completed before continuing operations in the main
        thread.
    </p>
      <pre class="programlisting">        // Initialize a pthread mutex. Used to help provide thread ids.
        (void)pthread_mutex_init(&amp;thread_num_lock, NULL);

// Start the writer threads.
        for (i = 0; i &lt; NUMWRITERS; i++)
            (void)pthread_create(&amp;writerThreads[i], NULL,
                writerThread, (void *)dbp);

// Join the writers
        for (i = 0; i &lt; NUMWRITERS; i++)
            (void)pthread_join(writerThreads[i], NULL);

} catch(DbException &amp;e) {
        std::cerr &lt;&lt; "Error opening database environment: "
                  &lt;&lt; dbHomeDir &lt;&lt; std::endl;
        std::cerr &lt;&lt; e.what() &lt;&lt; std::endl;
        return (EXIT_FAILURE);
    }  </pre>
      <p>
        Finally, to wrap up <tt class="function">main()</tt>, we close out our
        database and environment handle, as is normal for any DB
        application. Notice that this is where our <tt class="literal">err</tt>
        label is placed in our application. If any database operation prior
        to this point in the program returns an error status, the program
        simply jumps to this point and closes our handles if necessary
        before exiting the application completely.
    </p>
      <pre class="programlisting">    try {
        // Close our database handle if it was opened.
        if (dbp != NULL)
            dbp-&gt;close(0);

// Close our environment if it was opened.
        if (envp != NULL)
            envp-&gt;close(0);
    } catch(DbException &amp;e) {
        std::cerr &lt;&lt; "Error closing database and environment."
                  &lt;&lt; std::endl;
        std::cerr &lt;&lt; e.what() &lt;&lt; std::endl;
        return (EXIT_FAILURE);
    }

// Final status message and return.

std::cout &lt;&lt; "I'm all done." &lt;&lt; std::endl;
    return (EXIT_SUCCESS);
}  </pre>
      <p>
    Now that we have completed <tt class="function">main()</tt>, we need to
    implement the function that our writer threads will actually run. This
    is where the bulk of our transactional code resides.
</p>
      <p>
    We start as usual with variable declarations and initialization. 
</p>
      <pre class="programlisting">// A function that performs a series of writes to a
// Berkeley DB database. The information written
// to the database is largely nonsensical, but the
// mechanisms of transactional commit/abort and
// deadlock detection are illustrated here.
void *
writerThread(void *args)
{
    int j, thread_num;
    int max_retries = 20;   // Max retry on a deadlock
    char *key_strings[] = {"key 1", "key 2", "key 3", "key 4",
                           "key 5", "key 6", "key 7", "key 8",
                           "key 9", "key 10"};

Db *dbp = (Db *)args;
    DbEnv *envp = dbp-&gt;get_env();  </pre>
      <p>
    Now we want a thread number for reporting purposes. It is possible to
    use the <tt class="literal">pthread_t</tt> value directly for this purpose, 
    but how that is done unfortunately differs depending 
    on the pthread implementation you are using. So instead we use a
    mutex-protected global variable to obtain a simple integer for
    our reporting purposes.
</p>
      <p>
    Note that we are also use this thread id for initializing a random number generator, which we do here. 
    We use this random number generator for data generation.
</p>
      <pre class="programlisting">    // Get the thread number
    (void)pthread_mutex_lock(&amp;thread_num_lock);
    global_thread_num++;
    thread_num = global_thread_num;
    (void)pthread_mutex_unlock(&amp;thread_num_lock); 
    
    // Initialize the random number generator 
    srand((u_int)pthread_self());  </pre>
      <p>
        Now we begin the loop that we use to write data to the database.
        
        <span>
        Notice that in this top loop, we begin a new transaction. 
        </span>

We will actually use 50 transactions per writer
        thread, although we will only ever have one active transaction per
        thread at a time. Within each transaction, we will perform 10
        database writes.
    </p>
      <p>
        By combining multiple writes together under a single transaction,
        we increase the likelihood that a deadlock will occur. Normally,
        you want to reduce the potential for a deadlock and in this case
        the way to do that is to perform a single write per transaction. 
        To avoid deadlocks, we could be using auto commit to
        write to our database for this workload.
    </p>
      <p>
        However, we want to show deadlock handling and by performing
        multiple writes per transaction we can actually observe deadlocks
        occurring. We also want to underscore the idea that you can
        combing multiple database operations together in a single atomic
        unit of work in order to improve the efficiency of your writes. 
    </p>
      <pre class="programlisting">    // Perform 50 transactions
    for (int i=0; i&lt;50; i++) {
        DbTxn *txn;
        bool retry = true;
        int retry_count = 0;
        // while loop is used for deadlock retries
        while (retry) {
            // try block used for deadlock detection and
            // general db exception handling
            try {

// Begin our transaction. We group multiple writes in
                // this thread under a single transaction so as to
                // (1) show that you can atomically perform multiple 
                // writes at a time, and (2) to increase the chances 
                // of a deadlock occurring so that we can observe our 
                // deadlock detection at work.

// Normally we would want to avoid the potential for 
                // deadlocks, so for this workload the correct thing 
                // would be to perform our puts with auto commit. But 
                // that would excessively simplify our example, so we 
                // do the "wrong" thing here instead.
                txn = NULL;
                envp-&gt;txn_begin(NULL, &amp;txn, 0);  </pre>
      <p>
            Now we begin the inner loop that we use to actually
            perform the write.
        </p>
      <pre class="programlisting">                // Perform the database write for this transaction.
                for (j = 0; j &lt; 10; j++) {
                    Dbt key, value;
                    key.set_data(key_strings[j]);
                    key.set_size((strlen(key_strings[j]) + 1) *
                        sizeof(char));

int payload = rand() + i;
                    value.set_data(&amp;payload);
                    value.set_size(sizeof(int));

// Perform the database put
                    dbp-&gt;put(txn, &amp;key, &amp;value, 0);
                }  </pre>
      <p>
        Having completed the inner database write loop, we could simply
        commit the transaction and continue on to the next block of 10
        writes. However, we want to first illustrate a few points about
        transactional processing so instead we call our
         
        <tt class="function">countRecords()</tt>

function before calling the transaction
        commit. 
         
        <tt class="function">countRecords()</tt> 
        uses a cursor to read every
        record in the database and return a count of the number of records
        that it found. 
    </p>
      <pre class="programlisting">                // countRecords runs a cursor over the entire database.
                // We do this to illustrate issues of deadlocking
                std::cout &lt;&lt; thread_num &lt;&lt;  " : Found " 
                          &lt;&lt;  countRecords(dbp, NULL)    
                          &lt;&lt; " records in the database." &lt;&lt; std::endl;

std::cout &lt;&lt; thread_num &lt;&lt;  " : committing txn : " &lt;&lt; i 
                          &lt;&lt; std::endl;

// commit
                try {
                    txn-&gt;commit(0);
                    retry = false;
                    txn = NULL;
                } catch (DbException &amp;e) {
                    std::cout &lt;&lt; "Error on txn commit: "
                              &lt;&lt; e.what() &lt;&lt; std::endl;
                }  </pre>
      <p>
            Finally, we finish our try block. Notice how we examine the 
            exceptions to determine whether we need to 
            abort (or abort/retry in the case of a deadlock) our current
            transaction.
        </p>
      <pre class="programlisting">           } catch (DbDeadlockException &amp;de) {
                // First thing we MUST do is abort the transaction.
                if (txn != NULL)
                    (void)txn-&gt;abort();

// Now we decide if we want to retry the operation.
                // If we have retried less than max_retries,
                // increment the retry count and goto retry.
                if (retry_count &lt; max_retries) {
                    std::cout &lt;&lt; "############### Writer " &lt;&lt; thread_num
                              &lt;&lt; ": Got DB_LOCK_DEADLOCK.\n"
                              &lt;&lt; "Retrying write operation."
                              &lt;&lt; std::endl;
                    retry_count++;
                    retry = true;
                 } else {
                    // Otherwise, just give up.
                    std::cerr &lt;&lt; "Writer " &lt;&lt; thread_num
                              &lt;&lt; ": Got DeadLockException and out of "
                              &lt;&lt; "retries. Giving up." &lt;&lt; std::endl;
                    retry = false;
                 }
           } catch (DbException &amp;e) {
                std::cerr &lt;&lt; "db put failed" &lt;&lt; std::endl;
                std::cerr &lt;&lt; e.what() &lt;&lt; std::endl;
                if (txn != NULL)
                    txn-&gt;abort();
                retry = false;
           } catch (std::exception &amp;ee) {
            std::cerr &lt;&lt; "Unknown exception: " &lt;&lt; ee.what() &lt;&lt; std::endl;
            return (0);
          }
        }
    }
    return (0);
}  </pre>
      <p>
        <span>
        We want to back up for a moment and take a look at the call to <tt class="function">countRecords()</tt>.
        </span>
        If you look at the 
         
        <tt class="function">countRecords()</tt> 
        function prototype at the beginning of this example, you will see that the
        function's second parameter takes a transaction handle. However,
        our usage of the function here does not pass a transaction handle
        through to the function.
    </p>
      <p>

Because 
         
        <tt class="function">countRecords()</tt> 
        reads every record in the database, if used incorrectly the thread
        will self-deadlock.  The writer thread has just written 500 records
        to the database, but because the transaction used for that write
        has not yet been committed, each of those 500 records are still
        locked by the thread's transaction. If we then simply run a
        non-transactional cursor over the database from within the same
        thread that has locked those 500 records, the cursor will
        block when it tries to read one of those transactional
        protected records. The thread immediately stops operation at that
        point while the cursor waits for the read lock it has
        requested.  Because that read lock will never be released (the thread
        can never make any forward progress), this represents a
        self-deadlock for the the thread.
    </p>
      <p>
        There are three ways to prevent this self-deadlock:
    </p>
      <div class="orderedlist">
        <ol type="1">
          <li>
            <p>
                We can move the call to
                 
                <tt class="function">countRecords()</tt> 
                to a point after the thread's transaction has committed. 
            </p>
          </li>
          <li>
            <p>
                We can allow 
                     
                    <tt class="function">countRecords()</tt> 
                to operate under the same transaction as all of the writes
                were performed (this is what the transaction parameter for
                the function is for).
            </p>
          </li>
          <li>
            <p>
                We can reduce our isolation guarantee for the application
                by allowing uncommitted reads.
            </p>
          </li>
        </ol>
      </div>
      <p>
        For this example, we choose to use option 3 (uncommitted reads) to avoid
        the deadlock. This means that we have to open our database such
        that it supports uncommitted reads, and we have to open our cursor handle
        so that it knows to perform uncommitted reads.
    </p>
      <p>
        Note that in <a href="inmem_txnexample_c.html">In-Memory Transaction Example</a>, 
        we simply perform the cursor operation using the same transaction 
        as is used for the thread's writes. 
    </p>
      <p>
        The following is the 
            
            <tt class="function">countRecords()</tt>
        implementation. There is not anything particularly interesting
        about this function other than specifying uncommitted reads when 
        we open the cursor handle, but we include the function here anyway 
        for the sake of completeness.
    </p>
      <pre class="programlisting">// This simply counts the number of records contained in the
// database and returns the result.
//
// Note that this method exists only for illustrative purposes.
// A more straight-forward way to count the number of records in
// a database is to use the Database.getStats() method.
int
countRecords(Db *dbp, DbTxn *txn)
{

Dbc *cursorp = NULL;
    int count = 0;

try {
        // Get the cursor
        dbp-&gt;cursor(txn, &amp;cursorp, DB_READ_UNCOMMITTED);

Dbt key, value;
        while (cursorp-&gt;get(&amp;key, &amp;value, DB_NEXT) == 0) {
            count++;
        }
    } catch (DbDeadlockException &amp;de) {
        std::cerr &lt;&lt; "countRecords: got deadlock" &lt;&lt; std::endl;
        cursorp-&gt;close();
        throw de;
    } catch (DbException &amp;e) {
        std::cerr &lt;&lt; "countRecords error:" &lt;&lt; std::endl;
        std::cerr &lt;&lt; e.what() &lt;&lt; std::endl;
    }

if (cursorp != NULL) {
        try {
            cursorp-&gt;close();
        } catch (DbException &amp;e) {
            std::cerr &lt;&lt; "countRecords: cursor close failed:" &lt;&lt; std::endl;
            std::cerr &lt;&lt; e.what() &lt;&lt; std::endl;
        }
    }

return (count);
}  </pre>
      <p>
        Finally, we provide the implementation of our
         
        <tt class="function">openDb()</tt> 
        function. This function should hold
        no surprises for you. Note, however, that we do specify uncommitted reads
        when we open the database. If we did not do this, then our
         
        <tt class="function">countRecords()</tt> 
        function would cause our
        thread to self-deadlock because the cursor could not be opened to
        support uncommitted reads (that flag on the cursor open would, in fact, 
        be silently ignored by DB).
    </p>
      <pre class="programlisting">// Open a Berkeley DB database
int
openDb(Db **dbpp, const char *progname, const char *fileName,
  DbEnv *envp, u_int32_t extraFlags)
{
    int ret;
    u_int32_t openFlags;

try {
        Db *dbp = new Db(envp, 0);

// Point to the new'd Db
        *dbpp = dbp;

if (extraFlags != 0)
            ret = dbp-&gt;set_flags(extraFlags);

// Now open the database
        openFlags = DB_CREATE              | // Allow database creation
                    DB_READ_UNCOMMITTED    | // Allow uncommitted reads
                    DB_AUTO_COMMIT;          // Allow auto commit

dbp-&gt;open(NULL,       // Txn pointer
                  fileName,   // File name
                  NULL,       // Logical db name
                  DB_BTREE,   // Database type (using btree)
                  openFlags,  // Open flags
                  0);         // File mode. Using defaults
    } catch (DbException &amp;e) {
        std::cerr &lt;&lt; progname &lt;&lt; "open_db: db open failed:" &lt;&lt; std::endl;
        std::cerr &lt;&lt; e.what() &lt;&lt; std::endl;
        return (EXIT_FAILURE);
    }

return (EXIT_SUCCESS);
}  </pre>
      <p>
    This completes our transactional example. If you would like to
    experiment with this code, you can find the example in the following
    location in your DB distribution:
</p>
      <pre class="programlisting"><span class="emphasis"><em>DB_INSTALL</em></span>/examples_cxx/txn_guide</pre>
    </div>
    <div class="navfooter">
      <hr />
      <table width="100%" summary="Navigation footer">
        <tr>
          <td width="40%" align="left"><a accesskey="p" href="wrapup.html">Prev</a> </td>
          <td width="20%" align="center">
            <a accesskey="u" href="wrapup.html">Up</a>
          </td>
          <td width="40%" align="right"> <a accesskey="n" href="inmem_txnexample_c.html">Next</a></td>
        </tr>
        <tr>
          <td width="40%" align="left" valign="top">Chapter 6. Summary and Examples </td>
          <td width="20%" align="center">
            <a accesskey="h" href="index.html">Home</a>
          </td>
          <td width="40%" align="right" valign="top"> In-Memory Transaction Example</td>
        </tr>
      </table>
    </div>
  </body>
</html>

Back to File Manager