Logo Search packages:      
Sourcecode: lbzip2 version File versions

lbzip2.c

/* lbzip2.c,v 1.13 2009-01-28 01:53:48 lacos Exp */

#include <inttypes.h>     /* uint64_t */
#include <assert.h>       /* assert() */
#include <unistd.h>       /* read() */
#include <errno.h>        /* errno */
#include <stdlib.h>       /* EXIT_FAILURE */
#include <stdio.h>        /* fprintf() */
#include <bzlib.h>        /* bz_stream */

#include "main.h"         /* pname */
#include "lbzip2.h"       /* lbzip2() */
#include "lacos_rbtree.h" /* struct lacos_rbtree_node */


struct s2w_blk                   /* Splitter to workers. */
{
  uint64_t id;                   /* Block serial number as read from stdin. */
  struct s2w_blk *next;          /* Next in queue. */
  size_t loaded;                 /* # of bytes in plain, may be 0 for 1st. */
  char unsigned plain[MX_PLAIN]; /* Data read from stdin. */
};


struct s2w_q
{
  struct cond av_or_eof; /* New block available or splitter done. */
  struct s2w_blk *tail,  /* Splitter will append here. */
      *head;             /* Next ready worker shall compress this. */
  int eof;               /* Splitter done. */
};


static void
s2w_q_init(struct s2w_q *s2w_q)
{
  xinit(&s2w_q->av_or_eof);
  s2w_q->tail = 0;
  s2w_q->head = 0;
  s2w_q->eof = 0;
}


static void
s2w_q_uninit(struct s2w_q *s2w_q)
{
  assert(0 != s2w_q->eof);
  assert(0 == s2w_q->head);
  assert(0 == s2w_q->tail);
  xdestroy(&s2w_q->av_or_eof);
}


struct w2m_blk                   /* Workers to muxer. */
{
  uint64_t id;                   /* Block index as read from stdin. */
  struct w2m_blk *next;          /* Next block in list (unordered). */
  size_t produced;               /* Number of bytes in compr. */
  char unsigned compr[MX_BZIP2]; /* Data to write to stdout. */
};


static int
w2m_blk_cmp(const void *v_a, const void *v_b)
{
  uint64_t a,
      b;

  a = ((const struct w2m_blk *)v_a)->id;
  b = ((const struct w2m_blk *)v_b)->id;

  return
        a < b ? -1
      : a > b ?  1
      : 0;
}


struct w2m_q
{
  struct cond av_or_exit; /* New block available or all workers exited. */
  uint64_t needed;        /* Block needed for resuming writing. */
  struct w2m_blk *head;   /* Block list (unordered). */
  unsigned working;       /* Number of workers still running. */
};


static void
w2m_q_init(struct w2m_q *w2m_q, unsigned num_worker)
{
  assert(0u < num_worker);
  xinit(&w2m_q->av_or_exit);
  w2m_q->needed = 0u;
  w2m_q->head = 0;
  w2m_q->working = num_worker;
}


static void
w2m_q_uninit(struct w2m_q *w2m_q)
{
  assert(0u == w2m_q->working);
  assert(0 == w2m_q->head);
  xdestroy(&w2m_q->av_or_exit);
}


struct m2s_q         /* Muxer to splitter. */
{
  struct cond av;    /* Free slot available. */
  unsigned num_free; /* Number of free slots. */
};


static void
m2s_q_init(struct m2s_q *m2s_q, unsigned num_free)
{
  assert(0u < num_free);
  xinit(&m2s_q->av);
  m2s_q->num_free = num_free;
}


static void
m2s_q_uninit(struct m2s_q *m2s_q, unsigned num_free)
{
  assert(m2s_q->num_free == num_free);
  xdestroy(&m2s_q->av);
}


