path: root/fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_queue.h

/*
 * librdkafka - The Apache Kafka C/C++ library
 *
 * Copyright (c) 2016 Magnus Edenhill
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#ifndef _RDKAFKA_QUEUE_H_
#define _RDKAFKA_QUEUE_H_

#include "rdkafka_op.h"
#include "rdkafka_int.h"

#ifdef _WIN32
#include <io.h> /* for _write() */
#endif

/** @brief Queueing strategy */
#define RD_KAFKA_QUEUE_FIFO 0
#define RD_KAFKA_QUEUE_LIFO 1

TAILQ_HEAD(rd_kafka_op_tailq, rd_kafka_op_s);

/**
 * @struct Queue for rd_kafka_op_t*.
 *
 * @remark All readers of the queue must call rd_kafka_q_mark_served()
 *         after reading the queue (while still holding the queue lock) to
 *         clear the wakeup-sent flag.
 */
struct rd_kafka_q_s {
        mtx_t rkq_lock;
        cnd_t rkq_cond;
        struct rd_kafka_q_s *rkq_fwdq; /* Forwarded/Routed queue.
                                        * Used in place of this queue
                                        * for all operations. */

        struct rd_kafka_op_tailq rkq_q; /* TAILQ_HEAD(, rd_kafka_op_s) */
        int rkq_qlen;                   /* Number of entries in queue */
        int64_t rkq_qsize;              /* Size of all entries in queue */
        int rkq_refcnt;
        int rkq_flags;
#define RD_KAFKA_Q_F_ALLOCATED 0x1 /* Allocated: rd_free on destroy */
#define RD_KAFKA_Q_F_READY                                                     \
        0x2 /* Queue is ready to be used.                                      \
             * Flag is cleared on destroy */
#define RD_KAFKA_Q_F_FWD_APP                                                   \
        0x4 /* Queue is being forwarded by a call                              \
             * to rd_kafka_queue_forward. */
#define RD_KAFKA_Q_F_YIELD                                                     \
        0x8 /* Have waiters return even if                                     \
             * no rko was enqueued.                                            \
             * This is used to wake up a waiter                                \
             * by triggering the cond-var                                      \
             * but without having to enqueue                                   \
             * an op. */

        rd_kafka_t *rkq_rk;
        struct rd_kafka_q_io *rkq_qio; /* FD-based application signalling */

        /* Op serve callback (optional).
         * Mainly used for forwarded queues to use the original queue's
         * serve function from the forwarded position.
         * Shall return 1 if op was handled, else 0. */
        rd_kafka_q_serve_cb_t *rkq_serve;
        void *rkq_opaque;

#if ENABLE_DEVEL
        char rkq_name[64]; /* Debugging: queue name (FUNC:LINE) */
#else
        const char *rkq_name; /* Debugging: queue name (FUNC) */
#endif
};


/* Application signalling state holder. */
struct rd_kafka_q_io {
        /* For FD-based signalling */
        rd_socket_t fd;
        void *payload;
        size_t size;
        rd_bool_t sent; /**< Wake-up has been sent.
                         *   This field is reset to false by the queue
                         *   reader, allowing a new wake-up to be sent by a
                         *   subsequent writer. */
        /* For callback-based signalling */
        void (*event_cb)(rd_kafka_t *rk, void *opaque);
        void *event_cb_opaque;
};



/**
 * @return true if queue is ready/enabled, else false.
 * @remark queue luck must be held by caller (if applicable)
 */
static RD_INLINE RD_UNUSED int rd_kafka_q_ready(rd_kafka_q_t *rkq) {
        return rkq->rkq_flags & RD_KAFKA_Q_F_READY;
}



void rd_kafka_q_init0(rd_kafka_q_t *rkq,
                      rd_kafka_t *rk,
                      const char *func,
                      int line);
#define rd_kafka_q_init(rkq, rk)                                               \
        rd_kafka_q_init0(rkq, rk, __FUNCTION__, __LINE__)
rd_kafka_q_t *rd_kafka_q_new0(rd_kafka_t *rk, const char *func, int line);
#define rd_kafka_q_new(rk) rd_kafka_q_new0(rk, __FUNCTION__, __LINE__)
void rd_kafka_q_destroy_final(rd_kafka_q_t *rkq);

#define rd_kafka_q_lock(rkqu)   mtx_lock(&(rkqu)->rkq_lock)
#define rd_kafka_q_unlock(rkqu) mtx_unlock(&(rkqu)->rkq_lock)

static RD_INLINE RD_UNUSED rd_kafka_q_t *rd_kafka_q_keep(rd_kafka_q_t *rkq) {
        mtx_lock(&rkq->rkq_lock);
        rkq->rkq_refcnt++;
        mtx_unlock(&rkq->rkq_lock);
        return rkq;
}

static RD_INLINE RD_UNUSED rd_kafka_q_t *
rd_kafka_q_keep_nolock(rd_kafka_q_t *rkq) {
        rkq->rkq_refcnt++;
        return rkq;
}


/**
 * @returns the queue's name (used for debugging)
 */
static RD_INLINE RD_UNUSED const char *rd_kafka_q_name(rd_kafka_q_t *rkq) {
        return rkq->rkq_name;
}

