Merge pull request #84 from MayaWolf/master

jonhoo · web-flow · commit 89c747761316 · 2020-12-10T22:02:50.000-05:00
Move direct_write into left_right, plus a few convenience changes.
diff --git a/evmap/src/write.rs b/evmap/src/write.rs
@@ -54,11 +54,6 @@ where
 {
     handle: left_right::WriteHandle<Inner<K, V, M, S>, Operation<K, V, M>>,
     r_handle: ReadHandle<K, V, M, S>,
-
-    /// If Some, write directly to the write handle map, since no publish has happened.
-    /// Some(false) indicates that the necessary `Operation::JustCloneRHandle` has not
-    /// yet been appended to the oplog for when a publish does happen.
-    direct_write: Option<bool>,
 }
 
 impl<K, V, M, S> fmt::Debug for WriteHandle<K, V, M, S>
@@ -71,7 +66,6 @@ where
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("WriteHandle")
             .field("handle", &self.handle)
-            .field("direct_write", &self.direct_write)
             .finish()
     }
 }
@@ -87,11 +81,7 @@ where
         handle: left_right::WriteHandle<Inner<K, V, M, S>, Operation<K, V, M>>,
     ) -> Self {
         let r_handle = ReadHandle::new(left_right::ReadHandle::clone(&*handle));
-        Self {
-            handle,
-            r_handle,
-            direct_write: Some(false),
-        }
+        Self { handle, r_handle }
     }
 
     /// Publish all changes since the last call to `publish` to make them visible to readers.
@@ -100,7 +90,6 @@ where
     /// are many of them.
     pub fn publish(&mut self) -> &mut Self {
         self.handle.publish();
-        self.direct_write = None;
         self
     }
 
@@ -117,26 +106,7 @@ where
     }
 
     fn add_op(&mut self, op: Operation<K, V, M>) -> &mut Self {
-        if let Some(ref mut queued_clone) = self.direct_write {
-            {
-                // Safety: we know there are no outstanding w_handle readers, since we haven't
-                // refreshed ever before, so we can modify it directly!
-                let mut w_inner = self.handle.raw_write_handle();
-                let w_inner = unsafe { w_inner.as_mut() };
-                let r_handle = self.handle.enter().expect("map has not yet been destroyed");
-                // Because we are operating directly on the map, and nothing is aliased, we do want
-                // to perform drops, so we invoke absorb_second.
-                Absorb::absorb_second(w_inner, op, &*r_handle);
-            }
-
-            if !*queued_clone {
-                // NOTE: since we didn't record this in the oplog, r_handle *must* clone w_handle
-                self.handle.append(Operation::JustCloneRHandle);
-                *queued_clone = true;
-            }
-        } else {
-            self.handle.append(op);
-        }
+        self.handle.append(op);
         self
     }
 
@@ -426,10 +396,6 @@ where
             Operation::SetMeta(ref m) => {
                 self.meta = m.clone();
             }
-            Operation::JustCloneRHandle => {
-                // This is applying the operation to the original write handle,
-                // which we already applied the first batch of operations to.
-            }
         }
     }
 
@@ -541,33 +507,6 @@ where
             Operation::SetMeta(m) => {
                 inner.meta = m;
             }
-            Operation::JustCloneRHandle => {
-                // This is applying the operation to the original read handle,
-                // which is empty, and needs to copy over all data from the
-                // write handle that we wrote to directly.
-
-                // XXX: it really is too bad that we can't just .clone() the data here and save
-                // ourselves a lot of re-hashing, re-bucketization, etc.
-                inner.data.extend(other.data.iter().map(|(k, vs)| {
-                    // # Safety (for aliasing):
-                    //
-                    // We are aliasing every value in the read map, and the oplog has no other
-                    // pending operations (by the semantics of JustCloneRHandle). For any of the
-                    // values we alias to be dropped, the operation that drops it must first be
-                    // enqueued to the oplog, at which point it will _first_ go through
-                    // absorb_first, which will remove the alias and leave only one alias left.
-                    // Only after that, when that operation eventually goes through absorb_second,
-                    // will the alias be dropped, and by that time it is the only value.
-                    //
-                    // # Safety (for NoDrop -> DoDrop):
-                    //
-                    // The oplog has only this one operation in it for the first call to `publish`,
-                    // so we are about to turn the alias back into NoDrop.
-                    (k.clone(), unsafe {
-                        ValuesInner::alias(vs, other.data.hasher())
-                    })
-                }));
-            }
         }
     }
 
@@ -588,6 +527,37 @@ where
             unsafe { Box::from_raw(Box::into_raw(self) as *mut _ as *mut _) };
         drop(inner);
     }
