trunk/src/msgq.c

/*
 * IRC - Internet Relay Chat, ircd/msgq.c
 * Copyright (C) 2000 Kevin L. Mitchell <klmitch@mit.edu>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 1, or (at your option)
 * any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */
/** @file
 * @brief Outbound message queue implementation.
 * @version $Id$
 */

#include "stdinc.h"
#include "msgq.h"
#include "list.h"
#include "log.h"
#include "send.h"
#include "memory.h"
#include "restart.h"
#include "client.h"
#include "ircd.h"
#include "numeric.h"
#include "xsnprintf.h"


#include <sys/uio.h>

#define FEAT_BUFFERPOOL 27000000 /* XXX */

#define MB_BASE_SHIFT  5  /**< Log2 of smallest message body to allocate. */
#define MB_MAX_SHIFT   9  /**< Log2 of largest message body to allocate. */

/** Buffer for a single message. */
struct MsgBuf
{
  struct MsgBuf *next;      /**< next msg in global queue */
  struct MsgBuf **prev_p;   /**< what points to us in linked list */
  struct MsgBuf *real;      /**< the actual MsgBuf we're attaching */
  unsigned int ref;         /**< reference count */
  unsigned int length;      /**< length of message */
  unsigned int power;       /**< size of buffer (power of 2) */
  char msg[1];              /**< the message */
};

/** Return allocated length of the buffer of \a buf. */
#define bufsize(buf)    (1 << (buf)->power)

/** Message body for a particular destination. */
struct Msg
{
  struct Msg *next;    /**< next msg */
  unsigned int sent;   /**< bytes in msg that have already been sent */
  struct MsgBuf *msg;  /**< actual message in queue */
};

/** Statistics tracking for message sizes. */
struct MsgSizes
{
  unsigned int msgs;  /**< total number of messages */
  unsigned int sizes[IRCD_BUFSIZE];  /**< histogram of message sizes */
};

/** Global tracking data for message buffers. */
static struct
{
  struct MsgBuf *msglist;      /**< list of in-use MsgBuf's */

  struct
  {
    unsigned int alloc;        /**< number of Msg's allocated */
    unsigned int used;         /**< number of Msg's in use */
    struct Msg *free;          /**< freelist of Msg's */
  } msgs;                      /**< tracking info for Msg structs */

  size_t tot_bufsize;          /**< total amount of memory in buffers */

  /** Array of MsgBuf information, one entry for each used bucket size. */
  struct
  {
    unsigned int alloc;        /**< total MsgBuf's of this size */
    unsigned int used;         /**< number of MsgBuf's of this size in use */
    struct MsgBuf *free;       /**< list of free MsgBuf's */
  } msgBufs[MB_MAX_SHIFT - MB_BASE_SHIFT + 1];

  struct MsgSizes sizes;       /**< histogram of message sizes */
} MQData;

/*
 * This routine is used to remove a certain amount of data from a given
 * queue and release the Msg (and MsgBuf) structure if needed
 */
/** Remove some data from a list within a message queue.
 * @param[in,out] mq Message queue to remove from.
 * @param[in,out] qlist Particular list within queue to remove from.
 * @param[in,out] length_p Number of bytes left to remove.
 */
static void
msgq_delmsg(struct MsgQ *mq, struct MsgQList *qlist, unsigned int *length_p)
{
  struct Msg *m;
  unsigned int msglen;

  assert(mq);
  assert(qlist);
  assert(qlist->head);
  assert(length_p);

  m = qlist->head;  /* find the msg we're deleting from */

  msglen = m->msg->length - m->sent;  /* calculate how much is left */

  if (*length_p >= msglen)
  {
    /* deleted it all? */
    mq->length -= msglen;  /* decrement length */
    mq->count--;  /* decrement the message count */
    *length_p -= msglen;

    msgq_clean(m->msg);  /* free up the struct MsgBuf */
    m->msg = 0;  /* don't let it point anywhere nasty, please */

    if (qlist->head == qlist->tail)  /* figure out if we emptied the queue */
      qlist->head = qlist->tail = 0;
    else
      qlist->head = m->next;  /* just shift the list down some */

    MQData.msgs.used--;  /* struct Msg is not in use anymore */

    m->next = MQData.msgs.free;  /* throw it onto the free list */
    MQData.msgs.free = m;
  }
  else
  {
    mq->length -= *length_p;  /* decrement queue length */
    m->sent += *length_p;  /* this much of the message has been sent */
    *length_p = 0;  /* we've dealt with it all */
  }
}

/** Initialize \a mq.
 * @param[in] mq MsgQ to initialize.
 */
void
msgq_init(struct MsgQ *mq)
{
  assert(mq);

  mq->length = 0;
  mq->count = 0;
  mq->queue.head = 0;
  mq->queue.tail = 0;
  mq->prio.head = 0;
  mq->prio.tail = 0;
}

/** Delete bytes from the front of a message queue.
 * @param[in] mq Queue to drop data from.
 * @param[in] length Number of bytes to drop.
 */
void
msgq_delete(struct MsgQ *mq, unsigned int length)
{
  assert(0 != mq);

  while (length > 0)
  {
    if (mq->queue.head && mq->queue.head->sent > 0)  /* partial msg on norm q */
      msgq_delmsg(mq, &mq->queue, &length);
    else if (mq->prio.head)  /* message (partial or complete) on prio queue */
      msgq_delmsg(mq, &mq->prio, &length);
    else if (mq->queue.head)  /* message on normal queue */
      msgq_delmsg(mq, &mq->queue, &length);
    else
      break;
  }
}

/** Map data from a message queue to an I/O vector.
 * @param[in] mq Message queue to send from.
 * @param[out] iov Output vector.
 * @param[in] count Number of elements in \a iov.
 * @param[out] len Number of bytes mapped from \a mq to \a iov.
 * @return Number of elements filled in \a iov.
 */
