xdasd_mcache.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_mcache.h"
#include "xdasd_parse.h"
#include "xdasd_log.h"

#include "xdas_base.h"
#include "xdas_wire.h"

#ifdef _WIN32

# define WIN32_LEAN_AND_MEAN
# include <windows.h>
# include <direct.h>
# include <io.h>
# include <fcntl.h>
# include <sys/types.h>
# include <sys/stat.h>
# define mkdir(x,y)        _mkdir(x)
# define open              _open
# define lseek             _lseek
# define read              _read
# define write             _write
# define close             _close
# define unlink            _unlink
# define MC_FILE_ATTRS     (_S_IREAD | _S_IWRITE)
# define MC_OPEN_FLAGS     (_O_RDWR | _O_BINARY)
# define MC_CREATE_FLAGS   (_O_RDWR | _O_CREAT | _O_BINARY)

#else /* UNIX */

# ifdef HAVE_CONFIG_H
#  include "config.h"
# endif
# include <unistd.h>
# include <sys/types.h>
# include <sys/stat.h>
# include <fcntl.h>
# include <dirent.h>
# if HAVE_INTTYPES_H
#  include <inttypes.h>
# else
#  if HAVE_STDINT_H
#   include <stdint.h>
#  endif
# endif
# define MC_FILE_ATTRS     (0644)
# define MC_DIR_ATTRS      (0744)
# define MC_OPEN_FLAGS     (O_RDWR)
# define MC_CREATE_FLAGS   (O_RDWR | O_CREAT)
#endif

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

/* The maximum size of a single message cache file (1 Mb). 
   The test versions are smaller so we don't have to write so 
   much to test file overflow, etc. 
 */
#ifdef XDASD_MCACHE_TEST
# define MCACHE_MAX_FILESZ  (1024 + 512)
# define MCACHE_MAX_RECSZ   (512 + 256)
#else
# define MCACHE_MAX_FILESZ  (1024 * 1024)
# define MCACHE_MAX_RECSZ   (64 * 1024)
#endif

#define MCACHE_BASE_FNAME  "xdasd.cache"

/* The mcache file header array index values. */
#define MCH_BOF   0
#define MCH_EOF   1
#define MCH_LIM   2

typedef struct mcfile_
{
   int fh;                    
   unsigned usecnt;           
   unsigned fileno;           
   uint32_t header[3];        
   char fname[FILENAME_MAX];  
} mcfile_t;

typedef struct mcache_
{
   unsigned lofno;            
   unsigned hifno;            
   mcfile_t * ifp;            
   mcfile_t * ofp;            
   char mcdir[FILENAME_MAX];  
} mcache_t;

#ifdef _WIN32

/* An implementation of opendir/readdir for Win32 */

#define dirent _finddata_t
#define d_name name

typedef struct DIR_tag
{
   intptr_t handle;
   struct _finddata_t fileinfo;
   char filespec[FILENAME_MAX];
} DIR;

static DIR * opendir(const char * name)
{
   DIR * dir;
   size_t dirlen;
   if ((dir = (DIR *)malloc(sizeof(*dir))) == 0)
      return 0;
   memset(dir, 0, sizeof(*dir));
   dir->handle = -1;
   strncpy(dir->filespec, name, sizeof(dir->filespec));
   dir->filespec[sizeof(dir->filespec) - 1] = 0;
   dirlen = strlen(dir->filespec);
   if (dirlen > 0 && (dir->filespec[dirlen - 1] == '\\' 
         || dir->filespec[dirlen - 1] == '/'))
      dir->filespec[--dirlen] = 0;
   strcpy(dir->filespec + dirlen, "\\*");
   return dir;
}

static struct dirent * readdir(DIR * dir)
{
   if ((dir->handle == -1 && (dir->handle = 
            _findfirst(dir->filespec, &dir->fileinfo)) == -1)
         || _findnext(dir->handle, &dir->fileinfo) != 0)
      return 0;
   return &dir->fileinfo;
}

static int closedir(DIR * dir)
{
   int rc = _findclose(dir->handle);
   free(dir);
   return rc;
}

#endif