+
+    fn sync_with(&mut self, first: &Self) {
+        let inner: &mut Inner<K, V, M, S, crate::aliasing::DoDrop> =
+            unsafe { &mut *(self as *mut _ as *mut _) };
+        inner.data.extend(first.data.iter().map(|(k, vs)| {
+            // # Safety (for aliasing):
+            //
+            // We are aliasing every value in the read map, and the oplog has no other
+            // pending operations (by the semantics of JustCloneRHandle). For any of the
+            // values we alias to be dropped, the operation that drops it must first be
+            // enqueued to the oplog, at which point it will _first_ go through
+            // absorb_first, which will remove the alias and leave only one alias left.
+            // Only after that, when that operation eventually goes through absorb_second,
+            // will the alias be dropped, and by that time it is the only value.
+            //
+            // # Safety (for hashing):
+            //
+            // Due to `RandomState` there can be subtle differences between the iteration order
+            // of two `HashMap` instances. We prevent this by using `left_right::new_with_empty`,
+            // which `clone`s the first map, making them use the same hasher.
+            //
+            // # Safety (for NoDrop -> DoDrop):
+            //
+            // The oplog has only this one operation in it for the first call to `publish`,
+            // so we are about to turn the alias back into NoDrop.
+            (k.clone(), unsafe {
+                ValuesInner::alias(vs, first.data.hasher())
+            })
+        }));
+        self.ready = true;
+    }
 }
 
 impl<K, V, M, S> Extend<(K, V)> for WriteHandle<K, V, M, S>
@@ -658,8 +628,6 @@ pub(super) enum Operation<K, V, M> {
     MarkReady,
     /// Set the value of the map meta.
     SetMeta(M),
-    /// Copy over the contents of the read map wholesale as the write map is empty.
-    JustCloneRHandle,
 }
 
 impl<K, V, M> fmt::Debug for Operation<K, V, M>
@@ -685,7 +653,6 @@ where
             Operation::Reserve(ref a, ref b) => f.debug_tuple("Reserve").field(a).field(b).finish(),
             Operation::MarkReady => f.debug_tuple("MarkReady").finish(),
             Operation::SetMeta(ref a) => f.debug_tuple("SetMeta").field(a).finish(),
-            Operation::JustCloneRHandle => f.debug_tuple("JustCloneRHandle").finish(),
         }
     }
 }
diff --git a/left-right/src/lib.rs b/left-right/src/lib.rs
@@ -93,6 +93,10 @@
 //!
 //!     // See the documentation of `Absorb::drop_first`.
 //!     fn drop_first(self: Box<Self>) {}
+//!
+//!     fn sync_with(&mut self, first: &Self) {
+//!         *self = *first
+//!     }
 //! }
 //!
 //! // Now, you can construct a new left-right over an instance of your data structure.
@@ -241,6 +245,19 @@ pub trait Absorb<O> {
     /// Defaults to calling `Self::drop`.
     #[allow(clippy::boxed_local)]
     fn drop_second(self: Box<Self>) {}
+
+    /// Sync the data from `first` into `self`.
+    ///
+    /// To improve initialization performance, before the first call to `publish` changes aren't
+    /// added to the internal oplog, but applied to the first copy directly using `absorb_second`.
+    /// The first `publish` then calls `sync_with` instead of `absorb_second`.
+    ///
+    /// `sync_with` should ensure that `self`'s state exactly matches that of `first` after it
+    /// returns. Be particularly mindful of non-deterministic implementations of traits that are
+    /// often assumed to be deterministic (like `Eq` and `Hash`), and of "hidden states" that
+    /// subtly affect results like the `RandomState` of a `HashMap` which can change iteration
+    /// order.
+    fn sync_with(&mut self, first: &Self);
 }
 
 /// Construct a new write and read handle pair from an empty data structure.
@@ -259,12 +276,23 @@ where
 
 /// Construct a new write and read handle pair from the data structure default.
 ///
-/// The type must implement `Clone` so we can construct the second copy from the first.
+/// The type must implement `Default` so we can construct two empty instances. You must ensure that
+/// the trait's `Default` implementation is deterministic and idempotent - that is to say, two
+/// instances created by it must behave _exactly_ the same. An example of where this is problematic
+/// is `HashMap` - due to `RandomState`, two instances returned by `Default` may have a different
+/// iteration order.
+///
+/// If your type's `Default` implementation does not guarantee this, you can use `new_from_empty`,
+/// which relies on `Clone` instead of `Default`.
 pub fn new<T, O>() -> (WriteHandle<T, O>, ReadHandle<T>)
 where
-    T: Absorb<O> + Default + Clone,
+    T: Absorb<O> + Default,
 {
-    new_from_empty(T::default())
+    let epochs = Default::default();
+
+    let r = ReadHandle::new(T::default(), Arc::clone(&epochs));
+    let w = WriteHandle::new(T::default(), epochs, r.clone());
+    (w, r)
 }
 
 #[cfg(test)]
@@ -281,4 +309,8 @@ impl Absorb<CounterAddOp> for i32 {
     }
 
     fn drop_first(self: Box<Self>) {}
+
+    fn sync_with(&mut self, first: &Self) {
+        *self = *first
+    }
 }