int
msgq_mapiov(const struct MsgQ *mq, struct iovec *iov, int count,
            unsigned int *len)
{
  struct Msg *queue;
  struct Msg *prio;
  int i = 0;

  assert(mq);
  assert(iov);
  assert(count);
  assert(len);

  if (mq->length <= 0)  /* no data to map */
    return 0;

  if (mq->queue.head && mq->queue.head->sent > 0)
  {
    /* partial msg on norm q */
    iov[i].iov_base = mq->queue.head->msg->msg + mq->queue.head->sent;
    iov[i].iov_len = mq->queue.head->msg->length - mq->queue.head->sent;
    *len += iov[i].iov_len;

    queue = mq->queue.head->next;  /* where we start later... */

    i++;  /* filled an iovec... */

    if (!--count)  /* check for space */
      return i;
  }
  else
    queue = mq->queue.head;  /* start at head of queue */

  if (mq->prio.head && mq->prio.head->sent > 0)
  {
    /* partial msg on prio q */
    iov[i].iov_base = mq->prio.head->msg->msg + mq->prio.head->sent;
    iov[i].iov_len = mq->prio.head->msg->length - mq->prio.head->sent;
    *len += iov[i].iov_len;

    prio = mq->prio.head->next;  /* where we start later... */

    i++;  /* filled an iovec... */

    if (!--count)  /* check for space */
      return i;
  }
  else
    prio = mq->prio.head;  /* start at head of prio */

  for (; prio; prio = prio->next)
  {
    /* go through prio queue */
    iov[i].iov_base = prio->msg->msg;  /* store message */
    iov[i].iov_len = prio->msg->length;
    *len += iov[i].iov_len;

    i++;  /* filled an iovec... */

    if (!--count)  /* check for space */
      return i;
  }

  for (; queue; queue = queue->next)
  {
    /* go through normal queue */
    iov[i].iov_base = queue->msg->msg;
    iov[i].iov_len = queue->msg->length;
    *len += iov[i].iov_len;

    i++;  /* filled an iovec... */

    if (!--count)  /* check for space */
      return i;
  }

  return i;
}

/** Allocate a message buffer large enough to hold \a length bytes.
 * TODO: \a in_mb needs better documentation.
 * @param[in] in_mb Some other message buffer(?).
 * @param[in] length Number of bytes of space to reserve in output.
 * @return Pointer to some usable message buffer.
 */
static struct MsgBuf *
msgq_alloc(struct MsgBuf *in_mb, unsigned int length)
{
  struct MsgBuf *mb;
  unsigned int power;

  /* Find the power of two size that will accommodate the message */
  for (power = MB_BASE_SHIFT; power < MB_MAX_SHIFT + 1; power++)
    if ((length - 1) >> power == 0)
      break;

  assert((1 << power) >= length);
  assert((1 << power) <= 512);
  length = 1 << power;  /* reset the length */

  /* If the message needs a buffer of exactly the existing size, just use it */
  if (in_mb && in_mb->power == power)
  {
    in_mb->real = in_mb;  /* real buffer is this buffer */
    return in_mb;
  }

  /* Try popping one off the freelist first */
  if ((mb = MQData.msgBufs[power - MB_BASE_SHIFT].free))
    MQData.msgBufs[power - MB_BASE_SHIFT].free = mb->next;
  else if (MQData.tot_bufsize < FEAT_BUFFERPOOL)
  {
    /* Allocate another if we won't bust the BUFFERPOOL */
    ilog(LOG_TYPE_DEBUG, "Allocating MsgBuf of length %d (total size %zu)",
         length, sizeof(struct MsgBuf) + length);

    mb = MyMalloc(sizeof(struct MsgBuf) + length);
    MQData.msgBufs[power - MB_BASE_SHIFT].alloc++;
    mb->power = power;  /* remember size */
    MQData.tot_bufsize += length;
  }

  if (mb)
  {
    MQData.msgBufs[power - MB_BASE_SHIFT].used++;  /* how many are we using? */

    mb->real = 0;  /* essential initializations */
    mb->ref = 1;

    if (in_mb)  /* remember who's the *real* buffer */
      in_mb->real = mb;
  }
  else if (in_mb)  /* just use the input buffer */
    mb = in_mb->real = in_mb;

  return mb;  /* return the buffer */
}

/** Deallocate unused message buffers.
 */
static void
msgq_clear_freembs(void)
{
  struct MsgBuf *mb;
  int i;

  /* Walk through the various size classes */
  for (i = MB_BASE_SHIFT; i < MB_MAX_SHIFT + 1; i++)
  {
    /* walk down the free list */
    while ((mb = MQData.msgBufs[i - MB_BASE_SHIFT].free))
    {
      MQData.msgBufs[i - MB_BASE_SHIFT].free = mb->next;  /* shift free list */
      MQData.msgBufs[i - MB_BASE_SHIFT].alloc--;  /* reduce allocation count */
      MQData.tot_bufsize -= 1 << i;  /* reduce total buffer allocation count */
      MyFree(mb);  /* and free the buffer */
    }
  }
}

/** Format a message buffer for a client from a format string.
 * @param[in] dest %Client that receives the data (may be NULL).
 * @param[in] format Format string for message.
 * @param[in] vl Argument list for \a format.
 * @return Allocated MsgBuf.
 */