static void mcache_calc_file_range(const char * mcdir, 
      unsigned * lowp, unsigned * highp)
{
   DIR * dp;
   unsigned low = (unsigned)(-1);
   unsigned high = 0;

   /* attempt to open message cache file directory */
   if ((dp = opendir(mcdir)) != 0)
   {
      struct dirent * de;

      /* read each entry, scan for high and low file numbers */
      while ((de = readdir(dp)) != 0)
      {
         size_t namelen = strlen(de->d_name);
         if (namelen > 12 && strncmp("xdasd.cache.", de->d_name, 12) == 0)
         {
            unsigned fcval = atoi(de->d_name + 12);
            if (fcval != 0)   /* skip anything wierd... */
            {
               if (fcval < low)
                  low = fcval;
               if (fcval > high)
                  high = fcval;
            }
         }
      }
      closedir(dp);
   }

   /* return low and high values, or defaults (1), if not found. */
   *lowp = low < (unsigned)(-1)? low: 1;
   *highp = high > 0? high: 1;
}

static void mcache_close(mcfile_t * fp, int ferase)
{
   if (fp != 0 && --fp->usecnt == 0)
   {
      close(fp->fh);
      if (ferase) 
         unlink(fp->fname);
      free(fp);
   }
}

static mcfile_t * mcache_dup(mcfile_t * fp)
{
   if (fp)
      fp->usecnt++;
   return fp;
}

static int mcache_read_file_header(mcfile_t * fp)
{
   char header[sizeof(fp->header)];
   char * cp = header;

   if (read(fp->fh, header, sizeof(header)) != sizeof(header))
      return -1;

   fp->header[MCH_BOF] = xdas_get_uint32(&cp);
   fp->header[MCH_EOF] = xdas_get_uint32(&cp);
   fp->header[MCH_LIM] = xdas_get_uint32(&cp);

   if (fp->header[MCH_LIM] > MCACHE_MAX_FILESZ
         || fp->header[MCH_EOF] > fp->header[MCH_LIM]
         || fp->header[MCH_BOF] > fp->header[MCH_EOF])
      return -1;

   return 0;
}

static int mcache_write_file_header(mcfile_t * fp)
{
   char header[sizeof(fp->header)];
   char * cp = header;

   xdas_put_uint32(&cp, fp->header[MCH_BOF]);
   xdas_put_uint32(&cp, fp->header[MCH_EOF]);
   xdas_put_uint32(&cp, fp->header[MCH_LIM]);

   if (lseek(fp->fh, 0, SEEK_SET) == (off_t)(-1)
         || write(fp->fh, header, sizeof(header)) != sizeof(header))
      return -1;

   return 0;
}

static int mcache_open(const char * mcdir, unsigned fileno, mcfile_t ** fpp)
{
   mcfile_t * fp;

   /* allocate and populate a new mcfile object */
   if ((fp = (mcfile_t *)malloc(sizeof(*fp))) == 0)
      return -1;
   memset(fp, 0, sizeof(*fp));
   fp->usecnt++;
   fp->fileno = fileno;

   /* build the file name, open file */
   sprintf(fp->fname, "%s/" MCACHE_BASE_FNAME ".%d", mcdir, fileno);
   if ((fp->fh = open(fp->fname, MC_OPEN_FLAGS)) < 0)
   {
      free(fp);
      return -1;
   }

   /* read header and do some sanity checks */
   if (mcache_read_file_header(fp) < 0)
   {
      mcache_close(fp, 0); /* close and erase the file */
      return -1;
   }
   *fpp = fp;
   return 0;
}

static int mcache_create(const char * mcdir, unsigned fileno, mcfile_t ** fpp)
{
   mcfile_t * fp;
   
   mkdir(mcdir, MC_DIR_ATTRS);  /* ensure cache directory exists */  

   /* allocate and populate a new mcfile object */
   if ((fp = (mcfile_t *)malloc(sizeof(*fp))) == 0)
      return -1;
   memset(fp, 0, sizeof(*fp));
   fp->usecnt++;
   fp->fileno = fileno;

   /* build the file name, create file and fill in and write file header */
   sprintf(fp->fname, "%s/" MCACHE_BASE_FNAME ".%d", mcdir, fileno);
   if ((fp->fh = open(fp->fname, MC_CREATE_FLAGS, MC_FILE_ATTRS)) < 0)
   {
      free(fp);
      return -1;
   }

   fp->header[MCH_BOF] = sizeof(fp->header);
   fp->header[MCH_EOF] = sizeof(fp->header);
   fp->header[MCH_LIM] = MCACHE_MAX_FILESZ;
   if (mcache_write_file_header(fp) < 0)
   {
      mcache_close(fp, 1); /* close and erase the file */
      return -1;
   }

   /* flush */

   *fpp = fp;     /* return mcfile object pointer */
   return 0;
}