/**
 * @returns the final destination queue name (after forwarding)
 * @remark rkq MUST NOT be locked
 */
static RD_INLINE RD_UNUSED const char *rd_kafka_q_dest_name(rd_kafka_q_t *rkq) {
        const char *ret;
        mtx_lock(&rkq->rkq_lock);
        if (rkq->rkq_fwdq)
                ret = rd_kafka_q_dest_name(rkq->rkq_fwdq);
        else
                ret = rd_kafka_q_name(rkq);
        mtx_unlock(&rkq->rkq_lock);
        return ret;
}

/**
 * @brief Disable a queue.
 *        Attempting to enqueue ops to the queue will destroy the ops.
 */
static RD_INLINE RD_UNUSED void rd_kafka_q_disable0(rd_kafka_q_t *rkq,
                                                    int do_lock) {
        if (do_lock)
                mtx_lock(&rkq->rkq_lock);
        rkq->rkq_flags &= ~RD_KAFKA_Q_F_READY;
        if (do_lock)
                mtx_unlock(&rkq->rkq_lock);
}
#define rd_kafka_q_disable(rkq) rd_kafka_q_disable0(rkq, 1 /*lock*/)

int rd_kafka_q_purge0(rd_kafka_q_t *rkq, int do_lock);
#define rd_kafka_q_purge(rkq) rd_kafka_q_purge0(rkq, 1 /*lock*/)
void rd_kafka_q_purge_toppar_version(rd_kafka_q_t *rkq,
                                     rd_kafka_toppar_t *rktp,
                                     int version);

/**
 * @brief Loose reference to queue, when refcount reaches 0 the queue
 *        will be destroyed.
 *
 * @param disable Also disable the queue, to be used by owner of the queue.
 */
static RD_INLINE RD_UNUSED void rd_kafka_q_destroy0(rd_kafka_q_t *rkq,
                                                    int disable) {
        int do_delete = 0;

        if (disable) {
                /* To avoid recursive locking (from ops being purged
                 * that reference this queue somehow),
                 * we disable the queue and purge it with individual
                 * locking. */
                rd_kafka_q_disable0(rkq, 1 /*lock*/);
                rd_kafka_q_purge0(rkq, 1 /*lock*/);
        }

        mtx_lock(&rkq->rkq_lock);
        rd_kafka_assert(NULL, rkq->rkq_refcnt > 0);
        do_delete = !--rkq->rkq_refcnt;
        mtx_unlock(&rkq->rkq_lock);

        if (unlikely(do_delete))
                rd_kafka_q_destroy_final(rkq);
}

#define rd_kafka_q_destroy(rkq) rd_kafka_q_destroy0(rkq, 0 /*dont-disable*/)

/**
 * @brief Queue destroy method to be used by the owner (poller) of
 *        the queue. The only difference to q_destroy() is that this
 *        method also disables the queue so that any q_enq() operations
 *        will fail.
 *        Failure to disable a queue on the poller when it destroys its
 *        queue reference results in ops being enqueued on the queue
 *        but there is noone left to poll it, possibly resulting in a
 *        hang on termination due to refcounts held by the op.
 */
static RD_INLINE RD_UNUSED void rd_kafka_q_destroy_owner(rd_kafka_q_t *rkq) {
        rd_kafka_q_destroy0(rkq, 1 /*disable*/);
}


/**
 * Reset a queue.
 * WARNING: All messages will be lost and leaked.
 * NOTE: No locking is performed.
 */
static RD_INLINE RD_UNUSED void rd_kafka_q_reset(rd_kafka_q_t *rkq) {
        TAILQ_INIT(&rkq->rkq_q);
        rd_dassert(TAILQ_EMPTY(&rkq->rkq_q));
        rkq->rkq_qlen  = 0;
        rkq->rkq_qsize = 0;
}



/**
 * Forward 'srcq' to 'destq'
 */
void rd_kafka_q_fwd_set0(rd_kafka_q_t *srcq,
                         rd_kafka_q_t *destq,
                         int do_lock,
                         int fwd_app);
#define rd_kafka_q_fwd_set(S, D)                                               \
        rd_kafka_q_fwd_set0(S, D, 1 /*lock*/, 0 /*no fwd_app*/)

/**
 * @returns the forward queue (if any) with its refcount increased.
 * @locks rd_kafka_q_lock(rkq) == !do_lock
 */
static RD_INLINE RD_UNUSED rd_kafka_q_t *rd_kafka_q_fwd_get(rd_kafka_q_t *rkq,
                                                            int do_lock) {
        rd_kafka_q_t *fwdq;
        if (do_lock)
                mtx_lock(&rkq->rkq_lock);

        if ((fwdq = rkq->rkq_fwdq))
                rd_kafka_q_keep(fwdq);

        if (do_lock)
                mtx_unlock(&rkq->rkq_lock);

        return fwdq;
}


/**
 * @returns true if queue is forwarded, else false.
 *
 * @remark Thread-safe.
 */
static RD_INLINE RD_UNUSED int rd_kafka_q_is_fwded(rd_kafka_q_t *rkq) {
        int r;
        mtx_lock(&rkq->rkq_lock);
        r = rkq->rkq_fwdq ? 1 : 0;
        mtx_unlock(&rkq->rkq_lock);
        return r;
}



