Create "persistent memory allocator" for persisting and sharing objects.

base/memory/shared_memory_allocator.cc

Index: base/memory/shared_memory_allocator.cc
diff --git a/base/memory/shared_memory_allocator.cc b/base/memory/shared_memory_allocator.cc
new file mode 100644
index 0000000000000000000000000000000000000000..670db66da47d9e60be1af25e681dc7ea43a42c6c
--- /dev/null
+++ b/base/memory/shared_memory_allocator.cc
@@ -0,0 +1,424 @@
+// Copyright (c) 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "base/memory/shared_memory_allocator.h"
+
+#include <assert.h>
+#include <algorithm>
+
+#include "base/logging.h"
+
+// All integer constants in this file are signed because Atomic32 is signed
+// and keeping all others consistent with this avoids a lot of unnecessary
+// casting to avoid signed/unsigned operations just to avoid compiler errors.
+// This means an occasonal cast of a constant from sizeof() to "int" but
+// is far simpler than the alternative.
+
+namespace {
+
+// All allocations and data-structures must be aligned to this byte boundary.
+// Alignment as large as the physical bus between CPU and RAM is _required_
+// for some architectures, is simply more efficient on other CPUs, and
+// generally a Good Idea(tm) for all platforms as it reduces/eliminates the
+// chance that a type will span cache lines. Alignment mustn't be less
+// than 8 to ensure proper alignment for all types. The rest is a balance
+// between reducing spans across multiple cache lines and wasted space spent
+// padding out allocations. An alignment of 16 would ensure that the block
+// header structure always sits in a single cache line. An average of about
+// 1/2 this value will be wasted with every allocation.
+const int32_t kAllocAlignment = 8;
+
+// A constant (random) value placed in the shared metadata to identify
+// an already initialized memory segment.
+const int32_t kGlobalCookie = 0x408305DC;
+
+// The current version of the metadata. If updates are made that change
+// the metadata, the version number can be queried to operate in a backward-
+// compatible manner until the memory segment is completely re-initalized.
+const int32_t kGlobalVersion = 1;
+
+// Constant values placed in the block headers to indicate its state.
+const int32_t kBlockCookieFree = 0;
+const int32_t kBlockCookieQueue = 1;
+const int32_t kBlockCookieWasted = -1;
+const int32_t kBlockCookieAllocated = 0xC8799269;
+
+// TODO(bcwhite): When acceptable, consider moving flags to std::atomic<char>
+// types rather than combined bitfield.
+
+enum {
+  kFlagCorrupted,
+  kFlagFull
+};
+
+bool CheckFlag(base::subtle::Atomic32* flags, int flag) {
+  base::subtle::Atomic32 loaded_flags = base::subtle::Acquire_Load(flags);
+  return (loaded_flags & 1 << flag) != 0;
+}
+
+void SetFlag(base::subtle::Atomic32* flags, int flag) {
+  for (;;) {
+    base::subtle::Atomic32 loaded_flags = base::subtle::Acquire_Load(flags);
+    base::subtle::Atomic32 new_flags =
+        (loaded_flags & ~(1 << flag)) | (1 << flag);
+    if (base::subtle::Release_CompareAndSwap(
+            flags, loaded_flags, new_flags) == loaded_flags) {
+      break;
+    }
+  }
+}
+
+}  // namespace
+
+namespace base {
+
+// The block-header is placed at the top of every allocation within the
+// segment to describe the data that follows it.
+struct SharedMemoryAllocator::BlockHeader {
+  int32_t size;           // Number of bytes in this block, including header.
+  int32_t cookie;         // Constant value indicating completed allocation.
+  int32_t type_id;        // A number provided by caller indicating data type.
+  subtle::Atomic32 next;  // Pointer to the next block when iterating.
+};
+
+// The shared metadata exists once at the top of the memory segment to
+// describe the state of the allocator to all processes.
+struct SharedMemoryAllocator::SharedMetadata {
+  int32_t cookie;     // Some value that indicates complete initialization.
+  int32_t size;       // Total size of memory segment.
+  int32_t page_size;  // Paging size within memory segment.
+  int32_t version;    // Version code so upgrades don't break.
+  subtle::Atomic32 freeptr;  // Offset to first free space in the segment.
+  subtle::Atomic32 flags;    // Bitfield of information flags.
+  int32_t reserved;   // Padding to ensure size is multiple of alignment.
+
+  // The "iterable" queue is an M&S Queue as described here, append-only:
+  // https://www.research.ibm.com/people/m/michael/podc-1996.pdf
+  subtle::Atomic32 tailptr;  // Last block available for iteration.
+  BlockHeader queue;  // Empty block for linked-list head/tail. (must be last)
+};
+
+// The "queue" block header is used to detect "last node" so that zero/null
+// can be used to indicate that it hasn't been added at all. It is part of
+// the SharedMetadata structure which itself is always located at offset zero.
+// This can't be a constant because SharedMetadata is a private definition.
+#define OFFSET_QUEUE offsetof(SharedMetadata, queue)
+#define OFFSET_NULL 0  // the equivalest NULL value for an offset
+
+SharedMemoryAllocator::SharedMemoryAllocator(void* base,
+                                             int32_t size,
+                                             int32_t page_size)
+    : shared_meta_(static_cast<SharedMetadata*>(base)),
+      mem_base_(static_cast<char*>(base)),
+      mem_size_(size),
+      mem_page_(page_size ? page_size : size),
+      corrupted_(0) {
+  static_assert(sizeof(BlockHeader) % kAllocAlignment == 0,
+                "BlockHeader is not a multiple of kAllocAlignment");
+  static_assert(sizeof(SharedMetadata) % kAllocAlignment == 0,
+                "SharedMetadata is not a multiple of kAllocAlignment");
+
+  CHECK(base && reinterpret_cast<uintptr_t>(base) % kAllocAlignment == 0);
+  CHECK(size >= 1 << 10 && size <= 1 << 20 &&  // 1 KiB <= size <= 1 MiB
+        size % kAllocAlignment == 0);
+  CHECK(page_size >= 0 && (page_size == 0 || size % page_size == 0));
+
+  if (shared_meta_->cookie != kGlobalCookie) {
+    // This block is only executed when a completely new memory segment is
+    // being initialized. It's unshared and single-threaded...
+    const BlockHeader* first_block = reinterpret_cast<BlockHeader*>(
+        mem_base_ + sizeof(SharedMetadata));
+    if (shared_meta_->cookie != 0 ||
+        shared_meta_->size != 0 ||
+        shared_meta_->version != 0 ||
+        subtle::NoBarrier_Load(&shared_meta_->freeptr) != 0 ||
+        subtle::NoBarrier_Load(&shared_meta_->flags) != 0 ||
+        shared_meta_->tailptr != 0 ||
+        shared_meta_->queue.cookie != 0 ||
+        subtle::NoBarrier_Load(&shared_meta_->queue.next) != 0 ||
+        first_block->size != 0 ||
+        first_block->cookie != 0 ||
+        first_block->type_id != 0 ||
+        first_block->next != 0) {
+      // ...or something malicious has been playing with the metadata.
+      NOTREACHED();
+      SetCorrupted();
+    }
+
+    // This is still safe to do even if corruption has been detected.
+    shared_meta_->cookie = kGlobalCookie;
+    shared_meta_->size = size;
+    shared_meta_->page_size = page_size;
+    shared_meta_->version = kGlobalVersion;
+    subtle::NoBarrier_Store(&shared_meta_->freeptr, sizeof(SharedMetadata));
+
+    // Set up the queue of iterable allocations.
+    shared_meta_->queue.size = sizeof(BlockHeader);
+    shared_meta_->queue.cookie = kBlockCookieQueue;
+    subtle::NoBarrier_Store(&shared_meta_->queue.next, OFFSET_QUEUE);
+    subtle::NoBarrier_Store(&shared_meta_->tailptr, OFFSET_QUEUE);
+  } else {
+    // The allocator is attaching to a previously initialized segment of
+    // memory. Make sure the embedded data matches what has been passed.
+    if (shared_meta_->size != size || shared_meta_->page_size != page_size) {
+      NOTREACHED();
+      SetCorrupted();
+    }
+  }
+}
+
+SharedMemoryAllocator::~SharedMemoryAllocator() {}
+
+int32_t SharedMemoryAllocator::Allocate(int32_t size, int32_t type_id) {
+  if (size < 0) {

[Dmitry Vyukov, 2015/11/04 13:52:29]

check that size != 0 as well
in GetNextIterable and in other places you assume that min block size is
(sizeof(BlockHeader) + kAllocAlignment)
zero size can, in particular, falsely fail the loop check during iteration

[bcwhite, 2015/11/04 17:18:55]

Done.

+    NOTREACHED();
+    return OFFSET_NULL;
+  }
+
+  // Round up the requested size, plus header, to the next allocation alignment.
+  size += sizeof(BlockHeader);

[Dmitry Vyukov, 2015/11/04 13:52:29]

check for overflow, rendered can pass INT_MAX-1
nothing good will come out of that
you can just check that size < page_size as first check

[bcwhite, 2015/11/04 17:18:55]

Done.

+  size = (size + (kAllocAlignment - 1)) & ~(kAllocAlignment - 1);
+  if (size > mem_page_)

[Dmitry Vyukov, 2015/11/04 13:52:29]

check that size <= page_size

+    return OFFSET_NULL;
+
+  // Allocation is lockless so we do all our caculation and then, if saving
+  // indicates a change has occurred since we started, scrap everything and
+  // start over.
+  for (;;) {
+    if (IsCorrupted())
+      return OFFSET_NULL;
+
+    int32_t freeptr = subtle::Acquire_Load(&shared_meta_->freeptr);

[Dmitry Vyukov, 2015/11/04 13:52:29]

What do we acquire here? Where is the pairing release?

[bcwhite, 2015/11/04 17:18:55]

It's the CAS on line 214 or 234 (only one is executed).  Unless I'm not
understanding something.

As an aside, why do they have to be paired?  Logically there is a symmetry but
in actual operation, aren't they completely independent?

[Dmitry Vyukov, 2015/11/05 10:49:42]

Acquire or release operation that is not paired with the opposite operation is a
no-op.

Ok, it is paired with CAS on line 214 or 234. But what is the data that we want
to synchronize with this Acquire? Or put another way: why did you use Acquire
instead of NoBarrier?

[bcwhite, 2015/11/05 14:37:15]

I've been thinking about acquire/release in terms of how they translate into
machine code.  At their base, they simply prevent re-ordering memory access from
after/before their location.  I wasn't considering any "higher level" reasoning
since obviously machine instructions don't actually know anything about what is
happening on other threads.

With a lock/mutex, you "release" a lock you previous "acquired".  But with
atomics, it seems that you "acquire" a value "released" by another thread, a
value that something else claims it is finished with, at least for the moment.

There are no _code_ requirements to match acquire/release (unlike a mutex) but
conceptually you should consider this kind of pairing... and it should be
documented in comments.

Am I thinking of this correctly?

If so, then...

It's acquiring the state indicating where unallocated memory (currently) begins.

[Dmitry Vyukov, 2015/11/05 16:38:12]

Correct.
Well, a NoBarrier_Load will perfectly return where unallocated memory begins.
Memory barriers are about _associated_ state. Consider:

// thread 1
data = 42;
Release_Store(&data_ready, 1);

// thread 2
if (Acquire_Load(&data_ready) == 1)
    assert(data == 42);

Here data_ready is synchronization variable and data is the associated state
that we synchronize by using acquire/release.
If there is no associated state, then there is no point in using memory
barriers. E.g.:

// thread 1
while (NoBarrier_Load(&stop) == 0) {
  ...
}

// thread 2
NoBarrier_Store(&stop, 1);


The question is: here shared_meta_->freeptr is the synchronization variable, if
we use acquire, then there must be some associated state (which is not freeptr
itself -- we already loaded it). What is that state?
Or put it another way: reordering of memory accesses to shared_meta_->freeptr
and what else we want to prevent by using barriers?


Excessive memory barriers won't break the algorithm (it is missing barrier that
can break it). My concern is more clarity and documentation. If I see an acquire
or release, then I start looking for the pairing operation and try to understand
what state we synchronize by using barriers. If I find them, then good. If I
don't, then I continue looking, at some point I will have to stop with
impression that I am missing something important here.

[bcwhite, 2015/11/05 17:06:30]

Ahhh!  So it's not that specific value we're acquiring or releasing, it's things
that are related to it.  The value is more a kind of pivot around which the
related variables must show integrity.

So because "freeptr" is all alone, we don't need any barriers; any atomic access
will do and the Acquire/Release can be changed to NoBarrier.

In the case of tailptr/next, then there must be acquire/release semantics to
ensure that access to those _two_ values happen in a guaranteed order.  Neither
the compiler nor the cpu actually know what values are related so they just
ensure that all memory access before/after remains before/after in the face of
optimizations.

I'll change these to NoBarrier ops.

+    if (freeptr + size > mem_size_) {
+      SetFlag(&shared_meta_->flags, kFlagFull);
+      return OFFSET_NULL;
+    }
+
+    // Get pointer to the "free" block. It doesn't even have a header; pass
+    // -sizeof(header) so accouting for that will yield an expected size of
+    // zero which is what will be stored at that location. If something
+    // has been allocated since the load of freeptr above, it is still safe
+    // as nothing will be written to that location until after the CAS below.
+    BlockHeader* block = GetBlock(freeptr, 0, -(int)sizeof(BlockHeader), true);
+    if (!block) {
+      SetCorrupted();
+      return OFFSET_NULL;
+    }
+
+    // An allocation cannot cross page boundaries. If it would, create a
+    // "wasted" block and begin again at the top of the next page.
+    int32_t page_free = mem_page_ - freeptr % mem_page_;
+    if (size > page_free) {

[Dmitry Vyukov, 2015/11/04 13:52:29]

%K returns value in [0, K), not [1, K]
check for page_free = 0

[bcwhite, 2015/11/04 17:18:55]

I want [0, K).  If "freeptr" points to the start of a 4KiB page then "page_free"
should be 4096.  page_free can never be zero.

[Dmitry Vyukov, 2015/11/04 17:33:07]

I may be missing something then.
If we get page_free==0, then (size > page_free) is true, and we will try to
advice freeptr by 0, which will cause an infinite loop. If page_free==0, we must
not even try to advice freeptr. We have a whole page.
So I think it should be:
if (page_free == 0)
  page_free = mem_page_;
No?

[bcwhite, 2015/11/04 18:40:16]

mem_page_ > 0  therefore
freeptr % mem_page_ < mem_page_  therefore
mem_page_ - freeptr % mem_page_ > 0  therefore
page_free > 0

[Dmitry Vyukov, 2015/11/05 10:49:42]

Aha! I missed "mem_page_ - " part. Sorry.

+      int32_t new_freeptr = freeptr + page_free;
+      if (subtle::Release_CompareAndSwap(

[Dmitry Vyukov, 2015/11/04 13:52:29]

What do we release here? Where is the pairing acquire?

+              &shared_meta_->freeptr, freeptr, new_freeptr) == freeptr) {
+        block->size = page_free;

[Dmitry Vyukov, 2015/11/04 13:52:29]

Why do we need this?
We not don't iterate the region itself, only through the queue. So it is OK to
just bump freeptr.

[bcwhite, 2015/11/04 17:18:55]

I suppose it could be omitted now.  It was part of the original design that
provided iteration and recovery by making it possible to step through all
allocated objects.  Race conditions proved that insufficient in an active
system, hence the queue.

It feels "right" to have it, though.  And since the block header has to be
padded for alignment anyway, removing it would just leave a "reserved" field in
its place; no savings there.  So I'd like to keep it if you don't object too
much.

[Dmitry Vyukov, 2015/11/04 17:33:08]

I don't object too much. But then add a comment.
Future readers won't know about "the original design". And also it is the only
usage of kBlockCookieWasted, which makes it look very suspicious.

[bcwhite, 2015/11/04 18:40:16]

Done.

+        block->cookie = kBlockCookieWasted;
+      }
+      continue;
+    }
+
+    // Don't leave a slice at the end of a page too small for anything. This
+    // can result in an allocation up to two alignment-sizes greater than the
+    // minimum required by requested-size + header + alignment.
+    if (page_free - size < (int)(sizeof(BlockHeader) + kAllocAlignment))
+      size = page_free;
+
+    int32_t new_freeptr = freeptr + size;
+    if (new_freeptr > mem_size_) {
+      SetCorrupted();
+      return OFFSET_NULL;
+    }
+
+    if (subtle::Release_CompareAndSwap(

[Dmitry Vyukov, 2015/11/04 13:52:29]

What do we release here? Where is the pairing acquire?

+            &shared_meta_->freeptr, freeptr, new_freeptr) != freeptr) {
+      // Another thread must have completed an allocation while we were working.
+      // Try again.
+      continue;
+    }
+
+    // Given that all memory was zeroed before ever being given to an instance
+    // of this class and given that we only allocate in a monotomic fashion
+    // going forward, it must be that the newly allocated block is completely
+    // full of zeros. If we find anything in the block header that is NOT a
+    // zero then something must have previously run amuck through memory,
+    // writing beyond the allocated space and into unallocated space.
+    if (block->size != 0 ||
+        block->cookie != kBlockCookieFree ||
+        block->type_id != 0 ||
+        subtle::NoBarrier_Load(&block->next) != 0) {
+      SetCorrupted();
+      return OFFSET_NULL;
+    }
+
+    block->size = size;

[Dmitry Vyukov, 2015/11/04 13:52:29]

These should be atomic stores as they race with allocation checks in another
thread.
Or alternatively you can just not check these fields on allocation.

[bcwhite, 2015/11/04 17:18:55]

You mean the checks on lines 247-250?

[Dmitry Vyukov, 2015/11/04 17:33:08]

I mean checks in GetBlock done by another thread which now tries to allocate
this block.

+    block->cookie = kBlockCookieAllocated;
+    block->type_id = type_id;
+    return freeptr;
+  }
+}
+
+void SharedMemoryAllocator::GetMemoryInfo(MemoryInfo* meminfo) {
+  int32_t remaining =
+      mem_size_ - subtle::NoBarrier_Load(&shared_meta_->freeptr);
+  meminfo->total = mem_size_;
+  meminfo->free = IsCorrupted() ? 0 : remaining - sizeof(BlockHeader);
+}
+
+void SharedMemoryAllocator::MakeIterable(int32_t offset) {
+  if (IsCorrupted())
+    return;
+  BlockHeader* block = GetBlock(offset, 0, 0, false);
+  if (!block)  // invalid offset
+    return;
+  if (subtle::NoBarrier_Load(&block->next) != 0)  // previously set iterable
+    return;
+  subtle::NoBarrier_Store(&block->next, OFFSET_QUEUE);  // will be tail block
+
+  // Try to add this block to the tail of the queue. May take multiple tries.
+  int32_t tail;
+  for (;;) {
+    tail = subtle::Acquire_Load(&shared_meta_->tailptr);
+    block = GetBlock(tail, 0, 0, true);
+    if (!block) {
+      SetCorrupted();
+      return;
+    }
+    int32_t next = subtle::NoBarrier_Load(&block->next);
+
+    // Ensure that the tail pointer didn't change while reading next. Only
+    // the read of the tail pointer is atomic but we need to read both the
+    // tail pointer and the next pointer from it in an atomic fashion. The
+    // way to do this is to read both non-atomically and then verify after
+    // the second read that the first read is still valid/unchanged.
+    if (tail == subtle::Release_Load(&shared_meta_->tailptr)) {

[Dmitry Vyukov, 2015/11/04 13:52:29]

Why do we need the atomic read of both fields?
next or (next and tail) can change right after this check, so this is pointless
just use the return value from the failed CAS:

tail = tailptr
next = CAS(&tail->next, OFFSET_QUEUE, offset)
if next == OFFSET_QUEUE
  enqueued
else
  CAS(&tailptr, tail, next) // help


Also, Release_Load is a weird function without clearly defined semantics. It
will make porting to C++ atomics harder and ThreadSanitizer does not understand
it. Please don't use it.

[bcwhite, 2015/11/04 17:18:55]

This is how it is done in the M&S Queue paper -- the PDF link is in the comments
at the top of this file.  I expanded this comment based on my understanding but
I could be off.
They can change but the algorithm will fail later on and retry.

I believe the original paper assumes CAS has only a true/false return code.  As
you say, it may be possible to modify the algorithm to take advantage of our
version returning the existing value before the attempt.  For example, it may be
possible to combine the two "if" statements on lines 298 and 300 but there are
so many subtleties with these things that I'm loath to go against a researched
paper by people who know far more than I because there seems a better way with
the tools I have that they didn't.
Interesting.  I was looking at it only with regard to before/after ordering. 
I'll change it to an Acquire.

[Dmitry Vyukov, 2015/11/04 17:33:08]

What will break if we remove the CAS?

[bcwhite, 2015/11/04 18:40:16]

As I understand it...

If we remove the block-next update CAS on line 300 then it's possible for
another thread to insert into the queue only to be overwritten by this thread. 
The previous entry will become disconnected.

If we remove the tailptr update CAS on line 318 then it's possible for tailptr
to point to a node that isn't actually the tail.  This should get corrected by
the CAS on line 310 during the next enqueue attempt but will be in a bad state
until then.

[Dmitry Vyukov, 2015/11/05 10:49:42]

We need to understand this algorithm well enough to be sure that it is correct
and to be able to do future modifications. These kind of papers known to contain
bugs as well. And they are usually sloppy about proper memory barrier.

I propose to change this logic to:

for (;;) {
  tail = tailptr
  next = CAS(&tail->next, OFFSET_QUEUE, offset)
  if next == OFFSET_QUEUE {
    CAS(&tailptr, tail, offset)
    return
  } else
    CAS(&tailptr, tail, next) // help
}

It is way simpler. I can actually understand how it works.
The current "atomic" load of 2 fields raises more questions:
- what if block->next transitioned from OFFSET_QUEUE to offset of some node, is
it OK? how does it affect algorithm operation?
- what if block->next changes right after we do Release_Load check? Is it ok?
- if above is OK, why do we do the check at all?
- what if shared_meta_->tailptr changes right after we do the check? Is it ok? 

[bcwhite, 2015/11/05 14:37:15]

Fair enough.  Let's see what happens.
Nice.  I don't see any flaws in it.  The parallel test hasn't crashed or failed;
I have it running continuously at the moment.  Check my comments to make sure I
understood everything correctly.

+      // Check if the found block is truely the last in the queue (i.e. it
+      // points back to the "queue" node).
+      if (next == OFFSET_QUEUE) {
+        // Yes. Try to append the passed block after the current tail block.
+        if (subtle::Release_CompareAndSwap(
+                &block->next, OFFSET_QUEUE, offset) == OFFSET_QUEUE) {
+          // Success!  The block is enqueued; need to update the tail pointer.
+          break;
+        }
+      } else {
+        // No. Another thread has stopped between the block-next update
+        // and the tail-pointer update. Try to update tailptr past the
+        // found block. That other thread may complete it first or it
+        // may have crashed. Be fail-safe.
+        subtle::Release_CompareAndSwap(&shared_meta_->tailptr, tail, next);
+      }
+    }
+  }
+
+  // Block has been enqueued. Now update the tail-pointer past it. This
+  // could fail if another thread has already completed the operation as
+  // part of being fail-safe.
+  subtle::Release_CompareAndSwap(&shared_meta_->tailptr, tail, offset);
+}
+
+void SharedMemoryAllocator::CreateIterator(Iterator* state) {
+  state->last = OFFSET_QUEUE;
+  state->niter = 0;
+}
+
+int32_t SharedMemoryAllocator::GetNextIterable(Iterator* state,
+                                               int32_t* type_id) {
+  const BlockHeader* block = GetBlock(state->last, 0, 0, true);
+  if (!block)  // invalid iterator state
+    return OFFSET_NULL;
+  int32_t next = subtle::NoBarrier_Load(&block->next);

[Dmitry Vyukov, 2015/11/04 13:52:28]

this needs to be Acquire_Load, this is what acquires visibility of new nodes
from threads that concurrently add new nodes
synchronizes with Release_CAS(&block->next) in MakeIterable

there is another super-subtle ordering issue here (worth a comment)
currently it is possible that we load shared_meta_->freeptr first, then another
thread allocates and enqueues a node, then we load block->next and observe the
new node, as the result we can falsely fail the loop check and falsely mark the
region as corrupted 
if we use acquire_load here, it synchronizes with enqueue of the node, which in
turn synchronizes with allocation of the node (presumably happens in the same
thread); so the result of the load of freeptr below must cover the newly
observed node

[bcwhite, 2015/11/04 17:18:55]

Whew!  I've added a comment according to my understanding.  Please check that I
did actually understand.  :-D

Note, though, that the calculation below will actually return max+1 because
there is a memory-segment header that's worth a minimum allocation.  A dozen
allocations could be made in between but we're only returning one at a time.

+  block = GetBlock(next, 0, 0, false);
+  if (!block)  // no next allocation in queue
+    return OFFSET_NULL;
+
+  // Memory corruption could cause a loop in the list. We need to detect
+  // that so as to not cause an infinite loop in the caller. We do this
+  // simply by making sure we don't iterate more than the absolute maximum
+  // number of allocations that could have been made. Callers are likely
+  // to loop multiple times before it is detected but at least it stops.
+  int32_t freeptr = std::min(subtle::Acquire_Load(&shared_meta_->freeptr),

[Dmitry Vyukov, 2015/11/04 13:52:29]

visibility over what do we acquire here?

[bcwhite, 2015/11/04 17:18:55]

There must be something I don't understand about acquire/release naming because
to me it's just an atomic load of the value.

[Dmitry Vyukov, 2015/11/05 10:49:42]

Just atomic load is NoBarrier_Load.
Acquire/Release also give you mutual ordering of memory access as perceived by
other threads.

+                             mem_size_);
+  if (state->niter > freeptr / (sizeof(BlockHeader) + kAllocAlignment)) {
+    SetCorrupted();
+    return OFFSET_NULL;
+  }
+
+  state->last = next;
+  state->niter++;
+  *type_id = block->type_id;
+
+  return next;
+}
+
+// The "corrupted" state is held both locally and globally (shared). The
+// shared flag can't be trusted since a malicious actor could overwrite it.
+// The local version is immune to foreign actors. Thus, if seen shared,
+// copy it locally and, once known, always restore it globally.
+void SharedMemoryAllocator::SetCorrupted() {
+  LOG(ERROR) << "Corruption detected in shared-memory segment.";
+  subtle::NoBarrier_Store(&corrupted_, 1);
+  SetFlag(&shared_meta_->flags, kFlagCorrupted);
+}
+
+bool SharedMemoryAllocator::IsCorrupted() {
+  if (subtle::NoBarrier_Load(&corrupted_) ||
+      CheckFlag(&shared_meta_->flags, kFlagCorrupted)) {
+    SetCorrupted();  // Make sure all indicators are set.
+    return true;
+  }
+  return false;
+}
+
+bool SharedMemoryAllocator::IsFull() {
+  return CheckFlag(&shared_meta_->flags, kFlagFull);
+}
+
+// Dereference a block |offset| and ensure that it's valid for the desired
+// |type_id| and |size|. |special| indicates that we may try to access block
+// headers not available to callers but still accessed by this module. By
+// having internal dereferences go through this same function, the allocator
+// is hardened against corruption.
+SharedMemoryAllocator::BlockHeader* SharedMemoryAllocator::GetBlock(
+    int32_t offset,
+    int32_t type_id,
+    int32_t size,
+    bool special) {

[Dmitry Vyukov, 2015/11/04 13:52:29]

Split special flag into two flags: one allows to get pointer to OFFSET_QUEUE,
second allows to get pointer to non-allocated blocks.
Different call sites want one or another.
This will also get rid of the -size hack in Allocate (don't need to check
cookie, type and size for new blocks).

[bcwhite, 2015/11/04 17:18:55]

Done.

+  // Validation of parameters.
+  if (offset % kAllocAlignment != 0)
+    return nullptr;
+  if (offset < (int)(special ? OFFSET_QUEUE : sizeof(SharedMetadata)))
+    return nullptr;
+  size += sizeof(BlockHeader);
+  if (offset + size > mem_size_)
+    return nullptr;
+  int32_t freeptr = subtle::NoBarrier_Load(&shared_meta_->freeptr);
+  if (offset + size > freeptr)
+    return nullptr;
+
+  // Validation of referenced block-header.
+  const BlockHeader* block = reinterpret_cast<BlockHeader*>(mem_base_ + offset);
+  if (block->size < size)
+    return nullptr;
+  if (!special && block->cookie != kBlockCookieAllocated)
+    return nullptr;
+  if (type_id != 0 && block->type_id != type_id)
+    return nullptr;
+
+  // Return pointer to block data.
+  return reinterpret_cast<BlockHeader*>(mem_base_ + offset);
+}
+
+void* SharedMemoryAllocator::GetBlockData(int32_t offset,
+                                          int32_t type_id,
+                                          int32_t size,
+                                          bool special) {
+  DCHECK(size > 0);
+  BlockHeader* block = GetBlock(offset, type_id, size, special);
+  if (!block)
+    return nullptr;
+  return reinterpret_cast<char*>(block) + sizeof(BlockHeader);
+}
+
+}  // namespace base