static int mcache_write_record(mcfile_t * fp, Parsed * parsed, size_t msgsz)
{
   char buf[2 * sizeof(uint32_t) + (XDAS_FIELD_COUNT + 1) * sizeof(uint16_t)];
   char * cp = buf;
   int i;

   /* Actual message size is record size minus overhead:
         uint32_t recsz (size of remaining fields)
         uint32_t flags
         uint16_t offsets[XDAS_FIELD_COUNT + 1]
         char msg[msgsz] 
    */
   xdas_put_uint32(&cp, sizeof(buf) - sizeof(uint32_t) + msgsz);
   xdas_put_uint32(&cp, parsed->flags & PMF_IMPORTED);
   for (i = 0; i < XDAS_FIELD_COUNT + 1; i++)
      xdas_put_uint16(&cp, (uint16_t)(parsed->parsed[i] - parsed->msg));

   /* seek and write the entire record to the current end of file */
   if (lseek(fp->fh, fp->header[MCH_EOF], SEEK_SET) == (off_t)(-1)
         || write(fp->fh, buf, sizeof(buf)) != sizeof(buf)
         || write(fp->fh, parsed->msg, (unsigned)msgsz) != (int)msgsz)
      return -1;

   /* update the end of file and write the new file header */
   fp->header[MCH_EOF] += (uint32_t)(sizeof(buf) + msgsz);
   if (mcache_write_file_header(fp) < 0)
      return -1;

   /* flush */

   return 0;
}

static Parsed * mcache_read_record(mcfile_t * fp)
{
   size_t recsz;
   uint32_t u32recsz;
   Parsed * parsed = 0;

   if (lseek(fp->fh, fp->header[MCH_BOF], SEEK_SET) != (off_t)(-1)
         && read(fp->fh, &u32recsz, sizeof(u32recsz)) == sizeof(u32recsz)
         && (recsz = AS_UINT32(&u32recsz)) < MCACHE_MAX_RECSZ)
   {
      char buf[sizeof(uint32_t) + (XDAS_FIELD_COUNT + 1) * sizeof(uint16_t)];
      size_t msgsz = recsz - sizeof(buf);
      size_t allocsz = offsetof(Parsed, msg) + msgsz;
      char * cp = buf;
      int i;

      /* Actual message size is record size minus overhead:
            uint32_t recsz (size of remaining fields)
            uint32_t flags
            uint16_t offsets[XDAS_FIELD_COUNT + 1]
            char msg[msgsz] 
       */
      if ((parsed = (Parsed *)malloc(allocsz)) == 0
            || read(fp->fh, buf, sizeof(buf)) != sizeof(buf)
            || read(fp->fh, parsed->msg, (unsigned)msgsz) != (int)msgsz)
      {
         free(parsed);
         return 0;
      }

      /* extract overhead and parse offsets - convert offsets to pointers */
      memset(parsed, 0, offsetof(Parsed, parsed));
      parsed->structsz = allocsz;
      parsed->flags = xdas_get_uint32(&cp);
      for (i = 0; i < XDAS_FIELD_COUNT + 1; i++)
      {
         parsed->parsed[i] = parsed->msg + xdas_get_uint16(&cp);
         if (parsed->parsed[i] < parsed->msg
               || parsed->parsed[i] > parsed->msg + msgsz)
         {
            free(parsed);
            return 0;
         }
      }
   }
   return parsed;
}