/**
 * @brief Trigger an IO event for this queue.
 *
 * @remark Queue MUST be locked
 */
static RD_INLINE RD_UNUSED void rd_kafka_q_io_event(rd_kafka_q_t *rkq) {

        if (likely(!rkq->rkq_qio))
                return;

        if (rkq->rkq_qio->event_cb) {
                rkq->rkq_qio->event_cb(rkq->rkq_rk,
                                       rkq->rkq_qio->event_cb_opaque);
                return;
        }


        /* Only one wake-up event should be sent per non-polling period.
         * As the queue reader calls poll/reads the channel it calls to
         * rd_kafka_q_mark_served() to reset the wakeup sent flag, allowing
         * further wakeups in the next non-polling period. */
        if (rkq->rkq_qio->sent)
                return; /* Wake-up event already written */

        rkq->rkq_qio->sent = rd_true;

        /* Write wake-up event to socket.
         * Ignore errors, not much to do anyway. */
        if (rd_socket_write(rkq->rkq_qio->fd, rkq->rkq_qio->payload,
                            (int)rkq->rkq_qio->size) == -1)
                ;
}


/**
 * @brief rko->rko_prio comparator
 * @remark: descending order: higher priority takes preceedence.
 */
static RD_INLINE RD_UNUSED int rd_kafka_op_cmp_prio(const void *_a,
                                                    const void *_b) {
        const rd_kafka_op_t *a = _a, *b = _b;

        return RD_CMP(b->rko_prio, a->rko_prio);
}


/**
 * @brief Wake up waiters without enqueuing an op.
 */
static RD_INLINE RD_UNUSED void rd_kafka_q_yield(rd_kafka_q_t *rkq) {
        rd_kafka_q_t *fwdq;

        mtx_lock(&rkq->rkq_lock);

        rd_dassert(rkq->rkq_refcnt > 0);

        if (unlikely(!(rkq->rkq_flags & RD_KAFKA_Q_F_READY))) {
                /* Queue has been disabled */
                mtx_unlock(&rkq->rkq_lock);
                return;
        }

        if (!(fwdq = rd_kafka_q_fwd_get(rkq, 0))) {
                rkq->rkq_flags |= RD_KAFKA_Q_F_YIELD;
                cnd_broadcast(&rkq->rkq_cond);
                if (rkq->rkq_qlen == 0)
                        rd_kafka_q_io_event(rkq);

                mtx_unlock(&rkq->rkq_lock);
        } else {
                mtx_unlock(&rkq->rkq_lock);
                rd_kafka_q_yield(fwdq);
                rd_kafka_q_destroy(fwdq);
        }
}

/**
 * @brief Low-level unprotected enqueue that only performs
 *        the actual queue enqueue and counter updates.
 * @remark Will not perform locking, signaling, fwdq, READY checking, etc.
 */
static RD_INLINE RD_UNUSED void
rd_kafka_q_enq0(rd_kafka_q_t *rkq, rd_kafka_op_t *rko, int at_head) {
        if (likely(!rko->rko_prio))
                TAILQ_INSERT_TAIL(&rkq->rkq_q, rko, rko_link);
        else if (at_head)
                TAILQ_INSERT_HEAD(&rkq->rkq_q, rko, rko_link);
        else
                TAILQ_INSERT_SORTED(&rkq->rkq_q, rko, rd_kafka_op_t *, rko_link,
                                    rd_kafka_op_cmp_prio);
        rkq->rkq_qlen++;
        rkq->rkq_qsize += rko->rko_len;
}


/**
 * @brief Enqueue \p rko either at head or tail of \p rkq.
 *
 * The provided \p rko is either enqueued or destroyed.
 *
 * \p orig_destq is the original (outermost) dest queue for which
 * this op was enqueued, before any queue forwarding has kicked in.
 * The rko_serve callback from the orig_destq will be set on the rko
 * if there is no rko_serve callback already set, and the \p rko isn't
 * failed because the final queue is disabled.
 *
 * @returns 1 if op was enqueued or 0 if queue is disabled and
 * there was no replyq to enqueue on in which case the rko is destroyed.
 *
 * @locality any thread.
 */
static RD_INLINE RD_UNUSED int rd_kafka_q_enq1(rd_kafka_q_t *rkq,
                                               rd_kafka_op_t *rko,
                                               rd_kafka_q_t *orig_destq,
                                               int at_head,
                                               int do_lock) {
        rd_kafka_q_t *fwdq;

        if (do_lock)
                mtx_lock(&rkq->rkq_lock);

        rd_dassert(rkq->rkq_refcnt > 0);

        if (unlikely(!(rkq->rkq_flags & RD_KAFKA_Q_F_READY))) {
                /* Queue has been disabled, reply to and fail the rko. */
                if (do_lock)
                        mtx_unlock(&rkq->rkq_lock);

                return rd_kafka_op_reply(rko, RD_KAFKA_RESP_ERR__DESTROY);
        }

        if (!(fwdq = rd_kafka_q_fwd_get(rkq, 0))) {
                if (!rko->rko_serve && orig_destq->rkq_serve) {
                        /* Store original queue's serve callback and opaque
                         * prior to forwarding. */
                        rko->rko_serve        = orig_destq->rkq_serve;
                        rko->rko_serve_opaque = orig_destq->rkq_opaque;
                }

                rd_kafka_q_enq0(rkq, rko, at_head);
                cnd_signal(&rkq->rkq_cond);
                if (rkq->rkq_qlen == 1)
                        rd_kafka_q_io_event(rkq);

                if (do_lock)
                        mtx_unlock(&rkq->rkq_lock);
        } else {
                if (do_lock)
                        mtx_unlock(&rkq->rkq_lock);
                rd_kafka_q_enq1(fwdq, rko, orig_destq, at_head, 1 /*do lock*/);
                rd_kafka_q_destroy(fwdq);
        }

        return 1;
}

