src/casefile.c

   1 /* PSPP - computes sample statistics.
   2    Copyright (C) 2004 Free Software Foundation, Inc.
   3    Written by Ben Pfaff <blp@gnu.org>.
   4
   5    This program is free software; you can redistribute it and/or
   6    modify it under the terms of the GNU General Public License as
   7    published by the Free Software Foundation; either version 2 of the
   8    License, or (at your option) any later version.
   9
  10    This program is distributed in the hope that it will be useful, but
  11    WITHOUT ANY WARRANTY; without even the implied warranty of
  12    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13    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., 51 Franklin Street, Fifth Floor, Boston, MA
  18    02110-1301, USA. */
  19
  20 #include <config.h>
  21 #include "casefile.h"
  22 #include <assert.h>
  23 #include <errno.h>
  24 #include <fcntl.h>
  25 #include <stdio.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28 #include <unistd.h>
  29 #include "alloc.h"
  30 #include "case.h"
  31 #include "error.h"
  32 #include "full-read.h"
  33 #include "full-write.h"
  34 #include "misc.h"
  35 #include "mkfile.h"
  36 #include "settings.h"
  37 #include "var.h"
  38
  39 #include "gettext.h"
  40 #define _(msgid) gettext (msgid)
  41
  42 #define IO_BUF_SIZE (8192 / sizeof (union value))
  43
  44 /* A casefile represents a sequentially accessible stream of
  45    immutable cases.
  46
  47    If workspace allows, a casefile is maintained in memory.  If
  48    workspace overflows, then the casefile is pushed to disk.  In
  49    either case the interface presented to callers is kept the
  50    same.
  51
  52    The life cycle of a casefile consists of up to three phases:
  53
  54        1. Writing.  The casefile initially contains no cases.  In
  55           this phase, any number of cases may be appended to the
  56           end of a casefile.  (Cases are never inserted in the
  57           middle or before the beginning of a casefile.)
  58
  59           Use casefile_append() or casefile_append_xfer() to
  60           append a case to a casefile.
  61
  62        2. Reading.  The casefile may be read sequentially,
  63           starting from the beginning, by "casereaders".  Any
  64           number of casereaders may be created, at any time,
  65           during the reading phase.  Each casereader has an
  66           independent position in the casefile.
  67
  68           Casereaders may only move forward.  They cannot move
  69           backward to arbitrary records or seek randomly.
  70           Cloning casereaders is possible, but it is not yet
  71           implemented.
  72
  73           Use casefile_get_reader() to create a casereader for
  74           use in phase 2.  This also transitions from phase 1 to
  75           phase 2.  Calling casefile_mode_reader() makes the same
  76           transition, without creating a casereader.
  77
  78           Use casereader_read(), casereader_read_xfer(), or
  79           casereader_read_xfer_assert() to read a case from a
  80           casereader.  Use casereader_destroy() to discard a
  81           casereader when it is no longer needed.
  82
  83        3. Destruction.  This phase is optional.  The casefile is
  84           also read with casereaders in this phase, but the
  85           ability to create new casereaders is curtailed.
  86
  87           In this phase, casereaders could still be cloned (once
  88           we eventually implement cloning).
  89
  90           To transition from phase 1 or 2 to phase 3 and create a
  91           casereader, call casefile_get_destructive_reader().
  92           The same functions apply to the casereader obtained
  93           this way as apply to casereaders obtained in phase 2.
  94
  95           After casefile_get_destructive_reader() is called, no
  96           more casereaders may be created with
  97           casefile_get_reader() or
  98           casefile_get_destructive_reader().  (If cloning of
  99           casereaders were implemented, it would still be
 100           possible.)
 101
 102           The purpose of the limitations applied to casereaders
 103           in phase 3 is to allow in-memory casefiles to fully
 104           transfer ownership of cases to the casereaders,
 105           avoiding the need for extra copies of case data.  For
 106           relatively static data sets with many variables, I
 107           suspect (without evidence) that this may be a big
 108           performance boost.
 109
 110    When a casefile is no longer needed, it may be destroyed with
 111    casefile_destroy().  This function will also destroy any
 112    remaining casereaders. */
 113
 114 /* In-memory cases are arranged in an array of arrays.  The top
 115    level is variable size and the size of each bottom level array
 116    is fixed at the number of cases defined here.  */
 117 #define CASES_PER_BLOCK 128
 118
 119 /* A casefile. */
 120 struct casefile
 121   {
 122     /* Basic data. */
 123     struct casefile *next, *prev;       /* Next, prev in global list. */
 124     size_t value_cnt;                   /* Case size in `union value's. */
 125     size_t case_acct_size;              /* Case size for accounting. */
 126     unsigned long case_cnt;             /* Number of cases stored. */
 127     enum { MEMORY, DISK } storage;      /* Where cases are stored. */
 128     enum { WRITE, READ } mode;          /* Is writing or reading allowed? */
 129     struct casereader *readers;         /* List of our readers. */
 130     int being_destroyed;                /* Does a destructive reader exist? */
 131
 132     /* Memory storage. */
 133     struct ccase **cases;               /* Pointer to array of cases. */
 134
 135     /* Disk storage. */
 136     int fd;                             /* File descriptor, -1 if none. */
 137     char *filename;                     /* Filename. */
 138     union value *buffer;                /* I/O buffer, NULL if none. */
 139     size_t buffer_used;                 /* Number of values used in buffer. */
 140     size_t buffer_size;                 /* Buffer size in values. */
 141   };
 142
 143 /* For reading out the cases in a casefile. */
 144 struct casereader
 145   {
 146     struct casereader *next, *prev;     /* Next, prev in casefile's list. */
 147     struct casefile *cf;                /* Our casefile. */
 148     unsigned long case_idx;             /* Case number of current case. */
 149     int destructive;                    /* Is this a destructive reader? */
 150
 151     /* Disk storage. */
 152     int fd;                             /* File descriptor. */
 153     union value *buffer;                /* I/O buffer. */
 154     size_t buffer_pos;                  /* Offset of buffer position. */
 155     struct ccase c;                     /* Current case. */
 156   };
 157
 158 /* Return the case number of the current case */
 159 unsigned long
 160 casereader_cnum(const struct casereader *r)
 161 {
 162   return r->case_idx;
 163 }
 164
 165 /* Doubly linked list of all casefiles. */
 166 static struct casefile *casefiles;
 167
 168 /* Number of bytes of case allocated in in-memory casefiles. */
 169 static size_t case_bytes;
 170
 171 static void register_atexit (void);
 172 static void exit_handler (void);
 173
 174 static void reader_open_file (struct casereader *reader);
 175 static void write_case_to_disk (struct casefile *cf, const struct ccase *c);
 176 static void flush_buffer (struct casefile *cf);
 177 static void fill_buffer (struct casereader *reader);
 178
 179 static int safe_open (const char *filename, int flags);
 180 static int safe_close (int fd);
 181
 182 /* Creates and returns a casefile to store cases of VALUE_CNT
 183    `union value's each. */
 184 struct casefile *
 185 casefile_create (size_t value_cnt)
 186 {
 187   struct casefile *cf = xmalloc (sizeof *cf);
 188   cf->next = casefiles;
 189   cf->prev = NULL;
 190   if (cf->next != NULL)
 191     cf->next->prev = cf;
 192   casefiles = cf;
 193   cf->value_cnt = value_cnt;
 194   cf->case_acct_size = (cf->value_cnt + 4) * sizeof *cf->buffer;
 195   cf->case_cnt = 0;
 196   cf->storage = MEMORY;
 197   cf->mode = WRITE;
 198   cf->readers = NULL;
 199   cf->being_destroyed = 0;
 200   cf->cases = NULL;
 201   cf->fd = -1;
 202   cf->filename = NULL;
 203   cf->buffer = NULL;
 204   cf->buffer_size = ROUND_UP (cf->value_cnt, IO_BUF_SIZE);
 205   if (cf->value_cnt > 0 && cf->buffer_size % cf->value_cnt > 64)
 206     cf->buffer_size = cf->value_cnt;
 207   cf->buffer_used = 0;
 208   register_atexit ();
 209   return cf;
 210 }
 211
 212 /* Destroys casefile CF. */
 213 void
 214 casefile_destroy (struct casefile *cf)
 215 {
 216   if (cf != NULL)
 217     {
 218       if (cf->next != NULL)
 219         cf->next->prev = cf->prev;
 220       if (cf->prev != NULL)
 221         cf->prev->next = cf->next;
 222       if (casefiles == cf)
 223         casefiles = cf->next;
 224
 225       while (cf->readers != NULL)
 226         casereader_destroy (cf->readers);
 227
 228       if (cf->cases != NULL)
 229         {
 230           size_t idx, block_cnt;
 231
 232           case_bytes -= cf->case_cnt * cf->case_acct_size;
 233           for (idx = 0; idx < cf->case_cnt; idx++)
 234             {
 235               size_t block_idx = idx / CASES_PER_BLOCK;
 236               size_t case_idx = idx % CASES_PER_BLOCK;
 237               struct ccase *c = &cf->cases[block_idx][case_idx];
 238               case_destroy (c);
 239             }
 240
 241           block_cnt = DIV_RND_UP (cf->case_cnt, CASES_PER_BLOCK);
 242           for (idx = 0; idx < block_cnt; idx++)
 243             free (cf->cases[idx]);
 244
 245           free (cf->cases);
 246         }
 247
 248       if (cf->fd != -1)
 249         safe_close (cf->fd);
 250
 251       if (cf->filename != NULL && remove (cf->filename) == -1)
 252         msg (ME, _("%s: Removing temporary file: %s."),
 253              cf->filename, strerror (errno));
 254       free (cf->filename);
 255
 256       free (cf->buffer);
 257
 258       free (cf);
 259     }
 260 }
 261
 262 /* Returns nonzero only if casefile CF is stored in memory (instead of on
 263    disk). */
 264 int
 265 casefile_in_core (const struct casefile *cf)
 266 {
 267   assert (cf != NULL);
 268
 269   return cf->storage == MEMORY;
 270 }
 271
 272 /* Puts a casefile to "sleep", that is, minimizes the resources
 273    needed for it by closing its file descriptor and freeing its
 274    buffer.  This is useful if we need so many casefiles that we
 275    might not have enough memory and file descriptors to go
 276    around.
 277
 278    For simplicity, this implementation always converts the
 279    casefile to reader mode.  If this turns out to be a problem,
 280    with a little extra work we could also support sleeping
 281    writers. */
 282 void
 283 casefile_sleep (const struct casefile *cf_)
 284 {
 285   struct casefile *cf = (struct casefile *) cf_;
 286   assert (cf != NULL);
 287
 288   casefile_mode_reader (cf);
 289   casefile_to_disk (cf);
 290   flush_buffer (cf);
 291
 292   if (cf->fd != -1)
 293     {
 294       safe_close (cf->fd);
 295       cf->fd = -1;
 296     }
 297   if (cf->buffer != NULL)
 298     {
 299       free (cf->buffer);
 300       cf->buffer = NULL;
 301     }
 302 }
 303
 304 /* Returns the number of `union value's in a case for CF. */
 305 size_t
 306 casefile_get_value_cnt (const struct casefile *cf)
 307 {
 308   assert (cf != NULL);
 309
 310   return cf->value_cnt;
 311 }
 312
 313 /* Returns the number of cases in casefile CF. */
 314 unsigned long
 315 casefile_get_case_cnt (const struct casefile *cf)
 316 {
 317   assert (cf != NULL);
 318
 319   return cf->case_cnt;
 320 }
 321
 322 /* Appends a copy of case C to casefile CF.  Not valid after any
 323    reader for CF has been created. */
 324 void
 325 casefile_append (struct casefile *cf, const struct ccase *c)
 326 {
 327   assert (cf != NULL);
 328   assert (c != NULL);
 329   assert (cf->mode == WRITE);
 330
 331   /* Try memory first. */
 332   if (cf->storage == MEMORY)
 333     {
 334       if (case_bytes < get_max_workspace ())
 335         {
 336           size_t block_idx = cf->case_cnt / CASES_PER_BLOCK;
 337           size_t case_idx = cf->case_cnt % CASES_PER_BLOCK;
 338           struct ccase new_case;
 339
 340           case_bytes += cf->case_acct_size;
 341           case_clone (&new_case, c);
 342           if (case_idx == 0)
 343             {
 344               if ((block_idx & (block_idx - 1)) == 0)
 345                 {
 346                   size_t block_cap = block_idx == 0 ? 1 : block_idx * 2;
 347                   cf->cases = xrealloc (cf->cases,
 348                                         sizeof *cf->cases * block_cap);
 349                 }
 350
 351               cf->cases[block_idx] = xmalloc (sizeof **cf->cases
 352                                               * CASES_PER_BLOCK);
 353             }
 354
 355           case_move (&cf->cases[block_idx][case_idx], &new_case);
 356         }
 357       else
 358         {
 359           casefile_to_disk (cf);
 360           assert (cf->storage == DISK);
 361           write_case_to_disk (cf, c);
 362         }
 363     }
 364   else
 365     write_case_to_disk (cf, c);
 366
 367   cf->case_cnt++;
 368 }
 369
 370 /* Appends case C to casefile CF, which takes over ownership of
 371    C.  Not valid after any reader for CF has been created. */
 372 void
 373 casefile_append_xfer (struct casefile *cf, struct ccase *c)
 374 {
 375   casefile_append (cf, c);
 376   case_destroy (c);
 377 }
 378
 379 /* Writes case C to casefile CF's disk buffer, first flushing the buffer to
 380    disk if it would otherwise overflow. */
 381 static void
 382 write_case_to_disk (struct casefile *cf, const struct ccase *c)
 383 {
 384   case_to_values (c, cf->buffer + cf->buffer_used, cf->value_cnt);
 385   cf->buffer_used += cf->value_cnt;
 386   if (cf->buffer_used + cf->value_cnt > cf->buffer_size)
 387     flush_buffer (cf);
 388 }
 389
 390 /* If any bytes in CF's output buffer are used, flush them to
 391    disk. */
 392 static void
 393 flush_buffer (struct casefile *cf)
 394 {
 395   if (cf->buffer_used > 0)
 396     {
 397       if (!full_write (cf->fd, cf->buffer,
 398                        cf->buffer_size * sizeof *cf->buffer))
 399         msg (FE, _("Error writing temporary file: %s."), strerror (errno));
 400
 401       cf->buffer_used = 0;
 402     }
 403 }
 404
 405
 406 /* If CF is currently stored in memory, writes it to disk.  Readers, if any,
 407    retain their current positions. */
 408 void
 409 casefile_to_disk (const struct casefile *cf_)
 410 {
 411   struct casefile *cf = (struct casefile *) cf_;
 412   struct casereader *reader;
 413
 414   assert (cf != NULL);
 415
 416   if (cf->storage == MEMORY)
 417     {
 418       size_t idx, block_cnt;
 419
 420       assert (cf->filename == NULL);
 421       assert (cf->fd == -1);
 422       assert (cf->buffer_used == 0);
 423
 424       cf->storage = DISK;
 425       if (!make_temp_file (&cf->fd, &cf->filename))
 426         err_failure ();
 427       cf->buffer = xmalloc (cf->buffer_size * sizeof *cf->buffer);
 428       memset (cf->buffer, 0, cf->buffer_size * sizeof *cf->buffer);
 429
 430       case_bytes -= cf->case_cnt * cf->case_acct_size;
 431       for (idx = 0; idx < cf->case_cnt; idx++)
 432         {
 433           size_t block_idx = idx / CASES_PER_BLOCK;
 434           size_t case_idx = idx % CASES_PER_BLOCK;
 435           struct ccase *c = &cf->cases[block_idx][case_idx];
 436           write_case_to_disk (cf, c);
 437           case_destroy (c);
 438         }
 439
 440       block_cnt = DIV_RND_UP (cf->case_cnt, CASES_PER_BLOCK);
 441       for (idx = 0; idx < block_cnt; idx++)
 442         free (cf->cases[idx]);
 443
 444       free (cf->cases);
 445       cf->cases = NULL;
 446
 447       if (cf->mode == READ)
 448         flush_buffer (cf);
 449
 450       for (reader = cf->readers; reader != NULL; reader = reader->next)
 451         reader_open_file (reader);
 452     }
 453 }
 454
 455 /* Changes CF to reader mode, ensuring that no more cases may be
 456    added.  Creating a casereader for CF has the same effect. */
 457 void
 458 casefile_mode_reader (struct casefile *cf)
 459 {
 460   assert (cf != NULL);
 461   cf->mode = READ;
 462 }
 463
 464 /* Creates and returns a casereader for CF.  A casereader can be used to
 465    sequentially read the cases in a casefile. */
 466 struct casereader *
 467 casefile_get_reader (const struct casefile *cf_)
 468 {
 469   struct casefile *cf = (struct casefile *) cf_;
 470   struct casereader *reader;
 471
 472   assert (cf != NULL);
 473   assert (!cf->being_destroyed);
 474
 475   /* Flush the buffer to disk if it's not empty. */
 476   if (cf->mode == WRITE && cf->storage == DISK)
 477     flush_buffer (cf);
 478
 479   cf->mode = READ;
 480
 481   reader = xmalloc (sizeof *reader);
 482   reader->next = cf->readers;
 483   if (cf->readers != NULL)
 484     reader->next->prev = reader;
 485   cf->readers = reader;
 486   reader->prev = NULL;
 487   reader->cf = cf;
 488   reader->case_idx = 0;
 489   reader->destructive = 0;
 490   reader->fd = -1;
 491   reader->buffer = NULL;
 492   reader->buffer_pos = 0;
 493   case_nullify (&reader->c);
 494
 495   if (reader->cf->storage == DISK)
 496     reader_open_file (reader);
 497
 498   return reader;
 499 }
 500
 501 /* Creates and returns a destructive casereader for CF.  Like a
 502    normal casereader, a destructive casereader sequentially reads
 503    the cases in a casefile.  Unlike a normal casereader, a
 504    destructive reader cannot operate concurrently with any other
 505    reader.  (This restriction could be relaxed in a few ways, but
 506    it is so far unnecessary for other code.) */
 507 struct casereader *
 508 casefile_get_destructive_reader (struct casefile *cf)
 509 {
 510   struct casereader *reader;
 511
 512   assert (cf->readers == NULL);
 513   reader = casefile_get_reader (cf);
 514   reader->destructive = 1;
 515   cf->being_destroyed = 1;
 516   return reader;
 517 }
 518
 519 /* Opens a disk file for READER and seeks to the current position as indicated
 520    by case_idx.  Normally the current position is the beginning of the file,
 521    but casefile_to_disk may cause the file to be opened at a different
 522    position. */
 523 static void
 524 reader_open_file (struct casereader *reader)
 525 {
 526   struct casefile *cf = reader->cf;
 527   off_t file_ofs;
 528
 529   if (reader->case_idx >= cf->case_cnt)
 530     return;
 531
 532   if (cf->fd != -1)
 533     {
 534       reader->fd = cf->fd;
 535       cf->fd = -1;
 536     }
 537   else
 538     {
 539       reader->fd = safe_open (cf->filename, O_RDONLY);
 540       if (reader->fd < 0)
 541         msg (FE, _("%s: Opening temporary file: %s."),
 542              cf->filename, strerror (errno));
 543     }
 544
 545   if (cf->buffer != NULL)
 546     {
 547       reader->buffer = cf->buffer;
 548       cf->buffer = NULL;
 549     }
 550   else
 551     {
 552       reader->buffer = xmalloc (cf->buffer_size * sizeof *cf->buffer);
 553       memset (reader->buffer, 0, cf->buffer_size * sizeof *cf->buffer);
 554     }
 555
 556   if (cf->value_cnt != 0)
 557     {
 558       size_t buffer_case_cnt = cf->buffer_size / cf->value_cnt;
 559       file_ofs = ((off_t) reader->case_idx / buffer_case_cnt
 560                   * cf->buffer_size * sizeof *cf->buffer);
 561       reader->buffer_pos = (reader->case_idx % buffer_case_cnt
 562                             * cf->value_cnt);
 563     }
 564   else
 565     file_ofs = 0;
 566   if (lseek (reader->fd, file_ofs, SEEK_SET) != file_ofs)
 567     msg (FE, _("%s: Seeking temporary file: %s."),
 568          cf->filename, strerror (errno));
 569
 570   if (cf->case_cnt > 0 && cf->value_cnt > 0)
 571     fill_buffer (reader);
 572
 573   case_create (&reader->c, cf->value_cnt);
 574 }
 575
 576 /* Fills READER's buffer by reading a block from disk. */
 577 static void
 578 fill_buffer (struct casereader *reader)
 579 {
 580   int retval = full_read (reader->fd, reader->buffer,
 581                           reader->cf->buffer_size * sizeof *reader->buffer);
 582   if (retval < 0)
 583     msg (FE, _("%s: Reading temporary file: %s."),
 584          reader->cf->filename, strerror (errno));
 585   else if (retval != reader->cf->buffer_size * sizeof *reader->buffer)
 586     msg (FE, _("%s: Temporary file ended unexpectedly."),
 587          reader->cf->filename);
 588 }
 589
 590 /* Returns the casefile that READER reads. */
 591 const struct casefile *
 592 casereader_get_casefile (const struct casereader *reader)
 593 {
 594   assert (reader != NULL);
 595
 596   return reader->cf;
 597 }
 598
 599 /* Reads a copy of the next case from READER into C.
 600    Caller is responsible for destroying C.
 601    Returns true if successful, false at end of file. */
 602 int
 603 casereader_read (struct casereader *reader, struct ccase *c)
 604 {
 605   assert (reader != NULL);
 606
 607   if (reader->case_idx >= reader->cf->case_cnt)
 608     return 0;
 609
 610   if (reader->cf->storage == MEMORY)
 611     {
 612       size_t block_idx = reader->case_idx / CASES_PER_BLOCK;
 613       size_t case_idx = reader->case_idx % CASES_PER_BLOCK;
 614
 615       case_clone (c, &reader->cf->cases[block_idx][case_idx]);
 616       reader->case_idx++;
 617       return 1;
 618     }
 619   else
 620     {
 621       if (reader->buffer_pos + reader->cf->value_cnt > reader->cf->buffer_size)
 622         {
 623           fill_buffer (reader);
 624           reader->buffer_pos = 0;
 625         }
 626
 627       case_from_values (&reader->c, reader->buffer + reader->buffer_pos,
 628                         reader->cf->value_cnt);
 629       reader->buffer_pos += reader->cf->value_cnt;
 630       reader->case_idx++;
 631
 632       case_clone (c, &reader->c);
 633       return 1;
 634     }
 635 }
 636
 637 /* Reads the next case from READER into C and transfers ownership
 638    to the caller.  Caller is responsible for destroying C.
 639    Returns true if successful, false at end of file. */
 640 int
 641 casereader_read_xfer (struct casereader *reader, struct ccase *c)
 642 {
 643   assert (reader != NULL);
 644
 645   if (reader->destructive == 0
 646       || reader->case_idx >= reader->cf->case_cnt
 647       || reader->cf->storage == DISK)
 648     return casereader_read (reader, c);
 649   else
 650     {
 651       size_t block_idx = reader->case_idx / CASES_PER_BLOCK;
 652       size_t case_idx = reader->case_idx % CASES_PER_BLOCK;
 653       struct ccase *read_case = &reader->cf->cases[block_idx][case_idx];
 654
 655       case_move (c, read_case);
 656       reader->case_idx++;
 657       return 1;
 658     }
 659 }
 660
 661 /* Reads the next case from READER into C and transfers ownership
 662    to the caller.  Caller is responsible for destroying C.
 663    Assert-fails at end of file. */
 664 void
 665 casereader_read_xfer_assert (struct casereader *reader, struct ccase *c)
 666 {
 667   bool success = casereader_read_xfer (reader, c);
 668   assert (success);
 669 }
 670
 671 /* Destroys READER. */
 672 void
 673 casereader_destroy (struct casereader *reader)
 674 {
 675   assert (reader != NULL);
 676
 677   if (reader->next != NULL)
 678     reader->next->prev = reader->prev;
 679   if (reader->prev != NULL)
 680     reader->prev->next = reader->next;
 681   if (reader->cf->readers == reader)
 682     reader->cf->readers = reader->next;
 683
 684   if (reader->cf->buffer == NULL)
 685     reader->cf->buffer = reader->buffer;
 686   else
 687     free (reader->buffer);
 688
 689   if (reader->fd != -1)
 690     {
 691       if (reader->cf->fd == -1)
 692         reader->cf->fd = reader->fd;
 693       else
 694         safe_close (reader->fd);
 695     }
 696
 697   case_destroy (&reader->c);
 698
 699   free (reader);
 700 }
 701
 702 /* Calls open(), passing FILENAME and FLAGS, repeating as necessary
 703    to deal with interrupted calls. */
 704 static int
 705 safe_open (const char *filename, int flags)
 706 {
 707   int fd;
 708
 709   do
 710     {
 711       fd = open (filename, flags);
 712     }
 713   while (fd == -1 && errno == EINTR);
 714
 715   return fd;
 716 }
 717
 718 /* Calls close(), passing FD, repeating as necessary to deal with
 719    interrupted calls. */
 720 static int safe_close (int fd)
 721 {
 722   int retval;
 723
 724   do
 725     {
 726       retval = close (fd);
 727     }
 728   while (retval == -1 && errno == EINTR);
 729
 730   return retval;
 731 }
 732
 733 /* Registers our exit handler with atexit() if it has not already
 734    been registered. */
 735 static void
 736 register_atexit (void)
 737 {
 738   static int registered = 0;
 739   if (!registered)
 740     {
 741       registered = 1;
 742       atexit (exit_handler);
 743     }
 744 }
 745
 746
 747
 748 /* atexit() handler that closes and deletes our temporary
 749    files. */
 750 static void
 751 exit_handler (void)
 752 {
 753   while (casefiles != NULL)
 754     casefile_destroy (casefiles);
 755 }
 756 \f
 757 #include <gsl/gsl_rng.h>
 758 #include <stdarg.h>
 759 #include "command.h"
 760 #include "lexer.h"
 761
 762 static void test_casefile (int pattern, size_t value_cnt, size_t case_cnt);
 763 static void get_random_case (struct ccase *, size_t value_cnt,
 764                              size_t case_idx);
 765 static void write_random_case (struct casefile *cf, size_t case_idx);
 766 static void read_and_verify_random_case (struct casefile *cf,
 767                                          struct casereader *reader,
 768                                          size_t case_idx);
 769 static void fail_test (const char *message, ...);
 770
 771 int
 772 cmd_debug_casefile (void)
 773 {
 774   static const size_t sizes[] =
 775     {
 776       1, 2, 3, 4, 5, 6, 7, 14, 15, 16, 17, 31, 55, 73,
 777       100, 137, 257, 521, 1031, 2053
 778     };
 779   int size_max;
 780   int case_max;
 781   int pattern;
 782
 783   size_max = sizeof sizes / sizeof *sizes;
 784   if (lex_match_id ("SMALL"))
 785     {
 786       size_max -= 4;
 787       case_max = 511;
 788     }
 789   else
 790     case_max = 4095;
 791   if (token != '.')
 792     return lex_end_of_command ();
 793
 794   for (pattern = 0; pattern < 6; pattern++)
 795     {
 796       const size_t *size;
 797
 798       for (size = sizes; size < sizes + size_max; size++)
 799         {
 800           size_t case_cnt;
 801
 802           for (case_cnt = 0; case_cnt <= case_max;
 803                case_cnt = (case_cnt * 2) + 1)
 804             test_casefile (pattern, *size, case_cnt);
 805         }
 806     }
 807   printf ("Casefile tests succeeded.\n");
 808   return CMD_SUCCESS;
 809 }
 810
 811 static void
 812 test_casefile (int pattern, size_t value_cnt, size_t case_cnt)
 813 {
 814   struct casefile *cf;
 815   struct casereader *r1, *r2;
 816   struct ccase c;
 817   gsl_rng *rng;
 818   size_t i, j;
 819
 820   rng = gsl_rng_alloc (gsl_rng_mt19937);
 821   cf = casefile_create (value_cnt);
 822   if (pattern == 5)
 823     casefile_to_disk (cf);
 824   for (i = 0; i < case_cnt; i++)
 825     write_random_case (cf, i);
 826   if (pattern == 5)
 827     casefile_sleep (cf);
 828   r1 = casefile_get_reader (cf);
 829   r2 = casefile_get_reader (cf);
 830   switch (pattern)
 831     {
 832     case 0:
 833     case 5:
 834       for (i = 0; i < case_cnt; i++)
 835         {
 836           read_and_verify_random_case (cf, r1, i);
 837           read_and_verify_random_case (cf, r2, i);
 838         }
 839       break;
 840     case 1:
 841       for (i = 0; i < case_cnt; i++)
 842         read_and_verify_random_case (cf, r1, i);
 843       for (i = 0; i < case_cnt; i++)
 844         read_and_verify_random_case (cf, r2, i);
 845       break;
 846     case 2:
 847     case 3:
 848     case 4:
 849       for (i = j = 0; i < case_cnt; i++)
 850         {
 851           read_and_verify_random_case (cf, r1, i);
 852           if (gsl_rng_get (rng) % pattern == 0)
 853             read_and_verify_random_case (cf, r2, j++);
 854           if (i == case_cnt / 2)
 855             casefile_to_disk (cf);
 856         }
 857       for (; j < case_cnt; j++)
 858         read_and_verify_random_case (cf, r2, j);
 859       break;
 860     }
 861   if (casereader_read (r1, &c))
 862     fail_test ("Casereader 1 not at end of file.");
 863   if (casereader_read (r2, &c))
 864     fail_test ("Casereader 2 not at end of file.");
 865   if (pattern != 1)
 866     casereader_destroy (r1);
 867   if (pattern != 2)
 868     casereader_destroy (r2);
 869   if (pattern > 2)
 870     {
 871       r1 = casefile_get_destructive_reader (cf);
 872       for (i = 0; i < case_cnt; i++)
 873         {
 874           struct ccase read_case, expected_case;
 875
 876           get_random_case (&expected_case, value_cnt, i);
 877           if (!casereader_read_xfer (r1, &read_case))
 878             fail_test ("Premature end of casefile.");
 879           for (j = 0; j < value_cnt; j++)
 880             {
 881               double a = case_num (&read_case, j);
 882               double b = case_num (&expected_case, j);
 883               if (a != b)
 884                 fail_test ("Case %lu fails comparison.", (unsigned long) i);
 885             }
 886           case_destroy (&expected_case);
 887           case_destroy (&read_case);
 888         }
 889       casereader_destroy (r1);
 890     }
 891   casefile_destroy (cf);
 892   gsl_rng_free (rng);
 893 }
 894
 895 static void
 896 get_random_case (struct ccase *c, size_t value_cnt, size_t case_idx)
 897 {
 898   int i;
 899   case_create (c, value_cnt);
 900   for (i = 0; i < value_cnt; i++)
 901     case_data_rw (c, i)->f = case_idx % 257 + i;
 902 }
 903
 904 static void
 905 write_random_case (struct casefile *cf, size_t case_idx)
 906 {
 907   struct ccase c;
 908   get_random_case (&c, casefile_get_value_cnt (cf), case_idx);
 909   casefile_append_xfer (cf, &c);
 910 }
 911
 912 static void
 913 read_and_verify_random_case (struct casefile *cf,
 914                              struct casereader *reader, size_t case_idx)
 915 {
 916   struct ccase read_case, expected_case;
 917   size_t value_cnt;
 918   size_t i;
 919
 920   value_cnt = casefile_get_value_cnt (cf);
 921   get_random_case (&expected_case, value_cnt, case_idx);
 922   if (!casereader_read (reader, &read_case))
 923     fail_test ("Premature end of casefile.");
 924   for (i = 0; i < value_cnt; i++)
 925     {
 926       double a = case_num (&read_case, i);
 927       double b = case_num (&expected_case, i);
 928       if (a != b)
 929         fail_test ("Case %lu fails comparison.", (unsigned long) case_idx);
 930     }
 931   case_destroy (&read_case);
 932   case_destroy (&expected_case);
 933 }
 934
 935 static void
 936 fail_test (const char *message, ...)
 937 {
 938   va_list args;
 939
 940   va_start (args, message);
 941   vprintf (message, args);
 942   putchar ('\n');
 943   va_end (args);
 944
 945   exit (1);
 946 }