text: /*
   * 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 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_ts_t ts_rate;  /**< How often the IO wakeup may be performed (us) */
          rd_ts_t ts_last;  /**< Last IO wakeup */
          /* 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.
   *
   * @param rate_limit if true, rate limit IO-based wakeups.
   *
   * @remark Queue MUST be locked
   */
  static RD_INLINE RD_UNUSED
  void rd_kafka_q_io_event (rd_kafka_q_t *rkq, rd_bool_t rate_limit) {
  
  	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;
          }
  
  
          if (rate_limit) {
                  rd_ts_t now = rd_clock();
                  if (likely(rkq->rkq_qio->ts_last + rkq->rkq_qio->ts_rate > now))
                          return;
  
                  rkq->rkq_qio->ts_last = now;
          }
  
          /* Ignore errors, not much to do anyway. */
          if (rd_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_bool_t rate_limit) {
          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_signal(&rkq->rkq_cond);
                  if (rkq->rkq_qlen == 0)
                          rd_kafka_q_io_event(rkq, rate_limit);
  
                  mtx_unlock(&rkq->rkq_lock);
          } else {
                  mtx_unlock(&rkq->rkq_lock);
                  rd_kafka_q_yield(fwdq, rate_limit);
                  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, rd_false/*no rate-limiting*/);
  
                  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;
  }
  
  /**
   * 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, rd_false/*no rate-limiting*/);
                  rkq->rkq_qlen += srcq->rkq_qlen;
                  rkq->rkq_qsize += srcq->rkq_qsize;
  		cnd_signal(&rkq->rkq_cond);
  
                  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, rd_false/*no rate-limiting*/);
                  rkq->rkq_qlen += srcq->rkq_qlen;
                  rkq->rkq_qsize += srcq->rkq_qsize;
  
                  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
  
  /**
   * 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 */
  };
  
  
  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 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_ */
  