Replace viable task delays with overdue task delays
ClosedAll Users
Actions

src/main/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiterActor.scala
38	while we're editing this line, would be nice to extract this constant and give it a meaningful name. perhaps something like `overdueDelayCleanupInterval`? and maybe this constant could be referenced in test cases to give some shape to the magic numbers that appear in such places?

src/main/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiterActor.scala
38	Great idea, James; second this

Closed by commit rMARATHON31942642d8f6: Replace viable task delays with overdue task delays (authored by ichernetsky). · Explain WhySep 2 2017, 7:03 AM

This revision was automatically updated to reflect the committed changes.

src/main/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiterActor.scala
38	Somehow I skipped your comment. I will address it in another diff. Although the constant doesn't have anything to do with the durations in the tests below.

Revision Contents

			Path	Packages
M			src/main/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiter.scala (4 lines)
M			src/main/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiterActor.scala (18 lines)
M			src/test/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiterTest.scala (20 lines)

Diff	ID	Base	Description	Created	Lint	Unit
Base			Base
Diff 1	4166	edf19c9		Aug 30 2017, 10:40 PM	★	★
Diff 2	4167	edf19c9	- Remove unused ResetDelayResponse	Aug 30 2017, 10:46 PM	★	★
Diff 3	4194	2497310	rMARATHON31942642d8f68d71be04a5aa8be87186d17fb0d3	Sep 2 2017, 7:03 AM	★	★

Commit	Tree	Parents	Author	Summary	Date
2d18edc78187	0957e90047ae	0e925ae28edc	Ivan Chernetsky	Remove unused ResetDelayResponse	Aug 30 2017, 10:46 PM
0e925ae28edc	131f1a6635ac	edf19c9b728c	Ivan Chernetsky	Replace viable task delays with overdue task delays (Show More…)	Aug 30 2017, 10:38 PM

Diff 4194

View Options

src/main/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiter.scala

Show All 15 Lines
16	16	* We do not keep the delays for every version because that would include scaling changes or manual restarts.
17	17	*/
18	18	private[launchqueue] class RateLimiter(clock: Clock) {
19	19	import RateLimiter._
20	20
21	21	/** The task launch delays per run spec and their last config change. */
22	22	private[this] var taskLaunchDelays = Map[(PathId, Timestamp), Delay]()
23	23
24		/*Reset delays for tasks that have reached the maximum launch delay threshold. /
25		def resetDelaysOfViableTasks(): Unit = {
	24	/** Reset delays for tasks that have reached the maximum launch delay threshold. */
	25	def cleanUpOverdueDelays(): Unit = {
26	26	val now = clock.now()
27	27	taskLaunchDelays = taskLaunchDelays.filter {
28	28	case (_, delay) =>
29	29	now <= delay.referenceTimestamp + delay.maxLaunchDelay
30	30	}
31	31	}
32	32
33	33	def getDeadline(spec: RunSpec): Timestamp =
▲ Show 20 Lines • Show All 66 Lines • Show Last 20 Lines

View Options

src/main/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiterActor.scala

Show All 14 Lines
15	15		launchQueueRef: ActorRef): Props =
16	16		Props(new RateLimiterActor(
17	17		rateLimiter, launchQueueRef
18	18		))
19	19
20	20		case class DelayUpdate(runSpec: RunSpec, delayUntil: Timestamp)
21	21
22	22		case class ResetDelay(runSpec: RunSpec)
23			case object ResetDelayResponse
24
25	23		case class GetDelay(runSpec: RunSpec)
26	24		private[impl] case class AddDelay(runSpec: RunSpec)
27	25		private[impl] case class DecreaseDelay(runSpec: RunSpec)
28	26		private[impl] case class AdvanceDelay(runSpec: RunSpec)
29	27
30			private case object ResetViableTasksDelays
	28		private case object CleanupOverdueDelays
