~drizzle-trunk/drizzle/development

/*

  azio is a modified version of gzio. It  makes use of mysys and removes mallocs.

    -Brian Aker

*/

/* gzio.c -- IO on .gz files

 * Copyright (C) 1995-2005 Jean-loup Gailly.

 * For conditions of distribution and use, see copyright notice in zlib.h

 *

 */

/* @(#) $Id$ */

#include "config.h"

#include "azio.h"

#include <fcntl.h>

#include <unistd.h>

#include <cstdio>

#include <cstring>

#include <cstdlib>

#include <cassert>

using namespace drizzled;

static int const az_magic[3] = {0xfe, 0x03, 0x01}; /* az magic header */

static unsigned int azwrite(azio_stream *s, void *buf, unsigned int len);

static int azrewind (azio_stream *s);

static unsigned int azio_enable_aio(azio_stream *s);

static int do_flush(azio_stream *file, int flush);

static int    get_byte(azio_stream *s);

static void   check_header(azio_stream *s);

static int write_header(azio_stream *s);

static int    destroy(azio_stream *s);

static void putLong(azio_stream *s, uLong x);

static uLong  getLong(azio_stream *s);

static void read_header(azio_stream *s, unsigned char *buffer);

static void get_block(azio_stream *s);

#ifdef AZIO_AIO

static void do_aio_cleanup(azio_stream *s);

#endif

extern "C" pthread_handler_t run_task(void *p);

extern "C" pthread_handler_t run_task(void *p)

{

  int fd;

  char *buffer;

  size_t offset;

  azio_stream *s= (azio_stream *)p;

  while (1)

  {

    pthread_mutex_lock(&s->container.thresh_mutex);

    while (s->container.ready == AZ_THREAD_FINISHED)

    {

      pthread_cond_wait(&s->container.threshhold, &s->container.thresh_mutex);

    }

    offset= s->container.offset;

    fd= s->container.fd;

    buffer= (char *)s->container.buffer;

    pthread_mutex_unlock(&s->container.thresh_mutex);

    if (s->container.ready == AZ_THREAD_DEAD)

      break;

    s->container.read_size= pread((int)fd, (void *)buffer,

                                  (size_t)AZ_BUFSIZE_READ, (off_t)offset);

    pthread_mutex_lock(&s->container.thresh_mutex);

    s->container.ready= AZ_THREAD_FINISHED;

    pthread_mutex_unlock(&s->container.thresh_mutex);

  }

  return 0;

}

static void azio_kill(azio_stream *s)

{

  pthread_mutex_lock(&s->container.thresh_mutex);

  s->container.ready= AZ_THREAD_DEAD;

  pthread_mutex_unlock(&s->container.thresh_mutex);

  pthread_cond_signal(&s->container.threshhold);

  pthread_join(s->container.mainthread, NULL);

}

static size_t azio_return(azio_stream *s)

{

  return s->container.read_size;

}

/*

  Worried about spin?

  Don't be. In tests it never has spun more then 1 times.

*/

static az_thread_type azio_ready(azio_stream *s)

{

  az_thread_type temp;

  while (1)

  {

    pthread_mutex_lock(&s->container.thresh_mutex);

    temp= s->container.ready;

    pthread_mutex_unlock(&s->container.thresh_mutex);

    if (temp == AZ_THREAD_FINISHED || temp == AZ_THREAD_DEAD)

      break;

  }

  return temp;

}

static int azio_start(azio_stream *s)

{

  int rc= 0;

  pthread_attr_t attr;          /* Thread attributes */

  pthread_attr_init(&attr);

  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

  s->container.ready= AZ_THREAD_FINISHED;

  /* If we don't create a thread, signal the caller */

  if (pthread_create(&s->container.mainthread, &attr, run_task,

                     (void *)s) != 0)

    rc= 1;

  pthread_attr_destroy(&attr);

  return rc;

}

static int azio_read(azio_stream *s)

{

  pthread_mutex_lock(&s->container.thresh_mutex);

  s->container.ready= AZ_THREAD_ACTIVE;

  pthread_mutex_unlock(&s->container.thresh_mutex);

  pthread_cond_broadcast(&s->container.threshhold);

  return 0;

}

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

  Opens a gzip (.gz) file for reading or writing. The mode parameter

  is as in fopen ("rb" or "wb"). The file is given either by file descriptor

  or path name (if fd == -1).

  az_open returns NULL if the file could not be opened or if there was

  insufficient memory to allocate the (de)compression state; errno

  can be checked to distinguish the two cases (if errno is zero, the

  zlib error is Z_MEM_ERROR).

