-
Notifications
You must be signed in to change notification settings - Fork 1.4k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
finagle-memcached: Implement compressing cache using lz4 compression
**Problem** Make lz4 compression easily available via supported memcache clients. The desired effect is to drastically reduce the size of blobs stored in memcache, with minimal CPU cost to the calling services. **Solution** The hypothesis is that we can drastically reduce the size of our typical memcache payloads, at very low CPU cost, by using a throughput-oriented compressor like lz4. This will drastically reduce the network utilization of caches, improving performance in high-load situations like site failovers. This will also improve the storage utilization of cache, which will reduce eviction (and perhaps increase cache hit rates of calling services). For larger blobs that span >1 packet, this might also reduce cache tail latencies, by reducing the number of packets needed to transmit an object and thereby reducing the chances of any one of them being delayed or needing a retransmit. By using a few bits in the memcache protocol's flags field, this can be implemented in a way that allows for a transparent upgrade. New blobs written to the store will be compressed - existing blobs will be unmodified. The bits in the flags field will be used to signal whether the blob is compressed. Transparent upgrades will make adoption of this significantly easier, as the existing cached data in the cluster can be used while the transition occurs. **Result** ~30% compression for the higher sized items and 15-20% from p40 onwards Differential Revision: https://phabricator.twitter.biz/D1130236
- Loading branch information
Showing
26 changed files
with
1,575 additions
and
24 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
27 changes: 27 additions & 0 deletions
27
finagle-core/src/main/scala/com/twitter/finagle/filter/ToggleAwareSimpleFilter.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| package com.twitter.finagle.filter | ||
|
|
||
| import com.twitter.finagle.Service | ||
| import com.twitter.finagle.SimpleFilter | ||
| import com.twitter.finagle.server.ServerInfo | ||
| import com.twitter.finagle.toggle.Toggle | ||
| import com.twitter.util.Future | ||
|
|
||
| /** | ||
| * Simple filter which calls the underlying filter if the toggle is enabled or passes the call through | ||
| */ | ||
| class ToggleAwareSimpleFilter[Req, Rep](underlyingFilter: SimpleFilter[Req, Rep], toggle: Toggle) | ||
| extends SimpleFilter[Req, Rep] { | ||
|
|
||
| def apply(request: Req, service: Service[Req, Rep]): Future[Rep] = { | ||
| if (ToggleEnabled(toggle)) { | ||
| underlyingFilter(request, service) | ||
| } else { | ||
| service(request) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private object ToggleEnabled { | ||
| def apply(toggle: Toggle): Boolean = | ||
| toggle.isEnabled(ServerInfo().id.hashCode) | ||
| } |
54 changes: 54 additions & 0 deletions
54
finagle-core/src/test/scala/com/twitter/finagle/filter/ToggleAwareSimpleFilterTest.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| package com.twitter.finagle.filter | ||
|
|
||
| import com.twitter.conversions.DurationOps.RichDuration | ||
| import com.twitter.finagle.CoreToggles | ||
| import com.twitter.finagle.Service | ||
| import com.twitter.finagle.SimpleFilter | ||
| import com.twitter.util.Await | ||
| import com.twitter.util.Awaitable | ||
| import com.twitter.util.Future | ||
| import org.scalatestplus.mockito.MockitoSugar | ||
| import org.scalatest.funsuite.AnyFunSuite | ||
|
|
||
| class ToggleAwareSimpleFilterTest extends AnyFunSuite with MockitoSugar { | ||
|
|
||
| def await[T](awaitable: Awaitable[T]): T = Await.result(awaitable, 5.seconds) | ||
|
|
||
| trait Fixture { | ||
| val underlyingFilter = new SimpleFilter[Long, String] { | ||
| override def apply( | ||
| request: Long, | ||
| service: Service[Long, String] | ||
| ): Future[String] = Future.value("underlying filter") | ||
| } | ||
|
|
||
| val service = new Service[Long, String] { | ||
| override def apply(request: Long): Future[String] = Future.value("service") | ||
| } | ||
|
|
||
| val toggleKey = "com.twitter.finagle.filter.TestToggleAwareUnderlying" | ||
| val toggle = CoreToggles(toggleKey) | ||
| val filter = new ToggleAwareSimpleFilter[Long, String](underlyingFilter, toggle) | ||
|
|
||
| } | ||
|
|
||
| test("calls underlying filter when toggle is enabled") { | ||
| new Fixture { | ||
|
|
||
| com.twitter.finagle.toggle.flag.overrides.let(toggleKey, 1) { | ||
| val result = filter.apply(0L, service) | ||
| assert(await(result) == "underlying filter") | ||
| } | ||
| } | ||
| } | ||
|
|
||
| test("calls actual service when toggle is disabled") { | ||
| new Fixture { | ||
|
|
||
| com.twitter.finagle.toggle.flag.overrides.let(toggleKey, 0) { | ||
| val result = filter.apply(0L, service) | ||
| assert(await(result) == "service") | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,8 @@ | ||
| resources( | ||
| sources = [ | ||
| "!**/*.pyc", | ||
| "!BUILD*", | ||
| "**/*", | ||
| ], | ||
| tags = ["bazel-compatible"], | ||
| ) |
9 changes: 9 additions & 0 deletions
9
...mcached/src/main/resources/com/twitter/toggles/configs/com.twitter.finagle.memcached.json
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,9 @@ | ||
| { | ||
| "toggles": [ | ||
| { | ||
| "id": "com.twitter.finagle.filter.CompressingMemcached", | ||
| "description": "Enable compressing filter for memcached values", | ||
| "fraction": 1.0 | ||
| } | ||
| ] | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
184 changes: 184 additions & 0 deletions
184
...e-memcached/src/main/scala/com/twitter/finagle/memcached/CompressingMemcachedFilter.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,184 @@ | ||
| package com.twitter.finagle.memcached | ||
|
|
||
| import com.twitter.finagle._ | ||
| import com.twitter.finagle.memcached.compressing.CompressionProvider | ||
| import com.twitter.finagle.memcached.compressing.param.CompressionParam | ||
| import com.twitter.finagle.memcached.compressing.scheme.CompressionScheme | ||
| import com.twitter.finagle.memcached.compressing.scheme.Lz4 | ||
| import com.twitter.finagle.memcached.compressing.scheme.MemcachedCompression.FlagsAndBuf | ||
| import com.twitter.finagle.memcached.compressing.scheme.Uncompressed | ||
| import com.twitter.finagle.memcached.protocol.Add | ||
| import com.twitter.finagle.memcached.protocol.Cas | ||
| import com.twitter.finagle.memcached.protocol.Command | ||
| import com.twitter.finagle.memcached.protocol.NonStorageCommand | ||
| import com.twitter.finagle.memcached.protocol.Response | ||
| import com.twitter.finagle.memcached.protocol.RetrievalCommand | ||
| import com.twitter.finagle.memcached.protocol.Set | ||
| import com.twitter.finagle.memcached.protocol.StorageCommand | ||
| import com.twitter.finagle.memcached.protocol.Value | ||
| import com.twitter.finagle.memcached.protocol.Values | ||
| import com.twitter.finagle.memcached.protocol.ValuesAndErrors | ||
| import com.twitter.finagle.server.ServerInfo | ||
| import com.twitter.finagle.stats.StatsReceiver | ||
| import com.twitter.finagle.toggle.Toggle | ||
| import com.twitter.io.Buf | ||
| import com.twitter.util.Future | ||
| import com.twitter.util.Return | ||
| import com.twitter.util.Throw | ||
| import scala.collection.mutable | ||
|
|
||
| private[finagle] object CompressingMemcachedFilter { | ||
|
|
||
| /** | ||
| * Apply the [[CompressingMemcachedFilter]] protocol specific annotations | ||
| */ | ||
| def memcachedCompressingModule: Stackable[ServiceFactory[Command, Response]] = | ||
| new Stack.Module2[CompressionParam, param.Stats, ServiceFactory[Command, Response]] { | ||
| override val role: Stack.Role = Stack.Role("MemcachedCompressing") | ||
| override val description: String = "Memcached filter with compression logic" | ||
|
|
||
| override def make( | ||
| _compressionParam: CompressionParam, | ||
| _stats: param.Stats, | ||
| next: ServiceFactory[Command, Response] | ||
| ): ServiceFactory[Command, Response] = { | ||
| _compressionParam.scheme match { | ||
| case Uncompressed => | ||
| new CompressingMemcachedFilter(Uncompressed, _stats.statsReceiver).andThen(next) | ||
| case Lz4 => | ||
| new CompressingMemcachedFilter(Lz4, _stats.statsReceiver).andThen(next) | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private[finagle] final class CompressingMemcachedFilter( | ||
| compressionScheme: CompressionScheme, | ||
| statsReceiver: StatsReceiver) | ||
| extends SimpleFilter[Command, Response] { | ||
| private final val compressionFactory = | ||
| CompressionProvider(compressionScheme, statsReceiver) | ||
|
|
||
| private val toggle: Toggle = Toggles("com.twitter.finagle.filter.CompressingMemcached") | ||
|
|
||
| private val serverInfo = ServerInfo() | ||
|
|
||
| override def apply(command: Command, service: Service[Command, Response]): Future[Response] = { | ||
| command match { | ||
| case storageCommand: StorageCommand => | ||
| if (compressionScheme == Uncompressed || !toggle.isEnabled(serverInfo.id.hashCode)) { | ||
| service(storageCommand) | ||
| } else { service(compress(storageCommand)) } | ||
| case nonStorageCommand: NonStorageCommand => | ||
| decompress(nonStorageCommand, service) | ||
| case otherCommand => service(otherCommand) | ||
| } | ||
| } | ||
|
|
||
| private def compressedBufAndFlags(value: Buf, flags: Int): FlagsAndBuf = { | ||
| val (compressionFlags, compressedBuf) = compressionFactory.compressor(value) | ||
| (CompressionScheme.flagsWithCompression(flags, compressionFlags), compressedBuf) | ||
| } | ||
|
|
||
| def compress(command: StorageCommand): StorageCommand = { | ||
| command match { | ||
| case Set(key, flags, expiry, value) => | ||
| val (flagsWithCompression, compressedBuf) = | ||
| compressedBufAndFlags(value, flags) | ||
|
|
||
| Set(key, flagsWithCompression, expiry, compressedBuf) | ||
|
|
||
| case Add(key, flags, expiry, value) => | ||
| val (flagsWithCompression, compressedBuf) = | ||
| compressedBufAndFlags(value, flags) | ||
|
|
||
| Add(key, flagsWithCompression, expiry, compressedBuf) | ||
|
|
||
| case Cas(key, flags, expiry, value, casUnique) => | ||
| val (flagsWithCompression, compressedBuf) = | ||
| compressedBufAndFlags(value, flags) | ||
|
|
||
| Cas(key, flagsWithCompression, expiry, compressedBuf, casUnique) | ||
|
|
||
| case unsupportedStorageCommand => unsupported(unsupportedStorageCommand.name) | ||
| } | ||
| } | ||
|
|
||
| def decompress( | ||
| command: NonStorageCommand, | ||
| service: Service[Command, Response] | ||
| ): Future[Response] = { | ||
| command match { | ||
| case retrievalCommand: RetrievalCommand => | ||
| service(retrievalCommand).map { | ||
| case values: Values => | ||
| val decompressedValues = | ||
| decompressValues(values.values) | ||
| Values(decompressedValues) | ||
| case valuesAndErrors: ValuesAndErrors => | ||
| val decompressedValuesAndErrors = | ||
| decompressValuesAndErrors(valuesAndErrors) | ||
| decompressedValuesAndErrors | ||
| case retrievalResponse => retrievalResponse | ||
| } | ||
| case otherNonStorageCommand => service(otherNonStorageCommand) | ||
| } | ||
| } | ||
|
|
||
| private[finagle] def decompressValues( | ||
| values: Seq[Value], | ||
| ): Seq[Value] = { | ||
| val decompressedValuesList = mutable.ArrayBuffer[Value]() | ||
|
|
||
| values.foreach { value => | ||
| val flagsInt = flagsFromValues(value) | ||
|
|
||
| compressionFactory.decompressor((flagsInt, value.value)) match { | ||
| case Throw(ex) => throw ex | ||
| case Return(uncompressedValue) => | ||
| decompressedValuesList.append(value.copy(value = uncompressedValue)) | ||
| } | ||
| } | ||
|
|
||
| decompressedValuesList.toSeq | ||
| } | ||
|
|
||
| private[finagle] def decompressValuesAndErrors( | ||
| valueAndErrors: ValuesAndErrors, | ||
| ): ValuesAndErrors = { | ||
| // Decompress values. If for some reason this fails, move the values to failures. | ||
| val decompressedValuesList = mutable.ArrayBuffer[Value]() | ||
| val failuresList = mutable.ListBuffer[(Buf, Throwable)]() | ||
|
|
||
| val values = valueAndErrors.values | ||
|
|
||
| values.foreach { value => | ||
| try { | ||
| val flagsInt = flagsFromValues(value) | ||
|
|
||
| compressionFactory.decompressor((flagsInt, value.value)) match { | ||
| case Return(decompressedValue) => | ||
| decompressedValuesList.append(value.copy(value = decompressedValue)) | ||
| case Throw(ex) => failuresList.append(value.key -> ex) | ||
| } | ||
| } catch { | ||
| case ex: Throwable => failuresList.append(value.key -> ex) | ||
| } | ||
| } | ||
|
|
||
| valueAndErrors.copy( | ||
| values = decompressedValuesList.toSeq, | ||
| errors = valueAndErrors.errors ++ failuresList.toMap) | ||
| } | ||
|
|
||
| private def flagsFromValues(value: Value): Int = { | ||
| // The flags value is stored as a literal string, from the memcache text protocol. | ||
| value.flags.flatMap(Buf.Utf8.unapply(_)) match { | ||
| case Some(s) => s.toInt | ||
| case None => 0 | ||
| } | ||
| } | ||
|
|
||
| private def unsupported(command: String) = throw new UnsupportedOperationException( | ||
| s"$command is unsupported for compressing cache") | ||
| } |
Oops, something went wrong.