xdasd_mqueue.c

Go to the documentation of this file.

/*----------------------------------------------------------------------------
 * Copyright (c) 2006, Novell, Inc.
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without 
 * modification, are permitted provided that the following conditions are 
 * met:
 * 
 *     * Redistributions of source code must retain the above copyright 
 *       notice, this list of conditions and the following disclaimer.
 *     * 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.
 *     * Neither the name of the Novell nor the names of its contributors 
 *       may be used to endorse or promote products derived from this 
 *       software without specific prior written permission.
 * 
 * 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.
 *--------------------------------------------------------------------------*/

#include "xdasd_mqueue.h"
#include "xdasd_thread.h"

#include <stdio.h>
#include <string.h>
#include <malloc.h>

typedef struct mqueue_tag
{
   int shutting_down;         
   thread_t bgproc_th;        
   mutex_t queue_mutex;       
   cv_t queue_cv;             
   MsgPut_t * msg_put;        
   MsgGet_t * msg_get;        
   MsgProc_t * msg_process;   
} mqueue_t;

static void * bg_proc(void * data)
{
   mqueue_t * mqp = (mqueue_t *)data;
   int last_chance = 0;

   mutex_acquire(&mqp->queue_mutex);

   do 
   {
      void * msg;
      while ((msg = mqp->msg_get()) != 0)
      {
         mutex_release(&mqp->queue_mutex);

         mqp->msg_process(msg);     /* process msg OUTSIDE the lock */

         mutex_acquire(&mqp->queue_mutex);
      }
      if (!mqp->shutting_down)
         cv_wait(&mqp->queue_cv, &mqp->queue_mutex);
   } while (!mqp->shutting_down || !last_chance++);

   mutex_release(&mqp->queue_mutex);

   return 0;
}

/*--------------------------------------------------------------------------*/

int xdasd_mqueue_append(MsgQueue mq, void * msg)
{
   int err;
   mqueue_t * mqp = (mqueue_t *)mq;

   mutex_acquire(&mqp->queue_mutex);

   if ((err = mqp->msg_put(msg)) == 0)
      cv_signal(&mqp->queue_cv);

   mutex_release(&mqp->queue_mutex);

   return err;
}

MsgQueue xdasd_mqueue_create(MsgPut_t * put, MsgGet_t * get, MsgProc_t * proc)
{
   mqueue_t * mqp;

   /* allocate queue object */
   if ((mqp = (mqueue_t *)malloc(sizeof(*mqp))) != 0)
   {
      memset(mqp, 0, sizeof(*mqp));

      /* store user-provided function pointers */
      mqp->msg_put = put;
      mqp->msg_get = get;
      mqp->msg_process = proc;

      /* create the queue cv */
      if (cv_create(&mqp->queue_cv) != 0)
      {
         free(mqp);
         return 0;
      }
   
      /* create the queue mutex */
      mutex_create(&mqp->queue_mutex);
   
      /* create background thread */
      if ((mqp->bgproc_th = thread_create(bg_proc, mqp)) == 0)
      {
         mutex_destroy(&mqp->queue_mutex);
         cv_destroy(&mqp->queue_cv);
         free(mqp);
         return 0;
      }
   }
   return (MsgQueue)mqp;
}

void xdasd_mqueue_destroy(MsgQueue mq)
{
   mqueue_t * mqp = (mqueue_t *)mq;

   /* signal background thread to shutdown */
   mutex_acquire(&mqp->queue_mutex);
   mqp->shutting_down = 1;
   cv_signal(&mqp->queue_cv);
   mutex_release(&mqp->queue_mutex);

   /* wait for background logger thread to terminate */
   (void)thread_wait(mqp->bgproc_th);

   /* destroy sync resources */
   cv_destroy(&mqp->queue_cv);
   mutex_destroy(&mqp->queue_mutex);
   free(mqp);
}

/*==========================================================================*/

#ifdef XDASD_MQUEUE_TEST

#include <stdarg.h>

static void output(const char * fmt, ... ) 
{
#ifdef LOG_TO_STDOUT
   va_list args;
   va_start(args, fmt);
   vprintf(fmt, args);
   va_end(args);
#else
   (void)fmt;
#endif
}

#define TESTDATA (void*)(-1)

static int msgque = 0;
static int processed = 0;

static int testput(void * data) 
{
   if (data == TESTDATA)
      ++msgque;
   return 0;
}
static void * testget(void) 
{
   if (msgque == 0)
      return 0;
   msgque--;
   return TESTDATA;
}
static void testproc(void * data) 
{
   if (data == TESTDATA)
      processed++;
}

int main(void)
{
#define LOOP_COUNT 50000

   MsgQueue mq;
   int rv = 0;
   int err = 0;

   if ((mq = xdasd_mqueue_create(testput, testget, testproc)) != 0)
   {
      int i;
      for (i = 0; i < LOOP_COUNT; i++)
         if ((err = xdasd_mqueue_append(mq, TESTDATA)) != 0)
            break;
      xdasd_mqueue_destroy(mq);
   }
   if (!mq) rv++;
   if (err) rv++;
   if (msgque) rv++;
   if (processed != LOOP_COUNT) rv++;
   if (rv)
      output("Unit test error count: %d.\n", rv);
   return rv;
}

#endif