*/

int azopen(azio_stream *s, const char *path, int Flags, az_method method)

{

  int err;

  int level = Z_DEFAULT_COMPRESSION ; /* compression level */

  int strategy = Z_DEFAULT_STRATEGY; /* compression strategy */

  int fd= -1;

  memset(s, 0, sizeof(azio_stream));

  s->stream.zalloc = (alloc_func)0;

  s->stream.zfree = (free_func)0;

  s->stream.opaque = (voidpf)0;

  s->container.offset= 0;

  s->container.buffer= (void *)s->buffer1;

  s->container.ready= AZ_THREAD_FINISHED;

  s->inbuf= s->buffer1;

  s->stream.next_in = s->inbuf;

  s->stream.next_out = s->outbuf;

  s->z_err = Z_OK;

  s->back = EOF;

  s->crc = crc32(0L, Z_NULL, 0);

  s->mode = 'r';

  s->version = (unsigned char)az_magic[1]; /* this needs to be a define to version */

  s->version = (unsigned char)az_magic[2]; /* minor version */

  s->method= method;

  /*

    We do our own version of append by nature.

    We must always have write access to take card of the header.

  */

  assert(Flags | O_APPEND);

  assert(Flags | O_WRONLY);

  if (Flags & O_RDWR)

    s->mode = 'w';

  if (s->mode == 'w')

  {

    err = deflateInit2(&(s->stream), level,

                       Z_DEFLATED, -MAX_WBITS, 8, strategy);

    /* windowBits is passed < 0 to suppress zlib header */

    s->stream.next_out = s->outbuf;

    if (err != Z_OK)

    {

      destroy(s);

      return Z_NULL;

    }

  } else {

    /* Threads are only used when we are running with azio */

    s->stream.next_in  = s->inbuf;

    err = inflateInit2(&(s->stream), -MAX_WBITS);

    /* windowBits is passed < 0 to tell that there is no zlib header.

     * Note that in this case inflate *requires* an extra "dummy" byte

     * after the compressed stream in order to complete decompression and

     * return Z_STREAM_END. Here the gzip CRC32 ensures that 4 bytes are

     * present after the compressed stream.

   */

    if (err != Z_OK)

    {

      destroy(s);

      return Z_NULL;

    }

  }

  s->stream.avail_out = AZ_BUFSIZE_WRITE;

  errno = 0;

  s->file = fd < 0 ? internal::my_open(path, Flags, MYF(0)) : fd;

#ifdef AZIO_AIO

  s->container.fd= s->file;

#endif

  if (s->file < 0 )

  {

    destroy(s);

    return Z_NULL;

  }

  if (Flags & O_CREAT || Flags & O_TRUNC)

  {

    s->rows= 0;

    s->forced_flushes= 0;

    s->shortest_row= 0;

    s->longest_row= 0;

    s->auto_increment= 0;

    s->check_point= 0;

    s->comment_start_pos= 0;

    s->comment_length= 0;

    s->frm_start_pos= 0;

    s->frm_length= 0;

    s->dirty= 1; /* We create the file dirty */

    s->start = AZHEADER_SIZE + AZMETA_BUFFER_SIZE;

    if(write_header(s))

      return Z_NULL;

    s->pos= (size_t)lseek(s->file, 0, SEEK_END);

  }

  else if (s->mode == 'w')

  {

    unsigned char buffer[AZHEADER_SIZE + AZMETA_BUFFER_SIZE];

    const ssize_t read_size= AZHEADER_SIZE + AZMETA_BUFFER_SIZE;

    if(pread(s->file, buffer, read_size, 0) < read_size)

      return Z_NULL;

    read_header(s, buffer);

    s->pos= (size_t)lseek(s->file, 0, SEEK_END);

  }

  else

  {

    check_header(s); /* skip the .az header */

  }

  switch (s->method)

  {

  case AZ_METHOD_AIO:

    azio_enable_aio(s);

    break;

  case AZ_METHOD_BLOCK:

  case AZ_METHOD_MAX:

    break;

  }

  return 1;

}

int write_header(azio_stream *s)