struct MsgBuf *
msgq_vmake(struct Client *dest, const char *format, va_list vl)
{
  struct MsgBuf *mb;

  assert(format);

  if (!(mb = msgq_alloc(0, IRCD_BUFSIZE)))
  {
    /*
     * from "Married With Children" episode were Al bought a REAL toilet
     * on the black market because he was tired of the wimpy water
     * conserving toilets they make these days --Bleep
     */
    /*
     * Apparently this doesn't work, the server _has_ to
     * dump a few clients to handle the load. A fully loaded
     * server cannot handle a net break without dumping some
     * clients. If we flush the connections here under a full
     * load we may end up starving the kernel for mbufs and
     * crash the machine
     */
    /*
     * attempt to recover from buffer starvation before
     * bailing this may help servers running out of memory
     */
    send_queued_all();
    mb = msgq_alloc(0, IRCD_BUFSIZE);

    if (!mb)
    { /* OK, try clearing the buffer free list */
      msgq_clear_freembs();
      mb = msgq_alloc(0, IRCD_BUFSIZE);
    }
#if 0
    if (!mb)
    {
      /* OK, try killing a client */
      kill_highest_sendq(0);  /* Don't kill any server connections */
      msgq_clear_freembs();  /* Release whatever was just freelisted */
      mb = msgq_alloc(0, IRCD_BUFSIZE);
    }

    if (!mb)
    {
      kill_highest_sendq(1);  /* Try killing a server connection now */
      msgq_clear_freembs();  /* Clear freelist again */
      mb = msgq_alloc(0, IRCD_BUFSIZE);
    }
#endif
    if (!mb)  /* AIEEEE! */
      restart("Unable to allocate buffers!");
  }

  mb->next = MQData.msglist;  /* initialize the msgbuf */
  mb->prev_p = &MQData.msglist;

  /* fill the buffer */
  mb->length = xvsnprintf(dest, mb->msg, bufsize(mb) - 1, format, vl);

  if (mb->length > bufsize(mb) - 2)
    mb->length = bufsize(mb) - 2;

  mb->msg[mb->length++] = '\r';  /* add \r\n to buffer */
  mb->msg[mb->length++] = '\n';
  mb->msg[mb->length] = '\0';  /* not strictly necessary */

  assert(mb->length <= bufsize(mb));

  if (MQData.msglist)  /* link it into the list */
    MQData.msglist->prev_p = &mb->next;
  MQData.msglist = mb;

  return mb;
}

/** Format a message buffer for a client from a format string.
 * @param[in] dest %Client that receives the data (may be NULL).
 * @param[in] format Format string for message.
 * @return Allocated MsgBuf.
 */
struct MsgBuf *
msgq_make(struct Client *dest, const char *format, ...)
{
  va_list vl;
  struct MsgBuf *mb;

  va_start(vl, format);
  mb = msgq_vmake(dest, format, vl);
  va_end(vl);

  return mb;
}

/** Append text to an existing message buffer.
 * @param[in] dest %Client for whom to format the message.
 * @param[in] mb Message buffer to append to.
 * @param[in] format Format string of what to append.
 */
void
msgq_append(struct Client *dest, struct MsgBuf *mb, const char *format, ...)
{
  va_list vl;

  assert(mb);
  assert(format);
  assert(0 == mb->real);

  assert(2 < mb->length);
  assert(bufsize(mb) >= mb->length);

  mb->length -= 2;  /* back up to before \r\n */

  va_start(vl, format);  /* append to the buffer */

  mb->length += xvsnprintf(dest, mb->msg + mb->length,
                           bufsize(mb) - mb->length - 1, format, vl);

  va_end(vl);

  if (mb->length > bufsize(mb) - 2)
    mb->length = bufsize(mb) - 2;

  mb->msg[mb->length++] = '\r';  /* add \r\n to buffer */
  mb->msg[mb->length++] = '\n';
  mb->msg[mb->length] = '\0';  /* not strictly necessary */

  assert(mb->length <= bufsize(mb));
}

/** Decrement the reference count on \a mb, freeing it if needed.
 * @param[in] mb MsgBuf to release.
 */
void
msgq_clean(struct MsgBuf *mb)
{
  assert(mb);
  assert(0 < mb->ref);

  if (!--mb->ref)
  {
    /* deallocate the message */
    if (mb->prev_p)
    {
      *mb->prev_p = mb->next;  /* clip it out of active MsgBuf's list */

      if (mb->next)
        mb->next->prev_p = mb->prev_p;
    }

    if (mb->real && mb->real != mb)  /* clean up the real buffer */
      msgq_clean(mb->real);

    mb->next = MQData.msgBufs[mb->power - MB_BASE_SHIFT].free;
    MQData.msgBufs[mb->power - MB_BASE_SHIFT].free = mb;
    MQData.msgBufs[mb->power - MB_BASE_SHIFT].used--;

    mb->prev_p = 0;
  }
}

/** Append a message to a peer's message queue.
 * @param[in] mq Message queue to append to.
 * @param[in] mb Message to append.
 * @param[in] prio If non-zero, use the high-priority (lag-busting) message list; else use the normal list.
 */
void
msgq_add(struct MsgQ *mq, struct MsgBuf *mb, int prio)
{
  struct MsgQList *qlist;
  struct Msg *msg;

  assert(mq);
  assert(mb);
  assert(0 < mb->ref);
  assert(0 < mb->length);

  ilog(LOG_TYPE_DEBUG, "Adding buffer %p [%.*s] length %u to %s queue", mb,
       mb->length - 2, mb->msg, mb->length, prio ? "priority" : "normal");

  qlist = prio ? &mq->prio : &mq->queue;

  if (!(msg = MQData.msgs.free))
  {
    /* do I need to allocate one? */
    msg = MyMalloc(sizeof(struct Msg));
    MQData.msgs.alloc++;  /* we allocated another */
  }
  else  /* shift the free list */
    MQData.msgs.free = MQData.msgs.free->next;

  MQData.msgs.used++;  /* we're using another */

  msg->next = 0; /* initialize the msg */
  msg->sent = 0;

  /* Get the real buffer, allocating one if necessary */
  if (!mb->real)
  {
    struct MsgBuf *tmp;

    MQData.sizes.msgs++; /* update histogram counts */
    MQData.sizes.sizes[mb->length - 1]++;

    tmp = msgq_alloc(mb, mb->length); /* allocate a close-fitting buffer */

    if (tmp != mb)
    {
      /* OK, prepare the new "real" buffer */
      ilog(LOG_TYPE_DEBUG, "Copying old buffer %p [%.*s] length %u into new "
           "buffer %p size %u", mb, mb->length - 2, mb->msg, mb->length,
           tmp, bufsize(tmp));
      memcpy(tmp->msg, mb->msg, mb->length + 1);  /* copy string over */
      tmp->length = mb->length;

      tmp->next = mb->next;  /* replace it in the list, now */

      if (tmp->next)
        tmp->next->prev_p = &tmp->next;
      tmp->prev_p = mb->prev_p;
      *tmp->prev_p = tmp;

      mb->next = 0;  /* this one's no longer in the list */
      mb->prev_p = 0;
    }
  }

  mb = mb->real;  /* work with the real buffer */
  mb->ref++;  /* increment the ref count on the buffer */

  msg->msg = mb;  /* point at the real message buffer now */

  if (!qlist->head)  /* queue list was empty; head and tail point to msg */
    qlist->head = qlist->tail = msg;
  else
  {
    assert(qlist->tail);

    qlist->tail->next = msg;  /* queue had something in it; add to end */
    qlist->tail = msg;
  }

  mq->length += mb->length;  /* update the queue length */
  mq->count++;  /* and the queue count */
}

