Berkeley DB, concurrent queues

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:

#include <unistd.h>
#include <sstream>
#include <db_cxx.h>

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->set_flags(DB_RENUMBER);
        db->open(NULL, "db", NULL, DB_RECNO, DB_CREATE | DB_AUTO_COMMIT | DB_THREAD, 0);
   开发者_如何学JAVA }
    virtual ~Queue ()
    {
        db->close(0);
        delete db;
        close(0);
    }

    protected:
    Db * db;
};

class Enqueue : public Queue
{
    public:
    Enqueue ( ) : Queue() { }
    virtual ~Enqueue () { }

    bool push(const std::string& s)
    {
        int res;
        DbTxn * txn;

        try {
            txn_begin(NULL, &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*)&k0, sizeof(k0[0]));
            key.set_ulen(sizeof(k0));
            key.set_flags(DB_DBT_USERMEM);

            res = db->put(txn, &key, &val, DB_APPEND);

            if( res == 0 ) {
                txn->commit(0);
                return true;

            } else {
                std::cerr << "push failed: " << res << std::endl;
                txn->abort();
                return false;

            }
        } catch( DbException e) {
            std::cerr << "DB What()" << e.what() << std::endl;
            txn->abort();
            return false;
        } catch( std::exception e) {
            std::cerr << "What()" << e.what() << std::endl;
            txn->abort();
            return false;
        } catch(...) {
            std::cerr << "Unknown error" << std::endl;
            txn->abort();
            return false;
        }
    }
};

using namespace std;

int main(int argc, const char *argv[])
{
    fork();

    Enqueue e;

    stringstream ss;
    for(int i = 0; i < 10; i++){
        ss.str("");
        ss << "asdf" << i;
        cout << ss.str() << endl;
        if( ! e.push(ss.str()) )
            break;
    }

    return 0;
}

Compiling it:

g++ test.cxx -I/usr/include/db4.8 -ldb_cxx-4.8

Create the db-dir

mkdir /tmp/db

And when I run it I get all kind of errors (segmentations fault, allocation error, and some times it actually works)

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.

Just for the record, here is the solution I have settled on after much googleing and trial'n'error.

The application is a call home process, where the producer is adding data, and consumer tries to send it home. If the consumer fails to send it home, it must try again. The database must not block for producer while the consumer is trying to sink data.

The code has a file lock, and will only allow one consumer process.

Here are the code:

#include <db_cxx.h>
#include <sstream>
#include <fstream>
#include <vector>

#include <boost/interprocess/sync/file_lock.hpp>

class Queue : public DbEnv
{
public:
    Queue ( bool sync ) :
        DbEnv(0),
        db(0)
    {
        set_flags(DB_CDB_ALLDB, 1);

        if( sync )
            set_flags(DB_TXN_NOSYNC, 0);
        else
            set_flags(DB_TXN_NOSYNC, 1);

        open("/tmp/db", DB_INIT_LOCK |
             DB_INIT_LOG | DB_INIT_TXN | DB_INIT_MPOOL |
             DB_REGISTER | DB_RECOVER | DB_CREATE | DB_THREAD,
             0);

        db = new Db(this, 0);
        db->set_flags(DB_RENUMBER);
        db->open(NULL, "db", NULL, DB_RECNO, DB_CREATE | DB_AUTO_COMMIT | DB_THREAD, 0);
    }
    virtual ~Queue ()
    {
        db->close(0);
        delete db;
        close(0);
    }

protected:
    Db * db;
};

struct Transaction
{
    Transaction() : t(0) { }

    bool init(DbEnv * dbenv ){
        try {
            dbenv->txn_begin(NULL, &t, 0);
        } catch( DbException e) {
            std::cerr << "DB What()" << e.what() << std::endl;
            return false;
        } catch( std::exception e) {
            std::cerr << "What()" << e.what() << std::endl;
            return false;
        } catch(...) {
            std::cerr << "Unknown error" << std::endl;
            return false;
        }
        return true;
    }

    ~Transaction(){ if( t!=0) t->abort(); }

    void abort() { t->abort(); t = 0; }
    void commit() { t->commit(0); t = 0; }

    DbTxn * t;
};

struct Cursor
{
    Cursor() : c(0) { }

