Implement two-stage remove

opencog/persist/api/BackingStore.h

@@ -116,12 +116,49 @@ class BackingStore
 		 * If the recursive flag is set, then incoming set of the Atom
 		 * will also be removed.  If the recursive flag is not set, and
 		 * the Atom has a non-empty incoming set, then this method will
-		 * (usually) not be called.  However, if another thread is adding
-		 * the same Atom at the same time, all bets are off. Thus, the
-		 * behavior of calling this with recursive==false and a non-empty
-		 * incoming set is undefined (and implementation-dependent).
+		 * (usually) not be called.  However, if another thread is
+		 * altering the incoming set, race conditions can result. Thus,
+		 * the behavior of calling this with recursive==false and a
+		 * non-empty incoming set is undefined. Implementations must not
+		 * crash or assert, but are otherwise free to do whatever.
+		 * Ignoring the call if the incoming-set is non-empty is generally
+		 * a safe response.
 		 */
-		virtual void removeAtom(AtomSpace*, const Handle&, bool recursive) = 0;
+		virtual void removeAtom(AtomSpace*, const Handle&, bool recursive)
+		{
+			throw IOException(TRACE_INFO, "Not implemented!");
+		}
+
+		/**
+		 * Same as above, except providing a two-step removal API.
+		 *
+		 * When the user asks to remove an Atom, the preRemoveAtom()
+		 * method is called first. When it is called, the Atom is still in
+		 * the AtomSpace. Next, the Atom is removed from the AtomSpace,
+		 * and finally, the postRemoveAtom() method is called. This
+		 * two-step process gives the backend the freedom to implement
+		 * algorithms that would be difficult with a one-step process.
+		 *
+		 * A default implementation is provided for backwards compat.
+		 */
+		virtual void preRemoveAtom(AtomSpace* as, const Handle& h,
+		                           bool recursive)
+		{
+			removeAtom(as, h, recursive);
+		}
+
+		/**
+		 * The postRemoveAtom() method is called after the Atom has been
+		 * extracted from the AtomSpace. That extraction might fail, for
+		 * a large number of reasons, having to do with read-only, framing
+		 * and shadowing. Thus, the result of the extraction is passed in
+		 * as an argument: if `extracted` is false, then the Atom was never
+		 * removed from the AtomSpace (and shouldn't be removed from the
+		 * backend, either.)
+		 */
+		virtual void postRemoveAtom(AtomSpace* as, const Handle& h,
+		                            bool recursive, bool extracted)
+		{}
 
 		/**
 		 * Store the value located at `key` on `atom` to the remote

opencog/persist/api/StorageNode.cc

@@ -100,15 +100,17 @@ void StorageNode::update_value(const Handle& h, const Handle& key,
 
 bool StorageNode::remove_atom(AtomSpace* as, Handle h, bool recursive)
 {
-	// Removal is ... tricky. We need to remove Atoms from storage first,
-	// and only then the AtomSpace. We need to do storage first because
-	// storage needs access to the incoming set, and other Atom things,
-	// which get destroyed by the AtomSpace removal. However, we should
-	// remove from storage only if the AtomSpace remove succeeds! The
-	// AtomSpace remove logic is complex, and we cannot replicate it
-	// here. Thus we have a chicken-and-egg situation. The solution used
-	// here is minimalist: do only as much as needed to stay compatible
-	// with the docs.
+	// Removal is done with a two-step process. First, we tell storage
+	// about the Atom that is going away. It's still in the AtomSpace at
+	// this point, so storage can grab whatever data it needs from the
+	// AtomSpace (e.g. grab the IncomingSet). Next, we remove from the
+	// AtomSpace, and then finally, we tell storage that we're done.
+	//
+	// The postRemove call is called only if the AtomSpace remove
+	// succeeds! One reason for this is that the AtomSpace remove logic
+	// is complex, and its too hard to ask each storage to try to
+	// replicate it. The solution used here is minimalist: do only as
+	// much as needed to stay compatible with the docs.
 	//
 	// Unfortunately, this means the code won't be thread-safe. If one
 	// thread is adding atoms while another is deleting them, we will
@@ -124,12 +126,15 @@ bool StorageNode::remove_atom(AtomSpace* as, Handle h, bool recursive)
 	// it is acting as a cache for the database, and removal is used
 	// used to free up RAM storage.
 	if (not _atom_space->get_read_only())
-		removeAtom(as, h, recursive);
+		preRemoveAtom(as, h, recursive);
+
+	bool exok = as->extract_atom(h, recursive);
+
+	// Tell the backend that we're done. Pass the status code.
+	if (not _atom_space->get_read_only())
+		postRemoveAtom(as, h, recursive, exok);
 
-	// XXX FIXME This is breaks the WriteThruProxy when there are two
-	// or more targets: after the first target runs, the Atom will be
-	// gone and the second target will fail badly. For now, punt.
-	return as->extract_atom(h, recursive);
+	return exok;
 }
 
 Handle StorageNode::fetch_atom(const Handle& h, AtomSpace* as)

opencog/persist/api/StorageNode.h

@@ -37,6 +37,10 @@ namespace opencog
  */
 class StorageNode : public Node, protected BackingStore
 {
+	// The write-thru proxies need to call into the BackingStore
+	// directly. We declare them as friends.
+	friend class WriteThruProxy;
+	friend class ReadWriteProxy;
 protected:
 	Handle add_nocheck(AtomSpace* as, const Handle& h) const
 		{ return as->add(h); }

opencog/persist/proxy/ReadWriteProxy.cc

@@ -142,13 +142,18 @@ void ReadWriteProxy::storeAtom(const Handle& h, bool synchronous)
 	_writer->barrier();
 }
 