/** Report memory statistics for message buffers.
 * @param[in] cptr Client requesting information.
 * @param[out] msg_alloc Receives number of bytes allocated in Msg structs.
 * @param[out] msgbuf_alloc Receives number of bytes allocated in MsgBuf structs.
 */
void
msgq_count_memory(struct Client *source_p, size_t *msg_alloc, size_t *msgbuf_alloc)
{
  int i;
  size_t total = 0, size;

  assert(client_p);
  assert(msg_alloc);
  assert(msgbuf_alloc);

  /* Data for Msg's is simple, so just send it */
  sendto_one(source_p, ":%s %d %s :Msgs allocated %d(%zu) used %d(%zu) text %zu",
             me.name, RPL_STATSDEBUG, source_p->name,
             MQData.msgs.alloc, MQData.msgs.alloc * sizeof(struct Msg),
             MQData.msgs.used,  MQData.msgs.used * sizeof(struct Msg),
             MQData.tot_bufsize);
  /* count_memory() wants to know the total */
  *msg_alloc = MQData.msgs.alloc * sizeof(struct Msg);

  /* Ok, now walk through each size class */
  for (i = MB_BASE_SHIFT; i < MB_MAX_SHIFT + 1; i++)
  {
    size = sizeof(struct MsgBuf) + (1 << i);  /* total size of a buffer */

    /* Send information for this buffer size class */
    sendto_one(source_p, ":%s %d %s :MsgBufs of size %zu allocated %d(%zu) used %d(%zu)",
               me.name, RPL_STATSDEBUG, source_p->name, 1 << i,
               MQData.msgBufs[i - MB_BASE_SHIFT].alloc,
               MQData.msgBufs[i - MB_BASE_SHIFT].alloc * size,
               MQData.msgBufs[i - MB_BASE_SHIFT].used,
               MQData.msgBufs[i - MB_BASE_SHIFT].used * size);

    /* count_memory() wants to know the total */
    total += MQData.msgBufs[i - MB_BASE_SHIFT].alloc * size;
  }

  *msgbuf_alloc = total;
}

/** Report remaining space in a MsgBuf.
 * @param[in] mb Message buffer to check.
 * @return Number of additional bytes that can be appended to the message.
 */
unsigned int
msgq_bufleft(struct MsgBuf *mb)
{
  assert(mb);

  return bufsize(mb) - mb->length;  /* \r\n counted in mb->length */
}

/** Send histogram of message lengths to a client.
 * @param[in] cptr Client requesting statistics.
 * @param[in] sd Stats descriptor for request (ignored).
 * @param[in] param Extra parameter from user (ignored).
 */