diff --git a/left-right/src/read.rs b/left-right/src/read.rs
@@ -198,7 +198,9 @@ impl<T> ReadHandle<T> {
     /// Note that it is only safe to read through this pointer if you _know_ that the writer will
     /// not start writing into it. This is most likely only the case if you are calling this method
     /// from inside a method that holds `&mut WriteHandle`.
-    pub fn raw_handle(&mut self) -> Option<NonNull<T>> {
+    ///
+    /// Casting this pointer to `&mut` is never safe.
+    pub fn raw_handle(&self) -> Option<NonNull<T>> {
         NonNull::new(self.inner.load(atomic::Ordering::Acquire))
     }
 }
diff --git a/left-right/src/write.rs b/left-right/src/write.rs
@@ -31,6 +31,10 @@ where
     last_epochs: Vec<usize>,
     #[cfg(test)]
     refreshes: usize,
+    /// Write directly to the write handle map, since no publish has happened.
+    first: bool,
+    /// A publish has happened, but the two copies have not been synchronized yet.
+    second: bool,
 }
 
 // safety: if a `WriteHandle` is sent across a thread boundary, we need to be able to take
@@ -57,6 +61,8 @@ where
             .field("oplog", &self.oplog)
             .field("swap_index", &self.swap_index)
             .field("r_handle", &self.r_handle)
+            .field("first", &self.first)
+            .field("second", &self.second)
             .finish()
     }
 }
@@ -123,6 +129,8 @@ where
             last_epochs: Vec::new(),
             #[cfg(test)]
             refreshes: 0,
+            first: true,
+            second: true,
         }
     }
 
@@ -194,7 +202,7 @@ where
 
         self.wait(&mut epochs);
 
-        {
+        if !self.first {
             // all the readers have left!
             // safety: we haven't freed the Box, and no readers are accessing the w_handle
             let w_handle = unsafe { self.w_handle.as_mut() };
@@ -208,6 +216,11 @@ where
                     .unwrap()
             };
 
+            if self.second {
+                Absorb::sync_with(w_handle, r_handle);
+                self.second = false
+            }
+
             // the w_handle copy has not seen any of the writes in the oplog
             // the r_handle copy has not seen any of the writes following swap_index
             if self.swap_index != 0 {
@@ -226,7 +239,9 @@ where
             // the w_handle copy is about to become the r_handle, and can ignore the oplog
             self.swap_index = self.oplog.len();
 
-            // w_handle (the old r_handle) is now fully up to date!
+        // w_handle (the old r_handle) is now fully up to date!
+        } else {
+            self.first = false
         }
 
         // at this point, we have exclusive access to w_handle, and it is up-to-date with all
@@ -262,6 +277,16 @@ where
         self
     }
 
+    /// Publish as necessary to ensure that all operations are visible to readers.
+    ///
+    /// `WriteHandle::publish` will *always* wait for old readers to depart and swap the maps.
+    /// This method will only do so if there are pending operations.
+    pub fn flush(&mut self) {
+        if self.has_pending_operations() {
+            self.publish();
+        }
+    }
+
     /// Returns true if there are operations in the operational log that have not yet been exposed
     /// to readers.
     pub fn has_pending_operations(&self) -> bool {
@@ -274,7 +299,18 @@ where
     ///
     /// Its effects will not be exposed to readers until you call [`publish`](Self::publish).
     pub fn append(&mut self, op: O) -> &mut Self {
-        self.oplog.push_back(op);
+        if self.first {
+            // Safety: we know there are no outstanding w_handle readers, since we haven't
+            // refreshed ever before, so we can modify it directly!
+            let mut w_inner = self.raw_write_handle();
+            let w_inner = unsafe { w_inner.as_mut() };
+            let r_handle = self.enter().expect("map has not yet been destroyed");
+            // Because we are operating directly on the map, and nothing is aliased, we do want
+            // to perform drops, so we invoke absorb_second.
+            Absorb::absorb_second(w_inner, op, &*r_handle);
+        } else {
+            self.oplog.push_back(op);
+        }
         self
     }
 
@@ -312,6 +348,7 @@ where
 /// struct Data;
 /// impl left_right::Absorb<()> for Data {
 ///     fn absorb_first(&mut self, _: &mut (), _: &Self) {}
+///     fn sync_with(&mut self, _: &Self) {}
 /// }
 ///
 /// fn is_send<T: Send>() {

Original file line number	Diff line number	Diff line change
`@@ -198,7 +198,9 @@ impl<T> ReadHandle<T> {`
`198`	`198`	`/// Note that it is only safe to read through this pointer if you _know_ that the writer will`
`199`	`199`	`/// not start writing into it. This is most likely only the case if you are calling this method`
`200`	`200`	/// from inside a method that holds `&mut WriteHandle`.
`201`		`- pub fn raw_handle(&mut self) -> Option<NonNull<T>> {`
	`201`	`+ ///`
	`202`	+ /// Casting this pointer to `&mut` is never safe.
	`203`	`+ pub fn raw_handle(&self) -> Option<NonNull<T>> {`
`202`	`204`	`NonNull::new(self.inner.load(atomic::Ordering::Acquire))`
`203`	`205`	`}`
`204`	`206`	`}`