当前位置： 文江博客话题详情

Berkeley DB，并发队列

发布于 2024-12-02 19:10:00 字数 2808 浏览 0 评论 0原文

我正在尝试使用 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-dir

mkdir /tmp/db

当我运行它时，我收到各种错误（分段错误，分配错误，有时它实际上有效）

我确信我错过了一些锁定，但我只是不知道该怎么做。因此，非常欢迎任何解决此问题的提示和/或建议。

原文

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);
    }
    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.

如果你对这篇内容有疑问，欢迎到本站社区发帖提问 参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论， 你可以免费 注册 一个本站的账号。

评论（1）

ゝ杯具 2024-12-09 19:10:00

仅供记录，这是我经过多次谷歌搜索和反复试验后确定的解决方案。

该应用程序是一个回拨进程，其中生产者添加数据，而消费者尝试将其发送回家。如果消费者未能将其送回家，则必须重试。当消费者尝试接收数据时，数据库不得阻塞生产者。

该代码具有文件锁，并且只允许一个消费者进程。

以下是代码：

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

如果您发现任何错误，或者知道更简单的方法，请告诉我。

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.

回复 原文
~没有更多了~

关于作者

oО清风挽发oО

暂无简介

0 文章
0 评论
24 人气
发私信

相关话题

更多

热门标签

操作系统 程序设计 IT运维 Linux系统管理 JavaScript 服务器应用 solaris C/C++ PHP Shell BSD Vue.js aix Oracle Python HTML 系统管理 HTML5 CSS 前端
更多

推荐作者

Gabu-gabumon

文章 0 评论 0

qq_CgiN62

文章 0 评论 0

荔枝明

文章 0 评论 0

赏烟花じ飞满天

文章 0 评论 0

独守阴晴ぅ圆缺

文章 0 评论 0

¤→小豸慧

文章 0 评论 0

更多

友情链接

文江博客
    我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站，即表示您同意使用 Cookies 和您的相关数据。
    原文