Add stores for read-through and write-through caching

storehaus-core/src/main/scala/com/twitter/storehaus/ReadThroughStore.scala

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2014 Twitter Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.twitter.storehaus
+
+import com.twitter.concurrent.AsyncMutex
+import com.twitter.util.Future
+
+/**
+ * Provides read-through caching on a readable store fronted by a cache.
+ *
+ * Keys are fetched from backing store on cache miss and cache read failures.
+ *
+ * All cache operations are best effort i.e. 'get' will return the key from
+ * backing store even if adding/updating the cached copy fails.
+ *
+ * On the other hand, any failure while reading from backing store
+ * is propagated to the client.
+ *
+ * Thread-safety is achieved using a mutex.
+ *
+ * @author Ruban Monu
+ */
+class ReadThroughStore[K, V](backingStore: ReadableStore[K, V], cache: Store[K, V])
+  extends ReadableStore[K, V] {
+
+  protected [this] lazy val mutex = new AsyncMutex
+
+  private [this] def getFromBackingStore(k: K) : Future[Option[V]] = {
+    // attempt to fetch the key from backing store and
+    // write the key to cache, best effort
+    backingStore.get(k).flatMap { storeValue =>
+      mutex.acquire.flatMap { p =>
+        cache.put((k, storeValue))
+          .map { u : Unit => storeValue }
+          .onFailure { case x: Exception => storeValue }
+          .ensure { p.release }
+      }
+    }
+  }
+
+  override def get(k: K): Future[Option[V]] =
+    cache.get(k).flatMap { cacheValue =>
+      cacheValue match {
+        case None => getFromBackingStore(k)
+        case some => Future.value(some)
+      }
+    } onFailure { case x: Exception =>
+      getFromBackingStore(k)
+    }
+
+  override def multiGet[K1 <: K](ks: Set[K1]): Map[K1, Future[Option[V]]] = {
+    // attempt to read from cache first
+    val cacheResults : Map[K1, Future[Either[Option[V], Exception]]] =
+      cache.multiGet(ks).map { case (k, f) =>
+        (k, f.map { optv => Left(optv) } onFailure { case x: Exception => Right(x) })
+      }
+
+    // attempt to read all failed keys and cache misses from backing store
+    val f: Future[Map[K1, Option[V]]] =
+      FutureOps.mapCollect(cacheResults).flatMap { cacheResult =>
+        val failedKeys = cacheResult.filter { _._2.isRight }.keySet
+        val responses = cacheResult.filter { _._2.isLeft }.map { case (k, r) => (k, r.left.get) }
+        val hits = responses.filter { !_._2.isEmpty }
+        val missedKeys = responses.filter { _._2.isEmpty }.keySet
+
+        FutureOps.mapCollect(backingStore.multiGet(missedKeys ++ failedKeys)).flatMap { storeResult =>
+          // write fetched keys to cache, best effort
+          mutex.acquire.flatMap { p =>
+            FutureOps.mapCollect(cache.multiPut(storeResult))(FutureCollector.bestEffort[(K1, Unit)])
+              .map { u => hits ++ storeResult }
+              .ensure { p.release }
+          }
+        }
+      }
+    FutureOps.liftValues(ks, f, { (k: K1) => Future.None })
+  }
+}
+

