Simple thread-safe loading cache based on RxJava

Question

I'm sketching a simple thread-safe cache which can load missing values. It is based on RxJava's Observable which also means that it should be possible for a client to join a request for value which is already in flight.

My major concern is how to make it thread-safe whilst keeping an eye at performance and possible deadlocks.

So far I've started with ReentrantReadWriteLock and ConcurrentHashMap to achieve thread safety. There are comments along code lines used to clarify / justify decisions.

class ObservableCacheDispatcher(val cache: ICache, val dataMapper: IDataMapper): ICacheDispatcher {

    /** storage for requests which are being loaded at moment */
    val requestsInFlight = ConcurrentHashMap<String, Observable<Any>>()

    val lock = ReentrantReadWriteLock()

    val readLock = lock.readLock()

    val writeLock = lock.writeLock()


    /**
     * Returns data for the given {@code key}. If data is present in {@code cache} and not expired, returns it
     * immediately; otherwise loads data using {@code loader}, saves it {@code cache} and returns to caller.
     *
     * @param key string key which is a unique identifier of data
     * @param clazz class of data
     * @param loader loader used to load data if it's not present in {@code cache}
     * @param expiration date and time in future until cached data remains valid
     *
     * @return {@link Observable} of data
     */
    override fun <T> get(key: String, clazz: Class<T>, loader: () -> T, expiration: Date): Observable<T> {
        readLock.lock()
        try {
            // even if request is removed from requestsInFlight after get() but before if(),
            // the returned Observable will still be valid to obtain data from
            val observable = requestsInFlight.get(key)
            if (observable != null) {
                return observable as Observable<T>
            }

            // If there were no requests in flight, these possibilities exist:
            //   1) no request for this key was even made, and no value is present in cache,
            //      so execution goes to the next block guarded by writeLock (below);
            //
            //   2) another request has just successfully completed, and it is guaranteed there is already value
            //      in cache (writing to cache is guarded by requestsInFlight.get(key) presence;
            //
            //   3) another request has failed or writing to cache has failed so this request goes
            //      through the path (1);
            //
            //   4) another request has reached the block with the writeLock, but hasn't put anything
            //      to requestsInFlight yet -- this request will be waiting in front of writeLock.
            val entry = cache.get(key)
            if (entry != null && !entry.isHardExpired()) {
                return Observable.just(dataMapper.fromBytes(entry.data, clazz))
            }

        } finally {
            readLock.unlock()
        }

        val requestInFlight: Observable<Any>

        writeLock.lock()
        try {
            // since with ReadWriteLock there is no possibility to upgrade readLock to writeLock
            // when needed, there will be threads racing for writeLock, and we need to check
            // the preconditions once again
            val observable = requestsInFlight.get(key)
            if (observable != null) {
                return observable as Observable<T>
            }

            val entry = cache.get(key)
            if (entry != null && !entry.isHardExpired()) {
                return Observable.just(dataMapper.fromBytes(entry.data, clazz))
            }

            requestInFlight = Observable.just(loader())
            requestsInFlight.put(key, requestInFlight)

        } finally {
            writeLock.unlock()
        }

        // so the whole point of using writeLock was to create a request atomically,
        // after that it is possible to continue without lock, because the next block of code
        // is guarded by presence of request in requestsInFlight
        val result = requestInFlight.toBlocking().first()
        val data = dataMapper.toBytes(result)
        saveToCache(key, data, expiration)

        requestsInFlight.remove(key)

        return Observable.just(result as T)
    }

    private fun saveToCache(key: String, data: ByteArray, expiration: Date) {
        val entry = ICache.Entry(data)
        entry.setExpiration(expiration)
        cache.put(key, entry)
    }
}

Questions:

1. Is this design really thread-safe and deadlock free? Is it enough / too much to use ReentrantReadWriteLock and ConcurrentHashMap?
2. Would it be better to use a per-key lock; Guava's Striped?
3. Preconditions duplication in the readLock block and writeLock block feels like a code smell, is it possible to get rid of it?

Here is a link to the repo, which also includes a couple of tests.

Answer 1

I believe you can significantly simplify your code. First of all you can eliminate locks by using ConcurentHashMap.putIfAbsent or Map.getOrPut. Both of them not guaranteed however in most cases it is acceptable for caches to double load data due to concurrency at get/contains/put of Map. The other thing you can note to simplify you code is the fact that both initial loader and deserializer are actually Observable<T> that returns the same result. So you can always create observable depends on the case: Observable.from(loader).onComplete(serializer) or Observable.from(deserializer), put it to concurrent map and then return as is to user as loading and deserialization is also potentially blocking operation.

You also have to note that you shouldn't use locks because of potential blocking in cache.get(): Entry and mapper.fromBytes. Also both could take time so other requests couldn't be handled.

And, finally, you can look at RxJava Kotlin bindings here, which is going to be released for M11 soon.