Function bdb_queue_get_int

bdb/queue.c:1279–1715

}

static int bdb_queue_get_int(bdb_state_type *bdb_state, int consumer,
                             const struct bdb_queue_cursor *prevcursor,
                             void **fnd, size_t *fnddtalen, size_t *fnddtaoff,
                             struct bdb_queue_cursor *fndcursor,
                             int *bdberr)
{
    DBT dbt_key, dbt_data;
    DBC *dbcp;

    uint8_t hdrbuf[QUEUE_HDR_LEN];
    uint8_t *p_buf = hdrbuf, *p_buf_end = (p_buf + sizeof(hdrbuf));
    uint8_t *p_item_buf, *p_item_buf_start, *p_item_buf_end;

    struct bdb_queue_header hdr;
    int getop;
    db_recno_t recno;
    db_recno_t *recnos;
    struct bdb_queue_found item = {0};
    size_t item_len;
    char *dtaptr;
    size_t next_frag;
    int rc;
    unsigned long long lastgenid = 0;
    int tmptid = 0;
    DB_TXN *tid = NULL;
    int scanmode = 0;

    if (consumer < 0 || consumer >= BDBQUEUE_MAX_CONSUMERS) {
        *bdberr = BDBERR_BADARGS;
        return -1;
    }

    bdb_state->qpriv->stats.n_logical_gets++;

    if (TXN_READ_HACK()) {
        rc = bdb_state->dbenv->txn_begin(bdb_state->dbenv, NULL, &tid, 0);
        if (rc != 0) {
            logmsg(LOGMSG_ERROR, "bdb_queue_get_int: txn_begin failed %d %s\n", rc,
                    db_strerror(rc));
            *bdberr = BDBERR_MISC;
            return -1;
        }
        tmptid = 1;
    }

    rc = bdb_state->dbp_data[0][0]->cursor(bdb_state->dbp_data[0][0], tid,
                                           &dbcp, 0);
    if (rc != 0) {
        switch (rc) {
        case DB_REP_HANDLE_DEAD:
        case DB_LOCK_DEADLOCK:
            bdb_state->qpriv->stats.n_get_deadlocks++;
            *bdberr = BDBERR_DEADLOCK;
            break;

        default:
            logmsg(LOGMSG_ERROR, "bdb_queue_get_int: cursor failed %d %s\n", rc,
                    db_strerror(rc));