    bool init( Db * db,  DbTxn * t) {
        try {
            db->cursor(t, &c, 0);
        } catch( DbException e) {
            std::cerr << "DB What()" << e.what() << std::endl;
            return false;
        } catch( std::exception e) {
            std::cerr << "What()" << e.what() << std::endl;
            return false;
        } catch(...) {
            std::cerr << "Unknown error" << std::endl;
            return false;
        }
        return true;
    }

    ~Cursor(){ if( c!=0) c->close(); }
    void close(){ c->close(); c = 0; }
    Dbc * c;
};

class Enqueue : public Queue
{
public:
    Enqueue ( bool sync ) : Queue(sync) { }
    virtual ~Enqueue () { }

    bool push(const std::string& s)
    {
        int res;
        Transaction transaction;

        if( ! transaction.init(this) )
            return false;

        try {
            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*)&k0, sizeof(k0[0]));
            key.set_ulen(sizeof(k0));
            key.set_flags(DB_DBT_USERMEM);

            res = db->put(transaction.t, &key, &val, DB_APPEND);

            if( res == 0 ) {
                transaction.commit();
                return true;

            } else {
                std::cerr << "push failed: " << res << std::endl;
                return false;

            }

        } catch( DbException e) {
            std::cerr << "DB What()" << e.what() << std::endl;
            return false;
        } catch( std::exception e) {
            std::cerr << "What()" << e.what() << std::endl;
            return false;
        } catch(...) {
            std::cerr << "Unknown error" << std::endl;
            return false;
        }
    }
};

const char * create_file(const char * f ){
    std::ofstream _f;
    _f.open(f, std::ios::out);
    _f.close();
    return f;
}

class Dequeue : public Queue
{
public:
    Dequeue ( bool sync ) :
        Queue(sync),
        lock(create_file("/tmp/db-test-pop.lock")),
        number_of_records_(0)
    {
        std::cout << "Trying to get exclusize access to database" << std::endl;
        lock.lock();
    }

    virtual ~Dequeue ()
    {
    }

    bool pop(size_t number_of_records, std::vector<std::string>& records)
    {
        if( number_of_records_ != 0 ) // TODO, warning
            abort();

        Cursor cursor;
        records.clear();

        if( number_of_records_ != 0 )
            abort(); // TODO, warning

        // Get a cursor
        try {
            db->cursor(0, &cursor.c, 0);
        } catch( DbException e) {
            std::cerr << "DB What()" << e.what() << std::endl;
            abort();
            return false;
        }

        // Read and delete
        try {
            Dbt val;

            db_recno_t k0 = 0;
            Dbt key((void*)&k0, sizeof(k0));

            for( size_t i = 0; i < number_of_records; i ++ ) {
                int get_res = cursor.c->get(&key, &val, DB_NEXT);

                if( get_res == 0 )
                    records.push_back(std::string((char *)val.get_data(), val.get_size()));
                else
                    break;
            }

            number_of_records_ = records.size();
            if( number_of_records_ == 0 ) {
                abort();
                return false;
            } else {
                return true;
            }

        } catch( DbException e) {
            std::cerr << "DB read/delete What() " << e.what() << std::endl;
            abort();
            return false;
        } catch( std::exception e) {
            std::cerr << "DB read/delete What() " << e.what() << std::endl;
            abort();
            return false;
        }
    }

    bool commit()
    {
        if( number_of_records_ == 0 )
            return true;

        Transaction transaction;
        Cursor      cursor;

        if( ! transaction.init(this) )
            return false;

        if( ! cursor.init(db, transaction.t) )
            return false;

        // Read and delete
        try {
            Dbt val;

            db_recno_t k0 = 0;
            Dbt key((void*)&k0, sizeof(k0));

            for( size_t i = 0; i < number_of_records_; i ++ ) {
                int get_res = cursor.c->get(&key, &val, DB_NEXT);

                if( get_res == 0 )
                    cursor.c->del(0);
                else
                    break; // this is bad!
            }

            number_of_records_ = 0;
            cursor.close();
            transaction.commit();

            return true;

        } catch( DbException e) {
            std::cerr << "DB read/delete What() " << e.what() << std::endl;
            return false;
        } catch( std::exception e) {
            std::cerr << "DB read/delete What() " << e.what() << std::endl;
            return false;
        }
    }

    void abort()
    {
        number_of_records_ = 0;
    }

private:
    boost::interprocess::file_lock lock;
    size_t  number_of_records_;
    sigset_t orig_mask;
};

Please let me know if you see any errors, or if know about an easier way to do this.