method-ml.cc 27.5 KB
Newer Older
1
/* Copyright (C) 2012-2016 Free Software Foundation, Inc.
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71
   Contributed by Torvald Riegel <triegel@redhat.com>.

   This file is part of the GNU Transactional Memory Library (libitm).

   Libitm is free software; you can redistribute it and/or modify it
   under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 3 of the License, or
   (at your option) any later version.

   Libitm is distributed in the hope that it will be useful, but WITHOUT ANY
   WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
   FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
   more details.

   Under Section 7 of GPL version 3, you are granted additional
   permissions described in the GCC Runtime Library Exception, version
   3.1, as published by the Free Software Foundation.

   You should have received a copy of the GNU General Public License and
   a copy of the GCC Runtime Library Exception along with this program;
   see the files COPYING3 and COPYING.RUNTIME respectively.  If not, see
   <http://www.gnu.org/licenses/>.  */

#include "libitm_i.h"

using namespace GTM;

namespace {

// This group consists of all TM methods that synchronize via multiple locks
// (or ownership records).
struct ml_mg : public method_group
{
  static const gtm_word LOCK_BIT = (~(gtm_word)0 >> 1) + 1;
  static const gtm_word INCARNATION_BITS = 3;
  static const gtm_word INCARNATION_MASK = 7;
  // Maximum time is all bits except the lock bit, the overflow reserve bit,
  // and the incarnation bits).
  static const gtm_word TIME_MAX = (~(gtm_word)0 >> (2 + INCARNATION_BITS));
  // The overflow reserve bit is the MSB of the timestamp part of an orec,
  // so we can have TIME_MAX+1 pending timestamp increases before we overflow.
  static const gtm_word OVERFLOW_RESERVE = TIME_MAX + 1;

  static bool is_locked(gtm_word o) { return o & LOCK_BIT; }
  static gtm_word set_locked(gtm_thread *tx)
  {
    return ((uintptr_t)tx >> 1) | LOCK_BIT;
  }
  // Returns a time that includes the lock bit, which is required by both
  // validate() and is_more_recent_or_locked().
  static gtm_word get_time(gtm_word o) { return o >> INCARNATION_BITS; }
  static gtm_word set_time(gtm_word time) { return time << INCARNATION_BITS; }
  static bool is_more_recent_or_locked(gtm_word o, gtm_word than_time)
  {
    // LOCK_BIT is the MSB; thus, if O is locked, it is larger than TIME_MAX.
    return get_time(o) > than_time;
  }
  static bool has_incarnation_left(gtm_word o)
  {
    return (o & INCARNATION_MASK) < INCARNATION_MASK;
  }
  static gtm_word inc_incarnation(gtm_word o) { return o + 1; }

  // The shared time base.
  atomic<gtm_word> time __attribute__((aligned(HW_CACHELINE_SIZE)));

  // The array of ownership records.
  atomic<gtm_word>* orecs __attribute__((aligned(HW_CACHELINE_SIZE)));
  char tailpadding[HW_CACHELINE_SIZE - sizeof(atomic<gtm_word>*)];

72 73 74 75 76 77 78 79 80 81 82 83
  // Location-to-orec mapping.  Stripes of 32B mapped to 2^16 orecs using
  // multiplicative hashing.  See Section 5.2.2 of Torvald Riegel's PhD thesis
  // for the background on this choice of hash function and parameters:
  // http://nbn-resolving.de/urn:nbn:de:bsz:14-qucosa-115596
  // We pick the Mult32 hash because it works well with fewer orecs (i.e.,
  // less space overhead and just 32b multiplication).
  // We may want to check and potentially change these settings once we get
  // better or just more benchmarks.
  static const gtm_word L2O_ORECS_BITS = 16;
  static const gtm_word L2O_ORECS = 1 << L2O_ORECS_BITS;
  // An iterator over the orecs covering the region [addr,addr+len).
  struct orec_iterator
84
  {
85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116
    static const gtm_word L2O_SHIFT = 5;
    static const uint32_t L2O_MULT32 = 81007;
    uint32_t mult;
    size_t orec;
    size_t orec_end;
    orec_iterator (const void* addr, size_t len)
    {
      uint32_t a = (uintptr_t) addr >> L2O_SHIFT;
      uint32_t ae = ((uintptr_t) addr + len + (1 << L2O_SHIFT) - 1)
	  >> L2O_SHIFT;
      mult = a * L2O_MULT32;
      orec = mult >> (32 - L2O_ORECS_BITS);
      // We can't really avoid this second multiplication unless we use a
      // branch instead or know more about the alignment of addr.  (We often
      // know len at compile time because of instantiations of functions
      // such as _ITM_RU* for accesses of specific lengths.
      orec_end = (ae * L2O_MULT32) >> (32 - L2O_ORECS_BITS);
    }
    size_t get() { return orec; }
    void advance()
    {
      // We cannot simply increment orec because L2O_MULT32 is larger than
      // 1 << (32 - L2O_ORECS_BITS), and thus an increase of the stripe (i.e.,
      // addr >> L2O_SHIFT) could increase the resulting orec index by more
      // than one; with the current parameters, we would roughly acquire a
      // fourth more orecs than necessary for regions covering more than orec.
      // Keeping mult around as extra state shouldn't matter much.
      mult += L2O_MULT32;
      orec = mult >> (32 - L2O_ORECS_BITS);
    }
    bool reached_end() { return orec == orec_end; }
  };
117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172