static void
split(struct m2s_q *m2s_q, struct s2w_q *s2w_q)
{
  uint64_t id;
  ssize_t rd;

  id = 0u;
  do {
    struct s2w_blk *s2w_blk;
    size_t vacant;

    /* Grab a free slot. */
    xlock_pred(&m2s_q->av);
    while (0u == m2s_q->num_free) {
      xwait(&m2s_q->av);
    }
    --m2s_q->num_free;
    xunlock(&m2s_q->av);
    s2w_blk = xalloc(sizeof *s2w_blk);

    /* Fill block. */
    vacant = sizeof s2w_blk->plain;
    do {
      rd = read(STDIN_FILENO, s2w_blk->plain + (sizeof s2w_blk->plain
          - vacant), vacant > (size_t)SSIZE_MAX ? (size_t)SSIZE_MAX : vacant);
    } while (0 < rd && 0u < (vacant -= (size_t)rd));

    /* Read error. */
    if (-1 == rd) {
      fail("read()", errno);
    }

    if (sizeof s2w_blk->plain == vacant && 0u < id) {
      /* EOF on first read, but not for first input block. */
      (*freef)(s2w_blk);
      xlock(&m2s_q->av);
      ++m2s_q->num_free;
      xunlock(&m2s_q->av);
    }
    else {
      s2w_blk->id = id;
      s2w_blk->next = 0;
      s2w_blk->loaded = sizeof s2w_blk->plain - vacant;
    }

    /* We either push a block, or set EOF, or both. */
    assert(sizeof s2w_blk->plain > vacant || 0 == rd);

    xlock(&s2w_q->av_or_eof);
    if (0 == s2w_q->head) {
      xbroadcast(&s2w_q->av_or_eof);
    }

    if (sizeof s2w_blk->plain > vacant || 0u == id) {
      if (0 == s2w_q->tail) {
        s2w_q->head = s2w_blk;
      }
      else {
        s2w_q->tail->next = s2w_blk;
      }
      s2w_q->tail = s2w_blk;
    }
    s2w_q->eof = (0 == rd);
    xunlock(&s2w_q->av_or_eof);

    /*
      If we didn't push a block, then this is bogus, but then we did set EOF,
      so it doesn't matter, because we'll leave immediately.
    */
    ++id;
  } while (0 < rd);
}


struct split_arg
{
  struct m2s_q *m2s_q;
  struct s2w_q *s2w_q;
};


static void *
split_wrap(void *v_split_arg)
{
  struct split_arg *split_arg;

  split_arg = v_split_arg;

  split(split_arg->m2s_q, split_arg->s2w_q);
  return 0;
}


static void
work_compr(struct s2w_blk *s2w_blk, struct w2m_q *w2m_q)
{
  struct w2m_blk *w2m_blk;
  bz_stream strm;
  int bzret;

  assert(0u < s2w_blk->loaded || 0u == s2w_blk->id);

  w2m_blk = xalloc(sizeof *w2m_blk);

  strm.next_in = (char *)s2w_blk->plain;
  strm.avail_in = s2w_blk->loaded;
  strm.next_out = (char *)w2m_blk->compr;
  strm.avail_out = sizeof w2m_blk->compr;
  strm.bzalloc = lbzallocf;
  strm.bzfree = lbzfreef;
  strm.opaque = 0;

  bzret = BZ2_bzCompressInit(
      &strm,
      9, /* blockSize100k */
      0, /* verbosity */
      0  /* workFactor */
  );
  assert(BZ_MEM_ERROR == bzret || BZ_OK == bzret);

  if (BZ_MEM_ERROR == bzret) {
    (void)fprintf(stderr, "%s: BZ2_bzCompressInit(): BZ_MEM_ERROR\n", pname);
    _exit(EXIT_FAILURE);
  }

  bzret = BZ2_bzCompress(&strm, BZ_FINISH);
  assert(BZ_STREAM_END == bzret);

  w2m_blk->produced = sizeof w2m_blk->compr - strm.avail_out;

  bzret = BZ2_bzCompressEnd(&strm);
  assert(bzret == BZ_OK);

  w2m_blk->id = s2w_blk->id;

  /* Push block to muxer. */
  xlock(&w2m_q->av_or_exit);
  w2m_blk->next = w2m_q->head;
  w2m_q->head = w2m_blk;
  if (w2m_blk->id == w2m_q->needed) {
    xsignal(&w2m_q->av_or_exit);
  }
  xunlock(&w2m_q->av_or_exit);
}


