Synced collapsed forwarding in rock store with working shared mem cache code.

diff --git a/src/fs/rock/RockIoState.cc b/src/fs/rock/RockIoState.cc
index 9704ce97238f7e478ac6c2ce3839df66432204c0..5646d075fba98ff730d13a4b2e6e28f3462f34c5 100644 (file)
--- a/src/fs/rock/RockIoState.cc
+++ b/src/fs/rock/RockIoState.cc
@@ -3,6 +3,7 @@
  */
 
 #include "squid.h"
+#include "CollapsedForwarding.h"
 #include "base/TextException.h"
 #include "DiskIO/DiskIOModule.h"
 #include "DiskIO/DiskIOStrategy.h"
@@ -283,6 +284,7 @@ Rock::IoState::finishedWriting(const int errFlag)
 {
     // we incremented offset_ while accumulating data in write()
     writeableAnchor_ = NULL;
+    CollapsedForwarding::Broadcast(*e);
     callBack(errFlag);
 }
 

diff --git a/src/fs/rock/RockSwapDir.cc b/src/fs/rock/RockSwapDir.cc
index 9753e2ac5401cc43098c49242151248f58d579bc..7c0c1effdbf79bc83a13c58b6b92bfe0b741f3fe 100644 (file)
--- a/src/fs/rock/RockSwapDir.cc
+++ b/src/fs/rock/RockSwapDir.cc
@@ -94,7 +94,7 @@ Rock::SwapDir::anchorCollapsed(StoreEntry &collapsed, bool &inSync)
 
     anchorEntry(collapsed, filen, *slot);
     inSync = updateCollapsedWith(collapsed, *slot);
-    return false;
+    return true; // even if inSync is false
 }
 
 bool
@@ -124,8 +124,6 @@ Rock::SwapDir::anchorEntry(StoreEntry &e, const sfileno filen, const Ipc::StoreM
     const Ipc::StoreMapAnchor::Basics &basics = anchor.basics;
 
     e.swap_file_sz = basics.swap_file_sz;
-    e.swap_dirn = index;
-    e.swap_filen = filen;
     e.lastref = basics.lastref;
     e.timestamp = basics.timestamp;
     e.expires = basics.expires;
@@ -133,14 +131,22 @@ Rock::SwapDir::anchorEntry(StoreEntry &e, const sfileno filen, const Ipc::StoreM
     e.refcount = basics.refcount;
     e.flags = basics.flags;
 
-    e.store_status = STORE_OK;
-    e.setMemStatus(NOT_IN_MEMORY);
-    e.swap_status = SWAPOUT_DONE;
+    if (anchor.complete()) {
+        e.store_status = STORE_OK;
+        e.swap_status = SWAPOUT_DONE;
+    } else {
+        e.store_status = STORE_PENDING;
+        e.swap_status = SWAPOUT_WRITING; // even though another worker writes?
+    }
+
     e.ping_status = PING_NONE;
 
     EBIT_CLR(e.flags, RELEASE_REQUEST);
     EBIT_CLR(e.flags, KEY_PRIVATE);
     EBIT_SET(e.flags, ENTRY_VALIDATED);
+
+    e.swap_dirn = index;
+    e.swap_filen = filen;
 }
 
 
@@ -157,13 +163,18 @@ void Rock::SwapDir::disconnect(StoreEntry &e)
     // since e has swap_filen, its slot is locked for reading and/or writing
     // but it is difficult to know whether THIS worker is reading or writing e
     if (e.swap_status == SWAPOUT_WRITING ||
-        (e.mem_obj && e.mem_obj->swapout.sio != NULL))
+        (e.mem_obj && e.mem_obj->swapout.sio != NULL)) {
         map->abortWriting(e.swap_filen);
-    else
+        e.swap_dirn = -1;
+        e.swap_filen = -1;
+        e.swap_status = SWAPOUT_NONE;
+        Store::Root().transientsAbandon(e); // broadcasts after the change
+    } else {
         map->closeForReading(e.swap_filen);
-    e.swap_dirn = -1;
-    e.swap_filen = -1;
-    e.swap_status = SWAPOUT_NONE;
+        e.swap_dirn = -1;
+        e.swap_filen = -1;
+        e.swap_status = SWAPOUT_NONE;
+    }
 }
 
 uint64_t