  virtual void init()
  {
    // We assume that an atomic<gtm_word> is backed by just a gtm_word, so
    // starting with zeroed memory is fine.
    orecs = (atomic<gtm_word>*) xcalloc(
        sizeof(atomic<gtm_word>) * L2O_ORECS, true);
    // This store is only executed while holding the serial lock, so relaxed
    // memory order is sufficient here.
    time.store(0, memory_order_relaxed);
  }

  virtual void fini()
  {
    free(orecs);
  }

  // We only re-initialize when our time base overflows.  Thus, only reset
  // the time base and the orecs but do not re-allocate the orec array.
  virtual void reinit()
  {
    // This store is only executed while holding the serial lock, so relaxed
    // memory order is sufficient here.  Same holds for the memset.
    time.store(0, memory_order_relaxed);
    memset(orecs, 0, sizeof(atomic<gtm_word>) * L2O_ORECS);
  }
};

static ml_mg o_ml_mg;


// The multiple lock, write-through TM method.
// Maps each memory location to one of the orecs in the orec array, and then
// acquires the associated orec eagerly before writing through.
// Writes require undo-logging because we are dealing with several locks/orecs
// and need to resolve deadlocks if necessary by aborting one of the
// transactions.
// Reads do time-based validation with snapshot time extensions.  Incarnation
// numbers are used to decrease contention on the time base (with those,
// aborted transactions do not need to acquire a new version number for the
// data that has been previously written in the transaction and needs to be
// rolled back).
// gtm_thread::shared_state is used to store a transaction's current
// snapshot time (or commit time). The serial lock uses ~0 for inactive
// transactions and 0 for active ones. Thus, we always have a meaningful
// timestamp in shared_state that can be used to implement quiescence-based
// privatization safety.
class ml_wt_dispatch : public abi_dispatch
{
protected:
  static void pre_write(gtm_thread *tx, const void *addr, size_t len)
  {
    gtm_word snapshot = tx->shared_state.load(memory_order_relaxed);
    gtm_word locked_by_tx = ml_mg::set_locked(tx);

    // Lock all orecs that cover the region.
173
    ml_mg::orec_iterator oi(addr, len);
174 175 176 177 178
    do
      {
        // Load the orec.  Relaxed memory order is sufficient here because
        // either we have acquired the orec or we will try to acquire it with
        // a CAS with stronger memory order.
179
        gtm_word o = o_ml_mg.orecs[oi.get()].load(memory_order_relaxed);
180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204

        // Check whether we have acquired the orec already.
        if (likely (locked_by_tx != o))
          {
            // If not, acquire.  Make sure that our snapshot time is larger or
            // equal than the orec's version to avoid masking invalidations of
            // our snapshot with our own writes.
            if (unlikely (ml_mg::is_locked(o)))
              tx->restart(RESTART_LOCKED_WRITE);

            if (unlikely (ml_mg::get_time(o) > snapshot))
              {
                // We only need to extend the snapshot if we have indeed read
                // from this orec before.  Given that we are an update
                // transaction, we will have to extend anyway during commit.
                // ??? Scan the read log instead, aborting if we have read
                // from data covered by this orec before?
                snapshot = extend(tx);
              }

            // We need acquire memory order here to synchronize with other
            // (ownership) releases of the orec.  We do not need acq_rel order
            // because whenever another thread reads from this CAS'
            // modification, then it will abort anyway and does not rely on
            // any further happens-before relation to be established.
205
            if (unlikely (!o_ml_mg.orecs[oi.get()].compare_exchange_strong(
206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221
                o, locked_by_tx, memory_order_acquire)))
              tx->restart(RESTART_LOCKED_WRITE);

            // We use an explicit fence here to avoid having to use release
            // memory order for all subsequent data stores.  This fence will
            // synchronize with loads of the data with acquire memory order.
            // See post_load() for why this is necessary.
            // Adding require memory order to the prior CAS is not sufficient,
            // at least according to the Batty et al. formalization of the
            // memory model.
            atomic_thread_fence(memory_order_release);

            // We log the previous value here to be able to use incarnation
            // numbers when we have to roll back.
            // ??? Reserve capacity early to avoid capacity checks here?
            gtm_rwlog_entry *e = tx->writelog.push();
222
            e->orec = o_ml_mg.orecs + oi.get();
223 224
            e->value = o;
          }
225
        oi.advance();
226
      }
227
    while (!oi.reached_end());
228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304

    // Do undo logging.  We do not know which region prior writes logged
    // (even if orecs have been acquired), so just log everything.
    tx->undolog.log(addr, len);
  }

