Repository: glotzerlab/signac
Pull request #462 (2021-01-01 20:59:46)

Implements an in-memory buffer for SyncedCollections
<!-- Provide a general summary of your changes in the Title above -->

## Description
<!-- Describe your changes in detail -->

In case readers want to skip to them, benchmarks are at the bottom of the motivation section. 
[Here's the notebook I used to run the benchmarks](https://github.com/glotzerlab/signac/files/5759435/TestDifferentBuffers.txt) (saved as a text file because Github doesn't know how to handle anything else, but just rename it to ipynb and it'll load fine).

This PR implements an alternative to the current buffering strategy used by SyncedCollections (and by the JSONDict in signac master). Rather than serializing and deserializing the data into a JSON-encoded buffer on every operation, a buffering strategy that saves on I/O costs but still has to do a lot of work, it allows all SyncedCollections pointing a given file to directly share their internal data so that any operation on one is transparently reflected by the others. In addition to skipping serialization, this method also avoids the need for recursive updates of nested collections, which can be expensive. In a sense it's no longer true buffering because we're not simulating a write by serializing the data in the same manner that we would if we were truly writing to an intermediate buffer, we're stopping at the data being stored in memory.

## Motivation and Context
<!-- Why is this change required? What problem does it solve? -->
<!-- If it fixes an open issue, please link to the issue here. -->

The current buffering implementation is quite expensive. Although it largely avoids I/O operations, maintains a fully JSON-encoded buffer of all the data associated with existing SyncedCollection objects, which has a few notable benefits relative to the approach in this PR:

1. Data in the buffer can be directly written to disk
2. JSON-encoded data can be hashed to perform relatively cheap equality checks.
3. The total buffer size is trivially the sum of all JSON-encoded bytes in the buffer, so a buffer capacity can be set explicitly to control buffer flushes. 

However, this approach also has significant performance implications:

1. Every write operation requires a JSON encoding step
2. Every read requires decoding the encoded JSON
3. Every read requires updating the collection's internal data store to match the one in the buffer, which requires an expensive recursive traversal of the (possibly nested) data structure.

The approach in this PR circumvents these three performance bottlenecks by making all collections synced with a particular file directly share their underlying data attribute (an object of the "base type" that contains the data, e.g. a dict for a `SyncedAttrDict`). Since all mutable collections in Python are references, any change to one instance is transparently conveyed to others.

There are some tradeoffs with this approach when considering the advantages of the old approach listed above, but they can be mitigated to some extent:

1. While data in the buffer cannot be directly written to disk, this just means that we're deferring the cost of JSON-encoding to when the buffer flush occurs, it does not change the total amount of work to be done.
2. The equality checks are primarily of use as a way to determine whether or not the buffered data has changed from the version on disk, which indicates that it needs to be written. The new approach can't directly check this without performing a dict equality check, which I have not implemented but I would expect to be slow. However, we can use a simpler approach (which is currently implemented) that just checks whether any mutating operation has been called (e.g. setting an element); if the only operations that occur are read operations, then there's no need to write back to disk. The only scenario where this method will be slower than the old approach is in code that sets elements of a collection to the exact original values, which I don't think we need to optimize for. If that use case truly becomes a bottleneck (e.g. in status updates in flow), it's simple enough to just add a conditional check in client code that verifies whether the value will actually be changed, e.g. `if value != synced_dict[key]: synced_dict[key] = value`.
3. Although we can no longer exactly calculate the buffer size, we can employ various techniques to approximate it. The crudest one that I can think of would just be to assume that the collection's size won't increase by more than some factor --  say an order of magnitude -- during one instance of buffering. We already use `stat` calls to get file metadata when a collection first inserts into the buffer, so we could just also insert the original file size (which is also provided by `os.stat`) and then use that as a benchmark for how much can be inserted before a buffer flush is necessary. I'm open to other suggestions for this as well.

The performance payoff is shown in the table below. To get a sense for how much I/O contributes to the performance characteristics of buffered mode, I tested on three separate file systems. I tested both read (`d[key]`) and write (`d[key] = value`) operations. To capture the balance between in-memory operations and disk reads/writes, I perform 1000 total operations in each benchmark, distributed over different numbers of files. For example, some benchmarks read 1000 values from the same JSON file, while others read 1 value from 1000 different files. Note that there's some significant variability in the results because I only ran 10 loops in the benchmark, but I think the numbers are accurate enough for the purposes of this PR. 


| System             |  Operation | Number of dicts | Elements per dict  | Speed multiplier |
|------------------|------|---------------|-------------------|---------------------------------|
| Mac SSD  | setitem   | 1 | 1000 | 24.72 |
| Mac SSD  | setitem   | 10 | 100 | 21.79 |
| Mac SSD | setitem   | 100 | 10 | 12.19 |
| Mac SSD | setitem   | 1000 | 1 | 3.97 |
| Mac SSD | getitem   | 1 | 1000 | 182.96 |
| Mac SSD  | getitem   | 10 | 100 | 187.33 |
| Mac SSD | getitem   | 100 | 10 | 166.74 |
| Mac SSD | getitem   | 1000 | 1 | 98.47 |
| Linux NFS  | setitem   | 1 | 1000 | 16.39 |
| Linux NFS  | setitem   | 10 | 100 | 7.58 |
| Linux NFS  | setitem   | 100 | 10 | 2.51 |
| Linux NFS  | setitem   | 1000 | 1 | 1.33 |
| Linux NFS  | getitem   | 1 | 1000 | 133.48 |
| Linux NFS  | getitem   | 10 | 100 | 138.15 |
| Linux NFS  | getitem   | 100 | 10 | 115.78 |
| Linux NFS  | getitem   | 1000 | 1 | 72.95 |
| Linux scratch  | setitem   | 1 | 1000 | 22.35 |
| Linux scratch  | setitem   | 10 | 100 | 20.48 |
| Linux scratch  | setitem   | 100 | 10 | 11.93 |
| Linux scratch  | setitem   | 1000 | 1 | 3.25 |
| Linux scratch  | getitem   | 1 | 1000 | 143.59 |
| Linux scratch  | getitem   | 10 | 100 | 141.50 |
| Linux scratch  | getitem   | 100 | 10 | 116.29 |
| Linux scratch  | getitem   | 1000 | 1 | 71.02 |



## Types of Changes
<!-- Please select all items that apply either now or after creating the pull request: -->
- [ ] Documentation update
- [ ] Bug fix
- [ ] New feature
- [ ] Breaking change<sup>1</sup>

<sup>1</sup>The change breaks (or has the potential to break) existing functionality.

## Checklist:
<!-- Please select all items that apply either now or after creating the pull request. -->
<!-- If you are unsure about any of these items, do not hesitate to ask! -->
- [ ] I am familiar with the [**Contributing Guidelines**](https://github.com/glotzerlab/signac/blob/master/CONTRIBUTING.md).
- [ ] I agree with the terms of the [**Contributor Agreement**](https://github.com/glotzerlab/signac/blob/master/ContributorAgreement.md).
- [ ] My name is on the [list of contributors](https://github.com/glotzerlab/signac/blob/master/contributors.yaml).
- [ ] My code follows the [code style guideline](https://github.com/glotzerlab/signac/blob/master/CONTRIBUTING.md#code-style) of this project.
- [ ] The changes introduced by this pull request are covered by existing or newly introduced tests.

If necessary:
- [ ] I have updated the API documentation as part of the package doc-strings.
- [ ] I have created a separate pull request to update the [framework documentation](https://docs.signac.io/) on [signac-docs](https://github.com/glotzerlab/signac-docs) and linked it here.
- [ ] I have updated the [changelog](https://github.com/glotzerlab/signac/blob/master/changelog.txt) and added all related issue and pull request numbers for future reference (if applicable). See example below.

<!-- Example for a changelog entry: `Fix issue with launching rockets to the moon (#101, #212).` -->


codecov[bot]: # [Codecov](https://codecov.io/gh/glotzerlab/signac/pull/462?src=pr&el=h1) Report
> Merging [#462](https://codecov.io/gh/glotzerlab/signac/pull/462?src=pr&el=desc) (9c470f4) into [feature/synced_collection/optimization](https://codecov.io/gh/glotzerlab/signac/commit/bde520e9c6b6a05a743114bea37b1037761f5bff?el=desc) (bde520e) will **decrease** coverage by `0.59%`.
> The diff coverage is `36.45%`.

[![Impacted file tree graph](https://codecov.io/gh/glotzerlab/signac/pull/462/graphs/tree.svg?width=650&height=150&src=pr&token=FghbjTyzgM)](https://codecov.io/gh/glotzerlab/signac/pull/462?src=pr&el=tree)

```diff
@@                            Coverage Diff                             @@
##           feature/synced_collection/optimization     #462      +/-   ##
==========================================================================
- Coverage                                   77.85%   77.25%   -0.60%     
==========================================================================
  Files                                          56       57       +1     
  Lines                                        6575     6671      +96     
  Branches                                     1232     1248      +16     
==========================================================================
+ Hits                                         5119     5154      +35     
- Misses                                       1155     1214      +59     
- Partials                                      301      303       +2     
```


| [Impacted Files](https://codecov.io/gh/glotzerlab/signac/pull/462?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...e/synced\_collections/memory\_buffered\_collection.py](https://codecov.io/gh/glotzerlab/signac/pull/462/diff?src=pr&el=tree#diff-c2lnbmFjL2NvcmUvc3luY2VkX2NvbGxlY3Rpb25zL21lbW9yeV9idWZmZXJlZF9jb2xsZWN0aW9uLnB5) | `31.70% <31.70%> (ø)` | |
| [signac/core/synced\_collections/collection\_json.py](https://codecov.io/gh/glotzerlab/signac/pull/462/diff?src=pr&el=tree#diff-c2lnbmFjL2NvcmUvc3luY2VkX2NvbGxlY3Rpb25zL2NvbGxlY3Rpb25fanNvbi5weQ==) | `93.18% <64.28%> (-5.47%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/glotzerlab/signac/pull/462?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/glotzerlab/signac/pull/462?src=pr&el=footer). Last update [bde520e...9c470f4](https://codecov.io/gh/glotzerlab/signac/pull/462?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

bdice: @vyasr The concept seems sound. I haven’t reviewed the code carefully but your description of the feature and the large speedups indicate this is probably a very good idea. The benchmarks for the slow network filesystems are the most important in my opinion. It makes the penalty of excess system calls most obvious.
vyasr: @bdice just to make sure that we're on the same page, this PR does not change the amount of I/O performed at all (except in the one edge case mentioned in point 2 above). The slow nfs benchmarks simply indicate that on nfs I/O costs dramatically outweigh all others to the point where removing JSON-serialization and internal updates only leads to marginal performance improvements. So the other benchmarks are much more representative of how expensive all of the _other_ tasks are on a fast filesystem. On the other hand, the nfs benchmarks indicate the limited impact those changes will have when the filesystem is slow. I'm not sure if that's what you mean by "most important".
bdice: @vyasr Ok! Thanks for clarifying, that was helpful.

Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index f3dadee1..ec5c66d6 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -47,6 +47,8 @@ jobs:
   linux-python-39: &linux-template
     docker:
       - image: circleci/python:3.9
+      - image: circleci/mongo:latest
+      - image: circleci/redis:latest
 
     environment:
       PYTHON: python
@@ -110,16 +112,22 @@ jobs:
     <<: *linux-template
     docker:
       - image: circleci/python:3.8
+      - image: circleci/mongo:latest
+      - image: circleci/redis:latest
 
   linux-python-37:
     <<: *linux-template
     docker:
       - image: circleci/python:3.7
+      - image: circleci/mongo:latest
+      - image: circleci/redis:latest
 
   linux-python-36:
     <<: *linux-template
     docker:
       - image: circleci/python:3.6
+      - image: circleci/mongo:latest
+      - image: circleci/redis:latest
 
   linux-pypy-3:
     <<: *linux-template
diff --git a/signac/core/synced_collections/collection_json.py b/signac/core/synced_collections/collection_json.py
index 0d0a54a6..df638f74 100644
--- a/signac/core/synced_collections/collection_json.py
+++ b/signac/core/synced_collections/collection_json.py
@@ -10,6 +10,7 @@ import uuid
 import warnings
 
 from .file_buffered_collection import FileBufferedCollection
+from .memory_buffered_collection import SharedMemoryFileBufferedCollection
 from .synced_attr_dict import SyncedAttrDict
 from .synced_collection import SyncedCollection
 from .synced_list import SyncedList
@@ -76,6 +77,7 @@ class JSONCollection(SyncedCollection):
     """
 
     _backend = __name__  # type: ignore
+    _supports_threading = True
 
     def __init__(
         self, filename=None, write_concern=False, parent=None, *args, **kwargs
@@ -107,8 +109,10 @@ class JSONCollection(SyncedCollection):
         # Serialize data
         blob = json.dumps(self, cls=SCJSONEncoder).encode()
         # When write_concern flag is set, we write the data into dummy file and then
-        # replace that file with original file.
-        if self._write_concern:
+        # replace that file with original file. We also enable this mode
+        # irrespective of the write_concern flag if we're running in
+        # multithreaded mode.
+        if self._write_concern or hasattr(type(self), "_locks"):
             dirname, filename = os.path.split(self._filename)
             fn_tmp = os.path.join(dirname, f"._{uuid.uuid4()}_{filename}")
             with open(fn_tmp, "wb") as tmpfile:
@@ -123,6 +127,10 @@ class JSONCollection(SyncedCollection):
         """str: The name of the file this collection is synchronized with."""
         return self._filename
 
+    @property
+    def _lock_id(self):
+        return self._filename
+
 
 JSONCollection.add_validator(json_format_validator, _convert_key_to_str)
 
@@ -141,6 +149,20 @@ class BufferedJSONCollection(FileBufferedCollection, JSONCollection):
         super().__init__(*args, **kwargs)
 
 
+class MemoryBufferedJSONCollection(SharedMemoryFileBufferedCollection, JSONCollection):
+    """A :class:`JSONCollection` that supports I/O buffering.
+
+    This class implements the buffer protocol defined by :class:`BufferedCollection`.
+    The concrete implementation of buffering behavior is defined by the
+    :class:`FileBufferedCollection`.
+    """
+
+    _backend = __name__ + ".memory_buffered"  # type: ignore
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+
 class JSONDict(JSONCollection, SyncedAttrDict):
     """A dict-like mapping interface to a persistent JSON file.
 
@@ -311,3 +333,55 @@ class BufferedJSONList(BufferedJSONCollection, SyncedList):
             *args,
             **kwargs,
         )
+
+
+class MemoryBufferedJSONDict(MemoryBufferedJSONCollection, SyncedAttrDict):
+    """A buffered :class:`JSONDict`."""
+
+    _PROTECTED_KEYS = SyncedAttrDict._PROTECTED_KEYS + (
+        "_filename",
+        "_buffered",
+        "_is_buffered",
+    )
+
+    def __init__(
+        self,
+        filename=None,
+        write_concern=False,
+        data=None,
+        parent=None,
+        *args,
+        **kwargs,
+    ):
+        self._validate_constructor_args({"filename": filename}, data, parent)
+        super().__init__(
+            filename=filename,
+            write_concern=write_concern,
+            data=data,
+            parent=parent,
+            *args,
+            **kwargs,
+        )
+
+
+class MemoryBufferedJSONList(MemoryBufferedJSONCollection, SyncedList):
+    """A buffered :class:`JSONList`."""
+
+    def __init__(
+        self,
+        filename=None,
+        write_concern=False,
+        data=None,
+        parent=None,
+        *args,
+        **kwargs,
+    ):
+        self._validate_constructor_args({"filename": filename}, data, parent)
+        super().__init__(
+            filename=filename,
+            write_concern=write_concern,
+            data=data,
+            parent=parent,
+            *args,
+            **kwargs,
+        )
diff --git a/signac/core/synced_collections/file_buffered_collection.py b/signac/core/synced_collections/file_buffered_collection.py
index f416197a..47dd3834 100644
--- a/signac/core/synced_collections/file_buffered_collection.py
+++ b/signac/core/synced_collections/file_buffered_collection.py
@@ -256,8 +256,8 @@ class FileBufferedCollection(BufferedCollection):
             # hash (which is used for the consistency check) with the hash of
             # the current data on disk. _initialize_data_in_cache always uses
             # the current metadata, so the only extra work here is to modify
-            # the hash after it's called (since it uses self._to_base()) to get
-            # the data to initialize the cache with.
+            # the hash after it's called (since it uses self._to_base() to get
+            # the data to initialize the cache with).
             self._initialize_data_in_cache()
             disk_data = self._load_from_resource()
             FileBufferedCollection._cache[self._filename]["hash"] = self._hash(
diff --git a/signac/core/synced_collections/memory_buffered_collection.py b/signac/core/synced_collections/memory_buffered_collection.py
new file mode 100644
index 00000000..0507b843
--- /dev/null
+++ b/signac/core/synced_collections/memory_buffered_collection.py
@@ -0,0 +1,384 @@
+# Copyright (c) 2020 The Regents of the University of Michigan
+# All rights reserved.
+# This software is licensed under the BSD 3-Clause License.
+"""A standardized buffering implementation for file-based backends.
+
+This module defines a variant of the buffering protocol in
+:py:mod:`file_buffered_collection`. This module exploits the same idea that
+all file-based backends can use a similar buffering protocol using integrity
+checks based on file modification times. However, that buffering approach
+mitigates I/O costs by serializing data into a single buffer that is then
+treated as the underlying data store by associated :class:`SyncedCollection` objects.
+That approach is expensive because it still requires constant encoding and
+decoding of data while checking for changes to the data in the buffer from other
+instances pointing to the same underlying files. The approach in this module circumvents
+this these performance bottlenecks by directly sharing the underlying data store
+in the buffer for all objects, completely removing the need for encoding, decoding,
+and updating in place.
+"""
+
+import errno
+import os
+from typing import Dict, Tuple, Union
+
+from .buffered_collection import BufferedCollection
+from .errors import MetadataError
+
+
+class SharedMemoryFileBufferedCollection(BufferedCollection):
+    """A :class:`SyncedCollection` that defers all I/O when buffered.
+
+    This class implements a variant of the buffering strategy defined in the
+    :class:`signac.core.synced_collections.file_buffered_collection.FileBufferedCollection`.
+    Like that class, this class enables buffering for file-based backends by storing
+    the last known modification time of the data on disk prior to entering buffered
+    mode, then checking whether that has changed when the buffer is flushed.
+    The buffering implemented by
+    :class:`~.FileBufferedCollection`
+    is true buffering in the sense that the entire write operation is
+    performed as normal, including serialization, except that data is written to
+    a different persistent store than the underlying resource. However, such
+    buffering incurs significant overhead associated all the non-I/O tasks
+    required prior to writing to the buffer, particularly data serialization.
+    This class exists to remove these performance bottlenecks, which can be
+    severe, and provide a more performant alternative.
+
+    The buffering method implemented in this class circumvents the
+    aforementioned performance bottlenecks by directly sharing the data between
+    multiple synced collections. Rather than encoding and decoding data, all
+    objects associated with a file share a single underlying in-memory data store,
+    allowing any changes to one to be transparently persisted to the others. When
+    the first collection associated with a particular file is accessed or
+    modified in buffered mode for the first time, its data is stored in a cache.
+    When future collections pointing to the same file access the cache, rather
+    than synchronizing their data with the contents of the cache, the underlying
+    data attribute of the collection is simply repointed at the data in the cache.
+    This method exploits the fact that all mutable collection types in Python are
+    references, so modifying one such collection results in modifying all of them,
+    thereby removing any need for more complicated synchronization protocols.
+
+    This approach has one principal downside relative to the other buffering
+    method. Since the data is not always encoded into a byte-string, the exact
+    size of the data in the buffer is not exactly known. This method simply counts
+    the number of objects in the buffer as a means to decide how much to store before
+    the buffer is flushed.
+
+    Parameters
+    ----------
+    filename : str, optional
+        The filename of the associated JSON file on disk (Default value = None).
+
+    .. note::
+        Important note for subclasses: This class should be inherited before
+        any other collections. This requirement is due to the extensive use of
+        multiple inheritance: since this class is designed to be combined with
+        other :class:`SyncedCollection` types without making those types aware
+        of buffering behavior, it transparently hooks into the initialization
+        process, but this is dependent on its constructor being called before
+        those of other classes.
+
+    """
+
+    _cache: Dict[str, Dict[str, Union[bytes, str, Tuple[int, float, int]]]] = {}
+    _cached_collections: Dict[int, BufferedCollection] = {}
+    # TODO: Do we really care about the total number of objects stored in the
+    # buffer, or do we only care about objects that have been modified (and
+    # therefore require a file write when flushing)? For files that have been
+    # read but not modified, I don't think that there's any reason not to just
+    # let them sit in memory, or at least to give the user an option to allow
+    # that in case they know that virtual memory exhaustion won't be an issue.
+    _BUFFER_CAPACITY = 1000  # The number of collections to store in the buffer.
+    _CURRENT_BUFFER_SIZE = 0
+
+    def __init__(self, filename=None, *args, **kwargs):
+        super().__init__(filename=filename, *args, **kwargs)
+        self._filename = filename
+
+    def _get_file_metadata(self):
+        """Return metadata of file.
+
+        Returns
+        -------
+        Tuple[int, float] or None
+            The size and last modification time of the associated file. If the
+            file does not exist, returns :code:`None`.
+
+        """
+        try:
+            metadata = os.stat(self._filename)
+            return metadata.st_size, metadata.st_mtime_ns
+        except OSError as error:
+            if error.errno != errno.ENOENT:
+                raise
+            # A return value of None indicates that the file does not
+            # exist. Since any non-``None`` value will return `False` when
+            # compared to ``None``, returning ``None`` provides a
+            # reasonable value to compare against for metadata-based
+            # validation checks.
+            return None
+
+    @staticmethod
+    def get_buffer_capacity():
+        """Get the current buffer capacity.
+
+        Returns
+        -------
+        int
+            The number of collections that can be stored before a flush is triggered.
+
+        """
+        return SharedMemoryFileBufferedCollection._BUFFER_CAPACITY
+
+    @staticmethod
+    def set_buffer_capacity(new_capacity):
+        """Update the buffer capacity.
+
+        Parameters
+        ----------
+        new_capacity : int
+            The number of collections that can be fit in the buffer.
+
+        """
+        SharedMemoryFileBufferedCollection._BUFFER_CAPACITY = new_capacity
+        if new_capacity < SharedMemoryFileBufferedCollection._CURRENT_BUFFER_SIZE:
+            SharedMemoryFileBufferedCollection._flush_buffer()
+
+    @staticmethod
+    def get_current_buffer_size():
+        """Get the total number of collections currently stored in the buffer.
+
+        Returns
+        -------
+        int
+            The number of collections contained in the buffer.
+
+        """
+        return SharedMemoryFileBufferedCollection._CURRENT_BUFFER_SIZE
+
+    def _flush(self, force=False):
+        """Save buffered changes to the underlying file.
+
+        Parameters
+        ----------
+        force : bool
+            If True, force a flush even in buffered mode (defaults to False). This
+            parameter is used when the buffer is filled to capacity.
+        stored_metadata : tuple[int, float]
+            The metadata stored for this particular collection in the cache. It
+            will have been removed by _flush_buffer when this method is called, so
+            it must be passed as an argument. Under normal circumstances it will be
+            equal to the metadata stored in _cache, but if there are multiple objects
+            pointing to the same file and one of them modifies it by exiting buffered
+            mode first and the other makes _no_ modifications, it is safe to simply
+            reload the data.
+
+        Raises
+        ------
+        MetadataError
+            If any file is detected to have changed on disk since it was
+            originally loaded into the buffer and modified.
+
+        """
+        if not self._is_buffered or force:
+            # TODO: If we have two objects pointing to the same filename in the
+            # cache and one of them flushes before the other, need to decide
+            # how to handle it.
+            try:
+                _, stored_metadata = self._cached_collections.pop(id(self))
+                cached_data = self._cache[self._filename]
+            except KeyError:
+                # There are valid reasons for nothing to be in the cache (the
+                # object was never actually accessed during global buffering,
+                # multiple collections pointing to the same file, etc).
+
+                # However, we do have to verify that the current metadata for
+                # the file is not newer than the originally stored metadata;
+                # otherwise we load the file from disk, because this indicates
+                # that something has modified the file since its data was
+                # originally stored in the buffer. A typical use case would be
+                # multiple collections pointing to the same file where only one
+                # of them has changed. The fact that this collection hasn't
+                # stored anything to the buffer is why this behavior is valid.
+                cur_metadata = self._get_file_metadata()
+                if stored_metadata[1] < cur_metadata[1]:
+                    self._data = self._load_from_resource()
+            else:
+                # If the contents have not been changed since the initial read,
+                # we don't need to rewrite it.
+                try:
+                    # Validate that the file hasn't been changed by
+                    # something else.
+                    if cached_data["modified"]:
+                        if cached_data["metadata"] != self._get_file_metadata():
+                            raise MetadataError(self._filename, cached_data["contents"])
+                        self._data = cached_data["contents"]
+                        self._save_to_resource()
+                finally:
+                    # Whether or not an error was raised, the cache must be
+                    # cleared to ensure a valid final buffer state.
+                    SharedMemoryFileBufferedCollection._CURRENT_BUFFER_SIZE -= 1
+                    del self._cache[self._filename]
+
+    def _load(self):
+        """Load data from the backend but buffer if needed.
+
+        Override the base buffered method to skip the _update and to let
+        _load_from_buffer happen "in place."
+        """
+        if self._suspend_sync_ <= 0:
+            if self._parent is None:
+                if self._is_buffered:
+                    self._load_from_buffer()
+                else:
+                    data = self._load_from_resource()
+                    with self._suspend_sync():
+                        self._update(data)
+            else:
+                self._parent._load()
+
+    def _save_to_buffer(self):
+        """Store data in buffer.
+
+        See :meth:`~._initialize_data_in_cache` for details on the data stored
+        in the buffer and the integrity checks performed.
+        """
+        if self._filename in self._cache:
+            # Need to check if we have multiple collections pointing to the
+            # same file, and if so, track it.
+            if id(self) not in SharedMemoryFileBufferedCollection._cached_collections:
+                SharedMemoryFileBufferedCollection._cached_collections[id(self)] = (
+                    self,
+                    self._cache[self._filename]["metadata"],
+                )
+            self._cache[self._filename]["modified"] = True
+        else:
+            self._initialize_data_in_cache(modified=True)
+
+        if (
+            SharedMemoryFileBufferedCollection._CURRENT_BUFFER_SIZE
+            > SharedMemoryFileBufferedCollection._BUFFER_CAPACITY
+        ):
+            SharedMemoryFileBufferedCollection._flush_buffer(force=True)
+
+    def _load_from_buffer(self):
+        """Read data from buffer.
+
+        See :meth:`~._initialize_data_in_cache` for details on the data stored
+        in the buffer and the integrity checks performed.
+
+        Returns
+        -------
+        Collection
+            A collection of the same base type as the SyncedCollection this
+            method is called for, corresponding to data loaded from the
+            underlying file.
+
+        """
+        if self._filename in self._cache:
+            # Need to check if we have multiple collections pointing to the
+            # same file, and if so, track it.
+            if id(self) not in SharedMemoryFileBufferedCollection._cached_collections:
+                SharedMemoryFileBufferedCollection._cached_collections[id(self)] = (
+                    self,
+                    self._cache[self._filename]["metadata"],
+                )
+        else:
+            # TODO: The first time we call _load_from_buffer we might need to call
+            # _load_from_resource. Otherwise, if something modified the file in memory
+            # since the last time that we performed any save/load operation, we could be
+            # putting an out-of-date state into the buffer. This also affects the
+            # FileBufferedCollection.
+            self._initialize_data_in_cache(modified=False)
+
+        # Set local data to the version in the buffer.
+        self._data = self._cache[self._filename]["contents"]
+
+        if (
+            SharedMemoryFileBufferedCollection._CURRENT_BUFFER_SIZE
+            > SharedMemoryFileBufferedCollection._BUFFER_CAPACITY
+        ):
+            SharedMemoryFileBufferedCollection._flush_buffer(force=True)
+
+    def _initialize_data_in_cache(self, modified):
+        """Create the initial entry for the data in the cache.
+
+        This method should be called the first time that a collection's data is
+        accessed in buffered mode. This method stores the encoded data in the
+        cache, along with a hash of the data and the metadata of the underlying
+        file. The hash is later used for quick checks of whether the data in
+        memory has changed since the initial load into the buffer. If so the
+        metadata is used to verify that the file on disk has not been modified
+        since the data was modified in memory.
+
+        We also maintain a separate set of all collections that are currently
+        in buffered mode. This extra storage is necessary because when leaving
+        a buffered context we need to make sure to only flush collections that
+        are no longer in buffered mode in cases of nested buffering. This
+        additional check helps prevent or transparently error on otherwise
+        unsafe access patterns.
+        """
+        metadata = self._get_file_metadata()
+        SharedMemoryFileBufferedCollection._cache[self._filename] = {
+            "contents": self._data,
+            "metadata": metadata,
+            "modified": modified,
+        }
+        SharedMemoryFileBufferedCollection._CURRENT_BUFFER_SIZE += 1
+        SharedMemoryFileBufferedCollection._cached_collections[id(self)] = (
+            self,
+            metadata,
+        )
+
+    @classmethod
+    def _flush_buffer(cls, force=False):
+        """Flush the data in the file buffer.
+
+        Parameters
+        ----------
+        force : bool
+            If True, force a flush even in buffered mode (defaults to False). This
+            parameter is used when the buffer is filled to capacity.
+
+        Returns
+        -------
+        dict
+            Mapping of filename and errors occured during flushing data.
+
+        Raises
+        ------
+        BufferedError
+            If there are any issues with flushing the data.
+
+        """
+        # All subclasses share a single cache rather than having separate
+        # caches for each instance, so we can exit early in subclasses.
+        if cls != SharedMemoryFileBufferedCollection:
+            return {}
+
+        issues = {}
+
+        # We need to use the list of buffered objects rather than directly
+        # looping over the local cache so that each collection can
+        # independently decide whether or not to flush based on whether it's
+        # still buffered (if buffered contexts are nested).
+        remaining_collections = {}
+        while SharedMemoryFileBufferedCollection._cached_collections:
+            col_id = next(iter(SharedMemoryFileBufferedCollection._cached_collections))
+            collection = SharedMemoryFileBufferedCollection._cached_collections[col_id][
+                0
+            ]
+
+            if collection._is_buffered and not force:
+                remaining_collections[
+                    col_id
+                ] = SharedMemoryFileBufferedCollection._cached_collections.pop(col_id)
+                continue
+            try:
+                collection._flush(force=force)
+            except (OSError, MetadataError) as err:
+                issues[collection._filename] = err
+        if not issues:
+            SharedMemoryFileBufferedCollection._cached_collections = (
+                remaining_collections
+            )
+        return issues
diff --git a/signac/core/synced_collections/synced_attr_dict.py b/signac/core/synced_collections/synced_attr_dict.py
index 8a76377e..d51a714b 100644
--- a/signac/core/synced_collections/synced_attr_dict.py
+++ b/signac/core/synced_collections/synced_attr_dict.py
@@ -65,7 +65,8 @@ class SyncedAttrDict(SyncedCollection, MutableMapping):
                     key: self._from_base(data=value, parent=self)
                     for key, value in data.items()
                 }
-            self._save()
+            with self._thread_lock():
+                self._save()
 
     def _to_base(self):
         """Convert the SyncedDict object to Dictionary.
@@ -174,11 +175,12 @@ class SyncedAttrDict(SyncedCollection, MutableMapping):
         # directly set using those rather than looping over data.
         data = {key: value}
         self._validate(data)
-        self._load()
-        with self._suspend_sync():
-            for key, value in data.items():
-                self._data[key] = self._from_base(value, parent=self)
-        self._save()
+        with self._thread_lock():
+            self._load()
+            with self._suspend_sync():
+                for key, value in data.items():
+                    self._data[key] = self._from_base(value, parent=self)
+            self._save()
 
     def reset(self, data=None):
         """Update the instance with new data.
@@ -203,7 +205,8 @@ class SyncedAttrDict(SyncedCollection, MutableMapping):
                     key: self._from_base(data=value, parent=self)
                     for key, value in data.items()
                 }
-            self._save()
+            with self._thread_lock():
+                self._save()
         else:
             raise ValueError(
                 "Unsupported type: {}. The data must be a mapping or None.".format(
@@ -228,20 +231,23 @@ class SyncedAttrDict(SyncedCollection, MutableMapping):
         return self._data.get(key, default)
 
     def pop(self, key, default=None):  # noqa: D102
-        self._load()
-        ret = self._data.pop(key, default)
-        self._save()
+        with self._thread_lock():
+            self._load()
+            ret = self._data.pop(key, default)
+            self._save()
         return ret
 
     def popitem(self):  # noqa: D102
-        self._load()
-        ret = self._data.popitem()
-        self._save()
+        with self._thread_lock():
+            self._load()
+            ret = self._data.popitem()
+            self._save()
         return ret
 
     def clear(self):  # noqa: D102
         self._data = {}
-        self._save()
+        with self._thread_lock():
+            self._save()
 
     def update(self, other=None, **kwargs):  # noqa: D102
         if other is not None:
@@ -251,30 +257,32 @@ class SyncedAttrDict(SyncedCollection, MutableMapping):
         else:
             other = {}
 
-        self._load()
-        # The order here is important to ensure that the promised sequence of
-        # overrides is obeyed: kwargs > other > existing data.
-        self._update({**self._data, **other, **kwargs})
-        self._save()
+        with self._thread_lock():
+            self._load()
+            # The order here is important to ensure that the promised sequence of
+            # overrides is obeyed: kwargs > other > existing data.
+            self._update({**self._data, **other, **kwargs})
+            self._save()
 
     def setdefault(self, key, default=None):  # noqa: D102
-        self._load()
-        if key in self._data:
-            ret = self._data[key]
-        else:
-            ret = self._from_base(default, parent=self)
-            # TODO: Remove in signac 2.0, currently we're constructing a dict
-            # to allow in-place modification by _convert_key_to_str, but
-            # validators should not have side effects once that backwards
-            # compatibility layer is removed, so we can validate a temporary
-            # dict {key: value} and directly set using those rather than
-            # looping over data.
-            data = {key: ret}
-            self._validate(data)
-            with self._suspend_sync():
-                for key, value in data.items():
-                    self._data[key] = value
-            self._save()
+        with self._thread_lock():
+            self._load()
+            if key in self._data:
+                ret = self._data[key]
+            else:
+                ret = self._from_base(default, parent=self)
+                # TODO: Remove in signac 2.0, currently we're constructing a dict
+                # to allow in-place modification by _convert_key_to_str, but
+                # validators should not have side effects once that backwards
+                # compatibility layer is removed, so we can validate a temporary
+                # dict {key: value} and directly set using those rather than
+                # looping over data.
+                data = {key: ret}
+                self._validate(data)
+                with self._suspend_sync():
+                    for key, value in data.items():
+                        self._data[key] = value
+                self._save()
         return ret
 
     def __getattr__(self, name):
diff --git a/signac/core/synced_collections/synced_collection.py b/signac/core/synced_collections/synced_collection.py
index 7fb47623..ef706dd3 100644
--- a/signac/core/synced_collections/synced_collection.py
+++ b/signac/core/synced_collections/synced_collection.py
@@ -7,6 +7,7 @@ from collections import defaultdict
 from collections.abc import Collection
 from contextlib import contextmanager
 from inspect import isabstract
+from threading import RLock
 from typing import Any, Callable, DefaultDict, List
 
 from .utils import AbstractTypeResolver
@@ -18,6 +19,7 @@ try:
 except ImportError:
     NUMPY = False
 
+
 # Identifies types of SyncedCollection, which are the base type for this class.
 _sc_resolver = AbstractTypeResolver(
     {
@@ -26,6 +28,27 @@ _sc_resolver = AbstractTypeResolver(
 )
 
 
+class _fake_lock:
+    """A nullary context manager to simulate a lock in backends that don't support them."""
+
+    def __enter__(self):
+        pass
+
+    def __exit__(self, type, value, traceback):
+        pass
+
+
+@contextmanager
+def _thread_lock(self):
+    """Prepare context for thread-safe operation.
+
+    All operations that can mutate an object should use this context
+    manager to ensure thread safety.
+    """
+    with type(self)._locks[self._lock_id]:
+        yield
+
+
 class SyncedCollection(Collection):
     """An abstract :class:`Collection` type that is synced with a backend.
 
@@ -87,6 +110,8 @@ class SyncedCollection(Collection):
 
     registry: DefaultDict[str, List[Any]] = defaultdict(list)
     _validators: List[Callable] = []
+    # Backends that support threading should modify this flag.
+    _supports_threading: bool = False
 
     def __init__(self, parent=None, *args, **kwargs):
         self._parent = parent
@@ -110,6 +135,41 @@ class SyncedCollection(Collection):
             SyncedCollection.registry[cls._backend].append(cls)
         cls._validators = []
 
+        # Monkey-patch subclasses that support locking.
+        if cls._supports_threading:
+            cls.enable_multithreading()
+
+    @classmethod
+    def enable_multithreading(cls):
+        """Allow multithreaded access to and modification of :class:`SyncedCollection`s.
+
+        Support for multithreaded execution can be disabled by calling
+        :meth:`~.disable_multithreading`; calling this method reverses that.
+
+        """
+        if cls._supports_threading:
+            cls._locks = defaultdict(RLock)
+            cls._thread_lock = _thread_lock
+        else:
+            raise ValueError("This class does not support multithreaded execution.")
+
+    @classmethod
+    def disable_multithreading(cls):
+        """Prevent multithreaded access to and modification of :class:`SyncedCollection`s.
+
+        The mutex locks required to enable multithreading introduce nontrivial performance
+        costs, so they can be disabled for classes that support it.
+
+        """
+        try:
+            del cls._locks
+            cls._thread_lock = _fake_lock
+        except AttributeError:
+            raise ValueError("This class does not support multithreaded execution.")
+
+    # By default, classes do not support locking.
+    _thread_lock = _fake_lock
+
     @property
     def validators(self):
         """List[Callable]: The validators that will be applied.
@@ -130,6 +190,15 @@ class SyncedCollection(Collection):
                 )
         return validators
 
+    @property
+    def _lock_id(self):
+        raise NotImplementedError(
+            "Backends must implement the _lock_id property to support multithreaded "
+            "execution. This property should return a hashable unique identifier for "
+            "all collections that will be used to maintain a resource-specific "
+            "set of locks."
+        )
+
     @classmethod
     def add_validator(cls, *args):
         r"""Register a validator to this class.
@@ -369,9 +438,10 @@ class SyncedCollection(Collection):
         return self._data[key]
 
     def __delitem__(self, item):
-        self._load()
-        del self._data[item]
-        self._save()
+        with self._thread_lock():
+            self._load()
+            del self._data[item]
+            self._save()
 
     def __iter__(self):
         self._load()
diff --git a/signac/core/synced_collections/synced_list.py b/signac/core/synced_collections/synced_list.py
index 932347c5..8488df79 100644
--- a/signac/core/synced_collections/synced_list.py
+++ b/signac/core/synced_collections/synced_list.py
@@ -59,7 +59,8 @@ class SyncedList(SyncedCollection, MutableSequence):
                 self._data = [
                     self._from_base(data=value, parent=self) for value in data
                 ]
-            self._save()
+            with self._thread_lock():
+                self._save()
 
     @classmethod
     def is_base_type(cls, data):
@@ -173,7 +174,8 @@ class SyncedList(SyncedCollection, MutableSequence):
                 self._data = [
                     self._from_base(data=value, parent=self) for value in data
                 ]
-            self._save()
+            with self._thread_lock():
+                self._save()
         else:
             raise ValueError(
                 "Unsupported type: {}. The data must be a non-string sequence or None.".format(
@@ -183,58 +185,69 @@ class SyncedList(SyncedCollection, MutableSequence):
 
     def __setitem__(self, key, value):
         self._validate(value)
-        self._load()
-        with self._suspend_sync():
-            self._data[key] = self._from_base(data=value, parent=self)
-        self._save()
+        with self._thread_lock():
+            self._load()
+            with self._suspend_sync():
+                self._data[key] = self._from_base(data=value, parent=self)
+            self._save()
 
     def __reversed__(self):
-        self._load()
+        with self._thread_lock():
+            self._load()
         return reversed(self._data)
 
     def __iadd__(self, iterable):
         # Convert input to a list so that iterators work as well as iterables.
         iterable_data = list(iterable)
         self._validate(iterable_data)
-        self._load()
-        with self._suspend_sync():
-            self._data += [
-                self._from_base(data=value, parent=self) for value in iterable_data
-            ]
-        self._save()
+        with self._thread_lock():
+            self._load()
+            with self._suspend_sync():
+                self._data += [
+                    self._from_base(data=value, parent=self) for value in iterable_data
+                ]
+            self._save()
         return self
 
     def insert(self, index, item):  # noqa: D102
         self._validate(item)
-        self._load()
-        with self._suspend_sync():
-            self._data.insert(index, self._from_base(data=item, parent=self))
-        self._save()
+        with self._thread_lock():
+            self._load()
+            with self._suspend_sync():
+                self._data.insert(index, self._from_base(data=item, parent=self))
+            self._save()
 
     def append(self, item):  # noqa: D102
         self._validate(item)
-        self._load()
-        with self._suspend_sync():
-            self._data.append(self._from_base(data=item, parent=self))
-        self._save()
+        with self._thread_lock():
+            self._load()
+            with self._suspend_sync():
+                self._data.append(self._from_base(data=item, parent=self))
+            self._save()
 
     def extend(self, iterable):  # noqa: D102
         # Convert iterable to a list to ensure generators are exhausted only once
         iterable_data = list(iterable)
         self._validate(iterable_data)
-        self._load()
-        with self._suspend_sync():
-            self._data.extend(
-                [self._from_base(data=value, parent=self) for value in iterable_data]
-            )
-        self._save()
+        with self._thread_lock():
+            self._load()
+            with self._suspend_sync():
+                self._data.extend(
+                    [
+                        self._from_base(data=value, parent=self)
+                        for value in iterable_data
+                    ]
+                )
+            self._save()
 
     def remove(self, value):  # noqa: D102
-        self._load()
-        with self._suspend_sync():
-            self._data.remove(self._from_base(data=value, parent=self))
-        self._save()
+        with self._thread_lock():
+            self._load()
+            with self._suspend_sync():
+                self._data.remove(self._from_base(data=value, parent=self))
+            self._save()
 
     def clear(self):  # noqa: D102
         self._data = []
-        self._save()
+        with self._thread_lock():
+            self._save()


Test patch:
diff --git a/tests/test_synced_collections/synced_collection_test.py b/tests/test_synced_collections/synced_collection_test.py
index c5553d3d..c054b1e2 100644
--- a/tests/test_synced_collections/synced_collection_test.py
+++ b/tests/test_synced_collections/synced_collection_test.py
@@ -1,6 +1,7 @@
 # Copyright (c) 2020 The Regents of the University of Michigan
 # All rights reserved.
 # This software is licensed under the BSD 3-Clause License.
+import platform
 from collections.abc import MutableMapping, MutableSequence
 from copy import deepcopy
 
@@ -17,6 +18,9 @@ except ImportError:
     NUMPY = False
 
 
+PYPY = "PyPy" in platform.python_implementation()
+
+
 class SyncedCollectionTest:
     def store(self, data):
         raise NotImplementedError("All backend tests must implement the store method.")
@@ -398,6 +402,47 @@ class SyncedDictTest(SyncedCollectionTest):
             with pytest.raises(TypeError):
                 synced_collection[key] = testdata
 
+    def test_multithreaded(self, synced_collection):
+        """Test multithreaded runs of synced dicts."""
+        if not type(synced_collection)._supports_threading:
+            return
+
+        from concurrent.futures import ThreadPoolExecutor
+        from threading import current_thread
+
+        def set_value(sd):
+            sd[current_thread().name] = current_thread().name
+
+        num_threads = 50
+        with ThreadPoolExecutor(max_workers=num_threads) as executor:
+            list(executor.map(set_value, [synced_collection] * num_threads * 10))
+
+        assert len(synced_collection) == num_threads
+
+        # The ways that PyPy fails here due to what appears to be optimization
+        # of the json library under the hood, so we skip testing the failure
+        # cases on PyPy.
+        if not PYPY:
+            # Now clear the data and try again with multithreading disabled. Unless
+            # we're very unlucky, some of these threads should overwrite each
+            # other.
+            type(synced_collection).disable_multithreading()
+            synced_collection.clear()
+
+            with ThreadPoolExecutor(max_workers=num_threads) as executor:
+                list(executor.map(set_value, [synced_collection] * num_threads * 10))
+
+            assert len(synced_collection) != num_threads
+
+            # For good measure, try reenabling multithreading and test to be safe.
+            type(synced_collection).enable_multithreading()
+            synced_collection.clear()
+
+            with ThreadPoolExecutor(max_workers=num_threads) as executor:
+                list(executor.map(set_value, [synced_collection] * num_threads * 10))
+
+            assert len(synced_collection) == num_threads
+
 
 class SyncedListTest(SyncedCollectionTest):
     @pytest.fixture(autouse=True)
@@ -627,3 +672,20 @@ class SyncedListTest(SyncedCollectionTest):
         child2 = synced_collection[0]["a"]
         assert isinstance(child2, SyncedCollection)
         assert isinstance(child1, SyncedCollection)
+
+    def test_multithreaded(self, synced_collection):
+        """Test multithreaded runs of synced lists."""
+        if not type(synced_collection)._supports_threading:
+            return
+
+        from concurrent.futures import ThreadPoolExecutor
+
+        def append_value(sl):
+            sl.append(0)
+
+        num_threads = 50
+        num_elements = num_threads * 10
+        with ThreadPoolExecutor(max_workers=num_threads) as executor:
+            list(executor.map(append_value, [synced_collection] * num_elements))
+
+        assert len(synced_collection) == num_elements
diff --git a/tests/test_synced_collections/test_json_buffered_collection.py b/tests/test_synced_collections/test_json_buffered_collection.py
index 2b1fbfc5..c9b61a4b 100644
--- a/tests/test_synced_collections/test_json_buffered_collection.py
+++ b/tests/test_synced_collections/test_json_buffered_collection.py
@@ -15,6 +15,9 @@ from signac.core.synced_collections.collection_json import (
     BufferedJSONCollection,
     BufferedJSONDict,
     BufferedJSONList,
+    MemoryBufferedJSONCollection,
+    MemoryBufferedJSONDict,
+    MemoryBufferedJSONList,
 )
 from signac.core.synced_collections.errors import BufferedError, MetadataError
 
@@ -39,7 +42,7 @@ class TestBufferedJSONDict(BufferedJSONCollectionTest, TestJSONDict):
         _tmp_dir2 = TemporaryDirectory(prefix="jsondict2_")
         _fn_2 = os.path.join(_tmp_dir2.name, "test2.json")
         _backend_kwargs2 = {"filename": _fn_2, "write_concern": False}
-        tmp = BufferedJSONDict(**_backend_kwargs2)
+        tmp = self._collection_type(**_backend_kwargs2)
         yield tmp
         _tmp_dir2.cleanup()
 
@@ -82,7 +85,7 @@ class TestBufferedJSONDict(BufferedJSONCollectionTest, TestJSONDict):
     def test_two_buffered(self, synced_collection, testdata):
         """Test that a non-buffered copy is not modified."""
         synced_collection["buffered"] = testdata
-        synced_collection2 = BufferedJSONDict(filename=synced_collection._filename)
+        synced_collection2 = self._collection_type(filename=synced_collection._filename)
 
         # Check that the non-buffered object is not modified.
         with synced_collection.buffered():
@@ -92,7 +95,7 @@ class TestBufferedJSONDict(BufferedJSONCollectionTest, TestJSONDict):
     def test_two_buffered_modify_unbuffered(self, synced_collection, testdata):
         """Test that in-memory changes raise errors in buffered mode."""
         synced_collection["buffered"] = testdata
-        synced_collection2 = BufferedJSONDict(filename=synced_collection._filename)
+        synced_collection2 = self._collection_type(filename=synced_collection._filename)
 
         # Check that the non-buffered object is not modified.
         with pytest.raises(MetadataError):
@@ -116,7 +119,7 @@ class TestBufferedJSONDict(BufferedJSONCollectionTest, TestJSONDict):
         # buffer when entering the context instead of waiting until the first
         # call to load, but for global buffering there's no equivalent.
         synced_collection["buffered"] = testdata
-        synced_collection2 = BufferedJSONDict(filename=synced_collection._filename)
+        synced_collection2 = self._collection_type(filename=synced_collection._filename)
 
         # Check that the non-buffered object is not modified.
         with synced_collection.buffered():
@@ -234,7 +237,7 @@ class TestBufferedJSONDict(BufferedJSONCollectionTest, TestJSONDict):
 
     def test_nested_copied_collection(self, synced_collection):
         """Test modifying two collections pointing to the same data."""
-        synced_collection2 = BufferedJSONDict(filename=synced_collection._filename)
+        synced_collection2 = self._collection_type(filename=synced_collection._filename)
 
         assert len(synced_collection) == 0
         assert len(synced_collection2) == 0
@@ -262,7 +265,7 @@ class TestBufferedJSONDict(BufferedJSONCollectionTest, TestJSONDict):
     @pytest.mark.skip("Not currently sure what the expected behavior is.")
     def test_nested_copied_collection_invalid(self, synced_collection):
         """Test the behavior of invalid modifications of copied objects."""
-        synced_collection2 = BufferedJSONDict(filename=synced_collection._filename)
+        synced_collection2 = self._collection_type(filename=synced_collection._filename)
 
         assert len(synced_collection) == 0
         assert len(synced_collection2) == 0
@@ -309,6 +312,35 @@ class TestBufferedJSONDict(BufferedJSONCollectionTest, TestJSONDict):
                 finished = True
         assert finished
 
+    def test_buffer_flush(self, synced_collection):
+        """Test that the buffer gets flushed when enough data is written."""
+        original_buffer_capacity = self._collection_type.get_buffer_capacity()
+
+        assert self._collection_type.get_current_buffer_size() == 0
+        self._collection_type.set_buffer_capacity(20)
+
+        # Ensure that the file exists on disk by executing a clear operation so
+        # that load operations work as expected.
+        assert len(synced_collection) == 0
+        synced_collection.clear()
+
+        with synced_collection.buffered():
+            synced_collection["foo"] = 1
+            assert self._collection_type.get_current_buffer_size() == len(
+                repr(synced_collection)
+            )
+            assert synced_collection != self.load(synced_collection)
+
+            # Add a long enough value to force a flush.
+            synced_collection["bar"] = 100
+            assert self._collection_type.get_current_buffer_size() == 0
+
+            # Make sure the file on disk now matches.
+            assert synced_collection == self.load(synced_collection)
+
+        # Reset buffer capacity for other tests.
+        self._collection_type.set_buffer_capacity(original_buffer_capacity)
+
 
 class TestBufferedJSONList(BufferedJSONCollectionTest, TestJSONList):
     """Tests of buffering JSONLists."""
@@ -383,6 +415,61 @@ class TestBufferedJSONList(BufferedJSONCollectionTest, TestJSONList):
         assert synced_collection == [1, 2, 3]
 
 
+class TestMemoryBufferedJSONDict(TestBufferedJSONDict):
+    """Tests of MemoryBufferedJSONDicts."""
+
+    _backend_collection = MemoryBufferedJSONCollection  # type: ignore
+    _collection_type = MemoryBufferedJSONDict  # type: ignore
+
+    def test_buffer_flush(self, synced_collection, synced_collection2):
+        """Test that the buffer gets flushed when enough data is written."""
+        original_buffer_capacity = self._collection_type.get_buffer_capacity()
+
+        assert self._collection_type.get_current_buffer_size() == 0
+        self._collection_type.set_buffer_capacity(1)
+
+        # Ensure that the file exists on disk by executing a clear operation so
+        # that load operations work as expected.
+        assert len(synced_collection) == 0
+        assert len(synced_collection2) == 0
+        synced_collection.clear()
+        synced_collection2.clear()
+
+        with buffer_all():
+            synced_collection["foo"] = 1
+            assert self._collection_type.get_current_buffer_size() == 1
+            assert synced_collection != self.load(synced_collection)
+
+            # This buffering mode is based on the number of files buffered, so
+            # we need to write to the second collection.
+            synced_collection2["bar"] = 2
+            assert self._collection_type.get_current_buffer_size() == 1
+
+            # The second file shouldn't match; the first file was flushed to
+            # make room for the second in the buffer.
+            assert synced_collection2 != self.load(synced_collection2)
+
+            # Make sure the first file on disk now matches. We must check this
+            # after the previous check, though, because the __eq__ check below
+            # results in synced_collection being loaded into the buffer,
+            # thereby causing a flush .
+            assert synced_collection == self.load(synced_collection)
+
+            # The second file will now match because the check above loaded
+            # synced_collection into the buffer and flushed synced_collection2.
+            assert synced_collection2 == self.load(synced_collection2)
+
+        # Reset buffer capacity for other tests.
+        self._collection_type.set_buffer_capacity(original_buffer_capacity)
+
+
+class TestMemoryBufferedJSONList(TestBufferedJSONList):
+    """Tests of MemoryBufferedJSONLists."""
+
+    _backend_collection = MemoryBufferedJSONCollection  # type: ignore
+    _collection_type = MemoryBufferedJSONList  # type: ignore
+
+
 class TestBufferedJSONDictWriteConcern(TestBufferedJSONDict):
     _write_concern = True