31	29		}
32	30
33	31		private class RateLimiterActor private (
34	32		rateLimiter: RateLimiter,
35	33		launchQueueRef: ActorRef) extends Actor with StrictLogging {
36	34		var cleanup: Cancellable = _
37	35
38	36		override def preStart(): Unit = {
39	37		import context.dispatcher
40			cleanup = context.system.scheduler.schedule(10.seconds, 10.seconds, self, ResetViableTasksDelays)
	38		cleanup = context.system.scheduler.schedule(10.seconds, 10.seconds, self, CleanupOverdueDelays)
		jdefUnsubmitted Not Done while we're editing this line, would be nice to extract this constant and give it a meaningful name. perhaps something like `overdueDelayCleanupInterval`? and maybe this constant could be referenced in test cases to give some shape to the magic numbers that appear in such places?
		timcharperUnsubmitted Not Done Great idea, James; second this
		ichernetskyAuthorUnsubmitted Not Done Somehow I skipped your comment. I will address it in another diff. Although the constant doesn't have anything to do with the durations in the tests below.
41	39		logger.info("started RateLimiterActor")
42	40		}
43	41
44	42		override def postStop(): Unit = {
45	43		cleanup.cancel()
46	44		}
47	45
48	46		override def receive: Receive = LoggingReceive {
49	47		Seq[Receive](
50	48		receiveCleanup,
51	49		receiveDelayOps
52	50		).reduceLeft(_.orElse[Any, Unit](_))
53	51		}
54	52
55			/**
56			* If an app gets removed or updated, the delay should be reset. If
57			* an app is considered viable, the delay should be reset too. We
58			* check and reset viable tasks' delays periodically.
59			*/
60	53		private[this] def receiveCleanup: Receive = {
61			case ResetViableTasksDelays =>
62			rateLimiter.resetDelaysOfViableTasks()
	54		case CleanupOverdueDelays =>
	55		// If a run spec gets removed or updated, the delay should be reset.
	56		// In addition to that we remove overdue delays to ensure there are no leaks,
	57		// by calling this periodically.
	58		rateLimiter.cleanUpOverdueDelays()
63	59		}
64	60
65	61		private[this] def receiveDelayOps: Receive = {
66	62		case GetDelay(runSpec) =>
67	63		sender() ! DelayUpdate(runSpec, rateLimiter.getDeadline(runSpec))
68	64
69	65		case AddDelay(runSpec) =>
70	66		rateLimiter.addDelay(runSpec)
Show All 13 Lines

View Options

src/test/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiterTest.scala

Show All 26 Lines
27	27	val app = AppDefinition(id = "test".toPath, backoffStrategy = BackoffStrategy(backoff = 10.seconds, factor = 2.0))
28	28
29	29	limiter.addDelay(app) // linter:ignore:IdenticalStatements
30	30	limiter.addDelay(app)
31	31
32	32	limiter.getDeadline(app) should be(clock.now() + 20.seconds)
33	33	}
34	34
35		"resetDelaysOfViableTasks" in {
	35	"cleanUpOverdueDelays" in {
36	36	val time_origin = clock.now()
37	37	val limiter = new RateLimiter(clock)
38	38	val threshold = 60.seconds
39	39
40		val app1 = AppDefinition(
41		id = "viable".toPath,
	40	val appWithOverdueDelay = AppDefinition(
	41	id = "overdue".toPath,
42	42	backoffStrategy = BackoffStrategy(backoff = 10.seconds, maxLaunchDelay = threshold))
43		limiter.addDelay(app1)
	43	limiter.addDelay(appWithOverdueDelay)
44	44
45		val app2 = AppDefinition(
46		id = "test".toPath,
	45	val appWithValidDelay = AppDefinition(
	46	id = "valid".toPath,
47	47	backoffStrategy = BackoffStrategy(backoff = 20.seconds, maxLaunchDelay = threshold + 10.seconds))
48		limiter.addDelay(app2)
	48	limiter.addDelay(appWithValidDelay)
49	49
50	50	// after advancing the clock by (threshold + 1), the existing delays
51	51	// with maxLaunchDelay < (threshold + 1) should be gone
52	52	clock += threshold + 1.seconds
53		limiter.resetDelaysOfViableTasks()
54		limiter.getDeadline(app1) should be(clock.now())
55		limiter.getDeadline(app2) should be(time_origin + 20.seconds)
	53	limiter.cleanUpOverdueDelays()
	54	limiter.getDeadline(appWithOverdueDelay) should be(clock.now())
	55	limiter.getDeadline(appWithValidDelay) should be(time_origin + 20.seconds)
56	56	}
57	57
58	58	"resetDelay" in {
59	59	val limiter = new RateLimiter(clock)
60	60	val app = AppDefinition(id = "test".toPath, backoffStrategy = BackoffStrategy(backoff = 10.seconds))
61	61
62	62	limiter.addDelay(app)
63	63	limiter.resetDelay(app)
64	64
65	65	limiter.getDeadline(app) should be(clock.now())
66	66	}
67	67	}
68	68	}

Replace viable task delays with overdue task delaysClosedAll UsersActions

Details

Diff Detail

＼\ ٩( ᐛ )و /／

＼\ ٩( ᐛ )و /／

Revision Contents

Diff 4194

src/main/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiter.scala

src/main/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiterActor.scala

src/test/scala/mesosphere/marathon/core/launchqueue/impl/RateLimiterTest.scala

Replace viable task delays with overdue task delays
ClosedAll Users
Actions