diff --git a/core/core.services.yml b/core/core.services.yml
index 2e186ae..1601e3d 100644
--- a/core/core.services.yml
+++ b/core/core.services.yml
@@ -113,10 +113,10 @@ services:
     factory_class: Drupal\Component\Utility\Settings
     factory_method: getSingleton
   state:
-    class: Drupal\Core\KeyValueStore\KeyValueStoreInterface
-    factory_method: get
-    factory_service: keyvalue
-    arguments: [state]
+    class: Drupal\Core\KeyValueStore\KeyValueCacheDecorator
+    arguments: ['@cache.cache', '@lock', '@keyvalue', state]
+    tags:
+      - { name: needs_destruction }
   queue:
     class: Drupal\Core\Queue\QueueFactory
     arguments: ['@settings']
diff --git a/core/includes/bootstrap.inc b/core/includes/bootstrap.inc
index aa0fbcb..354d5db 100644
--- a/core/includes/bootstrap.inc
+++ b/core/includes/bootstrap.inc
@@ -2517,7 +2517,7 @@ function module_hook($module, $hook) {
  * @return Drupal\Core\KeyValueStore\KeyValueStoreInterface
  */
 function state() {
-  return drupal_container()->get('keyvalue')->get('state');
+  return drupal_container()->get('state');
 }
 
 /**
diff --git a/core/lib/Drupal/Core/Cache/CacheCollector.php b/core/lib/Drupal/Core/Cache/CacheCollector.php
new file mode 100644
index 0000000..78373ec
--- /dev/null
+++ b/core/lib/Drupal/Core/Cache/CacheCollector.php
@@ -0,0 +1,268 @@
+<?php
+
+/**
+ * @file
+ * Contains \Drupal\Core\Cache\CacheCollector.
+ */
+
+namespace Drupal\Core\Cache;
+
+use Drupal\Core\Cache\CacheBackendInterface;
+use Drupal\Core\DestructableInterface;
+use Drupal\Core\Lock\LockBackendInterface;
+
+/**
+ * Default implementation for CacheCollectorInterface.
+ *
+ * By default, the class accounts for caches where calling functions might
+ * request keys that won't exist even after a cache rebuild. This prevents
+ * situations where a cache rebuild would be triggered over and over due to a
+ * 'missing' item. These cases are stored internally as a value of NULL. This
+ * means that the CacheCollector::get() method must be overridden if caching
+ * data where the values can legitimately be NULL, and where
+ * CacheCollector->has() needs to correctly return (equivalent to
+ * array_key_exists() vs. isset()). This should not be necessary in the majority
+ * of cases.
+ */
+abstract class CacheCollector implements CacheCollectorInterface, DestructableInterface {
+
+  /**
+   * A cid to pass to cache()->set() and cache()->get().
+   *
+   * @var string
+   */
+  protected $cid;
+
+  /**
+   * A tags array to pass to cache()->set().
+   *
+   * @var array
+   */
+  protected $tags;
+
+  /**
+   * The cache backend that should be used.
+   *
+   * @var \Drupal\Core\Cache\CacheBackendInterface
+   */
+  protected $cache;
+
+  /**
+   * The lock backend that should be used.
+   *
+   * @var \Drupal\Core\Lock\LockBackendInterface
+   */
+  protected $lock;
+
+  /**
+   * An array of keys to add to the cache on service termination.
+   *
+   * @var array
+   */
+  protected $keysToPersist = array();
+
+  /**
+   * An array of keys to remove from the cache on service termination.
+   *
+   * @var array
+   */
+  protected $keysToRemove = array();
+
+  /**
+   * Storage for the data itself.
+   *
+   * @var array
+   */
+  protected $storage = array();
+
+  /**
+   * Stores the cache creation time.
+   *
+   * This is used to check if an invalidated cache item has been overwritten in
+   * the meantime.
+   *
+   * @var int
+   */
+  protected $cacheCreated;
+
+  /**
+   * Flag that indicates of the cache has been invalidated.
+   *
+   * @var bool
+   */
+  protected $cacheInvalidated = FALSE;
+
+  /**
+   * Constructs a CacheArray object.
+   *
+   * @param string $cid
+   *   The cid for the array being cached.
+   * @param \Drupal\Core\Cache\CacheBackendInterface $cache
+   *   The cache backend.
+   * @param \Drupal\Core\Lock\LockBackendInterface $lock
+   *   The lock backend.
+   * @param array $tags
+   *   (optional) The tags to specify for the cache item.
+   */
+  public function __construct($cid, CacheBackendInterface $cache, LockBackendInterface $lock, $tags = array()) {
+    $this->cid = $cid;
+    $this->cache = $cache;
+    $this->tags = $tags;
+    $this->lock = $lock;
+
+    if ($cache = $this->cache->get($this->cid)) {
+      $this->cacheCreated = $cache->created;
+      $this->storage = $cache->data;
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function has($key) {
+    return $this->get($key) !== NULL;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function get($key) {
+    if (isset($this->storage[$key]) || array_key_exists($key, $this->storage)) {
+      return $this->storage[$key];
+    }
+    else {
+      return $this->resolveCacheMiss($key);
+    }
+  }
+
+  /**
+   * Implements CacheCollectorInterface::set().
+   *
+   * This is not persisted by default. In practice this means that setting a
+   * value will only apply while the object is in scope and will not be written
+   * back to the persistent cache. This follows a similar pattern to static vs.
+   * persistent caching in procedural code. Extending classes may wish to alter
+   * this behavior, for example by adding a call to persist().
+   */
+  public function set($key, $value) {
+    $this->storage[$key] = $value;
+    // The key might have been marked for deletion.
+    unset($this->keysToRemove[$key]);
+    // Invalidate the cache to make sure that other requests immediately see the
+    // deletion before this request is terminated.
+    $this->cache->invalidate($this->cid);
+    $this->cacheInvalidated = TRUE;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function delete($key) {
+    unset($this->storage[$key]);
+    $this->keysToRemove[$key] = $key;
+    // The key might have been marked for persisting.
+    unset($this->keysToPersist[$key]);
+    // Invalidate the cache to make sure that other requests immediately see the
+    // deletion before this request is terminated.
+    $this->cache->invalidate($this->cid);
+    $this->cacheInvalidated = TRUE;
+  }
+
+  /**
+   * Flags an offset value to be written to the persistent cache.
+   *
+   * @param string $key
+   *   The key that was request.
+   * @param bool $persist
+   *   (optional) Whether the offset should be persisted or not, defaults to
+   *   TRUE. When called with $persist = FALSE the offset will be unflagged so
+   *   that it will not written at the end of the request.
+   */
+  protected function persist($key, $persist = TRUE) {
+    $this->keysToPersist[$key] = $persist;
+  }
+
+  /**
+   * Resolves a cache miss.
+   *
+   * When an offset is not found in the object, this is treated as a cache
+   * miss. This method allows classes using this implementatio to look up the
+   * actual value and allow it to be cached.
+   *
+   * @param sring $key
+   *   The offset that was requested.
+   *
+   * @return mixed
+   *   The value of the offset, or NULL if no value was found.
+   */
+  abstract protected function resolveCacheMiss($key);
+
+  /**
+   * Writes a value to the persistent cache immediately.
+   *
+   * @param bool $lock
+   *   (optional) Whether to acquire a lock before writing to cache. Defaults to
+   *   TRUE.
+   */
+  protected function updateCache($lock = TRUE) {
+    $data = array();
+    foreach ($this->keysToPersist as $offset => $persist) {
+      if ($persist) {
+        $data[$offset] = $this->storage[$offset];
+      }
+    }
+    if (empty($data)) {
+      return;
+    }
+
+    // Lock cache writes to help avoid stampedes.
+    // To implement locking for cache misses, override __construct().
+    $lock_name = $this->cid . ':' . __CLASS__;
+    if (!$lock || $this->lock->acquire($lock_name)) {
+      // Set and delete operations invalidate the cache item. Try to also load
+      // an eventually invalidated cache entry, only update an invalidated cache
+      // entry if the creation date did not change as this could result in an
+      // inconsistent cache.
+      if ($cache = $this->cache->get($this->cid, $this->cacheInvalidated)) {
+        if ($this->cacheInvalidated && $cache->created != $this->cacheCreated) {
+          // We have invalidated the cache in this request and got a different
+          // cache entry, do not attempt to overwrite data that might have been
+          // changed in a different request. We'll let the cache rebuild in
+          // later requests.
+          $this->cache->delete($this->cid);
+          $this->lock->release($lock_name);
+          return;
+        }
+        $data = array_merge($cache->data, $data);
+      }
+      // Remove keys marked for deletion.
+      foreach ($this->keysToRemove as $delete_key) {
+        unset($data[$delete_key]);
+      }
+      $this->cache->set($this->cid, $data, CacheBackendInterface::CACHE_PERMANENT, $this->tags);
+      if ($lock) {
+        $this->lock->release($lock_name);
+      }
+    }
+
+    $this->keysToPersist = array();
+    $this->keysToRemove = array();
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function reset() {
+    $this->storage = array();
+    $this->keysToPersist = array();
+    $this->keysToRemove = array();
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function destruct() {
+    $this->updateCache();
+  }
+
+}
diff --git a/core/lib/Drupal/Core/Cache/CacheCollectorInterface.php b/core/lib/Drupal/Core/Cache/CacheCollectorInterface.php
new file mode 100644
index 0000000..6862543
--- /dev/null
+++ b/core/lib/Drupal/Core/Cache/CacheCollectorInterface.php
@@ -0,0 +1,72 @@
+<?php
+
+/**
+ * @file
+ * Contains \Drupal\Core\Cache\CacheCollectorInterface.
+ */
+
+namespace Drupal\Core\Cache;
+
+/**
+ * Provides a caching wrapper to be used in place of large structures.
+ *
+ * This should be extended by systems that need to cache large amounts of data
+ * to calling functions. These structures can become very large, so this
+ * class is used to allow different strategies to be used for caching internally
+ * (lazy loading, building caches over time etc.). This can dramatically reduce
+ * the amount of data that needs to be loaded from cache backends on each
+ * request, and memory usage from static caches of that same data.
+ *
+ * The default implementation is CacheCollector.
+ */
+Interface CacheCollectorInterface {
+
+  /**
+   * Gets value from the cache.
+   *
+   * @param string $key
+   *   Key that identifies the data.
+   *
+   * @return mixed
+   *   The corresponding cache data.
+   */
+  public function get($key);
+
+  /**
+   * Sets cache data.
+   *
+   * It depends on the specific case and implementation whether this has a
+   * permanent effect or if it just affects the current request.
+   *
+   * @param string $key
+   *   Key that identifies the data.
+   * @param mixed $value
+   *   The data to be set.
+   */
+  public function set($key, $value);
+
+  /**
+   * Deletes the element.
+   *
+   * It depends on the specific case and implementation whether this has a
+   * permanent effect or if it just affects the current request.
+   *
+   * @param string $key
+   *   Key that identifies the data.
+   */
+  public function delete($key);
+
+  /**
+   * Returns whether data exists for this key.
+   *
+   * @param string $key
+   *   Key that identifies the data.
+   */
+  public function has($key);
+
+  /**
+   * Resets the local cache.
+   */
+  public function reset();
+
+}
diff --git a/core/lib/Drupal/Core/Cache/MemoryBackend.php b/core/lib/Drupal/Core/Cache/MemoryBackend.php
index ac83a06..be1fb89 100644
--- a/core/lib/Drupal/Core/Cache/MemoryBackend.php
+++ b/core/lib/Drupal/Core/Cache/MemoryBackend.php
@@ -142,7 +142,9 @@ public function deleteAll() {
    * Implements Drupal\Core\Cache\CacheBackendInterface::invalidate().
    */
   public function invalidate($cid) {
-    $this->cache[$cid]->expire = REQUEST_TIME - 1;
+    if (isset($this->cache[$cid])) {
+      $this->cache[$cid]->expire = REQUEST_TIME - 1;
+    }
   }
 
   /**
diff --git a/core/lib/Drupal/Core/KeyValueStore/KeyValueCacheDecorator.php b/core/lib/Drupal/Core/KeyValueStore/KeyValueCacheDecorator.php
new file mode 100644
index 0000000..46151b6
--- /dev/null
+++ b/core/lib/Drupal/Core/KeyValueStore/KeyValueCacheDecorator.php
@@ -0,0 +1,142 @@
+<?php
+
+/**
+ * @file
+ * Contains \Drupal\Core\KeyValueStore\KeyValueCacheDecorator.
+ */
+
+namespace Drupal\Core\KeyValueStore;
+
+use Drupal\Core\Cache\CacheBackendInterface;
+use Drupal\Core\Cache\CacheCollector;
+use Drupal\Core\DestructableInterface;
+use Drupal\Core\Lock\LockBackendInterface;
+
+/**
+ * Provides a decorator for a key value store that caches all requested keys.
+ */
+class KeyValueCacheDecorator extends CacheCollector implements KeyValueStoreInterface, DestructableInterface {
+
+  /**
+   * The key value store to use.
+   *
+   * @var \Drupal\Core\KeyValueStore\KeyValueStoreInterface;
+   */
+  protected $keyValueStore;
+
+  /**
+   * Constructs a key value cache decorator.
+   *
+   * @param \Drupal\Core\Cache\CacheBackendInterface $cache
+   *   The cache backend to store the cache in.
+   * @param \Drupal\Core\Lock\LockBackendInterface $lock
+   *   The lock implementation to use when writing a changed cache storage.
+   * @param \Drupal\Core\KeyValueStore\KeyValueFactory $key_value_factory
+   *   The key value factory to get the key value storage from.
+   * @param string $collection
+   *   Name of the key value storage collection, also used as the cache id.
+   */
+  public function __construct(CacheBackendInterface $cache, LockBackendInterface $lock, KeyValueFactory $key_value_factory, $collection) {
+    parent::__construct($collection, $cache, $lock);
+    $this->keyValueStore = $key_value_factory->get($collection);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function resolveCacheMiss($offset) {
+    $this->storage[$offset] = $this->keyValueStore->get($offset);
+    $this->persist($offset);
+    return $this->storage[$offset];
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function delete($key) {
+    parent::delete($key);
+    $this->keyValueStore->delete($key);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function deleteMultiple(array $keys) {
+    foreach ($keys as $key) {
+      $this->delete($key);
+    }
+    $this->keyValueStore->deleteMultiple($keys);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getAll() {
+    // Caching this would mean that the whole store is added to the cache,
+    // this is expected to be a non-frequent operation that is not worth to be
+    // loaded from cache.
+    return $this->keyValueStore->getAll();
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getCollectionName() {
+    return $this->keyValueStore->getCollectionName();
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getMultiple(array $keys) {
+    $values = array();
+    foreach ($keys as $key) {
+      $value = $this->get($key);
+      // Only return keys with a value.
+      if ($value !== NULL) {
+        $values[$key] = $value;
+      }
+    }
+    return $values;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function setIfNotExists($key, $value) {
+    if ($this->keyValueStore->setIfNotExists($key, $value)) {
+      $this->set($key, $value);
+      return TRUE;
+    }
+    return FALSE;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function setMultiple(array $data) {
+    $this->keyValueStore->setMultiple($data);
+    foreach ($data as $key => $value) {
+      parent::set($key, $value);
+      $this->keysToPersist[$key] = $value;
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function set($key, $value) {
+    $this->keyValueStore->set($key, $value);
+    parent::set($key, $value);
+    $this->persist($key);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function deleteAll() {
+    $this->keyValueStore->deleteAll();
+    $this->cache->delete($this->cid);
+  }
+
+}
diff --git a/core/modules/simpletest/lib/Drupal/simpletest/DrupalUnitTestBase.php b/core/modules/simpletest/lib/Drupal/simpletest/DrupalUnitTestBase.php
index e5f33d4..67809f7 100644
--- a/core/modules/simpletest/lib/Drupal/simpletest/DrupalUnitTestBase.php
+++ b/core/modules/simpletest/lib/Drupal/simpletest/DrupalUnitTestBase.php
@@ -168,6 +168,11 @@ public function containerBuild(ContainerBuilder $container) {
       $container
         ->register('keyvalue', 'Drupal\Core\KeyValueStore\KeyValueFactory')
         ->addArgument(new Reference('service_container'));
+
+      $container->register('state', 'Drupal\Core\KeyValueStore\KeyValueStoreInterface')
+        ->setFactoryService(new Reference('keyvalue'))
+        ->setFactoryMethod('get')
+       ->addArgument('state');
     }
   }
 
diff --git a/core/modules/simpletest/lib/Drupal/simpletest/TestBase.php b/core/modules/simpletest/lib/Drupal/simpletest/TestBase.php
index a67d091..4c2fdf0 100644
--- a/core/modules/simpletest/lib/Drupal/simpletest/TestBase.php
+++ b/core/modules/simpletest/lib/Drupal/simpletest/TestBase.php
@@ -992,6 +992,15 @@ protected function tearDown() {
     // which means they may need to access its filesystem and database.
     drupal_static_reset();
 
+    if (($container = drupal_container()) && $container->has('keyvalue')) {
+      $captured_emails = state()->get('system.test_email_collector') ?: array();
+      $emailCount = count($captured_emails);
+      if ($emailCount) {
+        $message = format_plural($emailCount, '1 e-mail was sent during this test.', '@count e-mails were sent during this test.');
+        $this->pass($message, t('E-mail'));
+      }
+    }
+
     // Ensure that TestBase::changeDatabasePrefix() has run and TestBase::$setup
     // was not tricked into TRUE, since the following code would delete the
     // entire parent site otherwise.
@@ -1013,14 +1022,6 @@ protected function tearDown() {
     // In case a fatal error occurred that was not in the test process read the
     // log to pick up any fatal errors.
     simpletest_log_read($this->testId, $this->databasePrefix, get_class($this), TRUE);
-    if (($container = drupal_container()) && $container->has('keyvalue')) {
-      $captured_emails = state()->get('system.test_email_collector') ?: array();
-      $emailCount = count($captured_emails);
-      if ($emailCount) {
-        $message = format_plural($emailCount, '1 e-mail was sent during this test.', '@count e-mails were sent during this test.');
-        $this->pass($message, t('E-mail'));
-      }
-    }
 
     // Delete temporary files directory.
     file_unmanaged_delete_recursive($this->originalFileDirectory . '/simpletest/' . substr($this->databasePrefix, 10), array($this, 'filePreDeleteCallback'));
diff --git a/core/modules/simpletest/lib/Drupal/simpletest/WebTestBase.php b/core/modules/simpletest/lib/Drupal/simpletest/WebTestBase.php
index 4b66223..8ace9c3 100644
--- a/core/modules/simpletest/lib/Drupal/simpletest/WebTestBase.php
+++ b/core/modules/simpletest/lib/Drupal/simpletest/WebTestBase.php
@@ -926,7 +926,8 @@ protected function refreshVariables() {
     global $conf;
     cache('bootstrap')->delete('variables');
     $conf = variable_initialize();
-    drupal_container()->get('config.factory')->reset();
+    \Drupal::service('config.factory')->reset();
+    \Drupal::state()->reset();
   }
 
   /**
diff --git a/core/modules/system/lib/Drupal/system/Tests/KeyValueStore/CacheDecoratorTest.php b/core/modules/system/lib/Drupal/system/Tests/KeyValueStore/CacheDecoratorTest.php
new file mode 100644
index 0000000..a18b102
--- /dev/null
+++ b/core/modules/system/lib/Drupal/system/Tests/KeyValueStore/CacheDecoratorTest.php
@@ -0,0 +1,87 @@
+<?php
+
+/**
+ * @file
+ * Contains \Drupal\system\Tests\KeyValueStore\CacheDecoratorTest.
+ */
+
+namespace Drupal\system\Tests\KeyValueStore;
+
+use Drupal\Core\Cache\MemoryBackend;
+use Drupal\Core\KeyValueStore\KeyValueCacheDecorator;
+use Drupal\Core\Lock\NullLockBackend;
+
+/**
+ * Tests the key-value keyvalue database storage.
+ */
+class CacheDecoratorTest extends StorageTestBase {
+
+  /**
+   * The cache backend in which the caches are stored.
+   *
+   * @var \Drupal\Core\Cache\CacheBackendInterface
+   */
+  protected $cache;
+
+  public static function getInfo() {
+    return array(
+      'name' => 'Key value cache decorator',
+      'description' => 'Tests the key value cache decorator.',
+      'group' => 'Key-value store',
+    );
+  }
+
+  protected function setUp() {
+    parent::setUp();
+    $this->container
+      ->register('keyvalue.memory', 'Drupal\Core\KeyValueStore\KeyValueMemoryFactory');
+    global $conf;
+    $conf['keyvalue_default'] = 'keyvalue.memory';
+    $this->cache = new MemoryBackend('bin');
+  }
+
+  /**
+   * Tests that values are cached.
+   */
+  public function testCache() {
+    $stores = $this->createStorage();
+    $values = array();
+    // Set the value and test that it is correctly returned.
+    foreach ($this->collections as $i => $collection) {
+      $stores[$i]->set('key', $this->objects[$i]);
+      $this->assertEqual($stores[$i]->get('key'), $this->objects[$i]);
+      // Destruct the class to have it write the cache.
+      $stores[$i]->destruct();
+
+      // Delete the value from the key value storage.
+      $this->container->get($this->factory)->get($collection)->delete('key');
+    }
+
+    // Create new objects.
+    $stores = $this->createStorage();
+
+    // Verify that we get the cached state as we have not notified the decorator
+    // about the deletion.
+    foreach ($this->collections as $i => $collection) {
+      $this->assertEqual($stores[$i]->get('key'), $this->objects[$i]);
+
+      // Reset the cache and make sure the value was updated.
+      $stores[$i]->reset();
+      $this->assertNull($stores[$i]->get('key'));
+    }
+  }
+
+  /**
+   * Overrides StorageTestBase::createStorage()
+   */
+  protected function createStorage() {
+    $stores = array();
+    // Prepare the memory key value storages and decorated ones.
+    foreach ($this->collections as $i => $collection) {
+      $stores[$i] = new KeyValueCacheDecorator($this->cache, new NullLockBackend(), $this->container->get($this->factory), $collection);
+    }
+
+    return $stores;
+  }
+
+}