  static void pre_write(const void *addr, size_t len)
  {
    gtm_thread *tx = gtm_thr();
    pre_write(tx, addr, len);
  }

  // Returns true iff all the orecs in our read log still have the same time
  // or have been locked by the transaction itself.
  static bool validate(gtm_thread *tx)
  {
    gtm_word locked_by_tx = ml_mg::set_locked(tx);
    // ??? This might get called from pre_load() via extend().  In that case,
    // we don't really need to check the new entries that pre_load() is
    // adding.  Stop earlier?
    for (gtm_rwlog_entry *i = tx->readlog.begin(), *ie = tx->readlog.end();
        i != ie; i++)
      {
	// Relaxed memory order is sufficient here because we do not need to
	// establish any new synchronizes-with relationships.  We only need
	// to read a value that is as least as current as enforced by the
	// callers: extend() loads global time with acquire, and trycommit()
	// increments global time with acquire.  Therefore, we will see the
	// most recent orec updates before the global time that we load.
        gtm_word o = i->orec->load(memory_order_relaxed);
        // We compare only the time stamp and the lock bit here.  We know that
        // we have read only committed data before, so we can ignore
        // intermediate yet rolled-back updates presented by the incarnation
        // number bits.
        if (ml_mg::get_time(o) != ml_mg::get_time(i->value)
            && o != locked_by_tx)
          return false;
      }
    return true;
  }

  // Tries to extend the snapshot to a more recent time.  Returns the new
  // snapshot time and updates TX->SHARED_STATE.  If the snapshot cannot be
  // extended to the current global time, TX is restarted.
  static gtm_word extend(gtm_thread *tx)
  {
    // We read global time here, even if this isn't strictly necessary
    // because we could just return the maximum of the timestamps that
    // validate sees.  However, the potential cache miss on global time is
    // probably a reasonable price to pay for avoiding unnecessary extensions
    // in the future.
    // We need acquire memory oder because we have to synchronize with the
    // increment of global time by update transactions, whose lock
    // acquisitions we have to observe (also see trycommit()).
    gtm_word snapshot = o_ml_mg.time.load(memory_order_acquire);
    if (!validate(tx))
      tx->restart(RESTART_VALIDATE_READ);

    // Update our public snapshot time.  Probably useful to decrease waiting
    // due to quiescence-based privatization safety.
    // Use release memory order to establish synchronizes-with with the
    // privatizers; prior data loads should happen before the privatizers
    // potentially modify anything.
    tx->shared_state.store(snapshot, memory_order_release);
    return snapshot;
  }

