HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D627

wip: fixes history loading issue (option b)
AbandonedAll Users
Actions

Authored by timcharper on Mar 21 2017, 3:32 AM.

Details

Reviewers
aquamatthias
jenkins
• jasongilanfarr
JIRA Issues
JIRA MARATHON-7133 Marathon Application History is not loaded after Marathon restart
Summary

Marathon 1.4.x experienced a regression in which history would not load for an app definition. This is a slightly more complex, that retains similar performance properties and races.

Fixes MARATHON-7133

Test Plan

...

Diff Detail

Repository
rMARATHON marathon
Branch
tharper/fix-history-issue-2
Lint
No Linters Available
Unit
No Unit Test Coverage
Build Status
Buildable 1832
Build 3473: Marathon (deprecated)Jenkins
Build 3472: arc lint + arc unit
timcharper updated this revision to Diff 2324.Mar 21 2017, 3:32 AM
timcharper created this revision.

le compiles

jenkins requested changes to this revision.Mar 21 2017, 3:36 AM
This revision now requires changes to proceed.Mar 21 2017, 3:36 AM
jenkins accepted this revision.Mar 21 2017, 4:05 AM

Thanks.

This revision is now accepted and ready to land.Mar 21 2017, 4:05 AM
aquamatthias requested changes to this revision.Mar 21 2017, 10:53 AM
aquamatthias added inline comments.
src/main/scala/mesosphere/marathon/core/storage/store/impl/cache/LazyCachingPersistenceStore.scala
218

Without introducing a VersionCacheStorageReference we could simplify this to:

if (versionCache.contains((category, unversionedId))) {
  val cached = versionCache.getOrElse((category, unversionedId), Set.empty) // linter:ignore UndesirableTypeInference
  versionCache.put((category, unversionedId), cached + version)
}

Both solutions have a race problem, since the TrieMap does not have an update operation based on the existing value.

This revision now requires changes to proceed.Mar 21 2017, 10:53 AM
timcharper abandoned this revision.Mar 21 2017, 3:44 PM

Revision Contents

Diff 2324

View Options

src/main/scala/mesosphere/marathon/core/storage/store/impl/cache/LazyCachingPersistenceStore.scala

