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