-
Notifications
You must be signed in to change notification settings - Fork 1.5k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
finagle-core: Split the ClientDispatcher.scala file into parts
Problem We have a lot of different things going on in ClientDispatcher.scala that don't need to be in the same file. Solution Split it into multiple files. Differential Revision: https://phabricator.twitter.biz/D342883
- Loading branch information
Bryce Anderson
authored and
jenkins
committed
Jul 22, 2019
1 parent
44b11e3
commit 4b0493c
Showing
15 changed files
with
176 additions
and
85 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
107 changes: 107 additions & 0 deletions
107
finagle-core/src/main/scala/com/twitter/finagle/dispatch/GenSerialClientDispatcher.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,107 @@ | ||
| package com.twitter.finagle.dispatch | ||
|
|
||
| import com.twitter.concurrent.AsyncSemaphore | ||
| import com.twitter.finagle.{Failure, FailureFlags, Service, Status} | ||
| import com.twitter.finagle.stats.StatsReceiver | ||
| import com.twitter.finagle.tracing.Trace | ||
| import com.twitter.finagle.transport.Transport | ||
| import com.twitter.util.{Future, Promise, Return, Throw, Time} | ||
| import java.net.InetSocketAddress | ||
|
|
||
| /** | ||
| * Dispatches requests one at a time; concurrent requests are queued. | ||
| * | ||
| * @param statsReceiver typically scoped to `clientName/dispatcher` | ||
| */ | ||
| abstract class GenSerialClientDispatcher[Req, Rep, In, Out]( | ||
| trans: Transport[In, Out], | ||
| statsReceiver: StatsReceiver) | ||
| extends Service[Req, Rep] { | ||
|
|
||
| private[this] val semaphore = new AsyncSemaphore(1) | ||
|
|
||
| private[this] val queueSize = | ||
| statsReceiver.scope("serial").addGauge("queue_size") { | ||
| semaphore.numWaiters | ||
| } | ||
|
|
||
| private[this] val localAddress: InetSocketAddress = trans.context.localAddress match { | ||
| case ia: InetSocketAddress => ia | ||
| case _ => new InetSocketAddress(0) | ||
| } | ||
|
|
||
| // satisfy pending requests on transport close with a retryable failure | ||
| trans.onClose.respond { res => | ||
| val exc = res match { | ||
| case Return(exc) => exc | ||
| case Throw(exc) => exc | ||
| } | ||
|
|
||
| queueSize.remove() // ensure that we don't leak the gauge | ||
| semaphore.fail(Failure.retryable(exc)) | ||
| } | ||
|
|
||
| /** | ||
| * Dispatch a request, satisfying Promise `p` with the response; | ||
| * the returned Future is satisfied when the dispatch is complete: | ||
| * only one request is admitted at any given time. | ||
| * | ||
| * Note that GenSerialClientDispatcher manages interrupts, | ||
| * satisfying `p` should it be interrupted -- implementors beware: | ||
| * use only `updateIfEmpty` variants for satisfying the Promise. | ||
| * | ||
| * GenSerialClientDispatcher will also attempt to satisfy the promise | ||
| * if the returned `Future[Unit]` fails. | ||
| */ | ||
| protected def dispatch(req: Req, p: Promise[Rep]): Future[Unit] | ||
|
|
||
| private[this] def tryDispatch(req: Req, p: Promise[Rep]): Future[Unit] = | ||
| p.isInterrupted match { | ||
| case Some(intr) => | ||
| p.setException(Failure.adapt(intr, FailureFlags.Interrupted)) | ||
| Future.Done | ||
| case None => | ||
| Trace.recordClientAddr(localAddress) | ||
|
|
||
| p.setInterruptHandler { | ||
| case intr => | ||
| if (p.updateIfEmpty(Throw(intr))) | ||
| trans.close() | ||
| } | ||
|
|
||
| dispatch(req, p) | ||
| } | ||
|
|
||
| def apply(req: Req): Future[Rep] = { | ||
| val p = new Promise[Rep] | ||
|
|
||
| semaphore.acquire().respond { | ||
| case Return(permit) => | ||
| tryDispatch(req, p).respond { | ||
| case t @ Throw(_) => | ||
| p.updateIfEmpty(t.cast[Rep]) | ||
| permit.release() | ||
| case Return(_) => | ||
| permit.release() | ||
| } | ||
| case t @ Throw(_) => | ||
| p.update(t.cast[Rep]) | ||
| } | ||
|
|
||
| p | ||
| } | ||
|
|
||
| override def status: Status = trans.status | ||
|
|
||
| override def close(deadline: Time): Future[Unit] = trans.close() | ||
| } | ||
|
|
||
| object GenSerialClientDispatcher { | ||
|
|
||
| @deprecated("Use the value in `ClientDispatcher` object instead.", "2019-07-18") | ||
| val StatsScope: String = ClientDispatcher.StatsScope | ||
|
|
||
| @deprecated("Use the function in `ClientDispatcher` object instead.", "2019-07-18") | ||
| def wrapWriteException(exc: Throwable): Future[Nothing] = | ||
| ClientDispatcher.wrapWriteException(exc) | ||
| } |
28 changes: 28 additions & 0 deletions
28
finagle-core/src/main/scala/com/twitter/finagle/dispatch/SerialClientDispatcher.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,28 @@ | ||
| package com.twitter.finagle.dispatch | ||
|
|
||
| import com.twitter.finagle.stats.StatsReceiver | ||
| import com.twitter.finagle.transport.Transport | ||
| import com.twitter.util.{Future, Promise, Return, Throw, Try} | ||
|
|
||
| /** | ||
| * @param statsReceiver typically scoped to `clientName/dispatcher` | ||
| */ | ||
| class SerialClientDispatcher[Req, Rep](trans: Transport[Req, Rep], statsReceiver: StatsReceiver) | ||
| extends GenSerialClientDispatcher[Req, Rep, Req, Rep](trans, statsReceiver) { | ||
|
|
||
| import ClientDispatcher.wrapWriteException | ||
|
|
||
| private[this] val tryReadTheTransport: Try[Unit] => Future[Rep] = { | ||
| case Return(_) => trans.read() | ||
| case Throw(exc) => wrapWriteException(exc) | ||
| } | ||
|
|
||
| protected def dispatch(req: Req, p: Promise[Rep]): Future[Unit] = | ||
| trans | ||
| .write(req) | ||
| .transform(tryReadTheTransport) | ||
| .respond(rep => p.updateIfEmpty(rep)) | ||
| .unit | ||
|
|
||
| protected def write(req: Req): Future[Unit] = trans.write(req) | ||
| } |
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
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
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
2 changes: 1 addition & 1 deletion
2
...le-netty4-http/src/main/scala/com/twitter/finagle/netty4/http/Netty4StreamTransport.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
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
Oops, something went wrong.