Cache is not synchronised by key

[ExpensiveBelly]

If you run this code

runBlocking {
  launch { cache.get(0) { fetchUser() } }
  launch { cache.get(0) { fetchUser() } }
  launch { cache.get(0) { fetchUser() } }
  launch { cache.get(0) { fetchUser() } }
}

suspend fun fetchUser(): Int {
  delay(100)
  println("Hitting the network to fetch user!")
  return 0
}

then the message "Hitting the network to fetch user!" is printed 4 times.

I would expect this cache to be synchronised by key in a way that when multiple threads are trying to read a value from a key in which the value is suspend then it will await till the value is computed and then deliver it, rather than calling it once per thread and then race to see who writes to the cache first.

[ExpensiveBelly]

I would like to help you but I don't know how to do it with Kotlin-Native.

The way I would do it is:

private val mutexes = mutableMapOf<Key, Mutex>()

 override suspend fun get(key: Key, loader: suspend () -> Value): Value {
        return get(key) ?: mutexes.getOrPut(key, { Mutex() }).withLock {
            loader().let { loadedValue ->
                put(key, loadedValue)
                loadedValue
            }
        }
    }

This way you will have a Mutex per key, you can also do the clean-up once you remove the keys from cacheEntries by mutexes.remove(key) and mutexes.clear() if you wish to clean the cache.

Hope it helps

[ychescale9]

Thanks. The originally version I implemented in Dropbox/store supported this but it was jvm only. We might need to look at migrating to atomicfu and Kotlinx collections and migrate to the new memory model.

[ExpensiveBelly]

Did you deploy that original version? Can we use it? Thanks

[ychescale9]

It was published as part of 4.0 rc of store but was reverted as KMP support wasn't a priority.

[ychescale9]

I think this is the last release with my cache: com.dropbox.mobile.store:cache4:4.0.0-alpha07

[ExpensiveBelly]

Ok, I checked out that version of the Cache and it is true it is synchronised by key as you mentioned (loadersSynchronizer.synchronizedFor(key)) but it's not using suspend in the method signature, I think you guys added that later.

fun get(key: Key, loader: () -> Value): Value

[ychescale9]

Fixed in #15

[ExpensiveBelly]

Sorry to mention this again but I don't think put is synchronised, whereas in Dropbox one is:

@Nullable
    @Override
    public V put(@Nonnull K key, @Nonnull V value) {
        Preconditions.checkNotNull(key);
        Preconditions.checkNotNull(value);
        int hash = hash(key);
        return segmentFor(hash).put(key, hash, value, false);
    }
    
    @Nullable
        V put(@Nonnull K key, int hash, V value, boolean onlyIfAbsent) {
            lock();
            try {
                long now = map.ticker.read();
                preWriteCleanup(now);

                int newCount = this.count + 1;
                if (newCount > this.threshold) { // ensure capacity
                    expand();
                    newCount = this.count + 1;
                }

                AtomicReferenceArray<ReferenceEntry<K, V>> table = this.table;
                int index = hash & (table.length() - 1);
                ReferenceEntry<K, V> first = table.get(index);

                // Look for an existing entry.
                for (ReferenceEntry<K, V> e = first; e != null; e = e.getNext()) {
                    K entryKey = e.getKey();
                    if (e.getHash() == hash && entryKey != null
                            && map.keyEquivalence.equivalent(key, entryKey)) {
                        // We found an existing entry.

                        ValueReference<K, V> valueReference = e.getValueReference();
                        V entryValue = valueReference.get();

                        if (entryValue == null) {
                            ++modCount;
                            if (valueReference.isActive()) {
                                enqueueNotification(key, hash, valueReference, RemovalCause.COLLECTED);
                                setValue(e, key, value, now);
                                newCount = this.count; // count remains unchanged
                            } else {
                                setValue(e, key, value, now);
                                newCount = this.count + 1;
                            }
                            this.count = newCount; // write-volatile
                            evictEntries(e);
                            return null;
                        } else if (onlyIfAbsent) {
                            // Mimic
                            // "if (!map.containsKey(key)) ...
                            // else return map.get(key);
                            recordLockedRead(e, now);
                            return entryValue;
                        } else {
                            // clobber existing entry, count remains unchanged
                            ++modCount;
                            enqueueNotification(key, hash, valueReference, RemovalCause.REPLACED);
                            setValue(e, key, value, now);
                            evictEntries(e);
                            return entryValue;
                        }
                    }
                }

                // Create a new entry.
                ++modCount;
                ReferenceEntry<K, V> newEntry = newEntry(key, hash, first);
                setValue(newEntry, key, value, now);
                table.set(index, newEntry);
                newCount = this.count + 1;
                this.count = newCount; // write-volatile
                evictEntries(newEntry);
                return null;
            } finally {
                unlock();
                postWriteCleanup();
            }
        }
    ```
    
    Does this mean if we are going to populate the Cache with `put` from different threads we need to synchronise ourselves?