  // First pass over orecs.  Load and check all orecs that cover the region.
  // Write to read log, extend snapshot time if necessary.
  static gtm_rwlog_entry* pre_load(gtm_thread *tx, const void* addr,
      size_t len)
  {
    // Don't obtain an iterator yet because the log might get resized.
    size_t log_start = tx->readlog.size();
    gtm_word snapshot = tx->shared_state.load(memory_order_relaxed);
    gtm_word locked_by_tx = ml_mg::set_locked(tx);

305
    ml_mg::orec_iterator oi(addr, len);
306 307 308 309 310 311 312
    do
      {
        // We need acquire memory order here so that this load will
        // synchronize with the store that releases the orec in trycommit().
        // In turn, this makes sure that subsequent data loads will read from
        // a visible sequence of side effects that starts with the most recent
        // store to the data right before the release of the orec.
313
        gtm_word o = o_ml_mg.orecs[oi.get()].load(memory_order_acquire);
314 315 316 317 318

        if (likely (!ml_mg::is_more_recent_or_locked(o, snapshot)))
          {
            success:
            gtm_rwlog_entry *e = tx->readlog.push();
319
            e->orec = o_ml_mg.orecs + oi.get();
320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336
            e->value = o;
          }
        else if (!ml_mg::is_locked(o))
          {
            // We cannot read this part of the region because it has been
            // updated more recently than our snapshot time.  If we can extend
            // our snapshot, then we can read.
            snapshot = extend(tx);
            goto success;
          }
        else
          {
            // If the orec is locked by us, just skip it because we can just
            // read from it.  Otherwise, restart the transaction.
            if (o != locked_by_tx)
              tx->restart(RESTART_LOCKED_READ);
          }
337
        oi.advance();
338
      }
339
    while (!oi.reached_end());
340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515
    return &tx->readlog[log_start];
  }

  // Second pass over orecs, verifying that the we had a consistent read.
  // Restart the transaction if any of the orecs is locked by another
  // transaction.
  static void post_load(gtm_thread *tx, gtm_rwlog_entry* log)
  {
    for (gtm_rwlog_entry *end = tx->readlog.end(); log != end; log++)
      {
        // Check that the snapshot is consistent.  We expect the previous data
        // load to have acquire memory order, or be atomic and followed by an
        // acquire fence.
        // As a result, the data load will synchronize with the release fence
        // issued by the transactions whose data updates the data load has read
        // from.  This forces the orec load to read from a visible sequence of
        // side effects that starts with the other updating transaction's
        // store that acquired the orec and set it to locked.
        // We therefore either read a value with the locked bit set (and
        // restart) or read an orec value that was written after the data had
        // been written.  Either will allow us to detect inconsistent reads
        // because it will have a higher/different value.
	// Also note that differently to validate(), we compare the raw value
	// of the orec here, including incarnation numbers.  We must prevent
	// returning uncommitted data from loads (whereas when validating, we
	// already performed a consistent load).
        gtm_word o = log->orec->load(memory_order_relaxed);
        if (log->value != o)
          tx->restart(RESTART_VALIDATE_READ);
      }
  }

  template <typename V> static V load(const V* addr, ls_modifier mod)
  {
    // Read-for-write should be unlikely, but we need to handle it or will
    // break later WaW optimizations.
    if (unlikely(mod == RfW))
      {
	pre_write(addr, sizeof(V));
	return *addr;
      }
    if (unlikely(mod == RaW))
      return *addr;
    // ??? Optimize for RaR?

    gtm_thread *tx = gtm_thr();
    gtm_rwlog_entry* log = pre_load(tx, addr, sizeof(V));

    // Load the data.
    // This needs to have acquire memory order (see post_load()).
    // Alternatively, we can put an acquire fence after the data load but this
    // is probably less efficient.
    // FIXME We would need an atomic load with acquire memory order here but
    // we can't just forge an atomic load for nonatomic data because this
    // might not work on all implementations of atomics.  However, we need
    // the acquire memory order and we can only establish this if we link
    // it to the matching release using a reads-from relation between atomic
    // loads.  Also, the compiler is allowed to optimize nonatomic accesses
    // differently than atomic accesses (e.g., if the load would be moved to
    // after the fence, we potentially don't synchronize properly anymore).
    // Instead of the following, just use an ordinary load followed by an
    // acquire fence, and hope that this is good enough for now:
    // V v = atomic_load_explicit((atomic<V>*)addr, memory_order_acquire);
    V v = *addr;
    atomic_thread_fence(memory_order_acquire);

    // ??? Retry the whole load if it wasn't consistent?
    post_load(tx, log);

    return v;
  }