-void ReadWriteProxy::removeAtom(AtomSpace* as, const Handle& h, bool recursive)
+void ReadWriteProxy::preRemoveAtom(AtomSpace* as, const Handle& h,
+                                   bool recursive)
 {
 	CHECK_OPEN
-	// XXX FIXME this is deeply fundamentally broken if there's more
-	// than one target; because StorageNode::remove_atom() is broken.
-	// See the comments in that code for additional guidance.
-	_writer->remove_atom(as, h, recursive);
+	_writer->preRemoveAtom(as, h, recursive);
+}
+
+void ReadWriteProxy::postRemoveAtom(AtomSpace* as, const Handle& h,
+                                    bool recursive, bool extract_ok)
+{
+	CHECK_OPEN
+	_writer->postRemoveAtom(as, h, recursive, extract_ok);
 }
 
 void ReadWriteProxy::storeValue(const Handle& atom, const Handle& key)

opencog/persist/proxy/ReadWriteProxy.h

@@ -57,7 +57,9 @@ class ReadWriteProxy : public ProxyNode
 	virtual void fetchIncomingByType(AtomSpace*, const Handle&, Type);
 
 	virtual void storeAtom(const Handle&, bool synchronous = false);
-	virtual void removeAtom(AtomSpace*, const Handle&, bool recursive);
+	virtual void preRemoveAtom(AtomSpace*, const Handle&, bool recursive);
+	virtual void postRemoveAtom(AtomSpace*, const Handle&,
+	                            bool recursive, bool extracted_ok);
 	virtual void storeValue(const Handle& atom, const Handle& key);
 	virtual void updateValue(const Handle& atom, const Handle& key,
 	                         const ValuePtr& delta);

opencog/persist/proxy/WriteThruProxy.cc

@@ -78,13 +78,19 @@ void WriteThruProxy::storeAtom(const Handle& h, bool synchronous)
 		stnp->barrier();
 }
 
-void WriteThruProxy::removeAtom(AtomSpace* as, const Handle& h, bool recursive)
+// Two-step remove. Just pass the two steps down to the children.
+void WriteThruProxy::preRemoveAtom(AtomSpace* as, const Handle& h,
+                                   bool recursive)
 {
-	// XXX FIXME this is deeply fundamentally broken if there's more
-	// than one target; because StorageNode::remove_atom() is broken.
-	// See the comments in that code for additional guidance.
 	for (const StorageNodePtr& stnp : _targets)
-		stnp->remove_atom(as, h, recursive);
+		stnp->preRemoveAtom(as, h, recursive);
+}
+
+void WriteThruProxy::postRemoveAtom(AtomSpace* as, const Handle& h,
+                                    bool recursive, bool extracted_ok)
+{
+	for (const StorageNodePtr& stnp : _targets)
+		stnp->postRemoveAtom(as, h, recursive, extracted_ok);
 }
 
 void WriteThruProxy::storeValue(const Handle& atom, const Handle& key)

opencog/persist/proxy/WriteThruProxy.h

@@ -54,7 +54,9 @@ class WriteThruProxy : public ProxyNode
 	virtual void fetchIncomingSet(AtomSpace*, const Handle&) {}
 	virtual void fetchIncomingByType(AtomSpace*, const Handle&, Type) {}
 	virtual void storeAtom(const Handle&, bool synchronous = false);
-	virtual void removeAtom(AtomSpace*, const Handle&, bool recursive);
+	virtual void preRemoveAtom(AtomSpace*, const Handle&, bool recursive);
+	virtual void postRemoveAtom(AtomSpace*, const Handle&,
+	                            bool recursive, bool exok);
 	virtual void storeValue(const Handle& atom, const Handle& key);
 	virtual void updateValue(const Handle& atom, const Handle& key,
 	                         const ValuePtr& delta);