{

  char buffer[AZHEADER_SIZE + AZMETA_BUFFER_SIZE];

  char *ptr= buffer;

  s->block_size= AZ_BUFSIZE_WRITE;

  s->version = (unsigned char)az_magic[1];

  s->minor_version = (unsigned char)az_magic[2];

  /* Write a very simple .az header: */

  memset(buffer, 0, AZHEADER_SIZE + AZMETA_BUFFER_SIZE);

  *(ptr + AZ_MAGIC_POS)= az_magic[0];

  *(ptr + AZ_VERSION_POS)= (unsigned char)s->version;

  *(ptr + AZ_MINOR_VERSION_POS)= (unsigned char)s->minor_version;

  *(ptr + AZ_BLOCK_POS)= (unsigned char)(s->block_size/1024); /* Reserved for block size */

  *(ptr + AZ_STRATEGY_POS)= (unsigned char)Z_DEFAULT_STRATEGY; /* Compression Type */

  int4store(ptr + AZ_FRM_POS, s->frm_start_pos); /* FRM Block */

  int4store(ptr + AZ_FRM_LENGTH_POS, s->frm_length); /* FRM Block */

  int4store(ptr + AZ_COMMENT_POS, s->comment_start_pos); /* COMMENT Block */

  int4store(ptr + AZ_COMMENT_LENGTH_POS, s->comment_length); /* COMMENT Block */

  int4store(ptr + AZ_META_POS, 0); /* Meta Block */

  int4store(ptr + AZ_META_LENGTH_POS, 0); /* Meta Block */

  int8store(ptr + AZ_START_POS, (uint64_t)s->start); /* Start of Data Block Index Block */

  int8store(ptr + AZ_ROW_POS, (uint64_t)s->rows); /* Start of Data Block Index Block */

  int8store(ptr + AZ_FLUSH_POS, (uint64_t)s->forced_flushes); /* Start of Data Block Index Block */

  int8store(ptr + AZ_CHECK_POS, (uint64_t)s->check_point); /* Start of Data Block Index Block */

  int8store(ptr + AZ_AUTOINCREMENT_POS, (uint64_t)s->auto_increment); /* Start of Data Block Index Block */

  int4store(ptr+ AZ_LONGEST_POS , s->longest_row); /* Longest row */

  int4store(ptr+ AZ_SHORTEST_POS, s->shortest_row); /* Shorest row */

  int4store(ptr+ AZ_FRM_POS,

            AZHEADER_SIZE + AZMETA_BUFFER_SIZE); /* FRM position */

  *(ptr + AZ_DIRTY_POS)= (unsigned char)s->dirty; /* Start of Data Block Index Block */

  /* Always begin at the begining, and end there as well */

  const ssize_t write_size= AZHEADER_SIZE + AZMETA_BUFFER_SIZE;

  if(pwrite(s->file, (unsigned char*) buffer, write_size, 0)!=write_size)

    return -1;

  return 0;

}

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

  Read a byte from a azio_stream; update next_in and avail_in. Return EOF

  for end of file.

  IN assertion: the stream s has been sucessfully opened for reading.

*/

int get_byte(azio_stream *s)

{

  if (s->z_eof) return EOF;

  if (s->stream.avail_in == 0)

  {

    errno = 0;

    if (s->stream.avail_in == 0)

    {

      s->z_eof = 1;

      return EOF;

    }

    else if (s->stream.avail_in == (uInt) -1)

    {

      s->z_eof= 1;

      s->z_err= Z_ERRNO;

      return EOF;

    }

    s->stream.next_in = s->inbuf;

  }

  s->stream.avail_in--;

  return *(s->stream.next_in)++;

}

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

  Check the gzip header of a azio_stream opened for reading.

  IN assertion: the stream s has already been created sucessfully;

  s->stream.avail_in is zero for the first time, but may be non-zero

  for concatenated .gz files.

*/

void check_header(azio_stream *s)

{

  uInt len;

  /* Assure two bytes in the buffer so we can peek ahead -- handle case

    where first byte of header is at the end of the buffer after the last

    gzip segment */

  len = s->stream.avail_in;

  if (len < 2) {

    if (len) s->inbuf[0] = s->stream.next_in[0];

    errno = 0;

    len = (uInt)pread(s->file, (unsigned char *)s->inbuf + len, AZ_BUFSIZE_READ >> len, s->pos);

    s->pos+= len;

    if (len == (uInt)-1) s->z_err = Z_ERRNO;

    s->stream.avail_in += len;

    s->stream.next_in = s->inbuf;

  }

  /* Now we check the actual header */

  if ( s->stream.next_in[0] == az_magic[0]  && s->stream.next_in[1] == az_magic[1])

  {

    unsigned char buffer[AZHEADER_SIZE + AZMETA_BUFFER_SIZE];

    for (len = 0; len < (AZHEADER_SIZE + AZMETA_BUFFER_SIZE); len++)

      buffer[len]= get_byte(s);

    s->z_err = s->z_eof ? Z_DATA_ERROR : Z_OK;

    read_header(s, buffer);

    for (; len < s->start; len++)

      get_byte(s);

  }

  else

  {

    s->z_err = Z_OK;

    return;

  }

}