static int mcache_update_bof(mcfile_t * fp)
{
   uint32_t u32msgsz;

   /* read message size at current BOF position */
   if (fp->header[MCH_BOF] >= fp->header[MCH_EOF]
         || lseek(fp->fh, fp->header[MCH_BOF], SEEK_SET) == (off_t)(-1)
         || read(fp->fh, &u32msgsz, sizeof(u32msgsz)) != sizeof(u32msgsz))
      return -1;
   
   /* move BOF ahead one message, write new header */
   fp->header[MCH_BOF] += sizeof(u32msgsz) + AS_UINT32(&u32msgsz);
   if (mcache_write_file_header(fp) < 0)
      return -1;

   /* flush */

   return 0;
}

static int mcache_put_msg(mcache_t * mcp, Parsed * parsed)
{
   unsigned nextfno = mcp->hifno;
   size_t msgsz = parsed->parsed[XDAS_FIELD_COUNT] - parsed->parsed[0];
   size_t recsz = 2 * sizeof(uint32_t) 
         + (XDAS_FIELD_COUNT + 1) * sizeof(uint16_t) + msgsz;

   /* Actual record is:
         uint32_t recsz (size of remaining fields)
         uint32_t flags
         uint16_t offsets[XDAS_FIELD_COUNT + 1]
         char msg[msgsz] 
    */
   do
   {
      if (mcp->ifp != 0)
      {
         /* if this message is too big for current infile, close it */
         if (mcp->ifp->header[MCH_EOF] + recsz > mcp->ifp->header[MCH_LIM])
         {
            nextfno = mcp->ifp->fileno + 1;
            mcache_close(mcp->ifp, 0);       /* don't erase it */
            mcp->ifp = 0;
         }
      }

      /* open the next file, if there's not one open right now. */
      if (mcp->ifp == 0)
      {
         if (mcp->ofp && nextfno == mcp->ofp->fileno)
            mcp->ifp = mcache_dup(mcp->ofp);
         else if (mcache_open(mcp->mcdir, nextfno, &mcp->ifp) < 0
               && mcache_create(mcp->mcdir, nextfno, &mcp->ifp) < 0)
         {
            xdasd_log(0, "MCACHE: Unable to open message cache for write.\n");
            return -1;
         }
      }
   } while (mcp->ifp->header[MCH_EOF] + recsz > mcp->ifp->header[MCH_LIM]);

   /* write message to cache */
   if (mcache_write_record(mcp->ifp, parsed, msgsz) < 0)
   {
      xdasd_log(0, "MCACHE: Unable to write to message cache.\n");
      return -1;
   }
   return 0;
}

static Parsed * mcache_get_msg(mcache_t * mcp)
{
   unsigned nextfno = mcp->lofno;   /* init to lowest file number */

   /* if outfile is already open, move BOF pointer ahead one message */
   if (mcp->ofp != 0)
   {
      /* move to next message in current file */
      if (mcache_update_bof(mcp->ofp) < 0)
         return 0;

      /* if at end of file, close and erase this file */
      if (mcp->ofp->header[MCH_BOF] >= mcp->ofp->header[MCH_EOF])
      {
         /* if cache is empty, then close input file too -
            the next put_msg will reopen the starting input file */
         if (mcp->ifp != 0 && mcp->ofp->fileno == mcp->ifp->fileno)
         {
            mcp->lofno = mcp->hifno = 1;
            mcache_close(mcp->ifp, 0);
            mcp->ifp = 0;
         }

         /* close output file */
         nextfno = mcp->ofp->fileno + 1;
         mcache_close(mcp->ofp, 1);
         mcp->ofp = 0;
      }
   }

   /* if file is closed, open the next one if there is one */
   if (mcp->ofp == 0)
   {
      /* attempt to dup/open the next message file */
      if (mcp->ifp && nextfno == mcp->ifp->fileno)
         mcp->ofp = mcache_dup(mcp->ifp);
      else if (mcache_open(mcp->mcdir, nextfno, &mcp->ofp) < 0)
         return 0;
   }

   /* read and return the current parsed message */
   return mcache_read_record(mcp->ofp);
}

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

int xdasd_mcache_put_msg(MsgCache mc, Parsed * parsed)
{
   return mcache_put_msg((mcache_t *)mc, parsed);
}

Parsed * xdasd_mcache_get_msg(MsgCache mc)
{
   return mcache_get_msg((mcache_t *)mc);
}

