perf: allow concurrent decompress away from network loop

We want to ensure the transcode passes everything that is expected to be
run asynchronously over to the decode loop. In general, memcached calls
us back with gotData, then receivedStatus, then complete. We use gotData
to launch the work onto the transcode threadpool and return control to
memcached, which would immediately then call receivedStatus.

Previously, receivedStatus and complete were set up to interact and set
a value on the underlying future but only by synchronously blocking for
the transcode future. Given this callback is happening nearly immediately
after the gotData callback, we were firing the transcode and nearly
always performing a blocking get, which triggers a synchronous
decompression on the network IO loop. This is of course very detrimental
to evcache driver performance, since the driver cannot even accept new
requests to issue to them to memcached backends, and must wait until
decompression completes.

In this fix, we rearrange things a little to ensure that if the async
decode is requested, that we push the completion status updates to
happen only after the async decode completes. This is a little ugly
because of the current arrangement of the memcached decoder. A future
change might be to overhaul this integration and pull it out of the
memcached transcode framework and use something a bit more friendly.

Also, add a property to control sync decode threading behavior, default
to using the existing behaviors for now.

Author: jasonk000

evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java

@@ -13,14 +13,16 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiPredicate;
+import java.util.function.Consumer;
 
-import com.netflix.evcache.pool.EVCacheClientUtil;
+import com.netflix.archaius.api.PropertyRepository;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,6 +37,7 @@
 import com.netflix.evcache.operation.EVCacheLatchImpl;
 import com.netflix.evcache.operation.EVCacheOperationFuture;
 import com.netflix.evcache.pool.EVCacheClient;
+import com.netflix.evcache.pool.EVCacheClientUtil;
 import com.netflix.evcache.util.EVCacheConfig;
 import com.netflix.spectator.api.BasicTag;
 import com.netflix.spectator.api.DistributionSummary;