static void
work(struct s2w_q *s2w_q, struct w2m_q *w2m_q)
{
  for (;;) {
    struct s2w_blk *s2w_blk;

    /* Grab a block to work on. */
    xlock_pred(&s2w_q->av_or_eof);
    while (0 == s2w_q->head && !s2w_q->eof) {
      xwait(&s2w_q->av_or_eof);
    }
    if (0 == s2w_q->head) {
      /* No blocks available and splitter exited. */
      xunlock(&s2w_q->av_or_eof);
      break;
    }
    s2w_blk = s2w_q->head;
    s2w_q->head = s2w_blk->next;
    if (0 == s2w_q->head) {
      s2w_q->tail = 0;
    }
    xunlock(&s2w_q->av_or_eof);

    work_compr(s2w_blk, w2m_q);
    (*freef)(s2w_blk);
  }

  /* Notify muxer when last worker exits. */
  xlock(&w2m_q->av_or_exit);
  if (0u == --w2m_q->working && 0 == w2m_q->head) {
    xsignal(&w2m_q->av_or_exit);
  }
  xunlock(&w2m_q->av_or_exit);
}


struct work_arg
{
  struct s2w_q *s2w_q;
  struct w2m_q *w2m_q;
};


static void *
work_wrap(void *v_work_arg)
{
  struct work_arg *work_arg;

  work_arg = v_work_arg;
  work(work_arg->s2w_q, work_arg->w2m_q);
  return 0;
}


static void *
reord_alloc(size_t size, void *ignored)
{
  return (*mallocf)(size);
}


static void
reord_dealloc(void *ptr, void *ignored)
{
  (*freef)(ptr);
}


static void
mux_write(struct m2s_q *m2s_q, struct lacos_rbtree_node **reord,
    uint64_t *reord_needed)
{
  assert(0 != *reord);

  /*
    Go on until the tree becomes empty or the next block is found to be
    missing.
  */
  do {
    struct lacos_rbtree_node *reord_head;
    struct w2m_blk *reord_w2m_blk;

    reord_head = lacos_rbtree_min(*reord);
    assert(0 != reord_head);

    reord_w2m_blk = *(void **)reord_head;
    if (reord_w2m_blk->id != *reord_needed) {
      break;
    }

    /* Write out "reord_w2m_blk". */
    {
      char unsigned *cp;

      cp = reord_w2m_blk->compr;
      while (reord_w2m_blk->produced > 0u) {
        ssize_t written;

        written = write(STDOUT_FILENO, cp, reord_w2m_blk->produced
            > (size_t)SSIZE_MAX ? (size_t)SSIZE_MAX : reord_w2m_blk->produced);
        if (-1 == written) {
          fail("write()", errno);
        }

        reord_w2m_blk->produced -= (size_t)written;
        cp += written;
      }
    }

    ++*reord_needed;

    xlock(&m2s_q->av);
    if (0u == m2s_q->num_free++) {
      xsignal(&m2s_q->av);
    }
    xunlock(&m2s_q->av);

    lacos_rbtree_delete(
        reord,         /* new_root */
        reord_head,    /* old_node */
        0,             /* old_data */
        reord_dealloc, /* dealloc() */
        0              /* alloc_ctl */
    );

    /* Release "reord_w2m_blk". */
    (*freef)(reord_w2m_blk);
  } while (0 != *reord);
}


static void
mux(struct w2m_q *w2m_q, struct m2s_q *m2s_q)
{
  struct lacos_rbtree_node *reord;
  uint64_t reord_needed;

  reord = 0;
  reord_needed = 0u;

  xlock_pred(&w2m_q->av_or_exit);
  for (;;) {
    struct w2m_blk *w2m_blk;

    /* Grab all available compressed blocks in one step. */
    while (0 == w2m_q->head && 0u < w2m_q->working) {
      xwait(&w2m_q->av_or_exit);
    }

    if (0 == w2m_q->head) {
      /* w2m_q is empty and all workers exited */
      break;
    }

    w2m_blk = w2m_q->head;
    w2m_q->head = 0;
    xunlock(&w2m_q->av_or_exit);

    /* Merge blocks fetched this time into tree. */
    do {
      struct lacos_rbtree_node *new_node;
      struct w2m_blk *next;

      if (-1 == lacos_rbtree_insert(
          &reord,      /* new_root */
          &new_node,   /* new_node */
          w2m_blk,     /* new_data */
          w2m_blk_cmp, /* cmp() */
          reord_alloc, /* alloc() */
          0            /* alloc_ctl */
      )) {
        /* id collision shouldn't happen */
        assert(0 == new_node);
        (void)fprintf(stderr, "%s: lacos_rbtree_insert(): out of memory\n",
            pname);
        _exit(EXIT_FAILURE);
      }

      next = w2m_blk->next;
      w2m_blk->next = 0;
      w2m_blk = next;
    } while (0 != w2m_blk);

    /* Write out initial continuous sequence of reordered blocks. */
    mux_write(m2s_q, &reord, &reord_needed);

    xlock_pred(&w2m_q->av_or_exit);
    w2m_q->needed = reord_needed;
  }
  xunlock(&w2m_q->av_or_exit);

  assert(0 == reord);
}