/**
 * @brief Enqueue the 'rko' op at the tail of the queue 'rkq'.
 *
 * The provided 'rko' is either enqueued or destroyed.
 *
 * @returns 1 if op was enqueued or 0 if queue is disabled and
 * there was no replyq to enqueue on in which case the rko is destroyed.
 *
 * @locality any thread.
 * @locks rkq MUST NOT be locked
 */
static RD_INLINE RD_UNUSED int rd_kafka_q_enq(rd_kafka_q_t *rkq,
                                              rd_kafka_op_t *rko) {
        return rd_kafka_q_enq1(rkq, rko, rkq, 0 /*at tail*/, 1 /*do lock*/);
}


/**
 * @brief Re-enqueue rko at head of rkq.
 *
 * The provided 'rko' is either enqueued or destroyed.
 *
 * @returns 1 if op was enqueued or 0 if queue is disabled and
 * there was no replyq to enqueue on in which case the rko is destroyed.
 *
 * @locality any thread
 * @locks rkq MUST BE locked
 */
static RD_INLINE RD_UNUSED int rd_kafka_q_reenq(rd_kafka_q_t *rkq,
                                                rd_kafka_op_t *rko) {
        return rd_kafka_q_enq1(rkq, rko, rkq, 1 /*at head*/, 0 /*don't lock*/);
}


/**
 * Dequeue 'rko' from queue 'rkq'.
 *
 * NOTE: rkq_lock MUST be held
 * Locality: any thread
 */
static RD_INLINE RD_UNUSED void rd_kafka_q_deq0(rd_kafka_q_t *rkq,
                                                rd_kafka_op_t *rko) {
        rd_dassert(rkq->rkq_qlen > 0 &&
                   rkq->rkq_qsize >= (int64_t)rko->rko_len);

        TAILQ_REMOVE(&rkq->rkq_q, rko, rko_link);
        rkq->rkq_qlen--;
        rkq->rkq_qsize -= rko->rko_len;
}


/**
 * @brief Mark queue as served / read.
 *
 * This is currently used by the queue reader side to reset the io-event
 * wakeup flag.
 *
 * Should be called by all queue readers.
 *
 * @locks_required rkq must be locked.
 */
static RD_INLINE RD_UNUSED void rd_kafka_q_mark_served(rd_kafka_q_t *rkq) {
        if (rkq->rkq_qio)
                rkq->rkq_qio->sent = rd_false;
}


/**
 * Concat all elements of 'srcq' onto tail of 'rkq'.
 * 'rkq' will be be locked (if 'do_lock'==1), but 'srcq' will not.
 * NOTE: 'srcq' will be reset.
 *
 * Locality: any thread.
 *
 * @returns 0 if operation was performed or -1 if rkq is disabled.
 */
static RD_INLINE RD_UNUSED int
rd_kafka_q_concat0(rd_kafka_q_t *rkq, rd_kafka_q_t *srcq, int do_lock) {
        int r = 0;

        while (srcq->rkq_fwdq) /* Resolve source queue */
                srcq = srcq->rkq_fwdq;
        if (unlikely(srcq->rkq_qlen == 0))
                return 0; /* Don't do anything if source queue is empty */

        if (do_lock)
                mtx_lock(&rkq->rkq_lock);
        if (!rkq->rkq_fwdq) {
                rd_kafka_op_t *rko;

                rd_dassert(TAILQ_EMPTY(&srcq->rkq_q) || srcq->rkq_qlen > 0);
                if (unlikely(!(rkq->rkq_flags & RD_KAFKA_Q_F_READY))) {
                        if (do_lock)
                                mtx_unlock(&rkq->rkq_lock);
                        return -1;
                }
                /* First insert any prioritized ops from srcq
                 * in the right position in rkq. */
                while ((rko = TAILQ_FIRST(&srcq->rkq_q)) && rko->rko_prio > 0) {
                        TAILQ_REMOVE(&srcq->rkq_q, rko, rko_link);
                        TAILQ_INSERT_SORTED(&rkq->rkq_q, rko, rd_kafka_op_t *,
                                            rko_link, rd_kafka_op_cmp_prio);
                }

                TAILQ_CONCAT(&rkq->rkq_q, &srcq->rkq_q, rko_link);
                if (rkq->rkq_qlen == 0)
                        rd_kafka_q_io_event(rkq);
                rkq->rkq_qlen += srcq->rkq_qlen;
                rkq->rkq_qsize += srcq->rkq_qsize;
                cnd_signal(&rkq->rkq_cond);

                rd_kafka_q_mark_served(srcq);
                rd_kafka_q_reset(srcq);
        } else
                r = rd_kafka_q_concat0(rkq->rkq_fwdq ? rkq->rkq_fwdq : rkq,
                                       srcq, rkq->rkq_fwdq ? do_lock : 0);
        if (do_lock)
                mtx_unlock(&rkq->rkq_lock);

        return r;
}