  template <typename V> static void store(V* addr, const V value,
      ls_modifier mod)
  {
    if (likely(mod != WaW))
      pre_write(addr, sizeof(V));
    // FIXME We would need an atomic store here but we can't just forge an
    // atomic load for nonatomic data because this might not work on all
    // implementations of atomics.  However, we need this store to link the
    // release fence in pre_write() to the acquire operation in load, which
    // is only guaranteed if we have a reads-from relation between atomic
    // accesses.  Also, the compiler is allowed to optimize nonatomic accesses
    // differently than atomic accesses (e.g., if the store would be moved
    // to before the release fence in pre_write(), things could go wrong).
    // atomic_store_explicit((atomic<V>*)addr, value, memory_order_relaxed);
    *addr = value;
  }

public:
  static void memtransfer_static(void *dst, const void* src, size_t size,
      bool may_overlap, ls_modifier dst_mod, ls_modifier src_mod)
  {
    gtm_rwlog_entry* log = 0;
    gtm_thread *tx = 0;

    if (src_mod == RfW)
      {
        tx = gtm_thr();
        pre_write(tx, src, size);
      }
    else if (src_mod != RaW && src_mod != NONTXNAL)
      {
        tx = gtm_thr();
        log = pre_load(tx, src, size);
      }
    // ??? Optimize for RaR?

    if (dst_mod != NONTXNAL && dst_mod != WaW)
      {
        if (src_mod != RfW && (src_mod == RaW || src_mod == NONTXNAL))
          tx = gtm_thr();
        pre_write(tx, dst, size);
      }

    // FIXME We should use atomics here (see store()).  Let's just hope that
    // memcpy/memmove are good enough.
    if (!may_overlap)
      ::memcpy(dst, src, size);
    else
      ::memmove(dst, src, size);

    // ??? Retry the whole memtransfer if it wasn't consistent?
    if (src_mod != RfW && src_mod != RaW && src_mod != NONTXNAL)
      {
	// See load() for why we need the acquire fence here.
	atomic_thread_fence(memory_order_acquire);
	post_load(tx, log);
      }
  }

  static void memset_static(void *dst, int c, size_t size, ls_modifier mod)
  {
    if (mod != WaW)
      pre_write(dst, size);
    // FIXME We should use atomics here (see store()).  Let's just hope that
    // memset is good enough.
    ::memset(dst, c, size);
  }

  virtual gtm_restart_reason begin_or_restart()
  {
    // We don't need to do anything for nested transactions.
    gtm_thread *tx = gtm_thr();
    if (tx->parent_txns.size() > 0)
      return NO_RESTART;

    // Read the current time, which becomes our snapshot time.
    // Use acquire memory oder so that we see the lock acquisitions by update
    // transcations that incremented the global time (see trycommit()).
    gtm_word snapshot = o_ml_mg.time.load(memory_order_acquire);
    // Re-initialize method group on time overflow.
    if (snapshot >= o_ml_mg.TIME_MAX)
      return RESTART_INIT_METHOD_GROUP;

    // We don't need to enforce any ordering for the following store. There
    // are no earlier data loads in this transaction, so the store cannot
    // become visible before those (which could lead to the violation of
    // privatization safety). The store can become visible after later loads
    // but this does not matter because the previous value will have been
    // smaller or equal (the serial lock will set shared_state to zero when
    // marking the transaction as active, and restarts enforce immediate
    // visibility of a smaller or equal value with a barrier (see
    // rollback()).
    tx->shared_state.store(snapshot, memory_order_relaxed);
    return NO_RESTART;
  }

  virtual bool trycommit(gtm_word& priv_time)
  {
    gtm_thread* tx = gtm_thr();

    // If we haven't updated anything, we can commit.
    if (!tx->writelog.size())
      {
        tx->readlog.clear();
516 517 518 519 520 521 522 523 524 525 526 527 528 529 530
        // We still need to ensure privatization safety, unfortunately.  While
        // we cannot have privatized anything by ourselves (because we are not
        // an update transaction), we can have observed the commits of
        // another update transaction that privatized something.  Because any
        // commit happens before ensuring privatization, our snapshot and
        // commit can thus have happened before ensuring privatization safety
        // for this commit/snapshot time.  Therefore, before we can return to
        // nontransactional code that might use the privatized data, we must
        // ensure privatization safety for our snapshot time.
        // This still seems to be better than not allowing use of the
        // snapshot time before privatization safety has been ensured because
        // we at least can run transactions such as this one, and in the
        // meantime the transaction producing this commit time might have
        // finished ensuring privatization safety for it.
        priv_time = tx->shared_state.load(memory_order_relaxed);
531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621
        return true;
      }

    // Get a commit time.
    // Overflow of o_ml_mg.time is prevented in begin_or_restart().
    // We need acq_rel here because (1) the acquire part is required for our
    // own subsequent call to validate(), and the release part is necessary to
    // make other threads' validate() work as explained there and in extend().
    gtm_word ct = o_ml_mg.time.fetch_add(1, memory_order_acq_rel) + 1;

    // Extend our snapshot time to at least our commit time.
    // Note that we do not need to validate if our snapshot time is right
    // before the commit time because we are never sharing the same commit
    // time with other transactions.
    // No need to reset shared_state, which will be modified by the serial
    // lock right after our commit anyway.
    gtm_word snapshot = tx->shared_state.load(memory_order_relaxed);
    if (snapshot < ct - 1 && !validate(tx))
      return false;

    // Release orecs.
    // See pre_load() / post_load() for why we need release memory order.
    // ??? Can we use a release fence and relaxed stores?
    gtm_word v = ml_mg::set_time(ct);
    for (gtm_rwlog_entry *i = tx->writelog.begin(), *ie = tx->writelog.end();
        i != ie; i++)
      i->orec->store(v, memory_order_release);

    // We're done, clear the logs.
    tx->writelog.clear();
    tx->readlog.clear();

    // Need to ensure privatization safety. Every other transaction must
    // have a snapshot time that is at least as high as our commit time
    // (i.e., our commit must be visible to them).
    priv_time = ct;
    return true;
  }