void read_header(azio_stream *s, unsigned char *buffer)

{

  if (buffer[0] == az_magic[0]  && buffer[1] == az_magic[1])

  {

    s->version= (unsigned int)buffer[AZ_VERSION_POS];

    s->minor_version= (unsigned int)buffer[AZ_MINOR_VERSION_POS];

    s->block_size= 1024 * buffer[AZ_BLOCK_POS];

    s->start= (size_t)uint8korr(buffer + AZ_START_POS);

    s->rows= (uint64_t)uint8korr(buffer + AZ_ROW_POS);

    s->check_point= (uint64_t)uint8korr(buffer + AZ_CHECK_POS);

    s->forced_flushes= (uint64_t)uint8korr(buffer + AZ_FLUSH_POS);

    s->auto_increment= (uint64_t)uint8korr(buffer + AZ_AUTOINCREMENT_POS);

    s->longest_row= (unsigned int)uint4korr(buffer + AZ_LONGEST_POS);

    s->shortest_row= (unsigned int)uint4korr(buffer + AZ_SHORTEST_POS);

    s->frm_start_pos= (unsigned int)uint4korr(buffer + AZ_FRM_POS);

    s->frm_length= (unsigned int)uint4korr(buffer + AZ_FRM_LENGTH_POS);

    s->comment_start_pos= (unsigned int)uint4korr(buffer + AZ_COMMENT_POS);

    s->comment_length= (unsigned int)uint4korr(buffer + AZ_COMMENT_LENGTH_POS);

    s->dirty= (unsigned int)buffer[AZ_DIRTY_POS];

  }

  else

  {

    s->z_err = Z_OK;

    return;

  }

}

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

 * Cleanup then free the given azio_stream. Return a zlib error code.

 Try freeing in the reverse order of allocations.

 */

int destroy (azio_stream *s)

{

  int err = Z_OK;

  if (s->stream.state != NULL)

  {

    if (s->mode == 'w')

    {

      err = deflateEnd(&(s->stream));

      internal::my_sync(s->file, MYF(0));

    }

    else if (s->mode == 'r')

      err = inflateEnd(&(s->stream));

  }

  do_aio_cleanup(s);

  if (s->file > 0 && internal::my_close(s->file, MYF(0)))

      err = Z_ERRNO;

  s->file= -1;

  if (s->z_err < 0) err = s->z_err;

  return err;

}

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

  Reads the given number of uncompressed bytes from the compressed file.

  azread returns the number of bytes actually read (0 for end of file).

*/

/*

   This function is legacy, do not use.

     Reads the given number of uncompressed bytes from the compressed file.

   If the input file was not in gzip format, gzread copies the given number

   of bytes into the buffer.

     gzread returns the number of uncompressed bytes actually read (0 for

   end of file, -1 for error).

*/

static unsigned int azread_internal( azio_stream *s, voidp buf, unsigned int len, int *error)