#define rd_kafka_q_concat(dstq, srcq) rd_kafka_q_concat0(dstq, srcq, 1 /*lock*/)


/**
 * @brief Prepend all elements of 'srcq' onto head of 'rkq'.
 * 'rkq' will be be locked (if 'do_lock'==1), but 'srcq' will not.
 * 'srcq' will be reset.
 *
 * @remark Will not respect priority of ops, srcq will be prepended in its
 *         original form to rkq.
 *
 * @locality any thread.
 */
static RD_INLINE RD_UNUSED void
rd_kafka_q_prepend0(rd_kafka_q_t *rkq, rd_kafka_q_t *srcq, int do_lock) {
        if (do_lock)
                mtx_lock(&rkq->rkq_lock);
        if (!rkq->rkq_fwdq && !srcq->rkq_fwdq) {
                /* FIXME: prio-aware */
                /* Concat rkq on srcq */
                TAILQ_CONCAT(&srcq->rkq_q, &rkq->rkq_q, rko_link);
                /* Move srcq to rkq */
                TAILQ_MOVE(&rkq->rkq_q, &srcq->rkq_q, rko_link);
                if (rkq->rkq_qlen == 0 && srcq->rkq_qlen > 0)
                        rd_kafka_q_io_event(rkq);
                rkq->rkq_qlen += srcq->rkq_qlen;
                rkq->rkq_qsize += srcq->rkq_qsize;

                rd_kafka_q_mark_served(srcq);
                rd_kafka_q_reset(srcq);
        } else
                rd_kafka_q_prepend0(rkq->rkq_fwdq ? rkq->rkq_fwdq : rkq,
                                    srcq->rkq_fwdq ? srcq->rkq_fwdq : srcq,
                                    rkq->rkq_fwdq ? do_lock : 0);
        if (do_lock)
                mtx_unlock(&rkq->rkq_lock);
}

#define rd_kafka_q_prepend(dstq, srcq)                                         \
        rd_kafka_q_prepend0(dstq, srcq, 1 /*lock*/)


/* Returns the number of elements in the queue */
static RD_INLINE RD_UNUSED int rd_kafka_q_len(rd_kafka_q_t *rkq) {
        int qlen;
        rd_kafka_q_t *fwdq;
        mtx_lock(&rkq->rkq_lock);
        if (!(fwdq = rd_kafka_q_fwd_get(rkq, 0))) {
                qlen = rkq->rkq_qlen;
                mtx_unlock(&rkq->rkq_lock);
        } else {
                mtx_unlock(&rkq->rkq_lock);
                qlen = rd_kafka_q_len(fwdq);
                rd_kafka_q_destroy(fwdq);
        }
        return qlen;
}

/* Returns the total size of elements in the queue */
static RD_INLINE RD_UNUSED uint64_t rd_kafka_q_size(rd_kafka_q_t *rkq) {
        uint64_t sz;
        rd_kafka_q_t *fwdq;
        mtx_lock(&rkq->rkq_lock);
        if (!(fwdq = rd_kafka_q_fwd_get(rkq, 0))) {
                sz = rkq->rkq_qsize;
                mtx_unlock(&rkq->rkq_lock);
        } else {
                mtx_unlock(&rkq->rkq_lock);
                sz = rd_kafka_q_size(fwdq);
                rd_kafka_q_destroy(fwdq);
        }
        return sz;
}

/**
 * @brief Construct a temporary on-stack replyq with increased
 *        \p rkq refcount (unless NULL), version, and debug id.
 */
static RD_INLINE RD_UNUSED rd_kafka_replyq_t
rd_kafka_replyq_make(rd_kafka_q_t *rkq, int version, const char *id) {
        rd_kafka_replyq_t replyq = RD_ZERO_INIT;

        if (rkq) {
                replyq.q       = rd_kafka_q_keep(rkq);
                replyq.version = version;
#if ENABLE_DEVEL
                replyq._id = rd_strdup(id);
#endif
        }

        return replyq;
}

/* Construct temporary on-stack replyq with increased Q refcount and
 * optional VERSION. */
#define RD_KAFKA_REPLYQ(Q, VERSION)                                            \
        rd_kafka_replyq_make(Q, VERSION, __FUNCTION__)

/* Construct temporary on-stack replyq for indicating no replyq. */
#if ENABLE_DEVEL
#define RD_KAFKA_NO_REPLYQ                                                     \
        (rd_kafka_replyq_t) {                                                  \
                NULL, 0, NULL                                                  \
        }
#else
#define RD_KAFKA_NO_REPLYQ                                                     \
        (rd_kafka_replyq_t) {                                                  \
                NULL, 0                                                        \
        }
#endif


/**
 * @returns true if the replyq is valid, else false.
 */
static RD_INLINE RD_UNUSED rd_bool_t
rd_kafka_replyq_is_valid(rd_kafka_replyq_t *replyq) {
        rd_bool_t valid = rd_true;

        if (!replyq->q)
                return rd_false;

        rd_kafka_q_lock(replyq->q);
        valid = rd_kafka_q_ready(replyq->q);
        rd_kafka_q_unlock(replyq->q);

        return valid;
}



