Diff	ID	Base	Description	Created	Lint	Unit
Base			Base
Diff 1	2426	39456b0		Mar 29 2017, 2:01 AM	★	★
Diff 2	2427	39456b0	- Fix TimeoutTest too	Mar 29 2017, 2:21 AM	★	★
Diff 3	2430	39456b0	- Fix docs	Mar 29 2017, 3:52 AM	★	★
Diff 4	2445	39456b0	- Address Comments, retest unstable test case x 1000	Mar 29 2017, 11:59 PM	★	★
Diff 5	2452	0d73745	rMARATHON165e94ba65a25f629be94623a4a5a11e60bb709f	Mar 30 2017, 7:59 PM	★	★

Commit	Tree	Parents	Author	Summary	Date
88d07228cce0	6b254f1a315e	e67e978c5a0e	Jason Gilanfarr	Address Comments	Mar 29 2017, 11:59 PM
e67e978c5a0e	ebd7219e24a3	1354ff71a948	Jason Gilanfarr	Fix docs	Mar 29 2017, 3:49 AM
1354ff71a948	f0d3ed19a466	de0b45cec7e7	Jason Gilanfarr	Fix TimeoutTest too	Mar 29 2017, 2:21 AM
de0b45cec7e7	df2bdc7c5829	39456b09377c	Jason Gilanfarr	Propagate Context even for Unstarted Actors (Show More…)	Mar 29 2017, 1:59 AM

Diff 2452

src/main/resources/reference.conf

This file was deleted.

This file was completely deleted. Show File Contents

src/main/scala/akka/dispatch/ContextPropagatingDispatcher.scala

1	1		package akka.dispatch
2	2
3	3		import java.util
4	4
5			import akka.actor.ActorCell
6			import mesosphere.marathon.core.async.{ Context, ContextPropagatingExecutionContext, propagateContext }
	5		import mesosphere.marathon.core.async.{ Context, propagateContext }