void
msgq_histogram(struct Client *source_p)
{
  struct MsgSizes tmp = MQData.sizes;  /* All hail structure copy! */
  int i;

  sendto_one(source_p, ":%s %d %s :Histogram of message lengths (%lu messages)",
             me.name, RPL_STATSDEBUG, source_p->name, tmp.msgs);

  for (i = 0; i + 16 <= IRCD_BUFSIZE; i += 16)
    sendto_one(source_p, ":%s %d %s :% 4d: %u %u %u %u "
               "%u %u %u %u %u %u %u %u %u %u %u %u",
               me.name, RPL_STATSDEBUG, source_p->name, i + 1,
               tmp.sizes[i +  0], tmp.sizes[i +  1], tmp.sizes[i +  2],
               tmp.sizes[i +  3], tmp.sizes[i +  4], tmp.sizes[i +  5],
               tmp.sizes[i +  6], tmp.sizes[i +  7], tmp.sizes[i +  8],
               tmp.sizes[i +  9], tmp.sizes[i + 10], tmp.sizes[i + 11],
               tmp.sizes[i + 12], tmp.sizes[i + 13], tmp.sizes[i + 14],
               tmp.sizes[i + 15]);
}
Revision:	2318
Committed:	Fri Jun 21 17:47:52 2013 UTC (12 years, 2 months ago) by michael
Content type:	text/x-csrc
File size:	20237 byte(s)
Log Message:	- Stole ircu's outbound message queue implementation for later use
#	User	Rev	Content
1	michael	2318	/*
2			* IRC - Internet Relay Chat, ircd/msgq.c
3			* Copyright (C) 2000 Kevin L. Mitchell <klmitch@mit.edu>
4			*
5			* This program is free software; you can redistribute it and/or modify
6			* it under the terms of the GNU General Public License as published by
7			* the Free Software Foundation; either version 1, or (at your option)
8			* any later version.
9			*
10			* This program is distributed in the hope that it will be useful,
11			* but WITHOUT ANY WARRANTY; without even the implied warranty of
12			* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13			* GNU General Public License for more details.
14			*
15			* You should have received a copy of the GNU General Public License
16			* along with this program; if not, write to the Free Software
17			* Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
18			*/
19			/** @file
20			* @brief Outbound message queue implementation.
21			* @version $Id$
22			*/
23
24			#include "stdinc.h"
25			#include "msgq.h"
26			#include "list.h"
27			#include "log.h"
28			#include "send.h"
29			#include "memory.h"
30			#include "restart.h"
31			#include "client.h"
32			#include "ircd.h"
33			#include "numeric.h"
34			#include "xsnprintf.h"
35
36
37			#include <sys/uio.h>
38
39			#define FEAT_BUFFERPOOL 27000000 /* XXX */
40
41			#define MB_BASE_SHIFT 5 /*< Log2 of smallest message body to allocate. /
42			#define MB_MAX_SHIFT 9 /*< Log2 of largest message body to allocate. /
43
44			/** Buffer for a single message. */
45			struct MsgBuf
46			{
47			struct MsgBuf next; /< next msg in global queue /
48			struct MsgBuf prev_p; /< what points to us in linked list */
49			struct MsgBuf real; /< the actual MsgBuf we're attaching /
50			unsigned int ref; /*< reference count /
51			unsigned int length; /*< length of message /
52			unsigned int power; /*< size of buffer (power of 2) /
53			char msg[1]; /*< the message /
54			};
55
56			/** Return allocated length of the buffer of \a buf. */
57			#define bufsize(buf) (1 << (buf)->power)
58
59			/** Message body for a particular destination. */
60			struct Msg
61			{
62			struct Msg next; /< next msg /
63			unsigned int sent; /*< bytes in msg that have already been sent /
64			struct MsgBuf msg; /< actual message in queue /
65			};
66
67			/** Statistics tracking for message sizes. */
68			struct MsgSizes
69			{
70			unsigned int msgs; /*< total number of messages /
71			unsigned int sizes[IRCD_BUFSIZE]; /*< histogram of message sizes /
72			};
73
74			/** Global tracking data for message buffers. */
75			static struct
76			{
77			struct MsgBuf msglist; /< list of in-use MsgBuf's /
78
79			struct
80			{
81			unsigned int alloc; /*< number of Msg's allocated /
82			unsigned int used; /*< number of Msg's in use /
83			struct Msg free; /< freelist of Msg's /
84			} msgs; /*< tracking info for Msg structs /
85
86			size_t tot_bufsize; /*< total amount of memory in buffers /
87
88			/** Array of MsgBuf information, one entry for each used bucket size. */
89			struct
90			{
91			unsigned int alloc; /*< total MsgBuf's of this size /
92			unsigned int used; /*< number of MsgBuf's of this size in use /
93			struct MsgBuf free; /< list of free MsgBuf's /
94			} msgBufs[MB_MAX_SHIFT - MB_BASE_SHIFT + 1];
95
96			struct MsgSizes sizes; /*< histogram of message sizes /
97			} MQData;
98
99			/*
100			* This routine is used to remove a certain amount of data from a given
101			* queue and release the Msg (and MsgBuf) structure if needed
102			*/
103			/** Remove some data from a list within a message queue.
104			* @param[in,out] mq Message queue to remove from.
105			* @param[in,out] qlist Particular list within queue to remove from.
106			* @param[in,out] length_p Number of bytes left to remove.
107			*/
108			static void
109			msgq_delmsg(struct MsgQ mq, struct MsgQList qlist, unsigned int *length_p)
110			{
111			struct Msg *m;
112			unsigned int msglen;
113
114			assert(mq);
115			assert(qlist);
116			assert(qlist->head);
117			assert(length_p);
118
119			m = qlist->head; /* find the msg we're deleting from */
120
121			msglen = m->msg->length - m->sent; /* calculate how much is left */
122
123			if (*length_p >= msglen)
124			{
125			/* deleted it all? */
126			mq->length -= msglen; /* decrement length */
127			mq->count--; /* decrement the message count */
128			*length_p -= msglen;
129
130			msgq_clean(m->msg); /* free up the struct MsgBuf */
131			m->msg = 0; /* don't let it point anywhere nasty, please */
132
133			if (qlist->head == qlist->tail) /* figure out if we emptied the queue */
134			qlist->head = qlist->tail = 0;
135			else
136			qlist->head = m->next; /* just shift the list down some */
137
138			MQData.msgs.used--; /* struct Msg is not in use anymore */
139
140			m->next = MQData.msgs.free; /* throw it onto the free list */
141			MQData.msgs.free = m;
142			}
143			else
144			{
145			mq->length -= length_p; / decrement queue length */
146			m->sent += length_p; / this much of the message has been sent */
147			length_p = 0; / we've dealt with it all */
148			}
149			}
150
151			/** Initialize \a mq.
152			* @param[in] mq MsgQ to initialize.
153			*/
154			void
155			msgq_init(struct MsgQ *mq)
156			{
157			assert(mq);
158
159			mq->length = 0;
160			mq->count = 0;
161			mq->queue.head = 0;
162			mq->queue.tail = 0;
163			mq->prio.head = 0;
164			mq->prio.tail = 0;
165			}
166
167			/** Delete bytes from the front of a message queue.
168			* @param[in] mq Queue to drop data from.
169			* @param[in] length Number of bytes to drop.
170			*/
171			void
172			msgq_delete(struct MsgQ *mq, unsigned int length)
173			{
174			assert(0 != mq);
175
176			while (length > 0)
177			{
178			if (mq->queue.head && mq->queue.head->sent > 0) /* partial msg on norm q */
179			msgq_delmsg(mq, &mq->queue, &length);
180			else if (mq->prio.head) /* message (partial or complete) on prio queue */
181			msgq_delmsg(mq, &mq->prio, &length);
182			else if (mq->queue.head) /* message on normal queue */
183			msgq_delmsg(mq, &mq->queue, &length);
184			else
185			break;
186			}
187			}
188
189			/** Map data from a message queue to an I/O vector.
190			* @param[in] mq Message queue to send from.
191			* @param[out] iov Output vector.
192			* @param[in] count Number of elements in \a iov.
193			* @param[out] len Number of bytes mapped from \a mq to \a iov.
194			* @return Number of elements filled in \a iov.
195			*/
196			int
197			msgq_mapiov(const struct MsgQ mq, struct iovec iov, int count,
198			unsigned int *len)
199			{
200			struct Msg *queue;
201			struct Msg *prio;
202			int i = 0;
203
204			assert(mq);
205			assert(iov);
206			assert(count);
207			assert(len);
208
209			if (mq->length <= 0) /* no data to map */
210			return 0;
211
212			if (mq->queue.head && mq->queue.head->sent > 0)
213			{
214			/* partial msg on norm q */
215			iov[i].iov_base = mq->queue.head->msg->msg + mq->queue.head->sent;
216			iov[i].iov_len = mq->queue.head->msg->length - mq->queue.head->sent;
217			*len += iov[i].iov_len;
218
219			queue = mq->queue.head->next; /* where we start later... */
220
221			i++; /* filled an iovec... */
222
223			if (!--count) /* check for space */
224			return i;
225			}
226			else
227			queue = mq->queue.head; /* start at head of queue */
228
229			if (mq->prio.head && mq->prio.head->sent > 0)
230			{
231			/* partial msg on prio q */
232			iov[i].iov_base = mq->prio.head->msg->msg + mq->prio.head->sent;
233			iov[i].iov_len = mq->prio.head->msg->length - mq->prio.head->sent;
234			*len += iov[i].iov_len;
235
236			prio = mq->prio.head->next; /* where we start later... */
237
238			i++; /* filled an iovec... */
239
240			if (!--count) /* check for space */
241			return i;
242			}
243			else
244			prio = mq->prio.head; /* start at head of prio */
245
246			for (; prio; prio = prio->next)
247			{
248			/* go through prio queue */
249			iov[i].iov_base = prio->msg->msg; /* store message */
250			iov[i].iov_len = prio->msg->length;
251			*len += iov[i].iov_len;
252
253			i++; /* filled an iovec... */
254
255			if (!--count) /* check for space */
256			return i;
257			}
258
259			for (; queue; queue = queue->next)
260			{
261			/* go through normal queue */
262			iov[i].iov_base = queue->msg->msg;
263			iov[i].iov_len = queue->msg->length;
264			*len += iov[i].iov_len;
265
266			i++; /* filled an iovec... */
267
268			if (!--count) /* check for space */
269			return i;
270			}
271
272			return i;
273			}
274
275			/** Allocate a message buffer large enough to hold \a length bytes.
276			* TODO: \a in_mb needs better documentation.
277			* @param[in] in_mb Some other message buffer(?).
278			* @param[in] length Number of bytes of space to reserve in output.
279			* @return Pointer to some usable message buffer.
280			*/
281			static struct MsgBuf *
282			msgq_alloc(struct MsgBuf *in_mb, unsigned int length)
283			{
284			struct MsgBuf *mb;
285			unsigned int power;
286
287			/* Find the power of two size that will accommodate the message */
288			for (power = MB_BASE_SHIFT; power < MB_MAX_SHIFT + 1; power++)
289			if ((length - 1) >> power == 0)
290			break;
291
292			assert((1 << power) >= length);
293			assert((1 << power) <= 512);
294			length = 1 << power; /* reset the length */
295
296			/* If the message needs a buffer of exactly the existing size, just use it */
297			if (in_mb && in_mb->power == power)
298			{
299			in_mb->real = in_mb; /* real buffer is this buffer */
300			return in_mb;
301			}
302
303			/* Try popping one off the freelist first */
304			if ((mb = MQData.msgBufs[power - MB_BASE_SHIFT].free))
305			MQData.msgBufs[power - MB_BASE_SHIFT].free = mb->next;
306			else if (MQData.tot_bufsize < FEAT_BUFFERPOOL)
307			{
308			/* Allocate another if we won't bust the BUFFERPOOL */
309			ilog(LOG_TYPE_DEBUG, "Allocating MsgBuf of length %d (total size %zu)",
310			length, sizeof(struct MsgBuf) + length);
311
312			mb = MyMalloc(sizeof(struct MsgBuf) + length);
313			MQData.msgBufs[power - MB_BASE_SHIFT].alloc++;
314			mb->power = power; /* remember size */
315			MQData.tot_bufsize += length;
316			}
317
318			if (mb)
319			{
320			MQData.msgBufs[power - MB_BASE_SHIFT].used++; /* how many are we using? */
321
322			mb->real = 0; /* essential initializations */
323			mb->ref = 1;
324
325			if (in_mb) /* remember who's the real buffer */
326			in_mb->real = mb;
327			}
328			else if (in_mb) /* just use the input buffer */
329			mb = in_mb->real = in_mb;
330
331			return mb; /* return the buffer */
332			}
333
334			/** Deallocate unused message buffers.
335			*/
336			static void
337			msgq_clear_freembs(void)
338			{
339			struct MsgBuf *mb;
340			int i;
341
342			/* Walk through the various size classes */
343			for (i = MB_BASE_SHIFT; i < MB_MAX_SHIFT + 1; i++)
344			{
345			/* walk down the free list */
346			while ((mb = MQData.msgBufs[i - MB_BASE_SHIFT].free))
347			{
348			MQData.msgBufs[i - MB_BASE_SHIFT].free = mb->next; /* shift free list */
349			MQData.msgBufs[i - MB_BASE_SHIFT].alloc--; /* reduce allocation count */
350			MQData.tot_bufsize -= 1 << i; /* reduce total buffer allocation count */
351			MyFree(mb); /* and free the buffer */
352			}
353			}
354			}
355
356			/** Format a message buffer for a client from a format string.
357			* @param[in] dest %Client that receives the data (may be NULL).
358			* @param[in] format Format string for message.
359			* @param[in] vl Argument list for \a format.
360			* @return Allocated MsgBuf.
361			*/
362			struct MsgBuf *
363			msgq_vmake(struct Client dest, const char format, va_list vl)
364			{
365			struct MsgBuf *mb;
366
367			assert(format);
368
369			if (!(mb = msgq_alloc(0, IRCD_BUFSIZE)))
370			{
371			/*
372			* from "Married With Children" episode were Al bought a REAL toilet
373			* on the black market because he was tired of the wimpy water
374			* conserving toilets they make these days --Bleep
375			*/
376			/*
377			* Apparently this doesn't work, the server _has_ to
378			* dump a few clients to handle the load. A fully loaded
379			* server cannot handle a net break without dumping some
380			* clients. If we flush the connections here under a full
381			* load we may end up starving the kernel for mbufs and
382			* crash the machine
383			*/
384			/*
385			* attempt to recover from buffer starvation before
386			* bailing this may help servers running out of memory
387			*/
388			send_queued_all();
389			mb = msgq_alloc(0, IRCD_BUFSIZE);
390
391			if (!mb)
392			{ /* OK, try clearing the buffer free list */
393			msgq_clear_freembs();
394			mb = msgq_alloc(0, IRCD_BUFSIZE);
395			}
396			#if 0
397			if (!mb)
398			{
399			/* OK, try killing a client */
400			kill_highest_sendq(0); /* Don't kill any server connections */
401			msgq_clear_freembs(); /* Release whatever was just freelisted */
402			mb = msgq_alloc(0, IRCD_BUFSIZE);
403			}
404
405			if (!mb)
406			{
407			kill_highest_sendq(1); /* Try killing a server connection now */
408			msgq_clear_freembs(); /* Clear freelist again */
409			mb = msgq_alloc(0, IRCD_BUFSIZE);
410			}
411			#endif
412			if (!mb) /* AIEEEE! */
413			restart("Unable to allocate buffers!");
414			}
415
416			mb->next = MQData.msglist; /* initialize the msgbuf */
417			mb->prev_p = &MQData.msglist;
418
419			/* fill the buffer */
420			mb->length = xvsnprintf(dest, mb->msg, bufsize(mb) - 1, format, vl);
421
422			if (mb->length > bufsize(mb) - 2)
423			mb->length = bufsize(mb) - 2;
424
425			mb->msg[mb->length++] = '\r'; /* add \r\n to buffer */
426			mb->msg[mb->length++] = '\n';
427			mb->msg[mb->length] = '\0'; /* not strictly necessary */
428
429			assert(mb->length <= bufsize(mb));
430
431			if (MQData.msglist) /* link it into the list */
432			MQData.msglist->prev_p = &mb->next;
433			MQData.msglist = mb;
434
435			return mb;
436			}
437
438			/** Format a message buffer for a client from a format string.
439			* @param[in] dest %Client that receives the data (may be NULL).
440			* @param[in] format Format string for message.
441			* @return Allocated MsgBuf.
442			*/
443			struct MsgBuf *
444			msgq_make(struct Client dest, const char format, ...)
445			{
446			va_list vl;
447			struct MsgBuf *mb;
448
449			va_start(vl, format);
450			mb = msgq_vmake(dest, format, vl);
451			va_end(vl);
452
453			return mb;
454			}
455
456			/** Append text to an existing message buffer.
457			* @param[in] dest %Client for whom to format the message.
458			* @param[in] mb Message buffer to append to.
459			* @param[in] format Format string of what to append.
460			*/
461			void
462			msgq_append(struct Client dest, struct MsgBuf mb, const char *format, ...)
463			{
464			va_list vl;
465
466			assert(mb);
467			assert(format);
468			assert(0 == mb->real);
469
470			assert(2 < mb->length);
471			assert(bufsize(mb) >= mb->length);
472
473			mb->length -= 2; /* back up to before \r\n */
474
475			va_start(vl, format); /* append to the buffer */
476
477			mb->length += xvsnprintf(dest, mb->msg + mb->length,
478			bufsize(mb) - mb->length - 1, format, vl);
479
480			va_end(vl);
481
482			if (mb->length > bufsize(mb) - 2)
483			mb->length = bufsize(mb) - 2;
484
485			mb->msg[mb->length++] = '\r'; /* add \r\n to buffer */
486			mb->msg[mb->length++] = '\n';
487			mb->msg[mb->length] = '\0'; /* not strictly necessary */
488
489			assert(mb->length <= bufsize(mb));
490			}
491
492			/** Decrement the reference count on \a mb, freeing it if needed.
493			* @param[in] mb MsgBuf to release.
494			*/
495			void
496			msgq_clean(struct MsgBuf *mb)
497			{
498			assert(mb);
499			assert(0 < mb->ref);
500
501			if (!--mb->ref)
502			{
503			/* deallocate the message */
504			if (mb->prev_p)
505			{
506			mb->prev_p = mb->next; / clip it out of active MsgBuf's list */
507
508			if (mb->next)
509			mb->next->prev_p = mb->prev_p;
510			}
511
512			if (mb->real && mb->real != mb) /* clean up the real buffer */
513			msgq_clean(mb->real);
514
515			mb->next = MQData.msgBufs[mb->power - MB_BASE_SHIFT].free;
516			MQData.msgBufs[mb->power - MB_BASE_SHIFT].free = mb;
517			MQData.msgBufs[mb->power - MB_BASE_SHIFT].used--;
518
519			mb->prev_p = 0;
520			}
521			}
522
523			/** Append a message to a peer's message queue.
524			* @param[in] mq Message queue to append to.
525			* @param[in] mb Message to append.
526			* @param[in] prio If non-zero, use the high-priority (lag-busting) message list; else use the normal list.
527			*/
528			void
529			msgq_add(struct MsgQ mq, struct MsgBuf mb, int prio)
530			{
531			struct MsgQList *qlist;
532			struct Msg *msg;
533
534			assert(mq);
535			assert(mb);
536			assert(0 < mb->ref);
537			assert(0 < mb->length);
538
539			ilog(LOG_TYPE_DEBUG, "Adding buffer %p [%.*s] length %u to %s queue", mb,
540			mb->length - 2, mb->msg, mb->length, prio ? "priority" : "normal");
541
542			qlist = prio ? &mq->prio : &mq->queue;
543
544			if (!(msg = MQData.msgs.free))
545			{
546			/* do I need to allocate one? */
547			msg = MyMalloc(sizeof(struct Msg));
548			MQData.msgs.alloc++; /* we allocated another */
549			}
550			else /* shift the free list */
551			MQData.msgs.free = MQData.msgs.free->next;
552
553			MQData.msgs.used++; /* we're using another */
554
555			msg->next = 0; /* initialize the msg */
556			msg->sent = 0;
557
558			/* Get the real buffer, allocating one if necessary */
559			if (!mb->real)
560			{
561			struct MsgBuf *tmp;
562
563			MQData.sizes.msgs++; /* update histogram counts */
564			MQData.sizes.sizes[mb->length - 1]++;
565
566			tmp = msgq_alloc(mb, mb->length); /* allocate a close-fitting buffer */
567
568			if (tmp != mb)
569			{
570			/* OK, prepare the new "real" buffer */
571			ilog(LOG_TYPE_DEBUG, "Copying old buffer %p [%.*s] length %u into new "
572			"buffer %p size %u", mb, mb->length - 2, mb->msg, mb->length,
573			tmp, bufsize(tmp));
574			memcpy(tmp->msg, mb->msg, mb->length + 1); /* copy string over */
575			tmp->length = mb->length;
576
577			tmp->next = mb->next; /* replace it in the list, now */
578
579			if (tmp->next)
580			tmp->next->prev_p = &tmp->next;
581			tmp->prev_p = mb->prev_p;
582			*tmp->prev_p = tmp;
583
584			mb->next = 0; /* this one's no longer in the list */
585			mb->prev_p = 0;
586			}
587			}
588
589			mb = mb->real; /* work with the real buffer */
590			mb->ref++; /* increment the ref count on the buffer */
591
592			msg->msg = mb; /* point at the real message buffer now */
593
594			if (!qlist->head) /* queue list was empty; head and tail point to msg */
595			qlist->head = qlist->tail = msg;
596			else
597			{
598			assert(qlist->tail);
599
600			qlist->tail->next = msg; /* queue had something in it; add to end */
601			qlist->tail = msg;
602			}
603
604			mq->length += mb->length; /* update the queue length */
605			mq->count++; /* and the queue count */
606			}
607
608			/** Report memory statistics for message buffers.
609			* @param[in] cptr Client requesting information.
610			* @param[out] msg_alloc Receives number of bytes allocated in Msg structs.
611			* @param[out] msgbuf_alloc Receives number of bytes allocated in MsgBuf structs.
612			*/
613			void
614			msgq_count_memory(struct Client source_p, size_t msg_alloc, size_t *msgbuf_alloc)
615			{
616			int i;
617			size_t total = 0, size;
618
619			assert(client_p);
620			assert(msg_alloc);
621			assert(msgbuf_alloc);
622
623			/* Data for Msg's is simple, so just send it */
624			sendto_one(source_p, ":%s %d %s :Msgs allocated %d(%zu) used %d(%zu) text %zu",
625			me.name, RPL_STATSDEBUG, source_p->name,
626			MQData.msgs.alloc, MQData.msgs.alloc * sizeof(struct Msg),
627			MQData.msgs.used, MQData.msgs.used * sizeof(struct Msg),
628			MQData.tot_bufsize);
629			/* count_memory() wants to know the total */
630			msg_alloc = MQData.msgs.alloc sizeof(struct Msg);
631
632			/* Ok, now walk through each size class */
633			for (i = MB_BASE_SHIFT; i < MB_MAX_SHIFT + 1; i++)
634			{
635			size = sizeof(struct MsgBuf) + (1 << i); /* total size of a buffer */
636
637			/* Send information for this buffer size class */
638			sendto_one(source_p, ":%s %d %s :MsgBufs of size %zu allocated %d(%zu) used %d(%zu)",
639			me.name, RPL_STATSDEBUG, source_p->name, 1 << i,
640			MQData.msgBufs[i - MB_BASE_SHIFT].alloc,
641			MQData.msgBufs[i - MB_BASE_SHIFT].alloc * size,
642			MQData.msgBufs[i - MB_BASE_SHIFT].used,
643			MQData.msgBufs[i - MB_BASE_SHIFT].used * size);
644
645			/* count_memory() wants to know the total */
646			total += MQData.msgBufs[i - MB_BASE_SHIFT].alloc * size;
647			}
648
649			*msgbuf_alloc = total;
650			}
651
652			/** Report remaining space in a MsgBuf.
653			* @param[in] mb Message buffer to check.
654			* @return Number of additional bytes that can be appended to the message.
655			*/
656			unsigned int
657			msgq_bufleft(struct MsgBuf *mb)
658			{
659			assert(mb);
660
661			return bufsize(mb) - mb->length; /* \r\n counted in mb->length */
662			}
663
664			/** Send histogram of message lengths to a client.
665			* @param[in] cptr Client requesting statistics.
666			* @param[in] sd Stats descriptor for request (ignored).
667			* @param[in] param Extra parameter from user (ignored).
668			*/
669			void
670			msgq_histogram(struct Client *source_p)
671			{
672			struct MsgSizes tmp = MQData.sizes; /* All hail structure copy! */
673			int i;
674
675			sendto_one(source_p, ":%s %d %s :Histogram of message lengths (%lu messages)",
676			me.name, RPL_STATSDEBUG, source_p->name, tmp.msgs);
677
678			for (i = 0; i + 16 <= IRCD_BUFSIZE; i += 16)
679			sendto_one(source_p, ":%s %d %s :% 4d: %u %u %u %u "
680			"%u %u %u %u %u %u %u %u %u %u %u %u",
681			me.name, RPL_STATSDEBUG, source_p->name, i + 1,
682			tmp.sizes[i + 0], tmp.sizes[i + 1], tmp.sizes[i + 2],
683			tmp.sizes[i + 3], tmp.sizes[i + 4], tmp.sizes[i + 5],
684			tmp.sizes[i + 6], tmp.sizes[i + 7], tmp.sizes[i + 8],
685			tmp.sizes[i + 9], tmp.sizes[i + 10], tmp.sizes[i + 11],
686			tmp.sizes[i + 12], tmp.sizes[i + 13], tmp.sizes[i + 14],
687			tmp.sizes[i + 15]);
688			}