/**
 * Set up replyq.
 * Q refcnt is increased.
 */
static RD_INLINE RD_UNUSED void rd_kafka_set_replyq(rd_kafka_replyq_t *replyq,
                                                    rd_kafka_q_t *rkq,
                                                    int32_t version) {
        replyq->q       = rkq ? rd_kafka_q_keep(rkq) : NULL;
        replyq->version = version;
#if ENABLE_DEVEL
        replyq->_id = rd_strdup(__FUNCTION__);
#endif
}

/**
 * Set rko's replyq with an optional version (versionptr != NULL).
 * Q refcnt is increased.
 */
static RD_INLINE RD_UNUSED void
rd_kafka_op_set_replyq(rd_kafka_op_t *rko,
                       rd_kafka_q_t *rkq,
                       rd_atomic32_t *versionptr) {
        rd_kafka_set_replyq(&rko->rko_replyq, rkq,
                            versionptr ? rd_atomic32_get(versionptr) : 0);
}

/* Set reply rko's version from replyq's version */
#define rd_kafka_op_get_reply_version(REPLY_RKO, ORIG_RKO)                     \
        do {                                                                   \
                (REPLY_RKO)->rko_version = (ORIG_RKO)->rko_replyq.version;     \
        } while (0)


/* Clear replyq holder without decreasing any .q references. */
static RD_INLINE RD_UNUSED void
rd_kafka_replyq_clear(rd_kafka_replyq_t *replyq) {
        memset(replyq, 0, sizeof(*replyq));
}

/**
 * @brief Make a copy of \p src in \p dst, with its own queue reference
 */
static RD_INLINE RD_UNUSED void rd_kafka_replyq_copy(rd_kafka_replyq_t *dst,
                                                     rd_kafka_replyq_t *src) {
        dst->version = src->version;
        dst->q       = src->q;
        if (dst->q)
                rd_kafka_q_keep(dst->q);
#if ENABLE_DEVEL
        if (src->_id)
                dst->_id = rd_strdup(src->_id);
        else
                dst->_id = NULL;
#endif
}


/**
 * Clear replyq holder and destroy any .q references.
 */
static RD_INLINE RD_UNUSED void
rd_kafka_replyq_destroy(rd_kafka_replyq_t *replyq) {
        if (replyq->q)
                rd_kafka_q_destroy(replyq->q);
#if ENABLE_DEVEL
        if (replyq->_id) {
                rd_free(replyq->_id);
                replyq->_id = NULL;
        }
#endif
        rd_kafka_replyq_clear(replyq);
}


/**
 * @brief Wrapper for rd_kafka_q_enq() that takes a replyq,
 *        steals its queue reference, enqueues the op with the replyq version,
 *        and then destroys the queue reference.
 *
 *        If \p version is non-zero it will be updated, else replyq->version.
 *
 * @returns Same as rd_kafka_q_enq()
 */
static RD_INLINE RD_UNUSED int rd_kafka_replyq_enq(rd_kafka_replyq_t *replyq,
                                                   rd_kafka_op_t *rko,
                                                   int version) {
        rd_kafka_q_t *rkq = replyq->q;
        int r;

        if (version)
                rko->rko_version = version;
        else
                rko->rko_version = replyq->version;

        /* The replyq queue reference is done after we've enqueued the rko
         * so clear it here. */
        replyq->q = NULL; /* destroyed separately below */

#if ENABLE_DEVEL
        if (replyq->_id) {
                rd_free(replyq->_id);
                replyq->_id = NULL;
        }
#endif

        /* Retain replyq->version since it is used by buf_callback
         * when dispatching the callback. */

        r = rd_kafka_q_enq(rkq, rko);

        rd_kafka_q_destroy(rkq);

        return r;
}



rd_kafka_op_t *rd_kafka_q_pop_serve(rd_kafka_q_t *rkq,
                                    rd_ts_t timeout_us,
                                    int32_t version,
                                    rd_kafka_q_cb_type_t cb_type,
                                    rd_kafka_q_serve_cb_t *callback,
                                    void *opaque);
rd_kafka_op_t *
rd_kafka_q_pop(rd_kafka_q_t *rkq, rd_ts_t timeout_us, int32_t version);
int rd_kafka_q_serve(rd_kafka_q_t *rkq,
                     int timeout_ms,
                     int max_cnt,
                     rd_kafka_q_cb_type_t cb_type,
                     rd_kafka_q_serve_cb_t *callback,
                     void *opaque);


int rd_kafka_q_move_cnt(rd_kafka_q_t *dstq,
                        rd_kafka_q_t *srcq,
                        int cnt,
                        int do_locks);

int rd_kafka_q_serve_rkmessages(rd_kafka_q_t *rkq,
                                int timeout_ms,
                                rd_kafka_message_t **rkmessages,
                                size_t rkmessages_size);
rd_kafka_resp_err_t rd_kafka_q_wait_result(rd_kafka_q_t *rkq, int timeout_ms);

int rd_kafka_q_apply(rd_kafka_q_t *rkq,
                     int (*callback)(rd_kafka_q_t *rkq,
                                     rd_kafka_op_t *rko,
                                     void *opaque),
                     void *opaque);

