src/procedure.c

   1 /* PSPP - computes sample statistics.
   2    Copyright (C) 1997-9, 2000, 2006 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
  22 #include <procedure.h>
  23
  24 #include <errno.h>
  25 #include <stdio.h>
  26 #include <stdlib.h>
  27 #include <unistd.h>
  28
  29 #include "expressions/public.h"
  30 #include <data/case-source.h>
  31 #include <data/case-sink.h>
  32 #include <data/case.h>
  33 #include <data/casefile.h>
  34 #include <data/dictionary.h>
  35 #include <data/file-handle-def.h>
  36 #include <data/settings.h>
  37 #include <data/storage-stream.h>
  38 #include <data/value-labels.h>
  39 #include <data/variable.h>
  40 #include <language/control/control-stack.h>
  41 #include <libpspp/alloc.h>
  42 #include <libpspp/message.h>
  43 #include <libpspp/message.h>
  44 #include <libpspp/misc.h>
  45 #include <libpspp/str.h>
  46 #include <output/manager.h>
  47 #include <output/table.h>
  48
  49 #include "gettext.h"
  50 #define _(msgid) gettext (msgid)
  51
  52 /*
  53    Virtual File Manager (vfm):
  54
  55    vfm is used to process data files.  It uses the model that
  56    data is read from one stream (the data source), processed,
  57    then written to another (the data sink).  The data source is
  58    then deleted and the data sink becomes the data source for the
  59    next procedure. */
  60
  61 /* Procedure execution data. */
  62 struct write_case_data
  63   {
  64     /* Function to call for each case. */
  65     bool (*proc_func) (struct ccase *, void *); /* Function. */
  66     void *aux;                                 /* Auxiliary data. */
  67
  68     struct ccase trns_case;     /* Case used for transformations. */
  69     struct ccase sink_case;     /* Case written to sink, if
  70                                    compaction is necessary. */
  71     size_t cases_written;       /* Cases output so far. */
  72     size_t cases_analyzed;      /* Cases passed to procedure so far. */
  73   };
  74
  75 /* The current active file, from which cases are read. */
  76 struct case_source *vfm_source;
  77
  78 /* The replacement active file, to which cases are written. */
  79 struct case_sink *vfm_sink;
  80
  81 /* The compactor used to compact a compact, if necessary;
  82    otherwise a null pointer. */
  83 static struct dict_compactor *compactor;
  84
  85 /* Time at which vfm was last invoked. */
  86 static time_t last_vfm_invocation;
  87
  88 /* Whether we're inside a procedure.
  89    For debugging purposes only. */
  90 static bool in_procedure;
  91
  92 /* Lag queue. */
  93 int n_lag;                      /* Number of cases to lag. */
  94 static int lag_count;           /* Number of cases in lag_queue so far. */
  95 static int lag_head;            /* Index where next case will be added. */
  96 static struct ccase *lag_queue; /* Array of n_lag ccase * elements. */
  97
  98 /* Active transformations. */
  99 struct transformation *t_trns;
 100 size_t n_trns, m_trns, f_trns;
 101
 102 static bool internal_procedure (bool (*proc_func) (struct ccase *, void *),
 103                                 void *aux);
 104 static void update_last_vfm_invocation (void);
 105 static void create_trns_case (struct ccase *, struct dictionary *);
 106 static void open_active_file (void);
 107 static bool write_case (struct write_case_data *wc_data);
 108 static int execute_transformations (struct ccase *c,
 109                                     struct transformation *trns,
 110                                     int first_idx, int last_idx,
 111                                     int case_num);
 112 static int filter_case (const struct ccase *c, int case_num);
 113 static void lag_case (const struct ccase *c);
 114 static void clear_case (struct ccase *c);
 115 static bool close_active_file (void);
 116 \f
 117 /* Public functions. */
 118
 119 /* Returns the last time the data was read. */
 120 time_t
 121 time_of_last_procedure (void)
 122 {
 123   if (last_vfm_invocation == 0)
 124     update_last_vfm_invocation ();
 125   return last_vfm_invocation;
 126 }
 127
 128 /* Reads the data from the input program and writes it to a new
 129    active file.  For each case we read from the input program, we
 130    do the following
 131
 132    1. Execute permanent transformations.  If these drop the case,
 133       start the next case from step 1.
 134
 135    2. N OF CASES.  If we have already written N cases, start the
 136       next case from step 1.
 137
 138    3. Write case to replacement active file.
 139
 140    4. Execute temporary transformations.  If these drop the case,
 141       start the next case from step 1.
 142
 143    5. FILTER, PROCESS IF.  If these drop the case, start the next
 144       case from step 1.
 145
 146    6. Post-TEMPORARY N OF CASES.  If we have already analyzed N
 147       cases, start the next case from step 1.
 148
 149    7. Pass case to PROC_FUNC, passing AUX as auxiliary data.
 150
 151    Returns true if successful, false if an I/O error occurred. */
 152 bool
 153 procedure (bool (*proc_func) (struct ccase *, void *), void *aux)
 154 {
 155   if (proc_func == NULL
 156       && case_source_is_class (vfm_source, &storage_source_class)
 157       && vfm_sink == NULL
 158       && !temporary
 159       && n_trns == 0)
 160     {
 161       /* Nothing to do. */
 162       update_last_vfm_invocation ();
 163       return true;
 164     }
 165   else
 166     {
 167       bool ok;
 168
 169       open_active_file ();
 170       ok = internal_procedure (proc_func, aux);
 171       if (!close_active_file ())
 172         ok = false;
 173
 174       return ok;
 175     }
 176 }
 177
 178 /* Executes a procedure, as procedure(), except that the caller
 179    is responsible for calling open_active_file() and
 180    close_active_file().
 181    Returns true if successful, false if an I/O error occurred. */
 182 static bool
 183 internal_procedure (bool (*proc_func) (struct ccase *, void *), void *aux)
 184 {
 185   struct write_case_data wc_data;
 186   bool ok;
 187
 188   wc_data.proc_func = proc_func;
 189   wc_data.aux = aux;
 190   create_trns_case (&wc_data.trns_case, default_dict);
 191   case_create (&wc_data.sink_case, dict_get_next_value_idx (default_dict));
 192   wc_data.cases_written = 0;
 193
 194   update_last_vfm_invocation ();
 195
 196   ok = (vfm_source == NULL
 197         || vfm_source->class->read (vfm_source,
 198                                     &wc_data.trns_case,
 199                                     write_case, &wc_data));
 200
 201   case_destroy (&wc_data.sink_case);
 202   case_destroy (&wc_data.trns_case);
 203
 204   return ok;
 205 }
 206
 207 /* Updates last_vfm_invocation. */
 208 static void
 209 update_last_vfm_invocation (void)
 210 {
 211   last_vfm_invocation = time (NULL);
 212 }
 213
 214 /* Creates and returns a case, initializing it from the vectors
 215    that say which `value's need to be initialized just once, and
 216    which ones need to be re-initialized before every case. */
 217 static void
 218 create_trns_case (struct ccase *trns_case, struct dictionary *dict)
 219 {
 220   size_t var_cnt = dict_get_var_cnt (dict);
 221   size_t i;
 222
 223   case_create (trns_case, dict_get_next_value_idx (dict));
 224   for (i = 0; i < var_cnt; i++)
 225     {
 226       struct variable *v = dict_get_var (dict, i);
 227       union value *value = case_data_rw (trns_case, v->fv);
 228
 229       if (v->type == NUMERIC)
 230         value->f = v->reinit ? 0.0 : SYSMIS;
 231       else
 232         memset (value->s, ' ', v->width);
 233     }
 234 }
 235
 236 /* Makes all preparations for reading from the data source and writing
 237    to the data sink. */
 238 static void
 239 open_active_file (void)
 240 {
 241   assert (!in_procedure);
 242   in_procedure = true;
 243
 244   /* Make temp_dict refer to the dictionary right before data
 245      reaches the sink */
 246   if (!temporary)
 247     {
 248       temp_trns = n_trns;
 249       temp_dict = default_dict;
 250     }
 251
 252   /* Figure out compaction. */
 253   compactor = (dict_needs_compaction (temp_dict)
 254                ? dict_make_compactor (temp_dict)
 255                : NULL);
 256
 257   /* Prepare sink. */
 258   if (vfm_sink == NULL)
 259     vfm_sink = create_case_sink (&storage_sink_class, temp_dict, NULL);
 260   if (vfm_sink->class->open != NULL)
 261     vfm_sink->class->open (vfm_sink);
 262
 263   /* Allocate memory for lag queue. */
 264   if (n_lag > 0)
 265     {
 266       int i;
 267
 268       lag_count = 0;
 269       lag_head = 0;
 270       lag_queue = xnmalloc (n_lag, sizeof *lag_queue);
 271       for (i = 0; i < n_lag; i++)
 272         case_nullify (&lag_queue[i]);
 273     }
 274
 275   /* Close any unclosed DO IF or LOOP constructs. */
 276   ctl_stack_clear ();
 277 }
 278
 279 /* Transforms trns_case and writes it to the replacement active
 280    file if advisable.  Returns true if more cases can be
 281    accepted, false otherwise.  Do not call this function again
 282    after it has returned false once.  */
 283 static bool
 284 write_case (struct write_case_data *wc_data)
 285 {
 286   int retval;
 287
 288   /* Execute permanent transformations.  */
 289   retval = execute_transformations (&wc_data->trns_case, t_trns, f_trns,
 290                                     temp_trns, wc_data->cases_written + 1);
 291   if (retval != 1)
 292     goto done;
 293
 294   /* N OF CASES. */
 295   if (dict_get_case_limit (default_dict)
 296       && wc_data->cases_written >= dict_get_case_limit (default_dict))
 297     goto done;
 298   wc_data->cases_written++;
 299
 300   /* Write case to LAG queue. */
 301   if (n_lag)
 302     lag_case (&wc_data->trns_case);
 303
 304   /* Write case to replacement active file. */
 305   if (vfm_sink->class->write != NULL)
 306     {
 307       if (compactor != NULL)
 308         {
 309           dict_compactor_compact (compactor, &wc_data->sink_case,
 310                                   &wc_data->trns_case);
 311           vfm_sink->class->write (vfm_sink, &wc_data->sink_case);
 312         }
 313       else
 314         vfm_sink->class->write (vfm_sink, &wc_data->trns_case);
 315     }
 316
 317   /* Execute temporary transformations. */
 318   retval = execute_transformations (&wc_data->trns_case, t_trns, temp_trns,
 319                                     n_trns, wc_data->cases_written);
 320   if (retval != 1)
 321     goto done;
 322
 323   /* FILTER, PROCESS IF, post-TEMPORARY N OF CASES. */
 324   if (filter_case (&wc_data->trns_case, wc_data->cases_written)
 325       || (dict_get_case_limit (temp_dict)
 326           && wc_data->cases_analyzed >= dict_get_case_limit (temp_dict)))
 327     goto done;
 328   wc_data->cases_analyzed++;
 329
 330   /* Pass case to procedure. */
 331   if (wc_data->proc_func != NULL)
 332     if (!wc_data->proc_func (&wc_data->trns_case, wc_data->aux))
 333       retval = -1;
 334
 335  done:
 336   clear_case (&wc_data->trns_case);
 337   return retval != -1;
 338 }
 339
 340 /* Transforms case C using the transformations in TRNS[] with
 341    indexes FIRST_IDX through LAST_IDX, exclusive.  Case C will
 342    become case CASE_NUM (1-based) in the output file.  Returns 1
 343    if the case was successfully transformed, 0 if it was filtered
 344    out by one of the transformations, or -1 if the procedure
 345    should be abandoned due to a fatal error. */
 346 static int
 347 execute_transformations (struct ccase *c,
 348                          struct transformation *trns,
 349                          int first_idx, int last_idx,
 350                          int case_num)
 351 {
 352   int idx;
 353
 354   for (idx = first_idx; idx != last_idx; )
 355     {
 356       struct transformation *t = &trns[idx];
 357       int retval = t->proc (t->private, c, case_num);
 358       switch (retval)
 359         {
 360         case TRNS_CONTINUE:
 361           idx++;
 362           break;
 363
 364         case TRNS_DROP_CASE:
 365           return 0;
 366
 367         case TRNS_ERROR:
 368           return -1;
 369
 370         case TRNS_NEXT_CASE:
 371           abort ();
 372
 373         case TRNS_END_FILE:
 374           abort ();
 375
 376         default:
 377           idx = retval;
 378           break;
 379         }
 380     }
 381
 382   return 1;
 383 }
 384
 385 /* Returns nonzero if case C with case number CASE_NUM should be
 386    excluded as specified on FILTER or PROCESS IF, otherwise
 387    zero. */
 388 static int
 389 filter_case (const struct ccase *c, int case_idx)
 390 {
 391   /* FILTER. */
 392   struct variable *filter_var = dict_get_filter (default_dict);
 393   if (filter_var != NULL)
 394     {
 395       double f = case_num (c, filter_var->fv);
 396       if (f == 0.0 || mv_is_num_missing (&filter_var->miss, f))
 397         return 1;
 398     }
 399
 400   /* PROCESS IF. */
 401   if (process_if_expr != NULL
 402       && expr_evaluate_num (process_if_expr, c, case_idx) != 1.0)
 403     return 1;
 404
 405   return 0;
 406 }
 407
 408 /* Add C to the lag queue. */
 409 static void
 410 lag_case (const struct ccase *c)
 411 {
 412   if (lag_count < n_lag)
 413     lag_count++;
 414   case_destroy (&lag_queue[lag_head]);
 415   case_clone (&lag_queue[lag_head], c);
 416   if (++lag_head >= n_lag)
 417     lag_head = 0;
 418 }
 419
 420 /* Clears the variables in C that need to be cleared between
 421    processing cases.  */
 422 static void
 423 clear_case (struct ccase *c)
 424 {
 425   size_t var_cnt = dict_get_var_cnt (default_dict);
 426   size_t i;
 427
 428   for (i = 0; i < var_cnt; i++)
 429     {
 430       struct variable *v = dict_get_var (default_dict, i);
 431       if (v->reinit)
 432         {
 433           if (v->type == NUMERIC)
 434             case_data_rw (c, v->fv)->f = SYSMIS;
 435           else
 436             memset (case_data_rw (c, v->fv)->s, ' ', v->width);
 437         }
 438     }
 439 }
 440
 441 /* Closes the active file. */
 442 static bool
 443 close_active_file (void)
 444 {
 445   /* Free memory for lag queue, and turn off lagging. */
 446   if (n_lag > 0)
 447     {
 448       int i;
 449
 450       for (i = 0; i < n_lag; i++)
 451         case_destroy (&lag_queue[i]);
 452       free (lag_queue);
 453       n_lag = 0;
 454     }
 455
 456   /* Dictionary from before TEMPORARY becomes permanent.. */
 457   if (temporary)
 458     {
 459       dict_destroy (default_dict);
 460       default_dict = temp_dict;
 461       temp_dict = NULL;
 462     }
 463
 464   /* Finish compaction. */
 465   if (compactor != NULL)
 466     {
 467       dict_compactor_destroy (compactor);
 468       dict_compact_values (default_dict);
 469     }
 470
 471   /* Free data source. */
 472   free_case_source (vfm_source);
 473   vfm_source = NULL;
 474
 475   /* Old data sink becomes new data source. */
 476   if (vfm_sink->class->make_source != NULL)
 477     vfm_source = vfm_sink->class->make_source (vfm_sink);
 478   free_case_sink (vfm_sink);
 479   vfm_sink = NULL;
 480
 481   /* Cancel TEMPORARY, PROCESS IF, FILTER, N OF CASES, vectors,
 482      and get rid of all the transformations. */
 483   cancel_temporary ();
 484   expr_free (process_if_expr);
 485   process_if_expr = NULL;
 486   dict_set_case_limit (default_dict, 0);
 487   dict_clear_vectors (default_dict);
 488
 489   assert (in_procedure);
 490   in_procedure = false;
 491
 492   return cancel_transformations ();
 493 }
 494 \f
 495 /* Returns a pointer to the lagged case from N_BEFORE cases before the
 496    current one, or NULL if there haven't been that many cases yet. */
 497 struct ccase *
 498 lagged_case (int n_before)
 499 {
 500   assert (n_before >= 1 );
 501   assert (n_before <= n_lag);
 502
 503   if (n_before <= lag_count)
 504     {
 505       int index = lag_head - n_before;
 506       if (index < 0)
 507         index += n_lag;
 508       return &lag_queue[index];
 509     }
 510   else
 511     return NULL;
 512 }
 513
 514 /* Appends TRNS to t_trns[], the list of all transformations to be
 515    performed on data as it is read from the active file. */
 516 void
 517 add_transformation (trns_proc_func *proc, trns_free_func *free, void *private)
 518 {
 519   struct transformation *trns;
 520
 521   assert (!in_procedure);
 522
 523   if (n_trns >= m_trns)
 524     t_trns = x2nrealloc (t_trns, &m_trns, sizeof *t_trns);
 525   trns = &t_trns[n_trns++];
 526   trns->proc = proc;
 527   trns->free = free;
 528   trns->private = private;
 529 }
 530
 531 /* Returns the index number that the next transformation added by
 532    add_transformation() will receive.  A trns_proc_func that
 533    returns this index causes control flow to jump to it. */
 534 size_t
 535 next_transformation (void)
 536 {
 537   return n_trns;
 538 }
 539
 540 /* Cancels all active transformations, including any transformations
 541    created by the input program.
 542    Returns true if successful, false if an I/O error occurred. */
 543 bool
 544 cancel_transformations (void)
 545 {
 546   bool ok = true;
 547   size_t i;
 548   for (i = 0; i < n_trns; i++)
 549     {
 550       struct transformation *t = &t_trns[i];
 551       if (t->free != NULL)
 552         {
 553           if (!t->free (t->private))
 554             ok = false;
 555         }
 556     }
 557   n_trns = f_trns = 0;
 558   free (t_trns);
 559   t_trns = NULL;
 560   m_trns = 0;
 561   return ok;
 562 }
 563 \f
 564 /* Represents auxiliary data for handling SPLIT FILE. */
 565 struct split_aux_data
 566   {
 567     size_t case_count;          /* Number of cases so far. */
 568     struct ccase prev_case;     /* Data in previous case. */
 569
 570     /* Functions to call... */
 571     void (*begin_func) (void *);               /* ...before data. */
 572     bool (*proc_func) (struct ccase *, void *); /* ...with data. */
 573     void (*end_func) (void *);                 /* ...after data. */
 574     void *func_aux;                            /* Auxiliary data. */
 575   };
 576
 577 static int equal_splits (const struct ccase *, const struct ccase *);
 578 static bool procedure_with_splits_callback (struct ccase *, void *);
 579 static void dump_splits (struct ccase *);
 580
 581 /* Like procedure(), but it automatically breaks the case stream
 582    into SPLIT FILE break groups.  Before each group of cases with
 583    identical SPLIT FILE variable values, BEGIN_FUNC is called.
 584    Then PROC_FUNC is called with each case in the group.
 585    END_FUNC is called when the group is finished.  FUNC_AUX is
 586    passed to each of the functions as auxiliary data.
 587
 588    If the active file is empty, none of BEGIN_FUNC, PROC_FUNC,
 589    and END_FUNC will be called at all.
 590
 591    If SPLIT FILE is not in effect, then there is one break group
 592    (if the active file is nonempty), and BEGIN_FUNC and END_FUNC
 593    will be called once.
 594
 595    Returns true if successful, false if an I/O error occurred. */
 596 bool
 597 procedure_with_splits (void (*begin_func) (void *aux),
 598                        bool (*proc_func) (struct ccase *, void *aux),
 599                        void (*end_func) (void *aux),
 600                        void *func_aux)
 601 {
 602   struct split_aux_data split_aux;
 603   bool ok;
 604
 605   split_aux.case_count = 0;
 606   case_nullify (&split_aux.prev_case);
 607   split_aux.begin_func = begin_func;
 608   split_aux.proc_func = proc_func;
 609   split_aux.end_func = end_func;
 610   split_aux.func_aux = func_aux;
 611
 612   open_active_file ();
 613   ok = internal_procedure (procedure_with_splits_callback, &split_aux);
 614   if (split_aux.case_count > 0 && end_func != NULL)
 615     end_func (func_aux);
 616   if (!close_active_file ())
 617     ok = false;
 618
 619   case_destroy (&split_aux.prev_case);
 620
 621   return ok;
 622 }
 623
 624 /* procedure() callback used by procedure_with_splits(). */
 625 static bool
 626 procedure_with_splits_callback (struct ccase *c, void *split_aux_)
 627 {
 628   struct split_aux_data *split_aux = split_aux_;
 629
 630   /* Start a new series if needed. */
 631   if (split_aux->case_count == 0
 632       || !equal_splits (c, &split_aux->prev_case))
 633     {
 634       if (split_aux->case_count > 0 && split_aux->end_func != NULL)
 635         split_aux->end_func (split_aux->func_aux);
 636
 637       dump_splits (c);
 638       case_destroy (&split_aux->prev_case);
 639       case_clone (&split_aux->prev_case, c);
 640
 641       if (split_aux->begin_func != NULL)
 642         split_aux->begin_func (split_aux->func_aux);
 643     }
 644
 645   split_aux->case_count++;
 646   if (split_aux->proc_func != NULL)
 647     return split_aux->proc_func (c, split_aux->func_aux);
 648   else
 649     return true;
 650 }
 651
 652 /* Compares the SPLIT FILE variables in cases A and B and returns
 653    nonzero only if they differ. */
 654 static int
 655 equal_splits (const struct ccase *a, const struct ccase *b)
 656 {
 657   return case_compare (a, b,
 658                        dict_get_split_vars (default_dict),
 659                        dict_get_split_cnt (default_dict)) == 0;
 660 }
 661
 662 /* Dumps out the values of all the split variables for the case C. */
 663 static void
 664 dump_splits (struct ccase *c)
 665 {
 666   struct variable *const *split;
 667   struct tab_table *t;
 668   size_t split_cnt;
 669   int i;
 670
 671   split_cnt = dict_get_split_cnt (default_dict);
 672   if (split_cnt == 0)
 673     return;
 674
 675   t = tab_create (3, split_cnt + 1, 0);
 676   tab_dim (t, tab_natural_dimensions);
 677   tab_vline (t, TAL_GAP, 1, 0, split_cnt);
 678   tab_vline (t, TAL_GAP, 2, 0, split_cnt);
 679   tab_text (t, 0, 0, TAB_NONE, _("Variable"));
 680   tab_text (t, 1, 0, TAB_LEFT, _("Value"));
 681   tab_text (t, 2, 0, TAB_LEFT, _("Label"));
 682   split = dict_get_split_vars (default_dict);
 683   for (i = 0; i < split_cnt; i++)
 684     {
 685       struct variable *v = split[i];
 686       char temp_buf[80];
 687       const char *val_lab;
 688
 689       assert (v->type == NUMERIC || v->type == ALPHA);
 690       tab_text (t, 0, i + 1, TAB_LEFT | TAT_PRINTF, "%s", v->name);
 691
 692       data_out (temp_buf, &v->print, case_data (c, v->fv));
 693
 694       temp_buf[v->print.w] = 0;
 695       tab_text (t, 1, i + 1, TAT_PRINTF, "%.*s", v->print.w, temp_buf);
 696
 697       val_lab = val_labs_find (v->val_labs, *case_data (c, v->fv));
 698       if (val_lab)
 699         tab_text (t, 2, i + 1, TAB_LEFT, val_lab);
 700     }
 701   tab_flags (t, SOMF_NO_TITLE);
 702   tab_submit (t);
 703 }
 704 \f
 705 /* Represents auxiliary data for handling SPLIT FILE in a
 706    multipass procedure. */
 707 struct multipass_split_aux_data
 708   {
 709     struct ccase prev_case;     /* Data in previous case. */
 710     struct casefile *casefile;  /* Accumulates data for a split. */
 711
 712     /* Function to call with the accumulated data. */
 713     bool (*split_func) (const struct casefile *, void *);
 714     void *func_aux;                            /* Auxiliary data. */
 715   };
 716
 717 static bool multipass_split_callback (struct ccase *c, void *aux_);
 718 static void multipass_split_output (struct multipass_split_aux_data *);
 719
 720 /* Returns true if successful, false if an I/O error occurred. */
 721 bool
 722 multipass_procedure_with_splits (bool (*split_func) (const struct casefile *,
 723                                                      void *),
 724                                  void *func_aux)
 725 {
 726   struct multipass_split_aux_data aux;
 727   bool ok;
 728
 729   assert (split_func != NULL);
 730
 731   open_active_file ();
 732
 733   case_nullify (&aux.prev_case);
 734   aux.casefile = NULL;
 735   aux.split_func = split_func;
 736   aux.func_aux = func_aux;
 737
 738   ok = internal_procedure (multipass_split_callback, &aux);
 739   if (aux.casefile != NULL)
 740     multipass_split_output (&aux);
 741   case_destroy (&aux.prev_case);
 742
 743   if (!close_active_file ())
 744     ok = false;
 745
 746   return ok;
 747 }
 748
 749 /* procedure() callback used by multipass_procedure_with_splits(). */
 750 static bool
 751 multipass_split_callback (struct ccase *c, void *aux_)
 752 {
 753   struct multipass_split_aux_data *aux = aux_;
 754
 755   /* Start a new series if needed. */
 756   if (aux->casefile == NULL || !equal_splits (c, &aux->prev_case))
 757     {
 758       /* Pass any cases to split_func. */
 759       if (aux->casefile != NULL)
 760         multipass_split_output (aux);
 761
 762       /* Start a new casefile. */
 763       aux->casefile = casefile_create (dict_get_next_value_idx (default_dict));
 764
 765       /* Record split values. */
 766       dump_splits (c);
 767       case_destroy (&aux->prev_case);
 768       case_clone (&aux->prev_case, c);
 769     }
 770
 771   return casefile_append (aux->casefile, c);
 772 }
 773
 774 static void
 775 multipass_split_output (struct multipass_split_aux_data *aux)
 776 {
 777   assert (aux->casefile != NULL);
 778   aux->split_func (aux->casefile, aux->func_aux);
 779   casefile_destroy (aux->casefile);
 780   aux->casefile = NULL;
 781 }
 782
 783
 784 /* Discards all the current state in preparation for a data-input
 785    command like DATA LIST or GET. */
 786 void
 787 discard_variables (void)
 788 {
 789   dict_clear (default_dict);
 790   fh_set_default_handle (NULL);
 791
 792   n_lag = 0;
 793
 794   if (vfm_source != NULL)
 795     {
 796       free_case_source (vfm_source);
 797       vfm_source = NULL;
 798     }
 799
 800   cancel_transformations ();
 801
 802   ctl_stack_clear ();
 803
 804   expr_free (process_if_expr);
 805   process_if_expr = NULL;
 806
 807   cancel_temporary ();
 808 }