src/data/casereader.c

   1 /* PSPP - a program for statistical analysis.
   2    Copyright (C) 2007, 2009 Free Software Foundation, Inc.
   3
   4    This program is free software: you can redistribute it and/or modify
   5    it under the terms of the GNU General Public License as published by
   6    the Free Software Foundation, either version 3 of the License, or
   7    (at your option) any later version.
   8
   9    This program is distributed in the hope that it will be useful,
  10    but WITHOUT ANY WARRANTY; without even the implied warranty of
  11    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12    GNU General Public License for more details.
  13
  14    You should have received a copy of the GNU General Public License
  15    along with this program.  If not, see <http://www.gnu.org/licenses/>. */
  16
  17 #include <config.h>
  18
  19 #include <data/casereader.h>
  20 #include <data/casereader-provider.h>
  21
  22 #include <stdlib.h>
  23
  24 #include <data/casewindow.h>
  25 #include <data/casewriter.h>
  26 #include <data/settings.h>
  27 #include <libpspp/assertion.h>
  28 #include <libpspp/heap.h>
  29 #include <libpspp/taint.h>
  30
  31 #include "xalloc.h"
  32
  33 /* A casereader. */
  34 struct casereader
  35   {
  36     struct taint *taint;                  /* Corrupted? */
  37     struct caseproto *proto;              /* Format of contained cases. */
  38     casenumber case_cnt;                  /* Number of cases,
  39                                              CASENUMBER_MAX if unknown. */
  40     const struct casereader_class *class; /* Class. */
  41     void *aux;                            /* Auxiliary data for class. */
  42   };
  43
  44 static void insert_shim (struct casereader *);
  45
  46 /* Reads and returns the next case from READER.  The caller owns
  47    the returned case and must call case_unref on it when its data
  48    is no longer needed.  Returns a null pointer if cases have
  49    been exhausted or upon detection of an I/O error.
  50
  51    The case returned is effectively consumed: it can never be
  52    read again through READER.  If this is inconvenient, READER
  53    may be cloned in advance with casereader_clone, or
  54    casereader_peek may be used instead. */
  55 struct ccase *
  56 casereader_read (struct casereader *reader)
  57 {
  58   if (reader->case_cnt != 0)
  59     {
  60       /* ->read may use casereader_swap to replace itself by
  61          another reader and then delegate to that reader by
  62          recursively calling casereader_read.  Currently only
  63          lazy_casereader does this and, with luck, nothing else
  64          ever will.
  65
  66          To allow this to work, however, we must decrement
  67          case_cnt before calling ->read.  If we decremented
  68          case_cnt after calling ->read, then this would actually
  69          drop two cases from case_cnt instead of one, and we'd
  70          lose the last case in the casereader. */
  71       struct ccase *c;
  72       if (reader->case_cnt != CASENUMBER_MAX)
  73         reader->case_cnt--;
  74       c = reader->class->read (reader, reader->aux);
  75       if (c != NULL)
  76         {
  77           size_t n_widths UNUSED = caseproto_get_n_widths (reader->proto);
  78           assert (case_get_value_cnt (c) >= n_widths);
  79           expensive_assert (caseproto_equal (case_get_proto (c), 0,
  80                                              reader->proto, 0, n_widths));
  81           return c;
  82         }
  83     }
  84   reader->case_cnt = 0;
  85   return NULL;
  86 }
  87
  88 /* Destroys READER.
  89    Returns false if an I/O error was detected on READER, true
  90    otherwise. */
  91 bool
  92 casereader_destroy (struct casereader *reader)
  93 {
  94   bool ok = true;
  95   if (reader != NULL)
  96     {
  97       reader->class->destroy (reader, reader->aux);
  98       ok = taint_destroy (reader->taint);
  99       caseproto_unref (reader->proto);
 100       free (reader);
 101     }
 102   return ok;
 103 }
 104
 105 /* Returns a clone of READER.  READER and its clone may be used
 106    to read the same sequence of cases in the same order, barring
 107    I/O errors. */
 108 struct casereader *
 109 casereader_clone (const struct casereader *reader_)
 110 {
 111   struct casereader *reader = (struct casereader *) reader_;
 112   struct casereader *clone;
 113   if ( reader == NULL )
 114     return NULL;
 115
 116   if (reader->class->clone == NULL)
 117     insert_shim (reader);
 118   clone = reader->class->clone (reader, reader->aux);
 119   assert (clone != NULL);
 120   assert (clone != reader);
 121   return clone;
 122 }
 123
 124 /* Makes a copy of ORIGINAL into *NEW1 (if NEW1 is non-null) and
 125    *NEW2 (if NEW2 is non-null), then destroys ORIGINAL. */
 126 void
 127 casereader_split (struct casereader *original,
 128                   struct casereader **new1, struct casereader **new2)
 129 {
 130   if (new1 != NULL && new2 != NULL)
 131     {
 132       *new1 = casereader_rename (original);
 133       *new2 = casereader_clone (*new1);
 134     }
 135   else if (new1 != NULL)
 136     *new1 = casereader_rename (original);
 137   else if (new2 != NULL)
 138     *new2 = casereader_rename (original);
 139   else
 140     casereader_destroy (original);
 141 }
 142
 143 /* Returns a copy of READER, which is itself destroyed.
 144    Useful for taking over ownership of a casereader, to enforce
 145    preventing the original owner from accessing the casereader
 146    again. */
 147 struct casereader *
 148 casereader_rename (struct casereader *reader)
 149 {
 150   struct casereader *new = xmemdup (reader, sizeof *reader);
 151   free (reader);
 152   return new;
 153 }
 154
 155 /* Exchanges the casereaders referred to by A and B. */
 156 void
 157 casereader_swap (struct casereader *a, struct casereader *b)
 158 {
 159   if (a != b)
 160     {
 161       struct casereader tmp = *a;
 162       *a = *b;
 163       *b = tmp;
 164     }
 165 }
 166
 167 /* Reads and returns the (IDX + 1)'th case from READER.  The
 168    caller owns the returned case and must call case_unref on it
 169    when it is no longer needed.  Returns a null pointer if cases
 170    have been exhausted or upon detection of an I/O error. */
 171 struct ccase *
 172 casereader_peek (struct casereader *reader, casenumber idx)
 173 {
 174   if (idx < reader->case_cnt)
 175     {
 176       struct ccase *c;
 177       if (reader->class->peek == NULL)
 178         insert_shim (reader);
 179       c = reader->class->peek (reader, reader->aux, idx);
 180       if (c != NULL)
 181         return c;
 182       else if (casereader_error (reader))
 183         reader->case_cnt = 0;
 184     }
 185   if (reader->case_cnt > idx)
 186     reader->case_cnt = idx;
 187   return NULL;
 188 }
 189
 190 /* Returns true if no cases remain to be read from READER, or if
 191    an error has occurred on READER.  (A return value of false
 192    does *not* mean that the next call to casereader_peek or
 193    casereader_read will return true, because an error can occur
 194    in the meantime.) */
 195 bool
 196 casereader_is_empty (struct casereader *reader)
 197 {
 198   if (reader->case_cnt == 0)
 199     return true;
 200   else
 201     {
 202       struct ccase *c = casereader_peek (reader, 0);
 203       if (c == NULL)
 204         return true;
 205       else
 206         {
 207           case_unref (c);
 208           return false;
 209         }
 210     }
 211 }
 212
 213 /* Returns true if an I/O error or another hard error has
 214    occurred on READER, a clone of READER, or on some object on
 215    which READER's data has a dependency, false otherwise. */
 216 bool
 217 casereader_error (const struct casereader *reader)
 218 {
 219   return taint_is_tainted (reader->taint);
 220 }
 221
 222 /* Marks READER as having encountered an error.
 223
 224    Ordinarily, this function should be called by the
 225    implementation of a casereader, not by the casereader's
 226    client.  Instead, casereader clients should usually ensure
 227    that a casereader's error state is correct by using
 228    taint_propagate to propagate to the casereader's taint
 229    structure, which may be obtained via casereader_get_taint. */
 230 void
 231 casereader_force_error (struct casereader *reader)
 232 {
 233   taint_set_taint (reader->taint);
 234 }
 235
 236 /* Returns READER's associate taint object, for use with
 237    taint_propagate and other taint functions. */
 238 const struct taint *
 239 casereader_get_taint (const struct casereader *reader)
 240 {
 241   return reader->taint;
 242 }
 243
 244 /* Returns the number of cases that will be read by successive
 245    calls to casereader_read for READER, assuming that no errors
 246    occur.  Upon an error condition, the case count drops to 0, so
 247    that no more cases can be obtained.
 248
 249    Not all casereaders can predict the number of cases that they
 250    will produce without actually reading all of them.  In that
 251    case, this function returns CASENUMBER_MAX.  To obtain the
 252    actual number of cases in such a casereader, use
 253    casereader_count_cases. */
 254 casenumber
 255 casereader_get_case_cnt (struct casereader *reader)
 256 {
 257   return reader->case_cnt;
 258 }
 259
 260 /* Returns the number of cases that will be read by successive
 261    calls to casereader_read for READER, assuming that no errors
 262    occur.  Upon an error condition, the case count drops to 0, so
 263    that no more cases can be obtained.
 264
 265    For a casereader that cannot predict the number of cases it
 266    will produce, this function actually reads (and discards) all
 267    of the contents of a clone of READER.  Thus, the return value
 268    is always correct in the absence of I/O errors. */
 269 casenumber
 270 casereader_count_cases (struct casereader *reader)
 271 {
 272   if (reader->case_cnt == CASENUMBER_MAX)
 273     {
 274       casenumber n_cases = 0;
 275       struct ccase *c;
 276
 277       struct casereader *clone = casereader_clone (reader);
 278
 279       for (; (c = casereader_read (clone)) != NULL; case_unref (c))
 280         n_cases++;
 281
 282       casereader_destroy (clone);
 283       reader->case_cnt = n_cases;
 284     }
 285
 286   return reader->case_cnt;
 287 }
 288
 289 /* Returns the prototype for the cases in READER.  The caller
 290    must not unref the returned prototype. */
 291 const struct caseproto *
 292 casereader_get_proto (const struct casereader *reader)
 293 {
 294   return reader->proto;
 295 }
 296
 297 /* Copies all the cases in READER to WRITER, propagating errors
 298    appropriately. */
 299 void
 300 casereader_transfer (struct casereader *reader, struct casewriter *writer)
 301 {
 302   struct ccase *c;
 303
 304   taint_propagate (casereader_get_taint (reader),
 305                    casewriter_get_taint (writer));
 306   while ((c = casereader_read (reader)) != NULL)
 307     casewriter_write (writer, c);
 308   casereader_destroy (reader);
 309 }
 310
 311 /* Creates and returns a new casereader.  This function is
 312    intended for use by casereader implementations, not by
 313    casereader clients.
 314
 315    This function is most suited for creating a casereader for a
 316    data source that is naturally sequential.
 317    casereader_create_random may be more appropriate for a data
 318    source that supports random access.
 319
 320    Ordinarily, specify a null pointer for TAINT, in which case
 321    the new casereader will have a new, unique taint object.  If
 322    the new casereader should have a clone of an existing taint
 323    object, specify that object as TAINT.  (This is most commonly
 324    useful in an implementation of the "clone" casereader_class
 325    function, in which case the cloned casereader should have the
 326    same taint object as the original casereader.)
 327
 328    PROTO must be the prototype for the cases that may be read
 329    from the casereader.  The caller retains its reference to
 330    PROTO.
 331
 332    CASE_CNT is an upper limit on the number of cases that
 333    casereader_read will return from the casereader in successive
 334    calls.  Ordinarily, this is the actual number of cases in the
 335    data source or CASENUMBER_MAX if the number of cases cannot be
 336    predicted in advance.
 337
 338    CLASS and AUX are a set of casereader implementation-specific
 339    member functions and auxiliary data to pass to those member
 340    functions, respectively. */
 341 struct casereader *
 342 casereader_create_sequential (const struct taint *taint,
 343                               const struct caseproto *proto,
 344                               casenumber case_cnt,
 345                               const struct casereader_class *class, void *aux)
 346 {
 347   struct casereader *reader = xmalloc (sizeof *reader);
 348   reader->taint = taint != NULL ? taint_clone (taint) : taint_create ();
 349   reader->proto = caseproto_ref (proto);
 350   reader->case_cnt = case_cnt;
 351   reader->class = class;
 352   reader->aux = aux;
 353   return reader;
 354 }
 355
 356 /* If READER is a casereader of the given CLASS, returns its
 357    associated auxiliary data; otherwise, returns a null pointer.
 358
 359    This function is intended for use from casereader
 360    implementations, not by casereader users.  Even within
 361    casereader implementations, its usefulness is quite limited,
 362    for at least two reasons.  First, every casereader member
 363    function already receives a pointer to the casereader's
 364    auxiliary data.  Second, a casereader's class can change
 365    (through a call to casereader_swap) and this is in practice
 366    quite common (e.g. any call to casereader_clone on a
 367    casereader that does not directly support clone will cause the
 368    casereader to be replaced by a shim caseader). */
 369 void *
 370 casereader_dynamic_cast (struct casereader *reader,
 371                          const struct casereader_class *class)
 372 {
 373   return reader->class == class ? reader->aux : NULL;
 374 }
 375 \f
 376 /* Random-access casereader implementation.
 377
 378    This is a set of wrappers around casereader_create_sequential
 379    and struct casereader_class to make it easy to create
 380    efficient casereaders for data sources that natively support
 381    random access. */
 382
 383 /* One clone of a random reader. */
 384 struct random_reader
 385   {
 386     struct random_reader_shared *shared; /* Data shared among clones. */
 387     struct heap_node heap_node; /* Node in shared data's heap of readers. */
 388     casenumber offset;          /* Number of cases already read. */
 389   };
 390
 391 /* Returns the random_reader in which the given heap_node is
 392    embedded. */
 393 static struct random_reader *
 394 random_reader_from_heap_node (const struct heap_node *node)
 395 {
 396   return heap_data (node, struct random_reader, heap_node);
 397 }
 398
 399 /* Data shared among clones of a random reader. */
 400 struct random_reader_shared
 401   {
 402     struct heap *readers;       /* Heap of struct random_readers. */
 403     casenumber min_offset;      /* Smallest offset of any random_reader. */
 404     const struct casereader_random_class *class;
 405     void *aux;
 406   };
 407
 408 static const struct casereader_class random_reader_casereader_class;
 409
 410 /* Creates and returns a new random_reader with the given SHARED
 411    data and OFFSET.  Inserts the new random reader into the
 412    shared heap. */
 413 static struct random_reader *
 414 make_random_reader (struct random_reader_shared *shared, casenumber offset)
 415 {
 416   struct random_reader *br = xmalloc (sizeof *br);
 417   br->offset = offset;
 418   br->shared = shared;
 419   heap_insert (shared->readers, &br->heap_node);
 420   return br;
 421 }
 422
 423 /* Compares random_readers A and B by offset and returns a
 424    strcmp()-like result. */
 425 static int
 426 compare_random_readers_by_offset (const struct heap_node *a_,
 427                                   const struct heap_node *b_,
 428                                   const void *aux UNUSED)
 429 {
 430   const struct random_reader *a = random_reader_from_heap_node (a_);
 431   const struct random_reader *b = random_reader_from_heap_node (b_);
 432   return a->offset < b->offset ? -1 : a->offset > b->offset;
 433 }
 434
 435 /* Creates and returns a new casereader.  This function is
 436    intended for use by casereader implementations, not by
 437    casereader clients.
 438
 439    This function is most suited for creating a casereader for a
 440    data source that supports random access.
 441    casereader_create_sequential is more appropriate for a data
 442    source that is naturally sequential.
 443
 444    PROTO must be the prototype for the cases that may be read
 445    from the casereader.  The caller retains its reference to
 446    PROTO.
 447
 448    CASE_CNT is an upper limit on the number of cases that
 449    casereader_read will return from the casereader in successive
 450    calls.  Ordinarily, this is the actual number of cases in the
 451    data source or CASENUMBER_MAX if the number of cases cannot be
 452    predicted in advance.
 453
 454    CLASS and AUX are a set of casereader implementation-specific
 455    member functions and auxiliary data to pass to those member
 456    functions, respectively. */
 457 struct casereader *
 458 casereader_create_random (const struct caseproto *proto, casenumber case_cnt,
 459                           const struct casereader_random_class *class,
 460                           void *aux)
 461 {
 462   struct random_reader_shared *shared = xmalloc (sizeof *shared);
 463   shared->readers = heap_create (compare_random_readers_by_offset, NULL);
 464   shared->class = class;
 465   shared->aux = aux;
 466   shared->min_offset = 0;
 467   return casereader_create_sequential (NULL, proto, case_cnt,
 468                                        &random_reader_casereader_class,
 469                                        make_random_reader (shared, 0));
 470 }
 471
 472 /* Reassesses the min_offset in SHARED based on the minimum
 473    offset in the heap.   */
 474 static void
 475 advance_random_reader (struct casereader *reader,
 476                        struct random_reader_shared *shared)
 477 {
 478   casenumber old, new;
 479
 480   old = shared->min_offset;
 481   new = random_reader_from_heap_node (heap_minimum (shared->readers))->offset;
 482   assert (new >= old);
 483   if (new > old)
 484     {
 485       shared->min_offset = new;
 486       shared->class->advance (reader, shared->aux, new - old);
 487     }
 488 }
 489
 490 /* struct casereader_class "read" function for random reader. */
 491 static struct ccase *
 492 random_reader_read (struct casereader *reader, void *br_)
 493 {
 494   struct random_reader *br = br_;
 495   struct random_reader_shared *shared = br->shared;
 496   struct ccase *c = shared->class->read (reader, shared->aux,
 497                                          br->offset - shared->min_offset);
 498   if (c != NULL)
 499     {
 500       br->offset++;
 501       heap_changed (shared->readers, &br->heap_node);
 502       advance_random_reader (reader, shared);
 503     }
 504   return c;
 505 }
 506
 507 /* struct casereader_class "destroy" function for random
 508    reader. */
 509 static void
 510 random_reader_destroy (struct casereader *reader, void *br_)
 511 {
 512   struct random_reader *br = br_;
 513   struct random_reader_shared *shared = br->shared;
 514
 515   heap_delete (shared->readers, &br->heap_node);
 516   if (heap_is_empty (shared->readers))
 517     {
 518       heap_destroy (shared->readers);
 519       shared->class->destroy (reader, shared->aux);
 520       free (shared);
 521     }
 522   else
 523     advance_random_reader (reader, shared);
 524
 525   free (br);
 526 }
 527
 528 /* struct casereader_class "clone" function for random reader. */
 529 static struct casereader *
 530 random_reader_clone (struct casereader *reader, void *br_)
 531 {
 532   struct random_reader *br = br_;
 533   struct random_reader_shared *shared = br->shared;
 534   return casereader_create_sequential (casereader_get_taint (reader),
 535                                        reader->proto,
 536                                        casereader_get_case_cnt (reader),
 537                                        &random_reader_casereader_class,
 538                                        make_random_reader (shared,
 539                                                            br->offset));
 540 }
 541
 542 /* struct casereader_class "peek" function for random reader. */
 543 static struct ccase *
 544 random_reader_peek (struct casereader *reader, void *br_, casenumber idx)
 545 {
 546   struct random_reader *br = br_;
 547   struct random_reader_shared *shared = br->shared;
 548
 549   return shared->class->read (reader, shared->aux,
 550                               br->offset - shared->min_offset + idx);
 551 }
 552
 553 /* Casereader class for random reader. */
 554 static const struct casereader_class random_reader_casereader_class =
 555   {
 556     random_reader_read,
 557     random_reader_destroy,
 558     random_reader_clone,
 559     random_reader_peek,
 560   };
 561 \f
 562 /* Buffering shim for implementing clone and peek operations.
 563
 564    The "clone" and "peek" operations aren't implemented by all
 565    types of casereaders, but we have to expose a uniform
 566    interface anyhow.  We do this by interposing a buffering
 567    casereader on top of the existing casereader on the first call
 568    to "clone" or "peek".  The buffering casereader maintains a
 569    window of cases that spans the positions of the original
 570    casereader and all of its clones (the "clone set"), from the
 571    position of the casereader that has read the fewest cases to
 572    the position of the casereader that has read the most.
 573
 574    Thus, if all of the casereaders in the clone set are at
 575    approximately the same position, only a few cases are buffered
 576    and there is little inefficiency.  If, on the other hand, one
 577    casereader is not used to read any cases at all, but another
 578    one is used to read all of the cases, the entire contents of
 579    the casereader is copied into the buffer.  This still might
 580    not be so inefficient, given that case data in memory is
 581    shared across multiple identical copies, but in the worst case
 582    the window implementation will write cases to disk instead of
 583    maintaining them in-memory. */
 584
 585 /* A buffering shim for a non-clonable or non-peekable
 586    casereader. */
 587 struct shim
 588   {
 589     struct casewindow *window;          /* Window of buffered cases. */
 590     struct casereader *subreader;       /* Subordinate casereader. */
 591   };
 592
 593 static const struct casereader_random_class shim_class;
 594
 595 /* Interposes a buffering shim atop READER. */
 596 static void
 597 insert_shim (struct casereader *reader)
 598 {
 599   const struct caseproto *proto = casereader_get_proto (reader);
 600   casenumber case_cnt = casereader_get_case_cnt (reader);
 601   struct shim *b = xmalloc (sizeof *b);
 602   b->window = casewindow_create (proto, settings_get_workspace_cases (proto));
 603   b->subreader = casereader_create_random (proto, case_cnt, &shim_class, b);
 604   casereader_swap (reader, b->subreader);
 605   taint_propagate (casewindow_get_taint (b->window),
 606                    casereader_get_taint (reader));
 607   taint_propagate (casereader_get_taint (b->subreader),
 608                    casereader_get_taint (reader));
 609 }
 610
 611 /* Ensures that B's window contains at least CASE_CNT cases.
 612    Return true if successful, false upon reaching the end of B's
 613    subreader or an I/O error. */
 614 static bool
 615 prime_buffer (struct shim *b, casenumber case_cnt)
 616 {
 617   while (casewindow_get_case_cnt (b->window) < case_cnt)
 618     {
 619       struct ccase *tmp = casereader_read (b->subreader);
 620       if (tmp == NULL)
 621         return false;
 622       casewindow_push_head (b->window, tmp);
 623     }
 624   return true;
 625 }
 626
 627 /* Reads the case at the given 0-based OFFSET from the front of
 628    the window into C.  Returns the case if successful, or a null
 629    pointer if OFFSET is beyond the end of file or upon I/O error.
 630    The caller must call case_unref() on the returned case when it
 631    is no longer needed. */
 632 static struct ccase *
 633 shim_read (struct casereader *reader UNUSED, void *b_,
 634            casenumber offset)
 635 {
 636   struct shim *b = b_;
 637   if (!prime_buffer (b, offset + 1))
 638     return NULL;
 639   return casewindow_get_case (b->window, offset);
 640 }
 641
 642 /* Destroys B. */
 643 static void
 644 shim_destroy (struct casereader *reader UNUSED, void *b_)
 645 {
 646   struct shim *b = b_;
 647   casewindow_destroy (b->window);
 648   casereader_destroy (b->subreader);
 649   free (b);
 650 }
 651
 652 /* Discards CNT cases from the front of B's window. */
 653 static void
 654 shim_advance (struct casereader *reader UNUSED, void *b_, casenumber case_cnt)
 655 {
 656   struct shim *b = b_;
 657   casewindow_pop_tail (b->window, case_cnt);
 658 }
 659
 660 /* Class for the buffered reader. */
 661 static const struct casereader_random_class shim_class =
 662   {
 663     shim_read,
 664     shim_destroy,
 665     shim_advance,
 666   };