void rd_kafka_q_fix_offsets(rd_kafka_q_t *rkq,
                            int64_t min_offset,
                            int64_t base_offset);

/**
 * @returns the last op in the queue matching \p op_type and \p allow_err (bool)
 * @remark The \p rkq must be properly locked before this call, the returned rko
 *         is not removed from the queue and may thus not be held for longer
 *         than the lock is held.
 */
static RD_INLINE RD_UNUSED rd_kafka_op_t *
rd_kafka_q_last(rd_kafka_q_t *rkq, rd_kafka_op_type_t op_type, int allow_err) {
        rd_kafka_op_t *rko;
        TAILQ_FOREACH_REVERSE(rko, &rkq->rkq_q, rd_kafka_op_tailq, rko_link) {
                if (rko->rko_type == op_type && (allow_err || !rko->rko_err))
                        return rko;
        }

        return NULL;
}

void rd_kafka_q_io_event_enable(rd_kafka_q_t *rkq,
                                rd_socket_t fd,
                                const void *payload,
                                size_t size);

/* Public interface */
struct rd_kafka_queue_s {
        rd_kafka_q_t *rkqu_q;
        rd_kafka_t *rkqu_rk;
        int rkqu_is_owner; /**< Is owner/creator of rkqu_q */
};


rd_kafka_queue_t *rd_kafka_queue_new0(rd_kafka_t *rk, rd_kafka_q_t *rkq);

void rd_kafka_q_dump(FILE *fp, rd_kafka_q_t *rkq);

extern int RD_TLS rd_kafka_yield_thread;



/**
 * @name Enqueue op once
 * @{
 */

/**
 * @brief Minimal rd_kafka_op_t wrapper that ensures that
 *        the op is only enqueued on the provided queue once.
 *
 * Typical use-case is for an op to be triggered from multiple sources,
 * but at most once, such as from a timer and some other source.
 */
typedef struct rd_kafka_enq_once_s {
        mtx_t lock;
        int refcnt;
        rd_kafka_op_t *rko;
        rd_kafka_replyq_t replyq;
} rd_kafka_enq_once_t;


/**
 * @brief Allocate and set up a new eonce and set the initial refcount to 1.
 * @remark This is to be called by the owner of the rko.
 */
static RD_INLINE RD_UNUSED rd_kafka_enq_once_t *
rd_kafka_enq_once_new(rd_kafka_op_t *rko, rd_kafka_replyq_t replyq) {
        rd_kafka_enq_once_t *eonce = rd_calloc(1, sizeof(*eonce));
        mtx_init(&eonce->lock, mtx_plain);
        eonce->rko    = rko;
        eonce->replyq = replyq; /* struct copy */
        eonce->refcnt = 1;
        return eonce;
}

/**
 * @brief Re-enable triggering of a eonce even after it has been triggered
 *        once.
 *
 * @remark This is to be called by the owner.
 */
static RD_INLINE RD_UNUSED void
rd_kafka_enq_once_reenable(rd_kafka_enq_once_t *eonce,
                           rd_kafka_op_t *rko,
                           rd_kafka_replyq_t replyq) {
        mtx_lock(&eonce->lock);
        eonce->rko = rko;
        rd_kafka_replyq_destroy(&eonce->replyq);
        eonce->replyq = replyq; /* struct copy */
        mtx_unlock(&eonce->lock);
}


/**
 * @brief Free eonce and its resources. Must only be called with refcnt==0
 *        and eonce->lock NOT held.
 */
static RD_INLINE RD_UNUSED void
rd_kafka_enq_once_destroy0(rd_kafka_enq_once_t *eonce) {
        /* This must not be called with the rko or replyq still set, which would
         * indicate that no enqueueing was performed and that the owner
         * did not clean up, which is a bug. */
        rd_assert(!eonce->rko);
        rd_assert(!eonce->replyq.q);
#if ENABLE_DEVEL
        rd_assert(!eonce->replyq._id);
#endif
        rd_assert(eonce->refcnt == 0);

        mtx_destroy(&eonce->lock);
        rd_free(eonce);
}


/**
 * @brief Increment refcount for source (non-owner), such as a timer.
 *
 * @param srcdesc a human-readable descriptive string of the source.
 *                May be used for future debugging.
 */
static RD_INLINE RD_UNUSED void
rd_kafka_enq_once_add_source(rd_kafka_enq_once_t *eonce, const char *srcdesc) {
        mtx_lock(&eonce->lock);
        eonce->refcnt++;
        mtx_unlock(&eonce->lock);
}


/**
 * @brief Decrement refcount for source (non-owner), such as a timer.
 *
 * @param srcdesc a human-readable descriptive string of the source.
 *                May be used for future debugging.
 *
 * @remark Must only be called from the owner with the owner
 *         still holding its own refcount.
 *         This API is used to undo an add_source() from the
 *         same code.
 */
static RD_INLINE RD_UNUSED void
rd_kafka_enq_once_del_source(rd_kafka_enq_once_t *eonce, const char *srcdesc) {
        int do_destroy;

        mtx_lock(&eonce->lock);
        rd_assert(eonce->refcnt > 0);
        eonce->refcnt--;
        do_destroy = eonce->refcnt == 0;
        mtx_unlock(&eonce->lock);

        if (do_destroy) {
                /* We're the last refcount holder, clean up eonce. */
                rd_kafka_enq_once_destroy0(eonce);
        }
}