7	6		import org.aspectj.lang.ProceedingJoinPoint
8	7		import org.aspectj.lang.annotation.{ Around, Aspect }
9	8		import org.slf4j.MDC
10	9
11			import scala.concurrent.duration.{ Duration, FiniteDuration }
12
13	10		// Note: we have to override behavior that is private to akka, so this has to be in the akka.dispatch package.
14	11
15	12		private class ContextWrapper(val invocation: Envelope, val context: Map[Context.ContextName[_], Any], val mdc: Option[util.Map[String, String]])
16	13		private object ContextWrapper {
17	14		def unapply(cw: ContextWrapper): Option[(Envelope, Map[Context.ContextName[_], Any], Option[util.Map[String, String]])] =
18	15		Some((cw.invocation, cw.context, cw.mdc))
19	16		}
20	17
21	18		/**
22			* Aspect that unwraps Akka Messages (in pair with [[ContextPropagatingDispatcher]]
	19		* Aspect that wraps and unwraps Akka Messages
23	20		* so that the Context/MDC are around the message.
24	21		*/
25	22		@Aspect
26	23		private class WeaveActorReceive {
27	24		@SuppressWarnings(Array("MethodReturningAny"))
28	25		@Around("execution(* akka.actor..ActorCell+.invoke(..)) && args(envelope)")
29	26		def contextInvoke(pjp: ProceedingJoinPoint, envelope: Envelope): Any = {
30	27		envelope match {
31	28		case Envelope(ContextWrapper(originalEnvelope, context, mdc), _) =>
		aquamatthiasUnsubmitted Not Done Would it be easier to check the message in `ContextPropagatingDispatcher.dispatch` to see if it is already a ContextWrapper?
		jasongilanfarrAuthorUnsubmitted Not Done in `unstartedSendMessage`? Yeah, definitely, I don't know why i didn't think of that. I got rid of the dispatcher as it isn't necessary.
32	29		propagateContext(context, mdc)(pjp.proceed(Array(originalEnvelope)))
33	30		case _ =>
34	31		pjp.proceed(Array(envelope))
35	32		}
36	33		}
37			}
38	34
39			/**
40			* Akka Dispatcher that copies the context and MDC when sending an actor a message.
41			*/
42			class ContextPropagatingDispatcher(
43			configurator: MessageDispatcherConfigurator,
44			id: String,
45			throughput: Int,
46			throughputDeadlineTime: Duration,
47			executorServiceFactoryProvider: ExecutorServiceFactoryProvider,
48			shutdownDeadlineTime: FiniteDuration)
49			extends Dispatcher(configurator, id, throughput, throughputDeadlineTime, executorServiceFactoryProvider, shutdownDeadlineTime)
50			with ContextPropagatingExecutionContext {
51
52			@SuppressWarnings(Array("NoOpOverride"))
53			override protected[akka] def dispatch(receiver: ActorCell, invocation: Envelope): Unit = {
54			super.dispatch(receiver, Envelope(new ContextWrapper(invocation, Context.copy(), Option(MDC.getCopyOfContextMap)), invocation.sender))
	35		// Unstarted actors use a chain of sendMessage, so we need to wrap only if we haven't already.
	36		@SuppressWarnings(Array("MethodReturningAny"))
	37		@Around("execution(* akka.actor..Cell+.sendMessage(..)) && args(envelope)")
	38		def sendMessage(pjp: ProceedingJoinPoint, envelope: Envelope): Any = {
	39		envelope match {
	40		case Envelope(ContextWrapper(_, _, _), _) =>
	41		// don't re-wrap
	42		pjp.proceed(Array(envelope))
	43		case _ =>
	44		pjp.proceed(Array(Envelope(new ContextWrapper(envelope, Context.copy(), Option(MDC.getCopyOfContextMap)), envelope.sender)))
55	45		}
56	46		}
	47		}
	48

View Options

src/main/scala/mesosphere/marathon/core/async/ExecutionContexts.scala

1	1	package mesosphere.marathon
2	2	package core.async
3	3
	4	import java.util.concurrent.Executor
	5
4	6	import org.slf4j.MDC
5	7
6	8	import scala.concurrent.{ ExecutionContext, ExecutionContextExecutor }
7		import java.util.concurrent.{ Executor, TimeUnit }
8	9
9		import akka.dispatch.{ ContextPropagatingDispatcher, DispatcherPrerequisites, MessageDispatcher, MessageDispatcherConfigurator }
10		import com.typesafe.config.Config
11
12		import scala.concurrent.duration.{ Duration, FiniteDuration }
13
14	10	/**
15	11	* Mixin that enables org.slf4j.MDC and [[Context]] propagation across threads.
16	12	*/
17	13	trait ContextPropagatingExecutionContext extends ExecutionContext { self =>
18	14	override def prepare(): ExecutionContext = new ExecutionContext {
19	15	val mdcContext = Option(MDC.getCopyOfContextMap)
20	16	val context = Context.copy() // linter:ignore
21	17
Show All 20 Lines
42	38	object CallerThreadExecutionContext {
43	39	val executor: Executor = new Executor {
44	40	override def execute(command: Runnable): Unit = command.run()
45	41	}
46	42
47	43	lazy val callerThreadExecutionContext: ExecutionContextExecutor = ExecutionContext.fromExecutor(executor)
48	44
49	45	def apply(): ExecutionContext = callerThreadExecutionContext
50		}
51
52		class ContextAwareDispatcherConfigurator(config: Config, prereqs: DispatcherPrerequisites)
53		extends MessageDispatcherConfigurator(config, prereqs) {
54
55		private val instance = {
56		new ContextPropagatingDispatcher(
57		this,
58		config.getString("id"),
59		config.getInt("throughput"),
60		Duration(config.getDuration("throughput-deadline-time", TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS),
61		configureExecutor(),
62		FiniteDuration(config.getDuration("shutdown-timeout", TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS)
63		)
64		}
65
66		override def dispatcher(): MessageDispatcher = instance
67	46	}
68	47
69	48	object ExecutionContexts {
70	49	/** Prefer this context over the default scala one as it can propagate org.slf4j.MDC and [[Context]] */
71	50	implicit lazy val global: ExecutionContext = ContextPropagatingExecutionContextWrapper(ExecutionContext.global)
72	51
73	52	lazy val callerThread: ExecutionContext = CallerThreadExecutionContext()
74	53	}

View Options

src/main/scala/mesosphere/marathon/core/async/package.scala

1	1	package mesosphere.marathon
2	2	package core
3	3
4	4	import org.slf4j.MDC
5	5	import java.util
6	6
7	7	package object async {
8		/** public only to enable [[akka.dispatch.ContextPropagatingDispatcher]] to use it */
	8	/** public only to enable our akka dispatch weaving to use it */
9	9	def propagateContext[T](context: Map[Context.ContextName[_], Any], mdc: Option[util.Map[String, String]])(f: => T): T = {
10	10	val oldMdc = Option(MDC.getCopyOfContextMap)
11	11	val oldContext = Context.copy // linter:ignore
12	12	try {
13	13	mdc.fold(MDC.clear())(MDC.setContextMap)
14	14	Context.set(context)
15	15	f
16	16	} finally {
17	17	oldMdc.fold(MDC.clear())(MDC.setContextMap)
18	18	Context.set(oldContext)
19	19	}
20	20	}
21	21	}

View Options

src/main/scala/mesosphere/marathon/util/Timeout.scala

Show All 22 Lines
23	23		* @param ctx The execution context to execute 'f' in
24	24		* @tparam T The result type of 'f'
25	25		* @return The eventual result of calling 'f' or TimeoutException if it didn't complete in time.
26	26		*/
27	27		def blocking[T](timeout: FiniteDuration)(f: => T)(implicit
28	28		scheduler: Scheduler,
29	29		ctx: ExecutionContext,
30	30		clock: Clock = Clock.systemDefaultZone()): Future[T] =
31			apply(timeout)(Future(blockingCall(f)))(scheduler, ctx, clock)
	31		apply(timeout)(Future(blockingCall(f))(ctx))(scheduler, ctx, clock)
		jeschkiesUnsubmitted Not Done Why is the context passed to Future now? It is hard for me to follow the change.
		jasongilanfarrAuthorUnsubmitted Not Done Its telling Future where to execute it.
32	32
33	33		/**
34	34		* Timeout a non-blocking call.
35	35		* @param timeout The maximum duration the method may execute in
36	36		* @param f The blocking call
37	37		* @param scheduler The akka scheduler
38	38		* @param ctx The execution context to execute 'f' in
39	39		* @tparam T The result type of 'f'
Show All 25 Lines

View Options

src/test/scala/mesosphere/marathon/core/async/ContextTest.scala

Show First 20 Lines • Show All 69 Lines • ▼ Show 20 Line(s)
70	70	val executor = Executors.newSingleThreadExecutor()
71	71	try {
72	72	val ctx = ContextPropagatingExecutionContextWrapper(ExecutionContext.fromExecutor(executor))
73	73	f(ctx)
74	74	} finally {
75	75	executor.shutdown()
76	76	}
77	77	}
	78
78	79	"ContextPropagatingExecutionContext" should {
79	80	"propagate the result across threads" in {
80	81	TestContext.withContext(7) {
81	82	val future = Future {
82	83	TestContext.value
83	84	}(ExecutionContexts.global)
84	85	future.futureValue.value should be(7)
85	86	}
▲ Show 20 Lines • Show All 216 Lines • Show Last 20 Lines

View Options

src/test/scala/mesosphere/marathon/util/TimeoutTest.scala

Show First 20 Lines • Show All 45 Lines • ▼ Show 20 Line(s)
46	46	val failure = Timeout.blocking(1.milli)(Thread.sleep(1000)).failed.futureValue
47	47	failure shouldBe a[TimeoutException]
48	48	}
49	49	"set the deadline" in {
50	50	implicit val clock = new SettableClock()
51	51
52	52	Timeout.blocking(1.second) {
53	53	clock.plus(2.second)
54		RunContext.state()
55		}.futureValue should be(Expired(Instant.now(clock).minusSeconds(2)))
	54	RunContext.state()(clock)
	55	}(scheduler, ExecutionContexts.global, clock).futureValue should be(Expired(Instant.now(clock).minusSeconds(2)))
56	56	}
57	57	}
58	58	}
59	59	}

		Path
D	M	src/main/resources/reference.conf (7 lines)
M		src/main/scala/akka/dispatch/ContextPropagatingDispatcher.scala (38 lines)
M		src/main/scala/mesosphere/marathon/core/async/ExecutionContexts.scala (25 lines)
M		src/main/scala/mesosphere/marathon/core/async/package.scala (2 lines)
M		src/main/scala/mesosphere/marathon/util/Timeout.scala (2 lines)
M		src/test/scala/mesosphere/marathon/core/async/ContextTest.scala (1 line)
M		src/test/scala/mesosphere/marathon/util/TimeoutTest.scala (4 lines)

Propagate Context even for Unstarted Actors
ClosedAll Users
Actions

Details

Diff Detail

Revision Contents

Diff 2452

src/main/resources/reference.conf

src/main/scala/akka/dispatch/ContextPropagatingDispatcher.scala

src/main/scala/mesosphere/marathon/core/async/ExecutionContexts.scala

src/main/scala/mesosphere/marathon/core/async/package.scala

src/main/scala/mesosphere/marathon/util/Timeout.scala

src/test/scala/mesosphere/marathon/core/async/ContextTest.scala

src/test/scala/mesosphere/marathon/util/TimeoutTest.scala

Propagate Context even for Unstarted ActorsClosedAll UsersActions

Details

Diff Detail

Revision Contents

Diff 2452

src/main/resources/reference.conf

src/main/scala/akka/dispatch/ContextPropagatingDispatcher.scala

src/main/scala/mesosphere/marathon/core/async/ExecutionContexts.scala

src/main/scala/mesosphere/marathon/core/async/package.scala

src/main/scala/mesosphere/marathon/util/Timeout.scala

src/test/scala/mesosphere/marathon/core/async/ContextTest.scala

src/test/scala/mesosphere/marathon/util/TimeoutTest.scala

Propagate Context even for Unstarted Actors
ClosedAll Users
Actions