myxce/src/controller/queue.c

#include <errno.h>
#include <stdlib.h>
#include <string.h>

#include "cluster.h"
#include "persist.h"
#include "qobj.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xstring.h"
#include "wal_types.h"

void queue_fini(queue_t* q)
{
    if (q != NULL) {
        free(q);
    }
}

queue_t* create_queue(cluster_t* c, xce_queue_t* p)
{
    queue_t* q;

    q = makeQueue(c->qmgr, p);
    return q;
}

int build_queues(void* c)
{
    cluster_t* cls = (cluster_t*)c;
    int        i;

    if (cls == NULL) {
        return -EINVAL;
    }

    if (cls->myconf == NULL) {
        return -EINVAL;
    }

    for (i = 0; i < cls->myconf->cluster_config->nqueues; i++) {
        xce_queue_t* q = cls->myconf->cluster_config->queues[i];

        create_queue(cls, q);
    }

    return 0;
}

int queue_ctrl(cluster_t* c, xce_queuectrl_t* ctrl)
{
    queue_t* q  = NULL;
    int      rc = 0;

    if (!c || !ctrl || !ctrl->queue || *ctrl->queue == '\0') {
        TRACE("%s invalid arguments.", __func__);
        return -EINVAL;
    }
    TRACE("%s queue_ctrl queue %s", __func__, ctrl->queue);

    lock(c->qmgr, RTTI_LOCK_READ);
    rc = key_find(c->qmgr, (void*)ctrl->queue, (void**)&q);
    if (q != NULL) {
        reference(q);
    }
    unlock(c->qmgr);

    if (rc < 0) {
        return rc;
    }

    if (q == NULL) {
        PERROR("qmgr_lockq");
        return -ENOENT;
    }

    /* bug-447: key_find returns 1 on found (binary librtti, unnormalizable
     * upstream). Normalize the found-rc to 0 here so the AFTER_EVENT hook
     * gate below fires (WAL append + replication) and the handler's
     * commit-barrier pattern sees 0. Real errors are untouched: rc < 0
     * returned above, the q == NULL miss returned -ENOENT, and control-
     * action failures set rc themselves. */
    rc = 0;

    lock(q, RTTI_LOCK_WRITE);
    switch (ctrl->action) {
    case QUEUE_OPEN:
        if (!(q->queue.status & QUEUE_STAT_CLOSED)) {
            goto clean;
        }
        q->queue.status &= ~QUEUE_STAT_CLOSED;
        break;
    case QUEUE_CLOSE:
        if (q->queue.status & QUEUE_STAT_CLOSED) {
            goto clean;
        }
        q->queue.status |= QUEUE_STAT_CLOSED;
        break;
    case QUEUE_INACTIVE:
        if (q->queue.status & QUEUE_STAT_INACT_ADMIN) {
            goto clean;
        }
        q->queue.status |= QUEUE_STAT_INACT_ADMIN;
        break;
    case QUEUE_ACTIVE:
        if (!(q->queue.status & QUEUE_STAT_INACT_ADMIN)) {
            goto clean;
        }
        q->queue.status &= ~QUEUE_STAT_INACT_ADMIN;
        break;
    default:
        rc = -1;
        break;
    }
    if (rc == 0) {
        xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_QUEUE,
                      HOOK_AFTER_EVENT, HOOK_DETAIL_QUEUE_UPDATE, c, q);
    }

clean:
    unlock(q);

    dereference(q);

    return rc;
}

int do_queuecontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int         rc = 0;
    xce_reply_t reply;

    init_reply_by_req(&reply, request);

    rc = queue_ctrl(c, request->body.queuectrlmsg);

    /* Commit barrier (ha-commit-barrier-rollout Req 1.1-1.4): the success
     * reply is sent only after the operation's WAL entry is majority-
     * committed; barrier failures (-ETIMEDOUT, ...) become the error
     * reply. Errors from queue_ctrl reply immediately — no barrier wait.
     * queue_ctrl released all entity locks before returning, so blocking
     * here holds no lock. timeout 0 = configured sync_timeout_ms. */
    if (rc == 0) {
        rc = persist_commit_barrier(&c->persist, 0);
        if (rc != 0)
            WARN("%s commit barrier failed, rc %d", __func__, rc);
    }

    reply.status     = rc;
    reply.data_which = RequestReply_data_nodata;

    xce_reply(ep, c->myhost, &reply);
    return rc;
}

int get_queue_exec(cluster_t* c, char* queue, char** pre_exec, char** post_exec)
{
    queue_t* q  = NULL;
    int      rc = 0;

    TRACE("entering %s", __func__);

    if ((c == NULL) || (queue == NULL) || (pre_exec == NULL)
        || (post_exec == NULL)) {
        return -EINVAL;
    }

    TRACE("%s queue %s", __func__, queue);

    lock(c->qmgr, RTTI_LOCK_READ);
    rc = key_find(c->qmgr, (void*)queue, (void**)&q);
    if (q != NULL) {
        reference(q);
    }
    unlock(c->qmgr);

    if (rc < 0) {
        return rc;
    }

    if (q == NULL) {
        PERROR("qmgr_lockq");
        return -ENOENT;
    }

    lock(q, RTTI_LOCK_READ);
    if (q->queue.pre_exec != NULL) {
        *pre_exec = makeString(q->queue.pre_exec);
    } else {
        *pre_exec = NULL;
    }

    if (q->queue.post_exec != NULL) {
        *post_exec = makeString(q->queue.post_exec);
    } else {
        *post_exec = NULL;
    }
    unlock(q);

    dereference(q);

    return 0;
}

/* Restore queue status flags from a WAL_QUEUE_CTRL entry during recovery.
 * Queues are built from config at startup; this re-applies admin state changes
 * (open/close/active/inactive) that were recorded in the WAL. */
int restore_queue_state(cluster_t* c, const char* name, uint32_t status)
{
    if (!name || name[0] == '\0')
        return -EINVAL;

    queue_t* q = NULL;
    lock(c->qmgr, RTTI_LOCK_READ);
    key_find(c->qmgr, (void*)name, (void**)&q);
    unlock(c->qmgr);
    if (!q)
        return -ENOENT; /* queue not yet built from config — skip */

    lock(q, RTTI_LOCK_WRITE);
    q->queue.status = status;
    unlock(q);
    return 0;
}