  virtual void rollback(gtm_transaction_cp *cp)
  {
    // We don't do anything for rollbacks of nested transactions.
    // ??? We could release locks here if we snapshot writelog size.  readlog
    // is similar.  This is just a performance optimization though.  Nested
    // aborts should be rather infrequent, so the additional save/restore
    // overhead for the checkpoints could be higher.
    if (cp != 0)
      return;

    gtm_thread *tx = gtm_thr();
    gtm_word overflow_value = 0;

    // Release orecs.
    for (gtm_rwlog_entry *i = tx->writelog.begin(), *ie = tx->writelog.end();
        i != ie; i++)
      {
        // If possible, just increase the incarnation number.
        // See pre_load() / post_load() for why we need release memory order.
	// ??? Can we use a release fence and relaxed stores?  (Same below.)
        if (ml_mg::has_incarnation_left(i->value))
          i->orec->store(ml_mg::inc_incarnation(i->value),
              memory_order_release);
        else
          {
            // We have an incarnation overflow.  Acquire a new timestamp, and
            // use it from now on as value for each orec whose incarnation
            // number cannot be increased.
            // Overflow of o_ml_mg.time is prevented in begin_or_restart().
            // See pre_load() / post_load() for why we need release memory
            // order.
            if (!overflow_value)
              // Release memory order is sufficient but required here.
              // In contrast to the increment in trycommit(), we need release
              // for the same reason but do not need the acquire because we
              // do not validate subsequently.
              overflow_value = ml_mg::set_time(
                  o_ml_mg.time.fetch_add(1, memory_order_release) + 1);
            i->orec->store(overflow_value, memory_order_release);
          }
      }

    // We need this release fence to ensure that privatizers see the
    // rolled-back original state (not any uncommitted values) when they read
    // the new snapshot time that we write in begin_or_restart().
    atomic_thread_fence(memory_order_release);

    // We're done, clear the logs.
    tx->writelog.clear();
    tx->readlog.clear();
  }

622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639
  virtual bool snapshot_most_recent()
  {
    // This is the same code as in extend() except that we do not restart
    // on failure but simply return the result, and that we don't validate
    // if our snapshot is already most recent.
    gtm_thread* tx = gtm_thr();
    gtm_word snapshot = o_ml_mg.time.load(memory_order_acquire);
    if (snapshot == tx->shared_state.load(memory_order_relaxed))
      return true;
    if (!validate(tx))
      return false;

    // Update our public snapshot time.  Necessary so that we do not prevent
    // other transactions from ensuring privatization safety.
    tx->shared_state.store(snapshot, memory_order_release);
    return true;
  }

640 641 642 643 644 645 646 647 648 649 650 651
  virtual bool supports(unsigned number_of_threads)
  {
    // Each txn can commit and fail and rollback once before checking for
    // overflow, so this bounds the number of threads that we can support.
    // In practice, this won't be a problem but we check it anyway so that
    // we never break in the occasional weird situation.
    return (number_of_threads * 2 <= ml_mg::OVERFLOW_RESERVE);
  }

  CREATE_DISPATCH_METHODS(virtual, )
  CREATE_DISPATCH_METHODS_MEM()

652
  ml_wt_dispatch() : abi_dispatch(false, true, false, false, 0, &o_ml_mg)
653 654 655 656 657 658 659 660 661 662 663 664
  { }
};

} // anon namespace

static const ml_wt_dispatch o_ml_wt_dispatch;

abi_dispatch *
GTM::dispatch_ml_wt ()
{
  return const_cast<ml_wt_dispatch *>(&o_ml_wt_dispatch);
}