MsgCache xdasd_mcache_create(const char * msgfdir)
{
   mcache_t * mcp;

   if ((mcp = (mcache_t *)malloc(sizeof(*mcp))) == 0)
       return 0;

   /* store msg file directory name, zero file object pointers */
   strncpy(mcp->mcdir, msgfdir, sizeof(mcp->mcdir));
   mcp->mcdir[sizeof(mcp->mcdir) - 1] = 0;
   mcp->ifp = mcp->ofp = 0;

   /* calculate the startup file range found in the mcache directory */
   mcache_calc_file_range(mcp->mcdir, &mcp->lofno, &mcp->hifno);

   return (MsgCache)mcp;
}

void xdasd_mcache_destroy(MsgCache mc)
{
   mcache_t * mcp = (mcache_t *)mc;

   /* close all cache files */
   mcache_close(mcp->ifp, 0);
   mcache_close(mcp->ofp, 0);
   free(mcp);
}

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

#ifdef XDASD_MCACHE_TEST

#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>

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

int xdasd_log_level(void) { return 0; }

int main(int argc, char ** argv)
{

#define TESTMSG1 \
   "HDR:00A8:OX1:1CB1B348:::time.nist.gov:MST7MDT:0x10000001:0:ORG:" \
   "originator_location_name:originator_location_address:" \
   "originator_service_type:originator_authentication_authority:" \
   "originator_principal_name:originator_principal_id:" \
   "INT:initiator_auth_authority:initiator_domain_name:initiator_domain_id:" \
   "TGT:target_location_name:target_location_address:target_service_type:" \
   "target_auth_authority:target_principal_name:target_principal_id:" \
   "SRC:pointer_to_source_domain:EVT:event_specific_info:END"

#define TESTMSG2 \
   "HDR:00B8:OX1:1FB7AC54:::time.ms.com:MST5MDT:0x10000002:0:ORG:" \
   "org_location_name:org_location_address:org_service_type:" \
   "org_auth_authority:org_principal_name:org_principal_id:INT:" \
   "int_auth_authority:int_domain_specific_name:int_domain_specific_id:" \
   "TGT:tgt_location_name:tgt_location_address:tgt_service_type:" \
   "tgt_auth_authority:tgt_principal_name:tgt_principal_id:" \
   "SRC:pointer_to_source_domain:EVT:event_specific_information:END"

   static struct 
   {
      char * msg;
      size_t msgsz;
   } msgs[] = 
   {
      {TESTMSG1, sizeof(TESTMSG1) - 1},
      {TESTMSG2, sizeof(TESTMSG2) - 1},
      {TESTMSG2, sizeof(TESTMSG2) - 1},
      {TESTMSG1, sizeof(TESTMSG1) - 1},
      {TESTMSG2, sizeof(TESTMSG2) - 1},
      {TESTMSG1, sizeof(TESTMSG1) - 1},
      {TESTMSG1, sizeof(TESTMSG1) - 1},
   };

   int i, j;
   DIR * dir;
   MsgCache mc;
   Parsed * parsed;
   struct dirent * de;
   char * cache_path = argc > 1? argv[1]: ".";

   /* start clean by removing all cache files from the cache directory */
   if ((dir = opendir(cache_path)) != 0)
   {
      while ((de = readdir(dir)) != 0)
         if (strncmp(de->d_name, MCACHE_BASE_FNAME, 
               sizeof(MCACHE_BASE_FNAME) - 1) == 0)
         {
            char fpath[FILENAME_MAX];
            sprintf(fpath, "%s\\%s", cache_path, de->d_name);
            unlink(fpath);
         }
      closedir(dir);
   }

   /* initialize the cache */
   mc = xdasd_mcache_create(cache_path);

   /* write half of the messages out to the cache */
   for (i = 0; i < (sizeof(msgs)/sizeof(*msgs)) / 2; i++)
   {
      int err;

      if ((err = xdasd_parse_message(PMF_IMPORTED, 
            msgs[i].msgsz, msgs[i].msg, &parsed)) != 0)
      {
         xdasd_log(0, "%s: error %d at: %s(%d), loop %d.\n", 
               argv[0], err, __FILE__, __LINE__, i);
         exit(1);
      }
      if ((err = xdasd_mcache_put_msg(mc, parsed)) != 0)
      {
         xdasd_log(0, "%s: error %d at: %s(%d), loop %d.\n", 
               argv[0], err, __FILE__, __LINE__, i);
         exit(1);
      }
      xdasd_parse_free(parsed);
   }

   /* simulate a crash */
   xdasd_mcache_destroy(mc);

   /* simulate restart with data in the cache */
   mc = xdasd_mcache_create(cache_path);

   /* retrieve two messages to see if they come out right */
   for (j = 0; j < 2; j++)
   {
      parsed = xdasd_mcache_get_msg(mc);
      if (parsed == 0 || parsed->parsed[XDAS_FIELD_COUNT] 
               - parsed->parsed[0] - 1 != (int)msgs[j].msgsz
            || strncmp(parsed->msg, msgs[j].msg, msgs[j].msgsz) != 0)
      {
         xdasd_log(0, "%s: corrupt cache detected at %s(%d), loop %d.\n", 
               argv[0], __FILE__, __LINE__, j);
         exit(1);                                  \
      }
      xdasd_parse_free(parsed);
   }

   /* simulate a crash */
   xdasd_mcache_destroy(mc);

   /* simulate restart with data in the cache */
   mc = xdasd_mcache_create(cache_path);

   /* write remaining elements to the cache */
   for ( ; i < sizeof(msgs)/sizeof(*msgs); i++)
   {
      int err;

      if ((err = xdasd_parse_message(PMF_IMPORTED, 
            msgs[i].msgsz, msgs[i].msg, &parsed)) != 0)
      {
         xdasd_log(0, "%s: error %d at: %s(%d), loop %d.\n", 
               argv[0], err, __FILE__, __LINE__, i);
         exit(1);
      }
      if ((err = xdasd_mcache_put_msg(mc, parsed)) != 0)
      {
         xdasd_log(0, "%s: error %d at: %s(%d), loop %d.\n", 
               argv[0], err, __FILE__, __LINE__, i);
         exit(1);
      }
      xdasd_parse_free(parsed);
   }

   /* simulate a crash */
   xdasd_mcache_destroy(mc);

   /* simulate restart with data in the cache */
   mc = xdasd_mcache_create(cache_path);

   /* retrieve all remaining messages to see if they come out right

      NOTE: Because of the (simulated) crash, we were not able to update
      the cache pointers to the next output message, so we will retrieve 
      that same output message again after re-initialization - thus we
      start by decrementing j so we are comparing with the correct entry 
      in the table on the first pass. 
    */
   for (--j; j < sizeof(msgs)/sizeof(*msgs); j++)
   {
      parsed = xdasd_mcache_get_msg(mc);
      if (parsed == 0 || parsed->parsed[XDAS_FIELD_COUNT] 
               - parsed->parsed[0] - 1 != (int)msgs[j].msgsz 
            || strncmp(parsed->msg, msgs[j].msg, msgs[j].msgsz) != 0)
      {
         xdasd_log(0, "%s: corrupt cache detected at %s(%d), loop %d.\n", 
               argv[0], __FILE__, __LINE__, j);
         exit(1);                                  \
      }
      xdasd_parse_free(parsed);
   }

   /* We have to read past the end in order to update the out pointer
      in the cache and cause the last file to be deleted - but it had
      better not return data! */
   if ((parsed = xdasd_mcache_get_msg(mc)) != 0)
   {
      xdasd_log(0, "%s: read past end of cache detected at %s(%d).\n", 
            argv[0], __FILE__, __LINE__);
      exit(1);                                  \
   }

   /* cleanup */
   xdasd_mcache_destroy(mc);

   /* See if any cache files still exist (they shouldn't) */
   if ((dir = opendir(cache_path)) != 0)
   {
      while ((de = readdir(dir)) != 0)
         if (strncmp(de->d_name, MCACHE_BASE_FNAME, 
               sizeof(MCACHE_BASE_FNAME) - 1) == 0)
         {
            xdasd_log(0, "%s: file %s in cache didn't get removed.\n", 
                  argv[0], de->d_name);
            exit(1);
         }
      closedir(dir);
   }
   return 0;
}

#endif   /* XDASD_MCACHE_TEST */