storehaus-core/src/main/scala/com/twitter/storehaus/WriteThroughStore.scala

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2014 Twitter Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.twitter.storehaus
+
+import com.twitter.util.Future
+
+/**
+ * Provides write-through caching on a store fronted by a cache.
+ *
+ * All cache operations are best effort. i.e. a write will succeed if the key
+ * was written to backing store even when adding/updating the cached copy of the key fails.
+ *
+ * On the other hand, any failure while writing to backing store
+ * is propagated to the client.
+ *
+ * If invalidate flag is set, any keys that fail to write to backing store
+ * are attempted to be removed from cache, rather than having the old value
+ * still in cache.
+ *
+ * Thread-safety is achieved using a mutex.
+ *
+ * @author Ruban Monu
+ */
+class WriteThroughStore[K, V](backingStore: Store[K, V], cache: Store[K, V], invalidate: Boolean = true)
+  extends ReadThroughStore[K, V](backingStore, cache) with Store[K, V] {
+
+  override def put(kv: (K, Option[V])): Future[Unit] = mutex.acquire.flatMap { p =>
+    // write key to backing store first
+    backingStore.put(kv).flatMap { u: Unit =>
+      // now write key to cache, best effort
+      cache.put(kv) onFailure { case x: Exception => u }
+    } onFailure { case x: Exception =>
+      // write to backing store failed
+      // now optionally invalidate the key in cache, best effort
+      if (invalidate) {
+        cache.put((kv._1, None)).flatMap { u: Unit => throw x } onFailure { throw x }
+      } else {
+        throw x
+      }
+    } ensure {
+      p.release
+    }
+  }
+
+  override def multiPut[K1 <: K](kvs: Map[K1, Option[V]]): Map[K1, Future[Unit]] = {
+    val f : Future[Map[K1, Either[Unit, Exception]]] = mutex.acquire.flatMap { p =>
+      // write keys to backing store first
+      val storeResults : Map[K1, Future[Either[Unit, Exception]]] =
+        backingStore.multiPut(kvs).map { case (k, f) =>
+          (k, f.map { u: Unit => Left(u) }.onFailure { case x: Exception => Right(x) })
+        }
+
+      // perform cache operations based on how writes to backing store go
+      FutureOps.mapCollect(storeResults).flatMap { storeResult : Map[K1, Either[Unit, Exception]] =>
+        val failedKeys = storeResult.filter { _._2.isRight }.keySet
+        val succeededKeys = storeResult.filter { _._2.isLeft }.keySet
+
+        // write updated keys to cache, best effort
+        val keysToUpdate = kvs.filterKeys { succeededKeys.contains(_) } ++ {
+            // optionally invalidate cached copy when write to backing store fails
+            if (invalidate) { failedKeys.map { k => (k, None) }.toMap }
+            else { Map.empty }
+          }
+
+        FutureOps.mapCollect(cache.multiPut(keysToUpdate))(FutureCollector.bestEffort[(K1, Unit)])
+          .map { f => storeResult }
+        // return original writes made to backing store
+        // once cache operations are complete
+      } ensure {
+        p.release
+      }
+    }
+
+    // throw original exception for any writes that failed
+    FutureOps.liftValues(kvs.keySet, f, { (k: K1) => Future.None })
+      .map { case kv : (K1, Future[Either[Unit, Exception]]) =>
+        val transform = kv._2.map { v =>
+          v match {
+            case Left(optv) => optv
+            case Right(x) => throw x
+          }
+        }
+        (kv._1, transform)
+      }
+  }
+}
+

storehaus-core/src/test/scala/com/twitter/storehaus/ReadThroughStoreProperties.scala

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2014 Twitter Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.twitter.storehaus
+
+import org.scalacheck.Properties
+import org.scalacheck.Prop._
+
+object ReadThroughStoreProperties extends Properties("ReadThroughStoreProperties") {
+  import ReadableStoreProperties.readableStoreLaws
+
+  property("ReadThroughStore obeys the ReadableStore laws") =
+    readableStoreLaws[String, Int] { m =>
+      new ReadThroughStore(ReadableStore.fromMap(m), new ConcurrentHashMapStore[String,Int])
+    }
+}
+

storehaus-core/src/test/scala/com/twitter/storehaus/WriteThroughStoreProperties.scala

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2014 Twitter Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.twitter.storehaus
+
+import org.scalacheck.Properties
+
+object WriteThroughStoreProperties extends Properties("WriteThroughStoreProperties") {
+  import StoreProperties.storeTest
+
+  property("WriteThroughStore with invalidation obeys the Store laws") =
+    storeTest {
+      new WriteThroughStore(new ConcurrentHashMapStore[String,Int],
+        new ConcurrentHashMapStore[String,Int], true)
+    }
+
+  property("WriteThroughStore with no invalidation obeys the Store laws") =
+    storeTest {
+      new WriteThroughStore(new ConcurrentHashMapStore[String,Int],
+        new ConcurrentHashMapStore[String,Int], false)
+    }
+}
+