src/mcas/replay.c

   1 /******************************************************************************
   2  * replay.c
   3  *
   4  * Replay the log output of search-structure runs.
   5  * Must build set_harness.c with DO_WRITE_LOG defined.
   6  *
   7  * Copyright (c) 2002-2003, K A Fraser
   8  *
   9
  10 Redistribution and use in source and binary forms, with or without
  11 modification, are permitted provided that the following conditions are
  12 met:
  13
  14     * Redistributions of source code must retain the above copyright
  15     * notice, this list of conditions and the following disclaimer.
  16     * Redistributions in binary form must reproduce the above
  17     * copyright notice, this list of conditions and the following
  18     * disclaimer in the documentation and/or other materials provided
  19     * with the distribution.  Neither the name of the Keir Fraser
  20     * nor the names of its contributors may be used to endorse or
  21     * promote products derived from this software without specific
  22     * prior written permission.
  23
  24 THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  25 "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  26 LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  27 A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  28 OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  29 SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  30 LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  31 DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  32 THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  33 (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  34 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  35  */
  36
  37 #include <stdio.h>
  38 #include <stdlib.h>
  39 #include <limits.h>
  40 #include <errno.h>
  41 #include <stdlib.h>
  42 #include <assert.h>
  43 #include <sys/types.h>
  44 #include <sys/wait.h>
  45 #include <sys/stat.h>
  46 #include <fcntl.h>
  47 #include <unistd.h>
  48
  49 #include "portable_defns.h"
  50
  51 #define RMAX_THREADS 256
  52 #define VERIFY_ORDERINGS
  53
  54 #define LOG_REPLAYED     (1<<26)
  55 #define LOG_KEY_MASK     0xffffff
  56
  57 typedef struct log_st
  58 {
  59     interval_t   start, end;
  60     unsigned int data;       /* key, and replay flag */
  61     void *val, *old_val;     /* op changed mapping from old_val to val */
  62 } log_t;
  63
  64 #define REPLAYED(_l) ((_l)->data & LOG_REPLAYED)
  65
  66 static log_t *global_log;
  67 static int nr_threads, nr_updates, nr_keys;
  68 static int *key_offsets;
  69 static int *success;
  70 static unsigned int next_key = 0;
  71 static pthread_mutex_t key_lock;
  72
  73
  74 /*
  75  * GLOBAL LOGS SORTED ON:
  76  *  1. Key value
  77  *  2. Start time
  78  *
  79  * Replayer deals with each key value in turn.
  80  */
  81 static int compare(const void *t1, const void *t2)
  82 {
  83     const log_t *l1 = t1;
  84     const log_t *l2 = t2;
  85     const int k1 = l1->data & LOG_KEY_MASK;
  86     const int k2 = l2->data & LOG_KEY_MASK;
  87
  88     if ( k1 < k2 ) return(-1);
  89     if ( k1 > k2 ) return(+1);
  90
  91     if ( l1->start < l2->start ) return(-1);
  92
  93     return(+1);
  94 }
  95
  96
  97 static int do_op(log_t *log, void **key_state)
  98 {
  99     if ( REPLAYED(log) || (log->old_val != *key_state) ) return(0);
 100     *key_state = log->val;
 101     log->data |= LOG_REPLAYED;
 102     return(1);
 103 }
 104
 105
 106 static void undo_op(log_t *log, void **key_state)
 107 {
 108     assert(REPLAYED(log));
 109     log->data &= ~LOG_REPLAYED;
 110     *key_state = log->old_val;
 111 }
 112
 113
 114 /* Sink down element @pos of @heap. */
 115 static void down_heap(log_t **heap, int *heap_offsets, log_t *log, int pos)
 116 {
 117     int sz = (int)heap[0], nxt;
 118     log_t *tmp;
 119     while ( (nxt = (pos << 1)) <= sz )
 120     {
 121         if ( ((nxt+1) <= sz) && (heap[nxt+1]->end < heap[nxt]->end) ) nxt++;
 122         if ( heap[nxt]->end > heap[pos]->end ) break;
 123         heap_offsets[heap[pos] - log] = nxt;
 124         heap_offsets[heap[nxt] - log] = pos;
 125         tmp = heap[pos];
 126         heap[pos] = heap[nxt];
 127         heap[nxt] = tmp;
 128         pos = nxt;
 129     }
 130 }
 131
 132 /* Float element @pos up @heap. */
 133 static void up_heap(log_t **heap, int *heap_offsets, log_t *log, int pos)
 134 {
 135     log_t *tmp;
 136     while ( pos > 1 )
 137     {
 138         if ( heap[pos]->end > heap[pos>>1]->end ) break;
 139         heap_offsets[heap[pos]    - log] = pos >> 1;
 140         heap_offsets[heap[pos>>1] - log] = pos;
 141         tmp = heap[pos];
 142         heap[pos]    = heap[pos>>1];
 143         heap[pos>>1] = tmp;
 144         pos >>= 1;
 145     }
 146 }
 147
 148
 149 /* Delete @entry from @heap. */
 150 static void remove_entry(log_t **heap, int *heap_offsets,
 151                          log_t *log, log_t *entry)
 152 {
 153     int sz = (int)heap[0];
 154     int pos = heap_offsets[entry - log];
 155     heap_offsets[heap[sz] - log] = pos;
 156     heap[pos] = heap[sz];
 157     heap[0] = (void *)(--sz);
 158     if ( (pos > 1) && (heap[pos]->end < heap[pos>>1]->end) )
 159     {
 160         up_heap(heap, heap_offsets, log, pos);
 161     }
 162     else
 163     {
 164         down_heap(heap, heap_offsets, log, pos);
 165     }
 166 }
 167
 168
 169 /* Add new entry @new to @heap. */
 170 static void add_entry(log_t **heap, int *heap_offsets, log_t *log, log_t *new)
 171 {
 172     int sz = (int)heap[0];
 173     heap[0] = (void *)(++sz);
 174     heap_offsets[new - log] = sz;
 175     heap[sz] = new;
 176     up_heap(heap, heap_offsets, log, sz);
 177 }
 178
 179
 180 /*
 181  * This linearisation algorithm is a depth-first search of all feasible
 182  * orderings. At each step, the next available operation is selected.
 183  * The set of "available" operations is those which:
 184  *  (1) have not already been selected on this search path
 185  *  (2) are operations whose results are correct given current state
 186  *      (eg. a failed delete couldn't be selected if the key is in the set!)
 187  *  (3) have start times <= the earliest end time in the set.
 188  * (1) ensures that each operation happens only once. (2) ensures that
 189  * abstract state is consistent between operations. (3) ensures that time
 190  * ordering is conserved.
 191  */
 192 static int linearise_ops_for_key(
 193     log_t *log, int nr_items, log_t **stack,
 194     log_t **cutoff_heap, int *heap_offsets, void **key_state)
 195 {
 196     int i;
 197     log_t **sp = stack;
 198     interval_t cutoff;
 199
 200     /* Construct cutoff heap. */
 201     cutoff_heap[0] = (void *)nr_items;
 202     for ( i = 0; i < nr_items; i++ )
 203     {
 204         cutoff_heap[i+1] = log + i;
 205         heap_offsets[i]  = i+1;
 206     }
 207     for ( i = nr_items>>1; i > 0; i-- )
 208     {
 209         down_heap(cutoff_heap, heap_offsets, log, i);
 210     }
 211
 212     cutoff = cutoff_heap[1]->end;
 213
 214     for ( i = 0; ; )
 215     {
 216         while ( (i < nr_items) && (log[i].start <= cutoff) )
 217         {
 218             if ( !do_op(&log[i], key_state) ) { i++; continue; }
 219
 220             *sp++ = &log[i];
 221
 222             /* Done? */
 223             if ( (sp - stack) == nr_items ) goto success;
 224
 225             remove_entry(cutoff_heap, heap_offsets, log, &log[i]);
 226             cutoff = cutoff_heap[1]->end;
 227             i = 0;
 228         }
 229
 230         /* Failure? */
 231         if ( (sp - stack) == 0 )
 232         {
 233             for ( i = -3; i < nr_items + 3; i++ )
 234             {
 235 #if 1
 236                 printf("%08x -> %08x -- %d: %08x -> %08x\n",
 237                        (unsigned int)log[i].start,
 238                        (unsigned int)log[i].end,
 239                        log[i].data & LOG_KEY_MASK,
 240                        (unsigned int)log[i].old_val,
 241                        (unsigned int)log[i].val);
 242 #endif
 243             }
 244             return(0);
 245         }
 246
 247         i = *--sp - log;
 248         undo_op(&log[i], key_state);
 249         add_entry(cutoff_heap, heap_offsets, log, &log[i]);
 250         cutoff = cutoff_heap[1]->end;
 251         i++;
 252     }
 253
 254  success:
 255     return(1);
 256 }
 257
 258
 259 static void *thread_start(void *arg)
 260 {
 261     unsigned long tid = (unsigned long)arg;
 262     unsigned int our_key;
 263     int ch_start, ch_end, start, end, nr_items, *heap_offsets;
 264     log_t **stack;
 265     log_t **cutoff_heap;
 266     interval_t cutoff;
 267     void *key_state;
 268 #ifdef VERIFY_ORDERINGS
 269     int i;
 270 #endif
 271
 272     stack        = malloc((nr_threads*nr_updates+1)*sizeof(log_t*));
 273     cutoff_heap  = malloc((nr_threads*nr_updates+1)*sizeof(*cutoff_heap));
 274     heap_offsets = malloc((nr_threads*nr_updates+1)*sizeof(*heap_offsets));
 275     if ( !stack || !cutoff_heap || !heap_offsets )
 276     {
 277         fprintf(stderr, "Error allocating space for stacks\n");
 278         return(NULL);
 279     }
 280
 281  again:
 282     pthread_mutex_lock(&key_lock);
 283     our_key = next_key++;
 284     pthread_mutex_unlock(&key_lock);
 285     if ( our_key >= nr_keys ) goto out;
 286
 287     start    = key_offsets[our_key];
 288     end      = key_offsets[our_key+1];
 289     nr_items = end - start;
 290
 291     printf("[Thread %lu] ++ Linearising key %d (%d events)\n",
 292            tid, our_key, nr_items);
 293
 294 #if 0
 295     {
 296         int i;
 297         for ( i = start; i < end; i++ )
 298         {
 299             printf("%04d/%04d -- %08x -> %08x -- %d: %08x -> %08x\n",
 300                    our_key, i - start,
 301                    (unsigned int)global_log[i].start,
 302                    (unsigned int)global_log[i].end,
 303                    global_log[i].data & LOG_KEY_MASK,
 304                    (unsigned int)global_log[i].old_val,
 305                    (unsigned int)global_log[i].val);
 306         }
 307     }
 308 #endif
 309
 310     /*
 311      * We divide operations into independent chunks. A chunk is a maximal
 312      * sequence of operations, ordered on start time, that does not
 313      * overlap with any operation in any other chunk. Clearly, finding
 314      * a linearisation for each chunk produces a total schedule.
 315      */
 316     success[our_key] = 1;
 317     key_state        = 0;
 318     for ( ch_start = start; ch_start < end; ch_start = ch_end )
 319     {
 320         cutoff = global_log[ch_start].end;
 321         for ( ch_end = ch_start; ch_end < end; ch_end++ )
 322         {
 323             if ( global_log[ch_end].start > cutoff ) break;
 324             if ( global_log[ch_end].end > cutoff )
 325                 cutoff = global_log[ch_end].end;
 326         }
 327
 328         /* Linearise chunk ch_start -> ch_end. */
 329         success[our_key] = linearise_ops_for_key(
 330             &global_log[ch_start],
 331             ch_end - ch_start,
 332             &stack[ch_start - start],
 333             cutoff_heap,
 334             heap_offsets,
 335             &key_state);
 336
 337         if ( !success[our_key] )
 338         {
 339             printf("[Thread %lu] -- Linearisation FAILED for key %d\n",
 340                    tid, our_key);
 341             goto again;
 342         }
 343     }
 344
 345     printf("[Thread %lu] -- Linearisation %s for key %d\n",
 346            tid, (success[our_key] ? "found" : "FAILED"), our_key);
 347
 348 #ifdef VERIFY_ORDERINGS
 349     printf("[Thread %lu] ++ Verifying key %d\n", tid, our_key);
 350     cutoff    = 0;
 351     key_state = 0;
 352     for ( i = 0; i < nr_items; i++ )
 353     {
 354         stack[i]->data &= ~LOG_REPLAYED; /* stop valid_op() from choking */
 355         if ( !do_op(stack[i], &key_state) || (stack[i]->end < cutoff) )
 356         {
 357             int j;
 358             fprintf(stderr, "\t*** INTERNAL ERROR: "
 359                     "Assigned ordering is invalid!\n");
 360             for ( j = (i < 2) ? 0 : (i-2); j < i+6; j++ )
 361             {
 362                 printf("%08x -> %08x -- %d: %08x -> %08x\n",
 363                        (unsigned int)stack[j]->start,
 364                        (unsigned int)stack[j]->end,
 365                        stack[j]->data & LOG_KEY_MASK,
 366                        (unsigned int)stack[j]->old_val,
 367                        (unsigned int)stack[j]->val);
 368             }
 369             exit(-1);
 370         }
 371         if ( stack[i]->start > cutoff ) cutoff = stack[i]->start;
 372     }
 373     printf("[Thread %lu] -- Verified key %d\n", tid, our_key);
 374 #endif
 375
 376     goto again;
 377
 378  out:
 379     return(NULL);
 380 }
 381
 382
 383 int main(int argc, char **argv)
 384 {
 385     pthread_t thread[RMAX_THREADS];
 386     int fd, i, j, failed = 0, nr_cpus;
 387     unsigned long log_header[3];
 388
 389     if ( argc != 2 )
 390     {
 391         fprintf(stderr, "%s <log name>\n", argv[0]);
 392         exit(1);
 393     }
 394
 395     nr_cpus = (int)sysconf(_SC_NPROCESSORS_ONLN);
 396     if ( nr_cpus > RMAX_THREADS ) nr_cpus = RMAX_THREADS;
 397
 398     if ( (fd = open(argv[1], O_RDONLY, 0)) == -1 )
 399     {
 400         fprintf(stderr, "Error opening log\n");
 401         exit(-1);
 402     }
 403
 404     /* Grok the log header. */
 405     read(fd, log_header, sizeof(log_header));
 406     nr_threads = log_header[0];
 407     nr_updates = log_header[1];
 408     nr_keys    = log_header[2];
 409     printf("Read log header: nr_updates=%d, nr_threads=%d, nr_keys=%d\n",
 410            nr_updates, nr_threads, nr_keys);
 411
 412     /* Allocate state for processing log entries. */
 413     global_log  = malloc((nr_threads*nr_updates+1)*sizeof(log_t));
 414     key_offsets = malloc((nr_keys+1)*sizeof(*key_offsets));
 415     success     = malloc(nr_keys*sizeof(*success));
 416     if ( !global_log || !key_offsets || !success )
 417     {
 418         fprintf(stderr, "Error allocating space for log\n");
 419         exit(-1);
 420     }
 421
 422     /* Read log entries, and sort into key and timestamp order. */
 423     read(fd, global_log, nr_threads*nr_updates*sizeof(log_t));
 424     global_log[nr_threads*nr_updates].data = LOG_KEY_MASK; /* sentinel */
 425
 426     printf("Sorting logs..."); fflush(stdout);
 427     qsort(global_log, nr_threads*nr_updates, sizeof(log_t), compare);
 428     printf(" done\n");
 429
 430     /* Find offsets of key regions in global table. */
 431     key_offsets[0] = 0;
 432     nr_keys        = 0;
 433     for ( i = 0; i < (nr_threads * nr_updates); i = j )
 434     {
 435         j = i+1;
 436         while ( (global_log[j].data & LOG_KEY_MASK) ==
 437                 (global_log[i].data & LOG_KEY_MASK) ) j++;
 438         key_offsets[++nr_keys] = j;
 439     }
 440
 441     /* Set up a bunch of worker threads.... */
 442     pthread_mutex_init(&key_lock, NULL);
 443     for ( i = 0; i < nr_cpus; i++ )
 444     {
 445         if ( pthread_create(&thread[i], NULL, thread_start, (void *)i) )
 446         {
 447             fprintf(stderr, "Error creating thread %d (%d)\n", i, errno);
 448             exit(1);
 449         }
 450     }
 451
 452     /* ...and wait for them all to complete. */
 453     for ( i = 0; i < nr_cpus; i++ )
 454     {
 455         pthread_join(thread[i], NULL);
 456     }
 457
 458     /* Summarise results from worker threads. */
 459     for ( i = 0; i < nr_keys; i++ )
 460     {
 461         if ( success[i] ) continue;
 462         printf("FAILED on key %d\n", i);
 463         failed++;
 464     }
 465
 466     if ( failed )
 467     {
 468         printf("Failed on %d keys\n", failed);
 469         return(1);
 470     }
 471
 472     printf("All assigned orderings are valid\n");
 473     return(0);
 474 }