{

  Bytef *start = (Bytef*)buf; /* starting point for crc computation */

  Byte  *next_out; /* == stream.next_out but not forced far (for MSDOS) */

  *error= 0;

  if (s->mode != 'r')

  {

    *error= Z_STREAM_ERROR;

    return 0;

  }

  if (s->z_err == Z_DATA_ERROR || s->z_err == Z_ERRNO)

  {

    *error= s->z_err;

    return 0;

  }

  if (s->z_err == Z_STREAM_END)  /* EOF */

  {

    return 0;

  }

  next_out = (Byte*)buf;

  s->stream.next_out = (Bytef*)buf;

  s->stream.avail_out = len;

  if (s->stream.avail_out && s->back != EOF) {

    *next_out++ = s->back;

    s->stream.next_out++;

    s->stream.avail_out--;

    s->back = EOF;

    s->out++;

    start++;

    if (s->last) {

      s->z_err = Z_STREAM_END;

      {

        return 1;

      }

    }

  }

  while (s->stream.avail_out != 0) {

    if (s->stream.avail_in == 0 && !s->z_eof) {

      errno = 0;

      get_block(s);

      if (s->stream.avail_in == 0)

      {

        s->z_eof = 1;

      }

      s->stream.next_in = (Bytef *)s->inbuf;

    }

    s->in += s->stream.avail_in;

    s->out += s->stream.avail_out;

    s->z_err = inflate(&(s->stream), Z_NO_FLUSH);

    s->in -= s->stream.avail_in;

    s->out -= s->stream.avail_out;

    if (s->z_err == Z_STREAM_END) {

      /* Check CRC and original size */

      s->crc = crc32(s->crc, start, (uInt)(s->stream.next_out - start));

      start = s->stream.next_out;

      if (getLong(s) != s->crc) {

        s->z_err = Z_DATA_ERROR;

      } else {

        (void)getLong(s);

        /* The uncompressed length returned by above getlong() may be

         * different from s->out in case of concatenated .gz files.

         * Check for such files:

       */

        check_header(s);

        if (s->z_err == Z_OK)

        {

          inflateReset(&(s->stream));

          s->crc = crc32(0L, Z_NULL, 0);

        }

      }

    }

    if (s->z_err != Z_OK || s->z_eof) break;

  }

  s->crc = crc32(s->crc, start, (uInt)(s->stream.next_out - start));

  if (len == s->stream.avail_out &&

      (s->z_err == Z_DATA_ERROR || s->z_err == Z_ERRNO))

  {

    *error= s->z_err;

    return 0;

  }

  return (len - s->stream.avail_out);

}

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

  Experimental Interface. We abstract out a concecpt of rows

*/

size_t azwrite_row(azio_stream *s, void *buf, unsigned int len)

{

  size_t length;

  /* First we write length */

  length= azwrite(s, &len, sizeof(unsigned int));

  if (length != sizeof(unsigned int))

    return length;

  /* Now we write the actual data */

  length= (size_t)azwrite(s, buf, len);

  if (length > 0)

    s->rows++;

  if (len > s->longest_row)

    s->longest_row= len;

  if (len < s->shortest_row || !(s->shortest_row))

    s->shortest_row= len;

  return length;

}

size_t azread_row(azio_stream *s, int *error)

{

  unsigned int row_length; /* Currently we are limited to this size for rows */

  char buffer[sizeof(unsigned int)];

  char *new_ptr;

  size_t read;

  read= azread_internal(s, buffer, sizeof(unsigned int), error);

  /* On error the return value will be zero as well */

  if (read == 0)

    return read;

  memcpy(&row_length, buffer, sizeof(unsigned int));

  new_ptr= (char *)realloc(s->row_ptr, (sizeof(char) * row_length));

  if (!new_ptr)

    return SIZE_MAX;

  s->row_ptr= new_ptr;

  /* TODO We should now adjust the length... */

  read= azread_internal(s, (voidp)s->row_ptr, row_length, error);

  return read;

}

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

  Writes the given number of uncompressed bytes into the compressed file.

  azwrite returns the number of bytes actually written (0 in case of error).

*/

static unsigned int azwrite(azio_stream *s, void *buf, unsigned int len)

{

  s->stream.next_in = (Bytef*)buf;

  s->stream.avail_in = len;

  while (s->stream.avail_in != 0)

  {

    if (s->stream.avail_out == 0)

    {

      s->stream.next_out = s->outbuf;

      if (pwrite(s->file, (unsigned char *)s->outbuf, AZ_BUFSIZE_WRITE, s->pos) != AZ_BUFSIZE_WRITE)

      {

        s->z_err = Z_ERRNO;

        break;

      }

      s->pos+= AZ_BUFSIZE_WRITE;

      s->stream.avail_out = AZ_BUFSIZE_WRITE;

    }

    s->in += s->stream.avail_in;

    s->out += s->stream.avail_out;

    s->z_err = deflate(&(s->stream), Z_NO_FLUSH);

    s->in -= s->stream.avail_in;

    s->out -= s->stream.avail_out;

    if (s->z_err != Z_OK) break;

  }

  s->crc = crc32(s->crc, (const Bytef *)buf, len);

  return (unsigned int)(len - s->stream.avail_in);

}

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

  Flushes all pending output into the compressed file. The parameter

  flush is as in the deflate() function.

*/

int do_flush (azio_stream *s, int flush)

{

  uInt len;

  int done = 0;

  size_t afterwrite_pos;

  if (s == NULL || s->mode != 'w') return Z_STREAM_ERROR;

  s->stream.avail_in = 0; /* should be zero already anyway */