void
lbzip2(unsigned num_worker, unsigned num_slot, int print_cctrs)
{
  struct s2w_q s2w_q;
  struct w2m_q w2m_q;
  struct m2s_q m2s_q;
  struct split_arg split_arg;
  pthread_t splitter;
  struct work_arg work_arg;
  pthread_t *worker;
  unsigned i;

  s2w_q_init(&s2w_q);
  w2m_q_init(&w2m_q, num_worker);
  m2s_q_init(&m2s_q, num_slot);

  split_arg.m2s_q = &m2s_q;
  split_arg.s2w_q = &s2w_q;
  xcreate(&splitter, split_wrap, &split_arg);

  work_arg.s2w_q = &s2w_q;
  work_arg.w2m_q = &w2m_q;

  assert(0u < num_worker);
  assert((size_t)-1 / sizeof *worker >= num_worker);
  worker = xalloc(num_worker * sizeof *worker);
  for (i = 0u; i < num_worker; ++i) {
    xcreate(&worker[i], work_wrap, &work_arg);
  }

  mux(&w2m_q, &m2s_q);

  i = num_worker;
  do {
    xjoin(worker[--i]);
  } while (0u < i);
  (*freef)(worker);

  xjoin(splitter);

  if (print_cctrs && 0 > fprintf(stderr,
    /*
      I have to replace this elegant construct with the lame one below, because
      the Tru64 system I tested on chokes on the %N$*M$lu conversion
      specification, even though it is certified UNIX 98 (I believe):

      $ uname -s -r -v -m
      OSF1 V5.1 2650 alpha
      $ c89 -V
      Compaq C V6.5-011 on Compaq Tru64 UNIX V5.1B (Rev. 2650)
      Compiler Driver V6.5-003 (sys) cc Driver

      http://www.opengroup.org/openbrand/register/brand2700.htm

      And no, I won't factor this out.
      28-JAN-2009 lacos
    */

#if 0
      "%1$s: any worker tried to consume from splitter: %3$*2$lu\n"
      "%1$s: any worker stalled                       : %4$*2$lu\n"
      "%1$s: muxer tried to consume from workers      : %5$*2$lu\n"
      "%1$s: muxer stalled                            : %6$*2$lu\n"
      "%1$s: splitter tried to consume from muxer     : %7$*2$lu\n"
      "%1$s: splitter stalled                         : %8$*2$lu\n",
      pname, (int)sizeof(long unsigned) * (int)CHAR_BIT / 3 + 1,
      s2w_q.av_or_eof.ccount, s2w_q.av_or_eof.wcount,
      w2m_q.av_or_exit.ccount, w2m_q.av_or_exit.wcount,
      m2s_q.av.ccount, m2s_q.av.wcount)
#else
#  define FW ((int)sizeof(long unsigned) * (int)CHAR_BIT / 3 + 1)
      "%s: any worker tried to consume from splitter: %*lu\n"
      "%s: any worker stalled                       : %*lu\n"
      "%s: muxer tried to consume from workers      : %*lu\n"
      "%s: muxer stalled                            : %*lu\n"
      "%s: splitter tried to consume from muxer     : %*lu\n"
      "%s: splitter stalled                         : %*lu\n",
      pname, FW, s2w_q.av_or_eof.ccount,
      pname, FW, s2w_q.av_or_eof.wcount,
      pname, FW, w2m_q.av_or_exit.ccount,
      pname, FW, w2m_q.av_or_exit.wcount,
      pname, FW, m2s_q.av.ccount,
      pname, FW, m2s_q.av.wcount)
#  undef FW
#endif
  ) {
    _exit(EXIT_FAILURE);
  }

  m2s_q_uninit(&m2s_q, num_slot);
  w2m_q_uninit(&w2m_q);
  s2w_q_uninit(&s2w_q);
}

Generated by  Doxygen 1.6.0   Back to index