@@ -80,16 +83,27 @@ public class EVCacheMemcachedClient extends MemcachedClient {
     private final Property<Integer> maxReadDuration, maxWriteDuration;
     private final Property<Boolean> enableDebugLogsOnWrongKey;
 
+    private volatile boolean alwaysDecodeSync;
+
     public EVCacheMemcachedClient(ConnectionFactory cf, List<InetSocketAddress> addrs,
                                   Property<Integer> readTimeout, EVCacheClient client) throws IOException {
         super(cf, addrs);
         this.connectionFactory = cf;
         this.readTimeout = readTimeout;
         this.client = client;
         this.appName = client.getAppName();
-        this.maxWriteDuration = EVCacheConfig.getInstance().getPropertyRepository().get(appName + ".max.write.duration.metric", Integer.class).orElseGet("evcache.max.write.duration.metric").orElse(50);
-        this.maxReadDuration = EVCacheConfig.getInstance().getPropertyRepository().get(appName + ".max.read.duration.metric", Integer.class).orElseGet("evcache.max.read.duration.metric").orElse(20);
-        this.enableDebugLogsOnWrongKey = EVCacheConfig.getInstance().getPropertyRepository().get(appName + ".enable.debug.logs.on.wrongkey", Boolean.class).orElse(false);
+        final PropertyRepository props = EVCacheConfig.getInstance().getPropertyRepository();
+        this.maxWriteDuration = props.get(appName + ".max.write.duration.metric", Integer.class).orElseGet("evcache.max.write.duration.metric").orElse(50);
+        this.maxReadDuration = props.get(appName + ".max.read.duration.metric", Integer.class).orElseGet("evcache.max.read.duration.metric").orElse(20);
+        this.enableDebugLogsOnWrongKey = props.get(appName + ".enable.debug.logs.on.wrongkey", Boolean.class).orElse(false);
+
+        // TODO in future remove this flag so that decode does not block the IO loop
+        // the default/legacy behavior (true) is effectively to decode on the IO loop, set to false to use the transcode threads
+        this.alwaysDecodeSync = true;
+        props.get(appName + ".get.alwaysDecodeSync", Boolean.class)
+                .orElseGet("evcache.get.alwaysDecodeSync")
+                .orElse(true)
+                .subscribe(v -> alwaysDecodeSync = v);
     }
 
     public NodeLocator getNodeLocator() {
@@ -126,62 +140,100 @@ private boolean isWrongKeyReturned(String original_key, String returned_key) {
     }
 
     public <T> EVCacheOperationFuture<T> asyncGet(final String key, final Transcoder<T> tc, EVCacheGetOperationListener<T> listener) {
-        final CountDownLatch latch = new CountDownLatch(1);
-        final EVCacheOperationFuture<T> rv = new EVCacheOperationFuture<T>(key, latch, new AtomicReference<T>(null), readTimeout.get().intValue(), executorService, client);
+        // we should only complete the latch when decode AND complete have completed
+        final CountDownLatch latch = new CountDownLatch(2);
+
+        final EVCacheOperationFuture<T> rv = new EVCacheOperationFuture<>(
+                key, latch, new AtomicReference<T>(null), readTimeout.get(), executorService, client);
+
+        final DistributionSummary dataSizeDS = getDataSizeDistributionSummary(
+                EVCacheMetricsFactory.GET_OPERATION,
+                EVCacheMetricsFactory.READ,
+                EVCacheMetricsFactory.IPC_SIZE_INBOUND);
+
+        @SuppressWarnings("unchecked")
+        final Transcoder<T> transcoder = (tc == null) ? (Transcoder<T>) getTranscoder() : tc;
+        final boolean shouldLog = log.isDebugEnabled() && client.getPool().getEVCacheClientPoolManager().shouldLog(appName);
+
         final Operation op = opFact.get(key, new GetOperation.Callback() {
-            private Future<T> val = null;
+            // not volatile since only ever used from memcached loop callbacks
+            private boolean asyncDecodeIssued = false;
 
-            public void receivedStatus(OperationStatus status) {
-                if (log.isDebugEnabled()) log.debug("Getting Key : " + key + "; Status : " + status.getStatusCode().name() + (log.isTraceEnabled() ?  " Node : " + getEVCacheNode(key) : "")
-                        + "; Message : " + status.getMessage() + "; Elapsed Time - " + (System.currentTimeMillis() - rv.getStartTime()));
-                try {
-                    if (val != null) {
-                        if (log.isTraceEnabled() && client.getPool().getEVCacheClientPoolManager().shouldLog(appName)) log.trace("Key : " + key + "; val : " + val.get());
-                        rv.set(val.get(), status);
+            // both volatile to ensure sync across transcode threads and memcached loop
+            private volatile T value;
+            private volatile OperationStatus status = null;
+
+            public void gotData(String k, int flags, byte[] data) {
+                if (isWrongKeyReturned(key, k)) {
+                    return;
+                }
+
+                if (shouldLog) {
+                    log.debug("Read data : key {}; flags : {}; data : {}", key, flags, data);
+                    if (data != null) {
+                        log.debug("Key : {}; val size : {}", key, data.length);
                     } else {
-                        if (log.isTraceEnabled() && client.getPool().getEVCacheClientPoolManager().shouldLog(appName)) log.trace("Key : " + key + "; val is null");
-                        rv.set(null, status);
+                        log.debug("Key : {}; val is null", key);
                     }
-                } catch (Exception e) {
-                    log.error(e.getMessage(), e);
-                    rv.set(null, status);
                 }
-            }
 
-            @SuppressWarnings("unchecked")
-            public void gotData(String k, int flags, byte[] data) {
+                if (data != null)  {
+                    dataSizeDS.record(data.length);
 
-                if (isWrongKeyReturned(key, k)) return;
+                    if (tcService == null) {
+                        log.error("tcService is null, will not be able to decode");
+                        throw new RuntimeException("TranscoderSevice is null. Not able to decode");
+                    }
 
-                if (log.isDebugEnabled() && client.getPool().getEVCacheClientPoolManager().shouldLog(appName)) log.debug("Read data : key " + key + "; flags : " + flags + "; data : " + data);
-                if (data != null)  {
-                    if (log.isDebugEnabled() && client.getPool().getEVCacheClientPoolManager().shouldLog(appName)) log.debug("Key : " + key + "; val size : " + data.length);
-                    getDataSizeDistributionSummary(EVCacheMetricsFactory.GET_OPERATION, EVCacheMetricsFactory.READ, EVCacheMetricsFactory.IPC_SIZE_INBOUND).record(data.length);
-                    if (tc == null) {
-                        if (tcService == null) {
-                            log.error("tcService is null, will not be able to decode");
-                            throw new RuntimeException("TranscoderSevice is null. Not able to decode");
-                        } else {
-                            final Transcoder<T> t = (Transcoder<T>) getTranscoder();
-                            val = tcService.decode(t, new CachedData(flags, data, t.getMaxSize()));
-                        }
+                    CachedData chunk = new CachedData(flags, data, transcoder.getMaxSize());
+                    boolean doSync = alwaysDecodeSync || (!transcoder.asyncDecode(chunk));
+
+                    if (doSync) {
+                        value = transcoder.decode(chunk);
+                        rv.set(value, status);
                     } else {
-                        if (tcService == null) {
-                            log.error("tcService is null, will not be able to decode");
-                            throw new RuntimeException("TranscoderSevice is null. Not able to decode");
-                        } else {
-                            val = tcService.decode(tc, new CachedData(flags, data, tc.getMaxSize()));
-                        }
+                        asyncDecodeIssued = true;
+                        final Transcoder<T> wrappedTranscoder = decodeAndThen(transcoder, (decoded) -> {
+                            value = decoded;
+                            rv.set(decoded, status);
+                            latch.countDown();
+                        });
+                        tcService.decode(wrappedTranscoder, chunk);
                     }
-                } else {
-                    if (log.isDebugEnabled() && client.getPool().getEVCacheClientPoolManager().shouldLog(appName)) log.debug("Key : " + key + "; val is null" );
+                }
+            }
+
+            public void receivedStatus(OperationStatus status) {
+                this.status = status;
+
+                // On rare occasion, it might be possible that transcoder finishes and starts to call rv.set(),
+                // at the exact time that receivedStatus here does a set(). This means that through unlucky timing
+                // here we might drop the decoded value that was set by the transcoder.
+                //
+                // We add a simple if check to see if the transcode thread has changed the value after we did rv.set(),
+                // and if it has, we will do it again. Since value is only set once (after decode), we need only a single
+                // check here. It is important that it is a separate volatile read from value.
+
+                T before = value;
+                rv.set(before, status);
+                T after = value;
+
+                if (after != before) {
+                    rv.set(after, status);
                 }
             }
 
             public void complete() {
+                // if an async decode was never issued, issue an extra countdown, since 2 latch values were set
+                if (!asyncDecodeIssued) {
+                    latch.countDown();
+                }
+
                 latch.countDown();
+
+                final String metricHit = (asyncDecodeIssued || value != null) ? EVCacheMetricsFactory.YES : EVCacheMetricsFactory.NO;
                 final String host = ((rv.getStatus().getStatusCode().equals(StatusCode.TIMEDOUT) && rv.getOperation() != null) ? getHostName(rv.getOperation().getHandlingNode().getSocketAddress()) : null);
-                getTimer(EVCacheMetricsFactory.GET_OPERATION, EVCacheMetricsFactory.READ, rv.getStatus(), (val != null ? EVCacheMetricsFactory.YES : EVCacheMetricsFactory.NO), host, getReadMetricMaxValue()).record((System.currentTimeMillis() - rv.getStartTime()), TimeUnit.MILLISECONDS);
+                getTimer(EVCacheMetricsFactory.GET_OPERATION, EVCacheMetricsFactory.READ, rv.getStatus(), metricHit, host, getReadMetricMaxValue()).record((System.currentTimeMillis() - rv.getStartTime()), TimeUnit.MILLISECONDS);
                 rv.signalComplete();
             }
         });
@@ -191,6 +243,37 @@ public void complete() {
         return rv;
     }
 
+    // A Transcode wrapper to allow an action to be performed after decode has completed.
+    static <T> Transcoder<T> decodeAndThen(Transcoder<T> transcoder, Consumer<T> completed) {
+        return new Transcoder<T>() {
+            @Override
+            public boolean asyncDecode(CachedData d) {
+                return transcoder.asyncDecode(d);
+            }
+
+            @Override
+            public CachedData encode(T o) {
+                throw new UnsupportedOperationException("encode");
+            }
+
+            @Override
+            public T decode(CachedData d) {
+                T decoded = null;
+                try {
+                    decoded = transcoder.decode(d);
+                    return decoded;
+                } finally {
+                    completed.accept(decoded);
+                }
+            }
+
+            @Override
+            public int getMaxSize() {
+                return transcoder.getMaxSize();
+            }
+        };
+    }
+
     public <T> EVCacheBulkGetFuture<T> asyncGetBulk(Collection<String> keys,
                                                     final Transcoder<T> tc,
                                                     EVCacheGetOperationListener<T> listener) {