title: Berkeley DB, concurrent queues
body: <p>I'm trying to implement a concurrent persistent queue using Berkeley DB. As a starter I tried to make two process which both appends to the DB:</p>
  
  <pre><code>#include &lt;unistd.h&gt;
  #include &lt;sstream&gt;
  #include &lt;db_cxx.h&gt;
  
  class Queue : public DbEnv
  {
      public:
      Queue ( ) :
          DbEnv(0),
          db(0)
      {
          set_flags(DB_CDB_ALLDB, 1);
          open("/tmp/db", DB_INIT_LOCK  |
                  DB_INIT_LOG   |
                  DB_INIT_TXN   |
                  DB_INIT_MPOOL |
                  DB_RECOVER    |
                  DB_CREATE     |
                  DB_THREAD,
                  0);
  
          db = new Db(this, 0); 
          db-&gt;set_flags(DB_RENUMBER);
          db-&gt;open(NULL, "db", NULL, DB_RECNO, DB_CREATE | DB_AUTO_COMMIT | DB_THREAD, 0);
      }
      virtual ~Queue ()
      {
          db-&gt;close(0);
          delete db;
          close(0);
      }
  
      protected:
      Db * db;
  };
  
  class Enqueue : public Queue
  {
      public:
      Enqueue ( ) : Queue() { }
      virtual ~Enqueue () { }
  
      bool push(const std::string&amp; s)
      {
          int res;
          DbTxn * txn;
  
          try {
              txn_begin(NULL, &amp;txn, DB_TXN_SYNC | DB_TXN_WAIT );
  
              db_recno_t k0[4]; // not sure how mutch data is needs???
              k0[0] = 0;
  
              Dbt val((void*)s.c_str(), s.length());
              Dbt key((void*)&amp;k0, sizeof(k0[0]));
              key.set_ulen(sizeof(k0));
              key.set_flags(DB_DBT_USERMEM);
  
              res = db-&gt;put(txn, &amp;key, &amp;val, DB_APPEND);
  
              if( res == 0 ) {
                  txn-&gt;commit(0);
                  return true;
  
              } else {
                  std::cerr &lt;&lt; "push failed: " &lt;&lt; res &lt;&lt; std::endl;
                  txn-&gt;abort();
                  return false;
  
              }
          } catch( DbException e) {
              std::cerr &lt;&lt; "DB What()" &lt;&lt; e.what() &lt;&lt; std::endl;
              txn-&gt;abort();
              return false;
          } catch( std::exception e) {
              std::cerr &lt;&lt; "What()" &lt;&lt; e.what() &lt;&lt; std::endl;
              txn-&gt;abort();
              return false;
          } catch(...) {
              std::cerr &lt;&lt; "Unknown error" &lt;&lt; std::endl;
              txn-&gt;abort();
              return false;
          }
      }
  };
  
  using namespace std;
  
  int main(int argc, const char *argv[])
  {
      fork();
  
      Enqueue e;
  
      stringstream ss;
      for(int i = 0; i &lt; 10; i++){
          ss.str("");
          ss &lt;&lt; "asdf" &lt;&lt; i;
          cout &lt;&lt; ss.str() &lt;&lt; endl;
          if( ! e.push(ss.str()) )
              break;
      }
  
      return 0;
  }
  </code></pre>
  
  <p>Compiling it:</p>
  
  <pre><code>g++ test.cxx -I/usr/include/db4.8 -ldb_cxx-4.8
  </code></pre>
  
  <p>Create the db-dir</p>
  
  <pre><code>mkdir /tmp/db
  </code></pre>
  
  <p>And when I run it I get all kind of errors (segmentations fault, allocation error, and some times it actually works)</p>
  
  <p>I'm sure that I have missed some locking, but I just do not know how to do it. So, any hints and/or suggestions to fix this are most welcome.</p>