/**
 * @brief Trigger a source's reference where the eonce resides on
 *        an rd_list_t. This is typically used as a free_cb for
 *        rd_list_destroy() and the trigger error code is
 *        always RD_KAFKA_RESP_ERR__DESTROY.
 */
void rd_kafka_enq_once_trigger_destroy(void *ptr);


/**
 * @brief Decrement refcount for source (non-owner) and return the rko
 *        if still set.
 *
 * @remark Must only be called by sources (non-owner) but only on the
 *         the owner's thread to make sure the rko is not freed.
 *
 * @remark The rko remains set on the eonce.
 */
static RD_INLINE RD_UNUSED rd_kafka_op_t *
rd_kafka_enq_once_del_source_return(rd_kafka_enq_once_t *eonce,
                                    const char *srcdesc) {
        rd_bool_t do_destroy;
        rd_kafka_op_t *rko;

        mtx_lock(&eonce->lock);

        rd_assert(eonce->refcnt > 0);
        /* Owner must still hold a eonce reference, or the eonce must
         * have been disabled by the owner (no rko) */
        rd_assert(eonce->refcnt > 1 || !eonce->rko);
        eonce->refcnt--;
        do_destroy = eonce->refcnt == 0;

        rko = eonce->rko;
        mtx_unlock(&eonce->lock);

        if (do_destroy) {
                /* We're the last refcount holder, clean up eonce. */
                rd_kafka_enq_once_destroy0(eonce);
        }

        return rko;
}

/**
 * @brief Trigger enqueuing of the rko (unless already enqueued)
 *        and drops the source's refcount.
 *
 * @remark Must only be called by sources (non-owner).
 */
static RD_INLINE RD_UNUSED void
rd_kafka_enq_once_trigger(rd_kafka_enq_once_t *eonce,
                          rd_kafka_resp_err_t err,
                          const char *srcdesc) {
        int do_destroy;
        rd_kafka_op_t *rko       = NULL;
        rd_kafka_replyq_t replyq = RD_ZERO_INIT;

        mtx_lock(&eonce->lock);

        rd_assert(eonce->refcnt > 0);
        eonce->refcnt--;
        do_destroy = eonce->refcnt == 0;

        if (eonce->rko) {
                /* Not already enqueued, do it.
                 * Detach the rko and replyq from the eonce and unlock the eonce
                 * before enqueuing rko on reply to avoid recursive locks
                 * if the replyq has been disabled and the ops
                 * destructor is called (which might then access the eonce
                 * to clean up). */
                rko    = eonce->rko;
                replyq = eonce->replyq;

                eonce->rko = NULL;
                rd_kafka_replyq_clear(&eonce->replyq);

                /* Reply is enqueued at the end of this function */
        }
        mtx_unlock(&eonce->lock);

        if (do_destroy) {
                /* We're the last refcount holder, clean up eonce. */
                rd_kafka_enq_once_destroy0(eonce);
        }

        if (rko) {
                rko->rko_err = err;
                rd_kafka_replyq_enq(&replyq, rko, replyq.version);
                rd_kafka_replyq_destroy(&replyq);
        }
}

/**
 * @brief Destroy eonce, must only be called by the owner.
 *        There may be outstanding refcounts by non-owners after this call
 */
static RD_INLINE RD_UNUSED void
rd_kafka_enq_once_destroy(rd_kafka_enq_once_t *eonce) {
        int do_destroy;

        mtx_lock(&eonce->lock);
        rd_assert(eonce->refcnt > 0);
        eonce->refcnt--;
        do_destroy = eonce->refcnt == 0;

        eonce->rko = NULL;
        rd_kafka_replyq_destroy(&eonce->replyq);

        mtx_unlock(&eonce->lock);

        if (do_destroy) {
                /* We're the last refcount holder, clean up eonce. */
                rd_kafka_enq_once_destroy0(eonce);
        }
}


/**
 * @brief Disable the owner's eonce, extracting, resetting and returning
 *        the \c rko object.
 *
 *        This is the same as rd_kafka_enq_once_destroy() but returning
 *        the rko.
 *
 *        Use this for owner-thread triggering where the enqueuing of the
 *        rko on the replyq is not necessary.
 *
 * @returns the eonce's rko object, if still available, else NULL.
 */
static RD_INLINE RD_UNUSED rd_kafka_op_t *
rd_kafka_enq_once_disable(rd_kafka_enq_once_t *eonce) {
        int do_destroy;
        rd_kafka_op_t *rko;

        mtx_lock(&eonce->lock);
        rd_assert(eonce->refcnt > 0);
        eonce->refcnt--;
        do_destroy = eonce->refcnt == 0;

        /* May be NULL */
        rko        = eonce->rko;
        eonce->rko = NULL;
        rd_kafka_replyq_destroy(&eonce->replyq);

        mtx_unlock(&eonce->lock);

        if (do_destroy) {
                /* We're the last refcount holder, clean up eonce. */
                rd_kafka_enq_once_destroy0(eonce);
        }

        return rko;
}


/**@}*/


#endif /* _RDKAFKA_QUEUE_H_ */