Беркли Д.Б., параллельные очереди - PullRequest
Новая
       8

Беркли Д.Б., параллельные очереди

1 голос
/ 05 сентября 2011

Я пытаюсь реализовать параллельную постоянную очередь с использованием Berkeley 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);
    }
    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;
}

Компиляция: 

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

Создать каталог db 

mkdir /tmp/db

И когда я запускаю его, я получаю всевозможные ошибки (ошибка сегментации, ошибка выделения, и иногда она действительно работает)

Я уверен, что пропустил некоторые блокировки, но я просто не знаю, каксделать это.Так что любые советы и / или предложения, чтобы это исправить, приветствуются.

1 Ответ

2 голосов
/ 05 сентября 2011

Просто для справки, вот решение, на котором я остановился после долгих поисков в Google и trial'n'error.

Приложение представляет собой процесс вызова домой, когда производитель добавляет данные, а потребитель пытается отправить их домой. Если потребитель не может отправить его домой, он должен повторить попытку. База данных не должна блокироваться для производителя, пока потребитель пытается получить данные.

Код имеет блокировку файла и допускает только один потребительский процесс.

Вот код: 

#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;
};

Пожалуйста, дайте мне знать, если вы видите какие-либо ошибки, или если знаете о более простом способе сделать это.

...