ghc/rts/STM.c

   1 /* -----------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team 1998-2005
   4  *
   5  * STM implementation.
   6  *
   7  * Overview
   8  * --------
   9  *
  10  * See the PPoPP 2005 paper "Composable memory transactions".  In summary,
  11  * each transcation has a TRec (transaction record) holding entries for each of the
  12  * TVars (transactional variables) that it has accessed.  Each entry records
  13  * (a) the TVar, (b) the expected value seen in the TVar, (c) the new value that
  14  * the transaction wants to write to the TVar, (d) during commit, the identity of
  15  * the TRec that wrote the expected value.
  16  *
  17  * Separate TRecs are used for each level in a nest of transactions.  This allows
  18  * a nested transaction to be aborted without condemning its enclosing transactions.
  19  * This is needed in the implementation of catchRetry.  Note that the "expected value"
  20  * in a nested transaction's TRec is the value expected to be *held in memory* if
  21  * the transaction commits -- not the "new value" stored in one of the enclosing
  22  * transactions.  This means that validation can be done without searching through
  23  * a nest of TRecs.
  24  *
  25  * Concurrency control
  26  * -------------------
  27  *
  28  * Three different concurrency control schemes can be built according to the settings
  29  * in STM.h:
  30  *
  31  * STM_UNIPROC assumes that the caller serialises invocations on the STM interface.
  32  * In the Haskell RTS this means it is suitable only for non-SMP builds.
  33  *
  34  * STM_CG_LOCK uses coarse-grained locking -- a single 'stm lock' is acquired during
  35  * an invocation on the STM interface.  Note that this does not mean that
  36  * transactions are simply serialized -- the lock is only held *within* the
  37  * implementation of stmCommitTransaction, stmWait etc.
  38  *
  39  * STM_FG_LOCKS uses fine-grained locking -- locking is done on a per-TVar basis
  40  * and, when committing a transaction, no locks are acquired for TVars that have
  41  * been read but not updated.
  42  *
  43  * Concurrency control is implemented in the functions:
  44  *
  45  *    lock_stm
  46  *    unlock_stm
  47  *    lock_tvar / cond_lock_tvar
  48  *    unlock_tvar
  49  *
  50  * The choice between STM_UNIPROC / STM_CG_LOCK / STM_FG_LOCKS affects the
  51  * implementation of these functions.
  52  *
  53  * lock_stm & unlock_stm are straightforward : they acquire a simple spin-lock
  54  * using STM_CG_LOCK, and otherwise they are no-ops.
  55  *
  56  * lock_tvar / cond_lock_tvar and unlock_tvar are more complex because they
  57  * have other effects (present in STM_UNIPROC and STM_CG_LOCK builds) as well
  58  * as the actual business of maniupultaing a lock (present only in STM_FG_LOCKS
  59  * builds).  This is because locking a TVar is implemented by writing the lock
  60  * holder's TRec into the TVar's current_value field:
  61  *
  62  *   lock_tvar - lock a specified TVar (STM_FG_LOCKS only), returning the value
  63  *               it contained.
  64  *
  65  *   cond_lock_tvar - lock a specified TVar (STM_FG_LOCKS only) if it
  66  *               contains a specified value.  Return TRUE if this succeeds,
  67  *               FALSE otherwise.
  68  *
  69  *   unlock_tvar - release the lock on a specified TVar (STM_FG_LOCKS only),
  70  *               storing a specified value in place of the lock entry.
  71  *
  72  * Using these operations, the typcial pattern of a commit/validate/wait operation
  73  * is to (a) lock the STM, (b) lock all the TVars being updated, (c) check that
  74  * the TVars that were only read from still contain their expected values,
  75  * (d) release the locks on the TVars, writing updates to them in the case of a
  76  * commit, (e) unlock the STM.
  77  *
  78  * Queues of waiting threads hang off the first_wait_queue_entry field of each
  79  * TVar.  This may only be manipulated when holding that TVar's lock.  In
  80  * particular, when a thread is putting itself to sleep, it mustn't release
  81  * the TVar's lock until it has added itself to the wait queue and marked its
  82  * TSO as BlockedOnSTM -- this makes sure that other threads will know to wake it.
  83  *
  84  * ---------------------------------------------------------------------------*/
  85
  86 #include "PosixSource.h"
  87 #include "Rts.h"
  88 #include "RtsFlags.h"
  89 #include "RtsUtils.h"
  90 #include "Schedule.h"
  91 #include "SMP.h"
  92 #include "STM.h"
  93 #include "Storage.h"
  94
  95 #include <stdlib.h>
  96 #include <stdio.h>
  97
  98 #define TRUE 1
  99 #define FALSE 0
 100
 101 // ACQ_ASSERT is used for assertions which are only required for SMP builds with
 102 // fine-grained locking.
 103
 104 #if defined(STM_FG_LOCKS)
 105 #define ACQ_ASSERT(_X) ASSERT(_X)
 106 #define NACQ_ASSERT(_X) /*Nothing*/
 107 #else
 108 #define ACQ_ASSERT(_X) /*Nothing*/
 109 #define NACQ_ASSERT(_X) ASSERT(_X)
 110 #endif
 111
 112 /*......................................................................*/
 113
 114 // If SHAKE is defined then validation will sometime spuriously fail.  They helps test
 115 // unusualy code paths if genuine contention is rare
 116
 117 #if defined(DEBUG)
 118 #define SHAKE
 119 #define TRACE(_x...) IF_DEBUG(stm, debugBelch ( _x ))
 120 #else
 121 #define TRACE(_x...) /*Nothing*/
 122 #endif
 123
 124 #ifdef SHAKE
 125 static const int do_shake = TRUE;
 126 #else
 127 static const int do_shake = FALSE;
 128 #endif
 129 static int shake_ctr = 0;
 130 static int shake_lim = 1;
 131
 132 static int shake(void) {
 133   if (do_shake) {
 134     if (((shake_ctr++) % shake_lim) == 0) {
 135       shake_ctr = 1;
 136       shake_lim ++;
 137       return TRUE;
 138     }
 139     return FALSE;
 140   } else {
 141     return FALSE;
 142   }
 143 }
 144
 145 /*......................................................................*/
 146
 147 // Helper macros for iterating over entries within a transaction
 148 // record
 149
 150 #define FOR_EACH_ENTRY(_t,_x,CODE) do {                                         \
 151   StgTRecHeader *__t = (_t);                                                    \
 152   StgTRecChunk *__c = __t -> current_chunk;                                     \
 153   StgWord __limit = __c -> next_entry_idx;                                      \
 154   TRACE("%p : FOR_EACH_ENTRY, current_chunk=%p limit=%ld\n", __t, __c, __limit); \
 155   while (__c != END_STM_CHUNK_LIST) {                                           \
 156     StgWord __i;                                                                \
 157     for (__i = 0; __i < __limit; __i ++) {                                      \
 158       TRecEntry *_x = &(__c -> entries[__i]);                                   \
 159       do { CODE } while (0);                                                    \
 160     }                                                                           \
 161     __c = __c -> prev_chunk;                                                    \
 162     __limit = TREC_CHUNK_NUM_ENTRIES;                                           \
 163   }                                                                             \
 164  exit_for_each:                                                                 \
 165   if (FALSE) goto exit_for_each;                                                \
 166 } while (0)
 167
 168 #define BREAK_FOR_EACH goto exit_for_each
 169
 170 /*......................................................................*/
 171
 172 #if defined(STM_UNIPROC)
 173 static const StgBool use_read_phase = FALSE;
 174
 175 static void lock_stm(StgTRecHeader *trec STG_UNUSED) {
 176   TRACE("%p : lock_stm()\n", trec);
 177 }
 178
 179 static void unlock_stm(StgTRecHeader *trec STG_UNUSED) {
 180   TRACE("%p : unlock_stm()\n", trec);
 181 }
 182
 183 static StgClosure *lock_tvar(StgTRecHeader *trec STG_UNUSED,
 184                              StgTVar *s STG_UNUSED) {
 185   StgClosure *result;
 186   TRACE("%p : lock_tvar(%p)\n", trec, s);
 187   result = s -> current_value;
 188   return result;
 189 }
 190
 191 static void unlock_tvar(StgTRecHeader *trec STG_UNUSED,
 192                         StgTVar *s STG_UNUSED,
 193                         StgClosure *c,
 194                         StgBool force_update) {
 195   TRACE("%p : unlock_tvar(%p)\n", trec, s);
 196   if (force_update) {
 197     s -> current_value = c;
 198   }
 199 }
 200
 201 static StgBool cond_lock_tvar(StgTRecHeader *trec STG_UNUSED,
 202                               StgTVar *s STG_UNUSED,
 203                               StgClosure *expected) {
 204   StgClosure *result;
 205   TRACE("%p : cond_lock_tvar(%p, %p)\n", trec, s, expected);
 206   result = s -> current_value;
 207   TRACE("%p : %d\n", (result == expected) ? "success" : "failure");
 208   return (result == expected);
 209 }
 210 #endif
 211
 212 #if defined(STM_CG_LOCK) /*........................................*/
 213
 214 static const StgBool use_read_phase = FALSE;
 215 static volatile StgTRecHeader *smp_locked = NULL;
 216
 217 static void lock_stm(StgTRecHeader *trec) {
 218   while (cas(&smp_locked, NULL, trec) != NULL) { }
 219   TRACE("%p : lock_stm()\n", trec);
 220 }
 221
 222 static void unlock_stm(StgTRecHeader *trec STG_UNUSED) {
 223   TRACE("%p : unlock_stm()\n", trec);
 224   ASSERT (smp_locked == trec);
 225   smp_locked = 0;
 226 }
 227
 228 static StgClosure *lock_tvar(StgTRecHeader *trec STG_UNUSED,
 229                              StgTVar *s STG_UNUSED) {
 230   StgClosure *result;
 231   TRACE("%p : lock_tvar(%p)\n", trec, s);
 232   ASSERT (smp_locked == trec);
 233   result = s -> current_value;
 234   return result;
 235 }
 236
 237 static void *unlock_tvar(StgTRecHeader *trec STG_UNUSED,
 238                          StgTVar *s STG_UNUSED,
 239                          StgClosure *c,
 240                          StgBool force_update) {
 241   TRACE("%p : unlock_tvar(%p, %p)\n", trec, s, c);
 242   ASSERT (smp_locked == trec);
 243   if (force_update) {
 244     s -> current_value = c;
 245   }
 246 }
 247
 248 static StgBool cond_lock_tvar(StgTRecHeader *trec STG_UNUSED,
 249                                StgTVar *s STG_UNUSED,
 250                                StgClosure *expected) {
 251   StgClosure *result;
 252   TRACE("%p : cond_lock_tvar(%p, %p)\n", trec, s, expected);
 253   ASSERT (smp_locked == trec);
 254   result = s -> current_value;
 255   TRACE("%p : %d\n", result ? "success" : "failure");
 256   return (result == expected);
 257 }
 258 #endif
 259
 260 #if defined(STM_FG_LOCKS) /*...................................*/
 261
 262 static const StgBool use_read_phase = TRUE;
 263
 264 static void lock_stm(StgTRecHeader *trec STG_UNUSED) {
 265   TRACE("%p : lock_stm()\n", trec);
 266 }
 267
 268 static void unlock_stm(StgTRecHeader *trec STG_UNUSED) {
 269   TRACE("%p : unlock_stm()\n", trec);
 270 }
 271
 272 static StgClosure *lock_tvar(StgTRecHeader *trec,
 273                              StgTVar *s STG_UNUSED) {
 274   StgClosure *result;
 275   TRACE("%p : lock_tvar(%p)\n", trec, s);
 276   do {
 277     do {
 278       result = s -> current_value;
 279     } while (GET_INFO(result) == &stg_TREC_HEADER_info);
 280   } while (cas(&(s -> current_value), result, trec) != result);
 281   return result;
 282 }
 283
 284 static void unlock_tvar(StgTRecHeader *trec STG_UNUSED,
 285                         StgTVar *s,
 286                         StgClosure *c,
 287                         StgBool force_update STG_UNUSED) {
 288   TRACE("%p : unlock_tvar(%p, %p)\n", trec, s, c);
 289   ASSERT(s -> current_value == trec);
 290   s -> current_value = c;
 291 }
 292
 293 static StgBool cond_lock_tvar(StgTRecHeader *trec,
 294                               StgTVar *s,
 295                               StgClosure *expected) {
 296   StgClosure *result;
 297   TRACE("%p : cond_lock_tvar(%p, %p)\n", trec, s, expected);
 298   result = cas(&(s -> current_value), expected, trec);
 299   TRACE("%p : %s\n", trec, result ? "success" : "failure");
 300   return (result == expected);
 301 }
 302 #endif
 303
 304 /*......................................................................*/
 305
 306 // Helper functions for thread blocking and unblocking
 307
 308 static void park_tso(StgTSO *tso) {
 309   ACQUIRE_LOCK(&sched_mutex);
 310   ASSERT(tso -> why_blocked == NotBlocked);
 311   tso -> why_blocked = BlockedOnSTM;
 312   tso -> block_info.closure = (StgClosure *) END_TSO_QUEUE;
 313   RELEASE_LOCK(&sched_mutex);
 314   TRACE("park_tso on tso=%p\n", tso);
 315 }
 316
 317 static void unpark_tso(StgTSO *tso) {
 318   // We will continue unparking threads while they remain on one of the wait
 319   // queues: it's up to the thread itself to remove it from the wait queues
 320   // if it decides to do so when it is scheduled.
 321   if (tso -> why_blocked == BlockedOnSTM) {
 322     TRACE("unpark_tso on tso=%p\n", tso);
 323     ACQUIRE_LOCK(&sched_mutex);
 324     tso -> why_blocked = NotBlocked;
 325     PUSH_ON_RUN_QUEUE(tso);
 326     RELEASE_LOCK(&sched_mutex);
 327   } else {
 328     TRACE("spurious unpark_tso on tso=%p\n", tso);
 329   }
 330 }
 331
 332 static void unpark_waiters_on(StgTVar *s) {
 333   StgTVarWaitQueue *q;
 334   TRACE("unpark_waiters_on tvar=%p\n", s);
 335   for (q = s -> first_wait_queue_entry;
 336        q != END_STM_WAIT_QUEUE;
 337        q = q -> next_queue_entry) {
 338     unpark_tso(q -> waiting_tso);
 339   }
 340 }
 341
 342 /*......................................................................*/
 343
 344 // Helper functions for allocation and initialization
 345
 346 static StgTVarWaitQueue *new_stg_tvar_wait_queue(StgRegTable *reg,
 347                                                  StgTSO *waiting_tso) {
 348   StgTVarWaitQueue *result;
 349   result = (StgTVarWaitQueue *)allocateLocal(reg, sizeofW(StgTVarWaitQueue));
 350   SET_HDR (result, &stg_TVAR_WAIT_QUEUE_info, CCS_SYSTEM);
 351   result -> waiting_tso = waiting_tso;
 352   return result;
 353 }
 354
 355 static StgTRecChunk *new_stg_trec_chunk(StgRegTable *reg) {
 356   StgTRecChunk *result;
 357   result = (StgTRecChunk *)allocateLocal(reg, sizeofW(StgTRecChunk));
 358   SET_HDR (result, &stg_TREC_CHUNK_info, CCS_SYSTEM);
 359   result -> prev_chunk = END_STM_CHUNK_LIST;
 360   result -> next_entry_idx = 0;
 361   return result;
 362 }
 363
 364 static StgTRecHeader *new_stg_trec_header(StgRegTable *reg,
 365                                           StgTRecHeader *enclosing_trec) {
 366   StgTRecHeader *result;
 367   result = (StgTRecHeader *) allocateLocal(reg, sizeofW(StgTRecHeader));
 368   SET_HDR (result, &stg_TREC_HEADER_info, CCS_SYSTEM);
 369
 370   result -> enclosing_trec = enclosing_trec;
 371   result -> current_chunk = new_stg_trec_chunk(reg);
 372
 373   if (enclosing_trec == NO_TREC) {
 374     result -> state = TREC_ACTIVE;
 375   } else {
 376     ASSERT(enclosing_trec -> state == TREC_ACTIVE ||
 377            enclosing_trec -> state == TREC_CONDEMNED);
 378     result -> state = enclosing_trec -> state;
 379   }
 380
 381   return result;
 382 }
 383
 384 /*......................................................................*/
 385
 386 // Helper functions for managing waiting lists
 387
 388 static void build_wait_queue_entries_for_trec(StgRegTable *reg,
 389                                       StgTSO *tso,
 390                                       StgTRecHeader *trec) {
 391   ASSERT(trec != NO_TREC);
 392   ASSERT(trec -> enclosing_trec == NO_TREC);
 393   ASSERT(trec -> state == TREC_ACTIVE);
 394
 395   TRACE("%p : build_wait_queue_entries_for_trec()\n", trec);
 396
 397   FOR_EACH_ENTRY(trec, e, {
 398     StgTVar *s;
 399     StgTVarWaitQueue *q;
 400     StgTVarWaitQueue *fq;
 401     s = e -> tvar;
 402     TRACE("%p : adding tso=%p to wait queue for tvar=%p\n", trec, tso, s);
 403     ACQ_ASSERT(s -> current_value == trec);
 404     NACQ_ASSERT(s -> current_value == e -> expected_value);
 405     fq = s -> first_wait_queue_entry;
 406     q = new_stg_tvar_wait_queue(reg, tso);
 407     q -> next_queue_entry = fq;
 408     q -> prev_queue_entry = END_STM_WAIT_QUEUE;
 409     if (fq != END_STM_WAIT_QUEUE) {
 410       fq -> prev_queue_entry = q;
 411     }
 412     s -> first_wait_queue_entry = q;
 413     e -> new_value = (StgClosure *) q;
 414   });
 415 }
 416
 417 static void remove_wait_queue_entries_for_trec(StgTRecHeader *trec) {
 418   ASSERT(trec != NO_TREC);
 419   ASSERT(trec -> enclosing_trec == NO_TREC);
 420   ASSERT(trec -> state == TREC_WAITING ||
 421          trec -> state == TREC_CONDEMNED);
 422
 423   TRACE("%p : remove_wait_queue_entries_for_trec()\n", trec);
 424
 425   FOR_EACH_ENTRY(trec, e, {
 426     StgTVar *s;
 427     StgTVarWaitQueue *pq;
 428     StgTVarWaitQueue *nq;
 429     StgTVarWaitQueue *q;
 430     s = e -> tvar;
 431     StgClosure *saw = lock_tvar(trec, s);
 432     q = (StgTVarWaitQueue *) (e -> new_value);
 433     TRACE("%p : removing tso=%p from wait queue for tvar=%p\n", trec, q -> waiting_tso, s);
 434     ACQ_ASSERT(s -> current_value == trec);
 435     nq = q -> next_queue_entry;
 436     pq = q -> prev_queue_entry;
 437     if (nq != END_STM_WAIT_QUEUE) {
 438       nq -> prev_queue_entry = pq;
 439     }
 440     if (pq != END_STM_WAIT_QUEUE) {
 441       pq -> next_queue_entry = nq;
 442     } else {
 443       ASSERT (s -> first_wait_queue_entry == q);
 444       s -> first_wait_queue_entry = nq;
 445     }
 446     unlock_tvar(trec, s, saw, FALSE);
 447   });
 448 }
 449
 450 /*......................................................................*/
 451
 452 static TRecEntry *get_new_entry(StgRegTable *reg,
 453                                 StgTRecHeader *t) {
 454   TRecEntry *result;
 455   StgTRecChunk *c;
 456   int i;
 457
 458   c = t -> current_chunk;
 459   i = c -> next_entry_idx;
 460   ASSERT(c != END_STM_CHUNK_LIST);
 461
 462   if (i < TREC_CHUNK_NUM_ENTRIES) {
 463     // Continue to use current chunk
 464     result = &(c -> entries[i]);
 465     c -> next_entry_idx ++;
 466   } else {
 467     // Current chunk is full: allocate a fresh one
 468     StgTRecChunk *nc;
 469     nc = new_stg_trec_chunk(reg);
 470     nc -> prev_chunk = c;
 471     nc -> next_entry_idx = 1;
 472     t -> current_chunk = nc;
 473     result = &(nc -> entries[0]);
 474   }
 475
 476   return result;
 477 }
 478
 479 /*......................................................................*/
 480
 481 static void merge_update_into(StgRegTable *reg,
 482                               StgTRecHeader *t,
 483                               StgTVar *tvar,
 484                               StgClosure *expected_value,
 485                               StgClosure *new_value) {
 486   int found;
 487
 488   // Look for an entry in this trec
 489   found = FALSE;
 490   FOR_EACH_ENTRY(t, e, {
 491     StgTVar *s;
 492     s = e -> tvar;
 493     if (s == tvar) {
 494       found = TRUE;
 495       if (e -> expected_value != expected_value) {
 496         // Must abort if the two entries start from different values
 497         TRACE("%p : entries inconsistent at %p (%p vs %p)\n",
 498               t, tvar, e -> expected_value, expected_value);
 499         t -> state = TREC_CONDEMNED;
 500       }
 501       e -> new_value = new_value;
 502       BREAK_FOR_EACH;
 503     }
 504   });
 505
 506   if (!found) {
 507     // No entry so far in this trec
 508     TRecEntry *ne;
 509     ne = get_new_entry(reg, t);
 510     ne -> tvar = tvar;
 511     ne -> expected_value = expected_value;
 512     ne -> new_value = new_value;
 513   }
 514 }
 515
 516 /*......................................................................*/
 517
 518 static StgBool entry_is_update(TRecEntry *e) {
 519   StgBool result;
 520   result = (e -> expected_value != e -> new_value);
 521   return result;
 522 }
 523
 524 static StgBool entry_is_read_only(TRecEntry *e) {
 525   StgBool result;
 526   result = (e -> expected_value == e -> new_value);
 527   return result;
 528 }
 529
 530 static StgBool tvar_is_locked(StgTVar *s, StgTRecHeader *h) {
 531   StgClosure *c;
 532   StgBool result;
 533   c = s -> current_value;
 534   result = (c == (StgClosure *) h);
 535   return result;
 536 }
 537
 538 // revert_ownership : release a lock on a TVar, storing back
 539 // the value that it held when the lock was acquired.  "revert_all"
 540 // is set in stmWait and stmReWait when we acquired locks on all of
 541 // the TVars involved.  "revert_all" is not set in commit operations
 542 // where we don't lock TVars that have been read from but not updated.
 543
 544 static void revert_ownership(StgTRecHeader *trec STG_UNUSED,
 545                              StgBool revert_all STG_UNUSED) {
 546 #if defined(STM_FG_LOCKS)
 547   FOR_EACH_ENTRY(trec, e, {
 548     if (revert_all || entry_is_update(e)) {
 549       StgTVar *s;
 550       s = e -> tvar;
 551       if (tvar_is_locked(s, trec)) {
 552         unlock_tvar(trec, s, e -> expected_value, TRUE);
 553       }
 554     }
 555   });
 556 #endif
 557 }
 558
 559 /*......................................................................*/
 560
 561 // validate_and_acquire_ownership : this performs the twin functions
 562 // of checking that the TVars referred to by entries in trec hold the
 563 // expected values and:
 564 //
 565 //   - locking the TVar (on updated TVars during commit, or all TVars
 566 //     during wait)
 567 //
 568 //   - recording the identity of the TRec who wrote the value seen in the
 569 //     TVar (on non-updated TVars during commit).  These values are
 570 //     stashed in the TRec entries and are then checked in check_read_only
 571 //     to ensure that an atomic snapshot of all of these locations has been
 572 //     seen.
 573
 574 static StgBool validate_and_acquire_ownership (StgTRecHeader *trec,
 575                                                int acquire_all,
 576                                                int retain_ownership) {
 577   StgBool result;
 578
 579   if (shake()) {
 580     TRACE("%p : shake, pretending trec is invalid when it may not be\n", trec);
 581     return FALSE;
 582   }
 583
 584   ASSERT ((trec -> state == TREC_ACTIVE) ||
 585           (trec -> state == TREC_WAITING) ||
 586           (trec -> state == TREC_CONDEMNED));
 587   result = !((trec -> state) == TREC_CONDEMNED);
 588   if (result) {
 589     FOR_EACH_ENTRY(trec, e, {
 590       StgTVar *s;
 591       s = e -> tvar;
 592       if (acquire_all || entry_is_update(e)) {
 593         TRACE("%p : trying to acquire %p\n", trec, s);
 594         if (!cond_lock_tvar(trec, s, e -> expected_value)) {
 595           TRACE("%p : failed to acquire %p\n", trec, s);
 596           result = FALSE;
 597           BREAK_FOR_EACH;
 598         }
 599       } else {
 600         TRACE("%p : will need to check %p\n", trec, s);
 601         if (s -> current_value != e -> expected_value) {
 602           TRACE("%p : doesn't match\n", trec);
 603           result = FALSE;
 604           BREAK_FOR_EACH;
 605         }
 606         e -> saw_update_by = s -> last_update_by;
 607         if (s -> current_value != e -> expected_value) {
 608           TRACE("%p : doesn't match (race)\n", trec);
 609           result = FALSE;
 610           BREAK_FOR_EACH;
 611         } else {
 612           TRACE("%p : need to check update by %p\n", trec, e -> saw_update_by);
 613         }
 614       }
 615     });
 616   }
 617
 618   if ((!result) || (!retain_ownership)) {
 619     revert_ownership(trec, acquire_all);
 620   }
 621
 622   return result;
 623 }
 624
 625 // check_read_only : check that we've seen an atomic snapshot of the
 626 // non-updated TVars accessed by a trec.  This checks that the last TRec to
 627 // commit an update to the TVar is unchanged since the value was stashed in
 628 // validate_and_acquire_ownership.  If no udpate is seen to any TVar than
 629 // all of them contained their expected values at the start of the call to
 630 // check_read_only.
 631 //
 632 // The paper "Concurrent programming without locks" (under submission), or
 633 // Keir Fraser's PhD dissertation "Practical lock-free programming" discuss
 634 // this kind of algorithm.
 635
 636 static StgBool check_read_only(StgTRecHeader *trec) {
 637   StgBool result = TRUE;
 638
 639   FOR_EACH_ENTRY(trec, e, {
 640     StgTVar *s;
 641     s = e -> tvar;
 642     if (entry_is_read_only(e)) {
 643       TRACE("%p : check_read_only for TVar %p, saw %p\n", trec, s, e -> saw_update_by);
 644       if (s -> last_update_by != e -> saw_update_by) {
 645         // ||s -> current_value != e -> expected_value) {
 646         TRACE("%p : mismatch\n", trec);
 647         result = FALSE;
 648         BREAK_FOR_EACH;
 649       }
 650     }
 651   });
 652
 653   return result;
 654 }
 655
 656
 657 /************************************************************************/
 658
 659 void stmPreGCHook() {
 660   lock_stm(NO_TREC);
 661   TRACE("stmPreGCHook\n");
 662   unlock_stm(NO_TREC);
 663 }
 664
 665 /************************************************************************/
 666
 667 void initSTM() {
 668   TRACE("initSTM, NO_TREC=%p\n", NO_TREC);
 669 }
 670
 671 /*......................................................................*/
 672
 673 StgTRecHeader *stmStartTransaction(StgRegTable *reg,
 674                                    StgTRecHeader *outer) {
 675   StgTRecHeader *t;
 676   TRACE("%p : stmStartTransaction\n", outer);
 677   t = new_stg_trec_header(reg, outer);
 678   TRACE("%p : stmStartTransaction()=%p\n", outer, t);
 679   return t;
 680 }
 681
 682 /*......................................................................*/
 683
 684 void stmAbortTransaction(StgTRecHeader *trec) {
 685   TRACE("%p : stmAbortTransaction\n", trec);
 686   ASSERT (trec != NO_TREC);
 687   ASSERT ((trec -> state == TREC_ACTIVE) ||
 688           (trec -> state == TREC_WAITING) ||
 689           (trec -> state == TREC_CONDEMNED));
 690
 691   lock_stm(trec);
 692   if (trec -> state == TREC_WAITING) {
 693     ASSERT (trec -> enclosing_trec == NO_TREC);
 694     TRACE("%p : stmAbortTransaction aborting waiting transaction\n", trec);
 695     remove_wait_queue_entries_for_trec(trec);
 696   }
 697   trec -> state = TREC_ABORTED;
 698   unlock_stm(trec);
 699
 700   TRACE("%p : stmAbortTransaction done\n", trec);
 701 }
 702
 703 /*......................................................................*/
 704
 705 void stmCondemnTransaction(StgTRecHeader *trec) {
 706   TRACE("%p : stmCondemnTransaction\n", trec);
 707   ASSERT (trec != NO_TREC);
 708   ASSERT ((trec -> state == TREC_ACTIVE) ||
 709           (trec -> state == TREC_WAITING) ||
 710           (trec -> state == TREC_CONDEMNED));
 711
 712   lock_stm(trec);
 713   if (trec -> state == TREC_WAITING) {
 714     ASSERT (trec -> enclosing_trec == NO_TREC);
 715     TRACE("%p : stmCondemnTransaction condemning waiting transaction\n", trec);
 716     remove_wait_queue_entries_for_trec(trec);
 717   }
 718   trec -> state = TREC_CONDEMNED;
 719   unlock_stm(trec);
 720
 721   TRACE("%p : stmCondemnTransaction done\n", trec);
 722 }
 723
 724 /*......................................................................*/
 725
 726 StgTRecHeader *stmGetEnclosingTRec(StgTRecHeader *trec) {
 727   StgTRecHeader *outer;
 728   TRACE("%p : stmGetEnclosingTRec\n", trec);
 729   outer = trec -> enclosing_trec;
 730   TRACE("%p : stmGetEnclosingTRec()=%p\n", trec, outer);
 731   return outer;
 732 }
 733
 734 /*......................................................................*/
 735
 736 StgBool stmValidateNestOfTransactions(StgTRecHeader *trec) {
 737   StgTRecHeader *t;
 738   StgBool result;
 739
 740   TRACE("%p : stmValidateNestOfTransactions\n", trec);
 741   ASSERT(trec != NO_TREC);
 742   ASSERT((trec -> state == TREC_ACTIVE) ||
 743          (trec -> state == TREC_WAITING) ||
 744          (trec -> state == TREC_CONDEMNED));
 745
 746   lock_stm(trec);
 747
 748   t = trec;
 749   result = TRUE;
 750   while (t != NO_TREC) {
 751     result &= validate_and_acquire_ownership(t, TRUE, FALSE);
 752     t = t -> enclosing_trec;
 753   }
 754
 755   if (!result && trec -> state != TREC_WAITING) {
 756     trec -> state = TREC_CONDEMNED;
 757   }
 758
 759   unlock_stm(trec);
 760
 761   TRACE("%p : stmValidateNestOfTransactions()=%d\n", trec, result);
 762   return result;
 763 }
 764
 765 /*......................................................................*/
 766
 767 StgBool stmCommitTransaction(StgRegTable *reg STG_UNUSED, StgTRecHeader *trec) {
 768   int result;
 769   TRACE("%p : stmCommitTransaction()\n", trec);
 770   ASSERT (trec != NO_TREC);
 771   ASSERT (trec -> enclosing_trec == NO_TREC);
 772   ASSERT ((trec -> state == TREC_ACTIVE) ||
 773           (trec -> state == TREC_CONDEMNED));
 774
 775   lock_stm(trec);
 776   result = validate_and_acquire_ownership(trec, (!use_read_phase), TRUE);
 777   if (result) {
 778     // We now know that all the updated locations hold their expected values.
 779     ASSERT (trec -> state == TREC_ACTIVE);
 780
 781     if (use_read_phase) {
 782       TRACE("%p : doing read check\n", trec);
 783       result = check_read_only(trec);
 784     }
 785
 786     if (result) {
 787       // We now know that all of the read-only locations held their exepcted values
 788       // at the end of the call to validate_and_acquire_ownership.  This forms the
 789       // linearization point of the commit.
 790
 791       TRACE("%p : read-check succeeded\n", trec);
 792       FOR_EACH_ENTRY(trec, e, {
 793         StgTVar *s;
 794         s = e -> tvar;
 795         if (e -> new_value != e -> expected_value) {
 796           // Entry is an update: write the value back to the TVar, unlocking it if
 797           // necessary.
 798
 799           ACQ_ASSERT(tvar_is_locked(s, trec));
 800           TRACE("%p : writing %p to %p, waking waiters\n", trec, e -> new_value, s);
 801           unpark_waiters_on(s);
 802           s -> last_update_by = trec;
 803           unlock_tvar(trec, s, e -> new_value, TRUE);
 804         }
 805         ACQ_ASSERT(!tvar_is_locked(s, trec));
 806       });
 807     } else {
 808       revert_ownership(trec, FALSE);
 809     }
 810   }
 811
 812   unlock_stm(trec);
 813
 814   TRACE("%p : stmCommitTransaction()=%d\n", trec, result);
 815
 816   return result;
 817 }
 818
 819 /*......................................................................*/
 820
 821 StgBool stmCommitNestedTransaction(StgRegTable *reg, StgTRecHeader *trec) {
 822   StgTRecHeader *et;
 823   int result;
 824   ASSERT (trec != NO_TREC && trec -> enclosing_trec != NO_TREC);
 825   TRACE("%p : stmCommitNestedTransaction() into %p\n", trec, trec -> enclosing_trec);
 826   ASSERT ((trec -> state == TREC_ACTIVE) || (trec -> state == TREC_CONDEMNED));
 827
 828   lock_stm(trec);
 829
 830   et = trec -> enclosing_trec;
 831   result = validate_and_acquire_ownership(trec, FALSE, TRUE);
 832   if (result) {
 833     // We now know that all the updated locations hold their expected values.
 834
 835     if (use_read_phase) {
 836       TRACE("%p : doing read check\n", trec);
 837       result = check_read_only(trec);
 838     }
 839     if (result) {
 840       // We now know that all of the read-only locations held their exepcted values
 841       // at the end of the call to validate_and_acquire_ownership.  This forms the
 842       // linearization point of the commit.
 843
 844       if (result) {
 845         TRACE("%p : read-check succeeded\n", trec);
 846         FOR_EACH_ENTRY(trec, e, {
 847           // Merge each entry into the enclosing transaction record, release all
 848           // locks.
 849
 850           StgTVar *s;
 851           s = e -> tvar;
 852           if (entry_is_update(e)) {
 853             unlock_tvar(trec, s, e -> expected_value, FALSE);
 854           }
 855           merge_update_into(reg, et, s, e -> expected_value, e -> new_value);
 856           ACQ_ASSERT(s -> current_value != trec);
 857         });
 858       } else {
 859         revert_ownership(trec, FALSE);
 860       }
 861     }
 862   }
 863
 864   unlock_stm(trec);
 865
 866   TRACE("%p : stmCommitNestedTransaction()=%d\n", trec, result);
 867
 868   return result;
 869 }
 870
 871 /*......................................................................*/
 872
 873 StgBool stmWait(StgRegTable *reg, StgTSO *tso, StgTRecHeader *trec) {
 874   int result;
 875   TRACE("%p : stmWait(%p)\n", trec, tso);
 876   ASSERT (trec != NO_TREC);
 877   ASSERT (trec -> enclosing_trec == NO_TREC);
 878   ASSERT ((trec -> state == TREC_ACTIVE) ||
 879           (trec -> state == TREC_CONDEMNED));
 880
 881   lock_stm(trec);
 882   result = validate_and_acquire_ownership(trec, TRUE, TRUE);
 883   if (result) {
 884     // The transaction is valid so far so we can actually start waiting.
 885     // (Otherwise the transaction was not valid and the thread will have to
 886     // retry it).
 887
 888     // Put ourselves to sleep.  We retain locks on all the TVars involved
 889     // until we are sound asleep : (a) on the wait queues, (b) BlockedOnSTM
 890     // in the TSO, (c) TREC_WAITING in the Trec.
 891     build_wait_queue_entries_for_trec(reg, tso, trec);
 892     park_tso(tso);
 893     trec -> state = TREC_WAITING;
 894
 895     // As soon as we start releasing ownership, another thread may find us
 896     // and wake us up.  This may happen even before we have finished
 897     // releasing ownership.
 898     revert_ownership(trec, TRUE);
 899   }
 900
 901   unlock_stm(trec);
 902
 903   TRACE("%p : stmWait(%p)=%d\n", trec, tso, result);
 904   return result;
 905 }
 906
 907 /*......................................................................*/
 908
 909 StgBool stmReWait(StgTSO *tso) {
 910   int result;
 911   StgTRecHeader *trec = tso->trec;
 912
 913   TRACE("%p : stmReWait\n", trec);
 914   ASSERT (trec != NO_TREC);
 915   ASSERT (trec -> enclosing_trec == NO_TREC);
 916   ASSERT ((trec -> state == TREC_WAITING) ||
 917           (trec -> state == TREC_CONDEMNED));
 918
 919   lock_stm(trec);
 920   result = validate_and_acquire_ownership(trec, TRUE, TRUE);
 921   TRACE("%p : validation %s\n", trec, result ? "succeeded" : "failed");
 922   if (result) {
 923     // The transaction remains valid -- do nothing because it is already on
 924     // the wait queues
 925     ASSERT (trec -> state == TREC_WAITING);
 926     park_tso(tso);
 927     revert_ownership(trec, TRUE);
 928   } else {
 929     // The transcation has become invalid.  We can now remove it from the wait
 930     // queues.
 931     if (trec -> state != TREC_CONDEMNED) {
 932       remove_wait_queue_entries_for_trec (trec);
 933     }
 934
 935   }
 936   unlock_stm(trec);
 937
 938   TRACE("%p : stmReWait()=%d\n", trec, result);
 939   return result;
 940 }
 941
 942 /*......................................................................*/
 943
 944 static TRecEntry *get_entry_for(StgTRecHeader *trec, StgTVar *tvar, StgTRecHeader **in) {
 945   TRecEntry *result = NULL;
 946
 947   TRACE("%p : get_entry_for TVar %p\n", trec, tvar);
 948   ASSERT(trec != NO_TREC);
 949
 950   do {
 951     FOR_EACH_ENTRY(trec, e, {
 952       if (e -> tvar == tvar) {
 953         result = e;
 954         if (in != NULL) {
 955           *in = trec;
 956         }
 957         BREAK_FOR_EACH;
 958       }
 959     });
 960     trec = trec -> enclosing_trec;
 961   } while (result == NULL && trec != NO_TREC);
 962
 963   return result;
 964 }
 965
 966 static StgClosure *read_current_value(StgTRecHeader *trec STG_UNUSED, StgTVar *tvar) {
 967   StgClosure *result;
 968   result = tvar -> current_value;
 969
 970 #if defined(STM_FG_LOCKS)
 971   while (GET_INFO(result) == &stg_TREC_HEADER_info) {
 972     TRACE("%p : read_current_value(%p) saw %p\n", trec, tvar, result);
 973     result = tvar -> current_value;
 974   }
 975 #endif
 976
 977   TRACE("%p : read_current_value(%p)=%p\n", trec, tvar, result);
 978   return result;
 979 }
 980
 981 /*......................................................................*/
 982
 983 StgClosure *stmReadTVar(StgRegTable *reg,
 984                         StgTRecHeader *trec,
 985                         StgTVar *tvar) {
 986   StgTRecHeader *entry_in;
 987   StgClosure *result = NULL;
 988   TRecEntry *entry = NULL;
 989   TRACE("%p : stmReadTVar(%p)\n", trec, tvar);
 990   ASSERT (trec != NO_TREC);
 991   ASSERT (trec -> state == TREC_ACTIVE ||
 992           trec -> state == TREC_CONDEMNED);
 993
 994   entry = get_entry_for(trec, tvar, &entry_in);
 995
 996   if (entry != NULL) {
 997     if (entry_in == trec) {
 998       // Entry found in our trec
 999       result = entry -> new_value;
1000     } else {
1001       // Entry found in another trec
1002       TRecEntry *new_entry = get_new_entry(reg, trec);
1003       new_entry -> tvar = tvar;
1004       new_entry -> expected_value = entry -> expected_value;
1005       new_entry -> new_value = entry -> new_value;
1006       result = new_entry -> new_value;
1007     }
1008   } else {
1009     // No entry found
1010     StgClosure *current_value = read_current_value(trec, tvar);
1011     TRecEntry *new_entry = get_new_entry(reg, trec);
1012     new_entry -> tvar = tvar;
1013     new_entry -> expected_value = current_value;
1014     new_entry -> new_value = current_value;
1015     result = current_value;
1016   }
1017
1018   TRACE("%p : stmReadTVar(%p)=%p\n", trec, tvar, result);
1019   return result;
1020 }
1021
1022 /*......................................................................*/
1023
1024 void stmWriteTVar(StgRegTable *reg,
1025                   StgTRecHeader *trec,
1026                   StgTVar *tvar,
1027                   StgClosure *new_value) {
1028
1029   StgTRecHeader *entry_in;
1030   TRecEntry *entry = NULL;
1031   TRACE("%p : stmWriteTVar(%p, %p)\n", trec, tvar, new_value);
1032   ASSERT (trec != NO_TREC);
1033   ASSERT (trec -> state == TREC_ACTIVE ||
1034           trec -> state == TREC_CONDEMNED);
1035
1036   entry = get_entry_for(trec, tvar, &entry_in);
1037
1038   if (entry != NULL) {
1039     if (entry_in == trec) {
1040       // Entry found in our trec
1041       entry -> new_value = new_value;
1042     } else {
1043       // Entry found in another trec
1044       TRecEntry *new_entry = get_new_entry(reg, trec);
1045       new_entry -> tvar = tvar;
1046       new_entry -> expected_value = entry -> expected_value;
1047       new_entry -> new_value = new_value;
1048     }
1049   } else {
1050     // No entry found
1051     StgClosure *current_value = read_current_value(trec, tvar);
1052     TRecEntry *new_entry = get_new_entry(reg, trec);
1053     new_entry -> tvar = tvar;
1054     new_entry -> expected_value = current_value;
1055     new_entry -> new_value = new_value;
1056   }
1057
1058   TRACE("%p : stmWriteTVar done\n", trec);
1059 }
1060
1061 /*......................................................................*/
1062