Show First 20 Lines • Show All 175 Lines • ▼ Show 20 Line(s)
176176}
177177
178178case class LazyVersionCachingPersistentStore[K, Category, Serialized](
179179 store: PersistenceStore[K, Category, Serialized],
180180 config: VersionCacheConfig = VersionCacheConfig.Default)(implicit
181181 mat: Materializer,
182182 ctx: ExecutionContext) extends PersistenceStore[K, Category, Serialized] with StrictLogging {
183183
184 private[store] val versionCache = TrieMap.empty[(Category, K), Set[OffsetDateTime]]
184 private[store] case class VersionCacheStorageReference(
185 complete: Boolean,
186 entries: Set[OffsetDateTime])
187 object VersionCacheStorageReference {
188 val empty = VersionCacheStorageReference(false, Set.empty) // linter:ignore UndesirableTypeInference
189 }
190
191 private[store] val versionCache = TrieMap.empty[(Category, K), VersionCacheStorageReference]
185192 private[store] val versionedValueCache = TrieMap.empty[(K, OffsetDateTime), Option[Any]]
186193
187194 private[cache] def maybePurgeCachedVersions(
188195 maxEntries: Int = config.maxEntries,
189196 purgeCount: Int = config.purgeCount,
190197 pRemove: Double = config.pRemove): Unit =
191198
192199 while (versionedValueCache.size > maxEntries) {
Show All 9 Lines
202209 id: Id,
203210 version: OffsetDateTime,
204211 v: Option[V])(implicit ir: IdResolver[Id, V, Category, K]): Unit = {
205212
206213 val category = ir.category
207214 val unversionedId = ir.toStorageId(id, None)
208215 maybePurgeCachedVersions()
209216 versionedValueCache.put((unversionedId, version), v)
210 val cached = versionCache.getOrElse((category, unversionedId), Set.empty) // linter:ignore UndesirableTypeInference
211 versionCache.put((category, unversionedId), cached + version)
217 val cached = versionCache.getOrElse((category, unversionedId), VersionCacheStorageReference.empty)
218 versionCache.put((category, unversionedId), cached.copy(entries = cached.entries + version))
aquamatthiasUnsubmitted
Not Done

Without introducing a VersionCacheStorageReference we could simplify this to:

if (versionCache.contains((category, unversionedId))) {
  val cached = versionCache.getOrElse((category, unversionedId), Set.empty) // linter:ignore UndesirableTypeInference
  versionCache.put((category, unversionedId), cached + version)
}

Both solutions have a race problem, since the TrieMap does not have an update operation based on the existing value.

212219 }
213220
214221 @SuppressWarnings(Array("all")) // async/await
215222 private def deleteCurrentOrAll[Id, V](
216223 id: Id, delete: () => Future[Done])(implicit ir: IdResolver[Id, V, Category, K]): Future[Done] = {
217224
218225 if (!ir.hasVersions) {
219226 delete()
▲ Show 20 Lines • Show All 96 Lines • ▼ Show 20 Line(s)
316323 version: OffsetDateTime)(implicit ir: IdResolver[Id, V, Category, K]): Future[Done] =
317324 deleteCurrentOrAll(k, () => store.deleteVersion(k, version))
318325
319326 @SuppressWarnings(Array("all")) // async/await
320327 override def versions[Id, V](id: Id)(implicit ir: IdResolver[Id, V, Category, K]): Source[OffsetDateTime, NotUsed] = {
321328 val versionsFuture = {
322329 val category = ir.category
323330 val storageId = ir.toStorageId(id, None)
324 if (versionCache.contains((category, storageId))) {
325 Future.successful(versionCache((category, storageId)))
331 if (versionCache.getOrElse((category, storageId), VersionCacheStorageReference.empty).complete) {
332 Future.successful(versionCache((category, storageId)).entries)
326333 } else {
327334 async {
328335 val children = await(store.versions(id).runWith(Sink.set))
329 versionCache((category, storageId)) = children
336 versionCache((category, storageId)) = VersionCacheStorageReference(true, children)
330337 children
331338 }
332339 }
333340 }
334341 Source.fromFuture(versionsFuture).mapConcat(identity)
335342 }
336343
337344 override def ids[Id, V]()(implicit ir: IdResolver[Id, V, Category, K]): Source[Id, NotUsed] = store.ids()
Show All 12 Lines
View Options

src/test/scala/mesosphere/marathon/core/storage/store/impl/cache/LazyCachingPersistenceStoreTest.scala

Show First 20 Lines • Show All 75 Lines • ▼ Show 20 Line(s)
7676 store.store("task-1", updated).futureValue should be(Done)
7777 store.store("task-1", updated).futureValue should be(Done) // redundant store should not lead to dup data
7878
7979 val storageId = ir.toStorageId("task-1", None)
8080 val cacheKey = (ir.category, storageId)
8181
8282 store.versionCache.size should be(1)
8383 store.versionCache.contains(cacheKey) should be(true)
84 store.versionCache(cacheKey) should contain theSameElementsAs Seq(original.version, updated.version)
84 store.versionCache(cacheKey).entries should contain theSameElementsAs Seq(original.version, updated.version)
8585
8686 store.versionedValueCache.size should be(2)
8787 store.versionedValueCache((storageId, original.version)) should be(Some(original))
8888 store.versionedValueCache((storageId, updated.version)) should be(Some(updated))
8989 }
9090
9191 "invalidates all cached versions upon deletion" in {
9292 implicit val clock = new SettableClock()
▲ Show 20 Lines • Show All 80 Lines • Show Last 20 Lines