You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@openwhisk.apache.org by ma...@apache.org on 2017/07/20 13:18:47 UTC
[incubator-openwhisk] branch master updated: Move load balancer
data into a separate class (#2477)
This is an automated email from the ASF dual-hosted git repository.
markusthoemmes pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-openwhisk.git
The following commit(s) were added to refs/heads/master by this push:
new 2397e2a Move load balancer data into a separate class (#2477)
2397e2a is described below
commit 2397e2a5721974fc77e237dc19e661d5ddf8d68d
Author: Vadim Raskin <ra...@gmail.com>
AuthorDate: Thu Jul 20 15:18:45 2017 +0200
Move load balancer data into a separate class (#2477)
Signed-off-by: Markus Thoemmes <ma...@de.ibm.com>
---
.../whisk/core/loadBalancer/LoadBalancerData.scala | 121 +++++++++++++
.../core/loadBalancer/LoadBalancerService.scala | 60 ++-----
.../loadBalancer/test/LoadBalancerDataTests.scala | 192 +++++++++++++++++++++
3 files changed, 332 insertions(+), 41 deletions(-)
diff --git a/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerData.scala b/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerData.scala
new file mode 100644
index 0000000..3fd3d96
--- /dev/null
+++ b/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerData.scala
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.loadBalancer
+
+import whisk.core.entity.{ActivationId, UUID, WhiskActivation}
+import scala.collection.concurrent.TrieMap
+import scala.concurrent.Promise
+
+/** Encapsulates data relevant for a single activation */
+case class ActivationEntry(id: ActivationId, namespaceId: UUID, invokerName: String, promise: Promise[Either[ActivationId, WhiskActivation]])
+
+/**
+ * Encapsulates data used for loadbalancer and active-ack bookkeeping.
+ *
+ * Note: The state keeping is backed by concurrent data-structures. As such,
+ * concurrent reads can return stale values (especially the counters returned).
+ */
+class LoadBalancerData() {
+
+ type TrieSet[T] = TrieMap[T, Unit]
+
+ private val activationByInvoker = new TrieMap[String, TrieSet[ActivationEntry]]
+ private val activationByNamespaceId = new TrieMap[UUID, TrieSet[ActivationEntry]]
+ private val activationsById = new TrieMap[ActivationId, ActivationEntry]
+
+ /**
+ * Get the number of activations for each namespace.
+ *
+ * @return a map (namespace -> number of activations in the system)
+ */
+ def activationCountByNamespace: Map[UUID, Int] = {
+ activationByNamespaceId.toMap.mapValues(_.size)
+ }
+
+ /**
+ * Get the number of activations for each invoker.
+ *
+ * @return a map (invoker -> number of activations queued for the invoker)
+ */
+ def activationCountByInvoker: Map[String, Int] = {
+ activationByInvoker.toMap.mapValues(_.size)
+ }
+
+ /**
+ * Get all active activations for a specific invoker.
+ *
+ * @param invokerName the invoker to get activations for
+ * @return all active activations for the given invoker
+ */
+ def activationsByInvoker(invokerName: String): TrieSet[ActivationEntry] = {
+ activationByInvoker.getOrElseUpdate(invokerName, new TrieSet[ActivationEntry])
+ }
+
+ /**
+ * Get an activation entry for a given activation id.
+ *
+ * @param activationId activation id to get data for
+ * @return the respective activation or None if it doesn't exist
+ */
+ def activationById(activationId: ActivationId): Option[ActivationEntry] = {
+ activationsById.get(activationId)
+ }
+
+ /**
+ * Adds an activation entry.
+ *
+ * @param id identifier to deduplicate the entry
+ * @param update block calculating the entry to add.
+ * Note: This is evaluated iff the entry
+ * didn't exist before.
+ * @return the entry calculated by the block or iff it did
+ * exist before the entry from the state
+ */
+ def putActivation(id: ActivationId, update: => ActivationEntry): ActivationEntry = {
+ activationsById.getOrElseUpdate(id, {
+ val entry = update
+ activationByNamespaceId.getOrElseUpdate(entry.namespaceId, new TrieSet[ActivationEntry]).put(entry, {})
+ activationByInvoker.getOrElseUpdate(entry.invokerName, new TrieSet[ActivationEntry]).put(entry, {})
+ entry
+ })
+ }
+
+ /**
+ * Removes the given entry.
+ *
+ * @param entry the entry to remove
+ * @return The deleted entry or None if nothing got deleted
+ */
+ def removeActivation(entry: ActivationEntry): Option[ActivationEntry] = {
+ activationsById.remove(entry.id).map { x =>
+ activationByNamespaceId.getOrElseUpdate(x.namespaceId, new TrieSet[ActivationEntry]).remove(entry)
+ activationByInvoker.getOrElseUpdate(x.invokerName, new TrieSet[ActivationEntry]).remove(entry)
+ x
+ }
+ }
+
+ /**
+ * Removes the activation identified by the given activation id.
+ *
+ * @param aid activation id to remove
+ * @return The deleted entry or None if nothing got deleted
+ */
+ def removeActivation(aid: ActivationId): Option[ActivationEntry] = {
+ activationsById.get(aid).flatMap(removeActivation)
+ }
+}
diff --git a/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala b/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala
index cdf16a3..9f78e45 100644
--- a/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala
+++ b/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala
@@ -18,9 +18,7 @@
package whisk.core.loadBalancer
import java.nio.charset.StandardCharsets
-import java.time.{ Clock, Instant }
-import scala.collection.concurrent.TrieMap
import scala.concurrent.Await
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
@@ -49,7 +47,7 @@ import whisk.core.connector.{ ActivationMessage, CompletionMessage }
import whisk.core.connector.MessageFeed
import whisk.core.connector.MessageProducer
import whisk.core.database.NoDocumentException
-import whisk.core.entity.{ ActivationId, WhiskAction, WhiskActivation }
+import whisk.core.entity.{ ActivationId, WhiskActivation }
import whisk.core.entity.InstanceId
import whisk.core.entity.ExecutableWhiskAction
import whisk.core.entity.UUID
@@ -99,7 +97,9 @@ class LoadBalancerService(
private val blackboxFraction: Double = Math.max(0.0, Math.min(1.0, config.controllerBlackboxFraction))
logging.info(this, s"blackboxFraction = $blackboxFraction")
- override def getActiveNamespaceActivationCounts: Map[UUID, Int] = activationByNamespaceId.toMap mapValues { _.size }
+ private val loadBalancerData = new LoadBalancerData()
+
+ override def getActiveNamespaceActivationCounts: Map[UUID, Int] = loadBalancerData.activationCountByNamespace
override def publish(action: ExecutableWhiskAction, msg: ActivationMessage)(
implicit transid: TransactionId): Future[Future[Either[ActivationId, WhiskActivation]]] = {
@@ -114,16 +114,6 @@ class LoadBalancerService(
def invokerHealth: Future[Map[String, InvokerState]] = invokerPool.ask(GetStatus)(Timeout(5.seconds)).mapTo[Map[String, InvokerState]]
/**
- * A map storing current activations based on ActivationId.
- * The promise value represents the obligation of writing the answer back.
- */
- case class ActivationEntry(id: ActivationId, namespaceId: UUID, invokerName: String, created: Instant, promise: Promise[Either[ActivationId, WhiskActivation]])
- type TrieSet[T] = TrieMap[T, Unit]
- private val activationById = new TrieMap[ActivationId, ActivationEntry]
- private val activationByInvoker = new TrieMap[String, TrieSet[ActivationEntry]]
- private val activationByNamespaceId = new TrieMap[UUID, TrieSet[ActivationEntry]]
-
- /**
* Tries to fill in the result slot (i.e., complete the promise) when a completion message arrives.
* The promise is removed form the map when the result arrives or upon timeout.
*
@@ -131,17 +121,14 @@ class LoadBalancerService(
*/
private def processCompletion(response: Either[ActivationId, WhiskActivation], tid: TransactionId, forced: Boolean): Unit = {
val aid = response.fold(l => l, r => r.activationId)
- activationById.remove(aid) match {
- case Some(entry @ ActivationEntry(_, namespaceId, invokerIndex, _, p)) =>
+ loadBalancerData.removeActivation(aid) match {
+ case Some(entry) =>
logging.info(this, s"${if (!forced) "received" else "forced"} active ack for '$aid'")(tid)
- activationByInvoker.getOrElseUpdate(invokerIndex, new TrieSet[ActivationEntry]).remove(entry)
- activationByNamespaceId.getOrElseUpdate(namespaceId, new TrieSet[ActivationEntry]).remove(entry)
if (!forced) {
- p.trySuccess(response)
+ entry.promise.trySuccess(response)
} else {
- p.tryFailure(new Throwable("no active ack received"))
+ entry.promise.tryFailure(new Throwable("no active ack received"))
}
-
case None =>
// the entry was already removed
logging.debug(this, s"received active ack for '$aid' which has no entry")(tid)
@@ -152,43 +139,34 @@ class LoadBalancerService(
* Creates an activation entry and insert into various maps.
*/
private def setupActivation(action: ExecutableWhiskAction, activationId: ActivationId, namespaceId: UUID, invokerName: String, transid: TransactionId): ActivationEntry = {
- // either create a new promise or reuse a previous one for this activation if it exists
+
val timeout = action.limits.timeout.duration + activeAckTimeoutGrace
- val entry = activationById.getOrElseUpdate(activationId, {
- // Install a timeout handler for the catastrophic case where an active ack is not received at all
- // (because say an invoker is down completely, or the connection to the message bus is disrupted) or when
- // the active ack is significantly delayed (possibly dues to long queues but the subject should not be penalized);
- // in this case, if the activation handler is still registered, remove it and update the books.
+ // Install a timeout handler for the catastrophic case where an active ack is not received at all
+ // (because say an invoker is down completely, or the connection to the message bus is disrupted) or when
+ // the active ack is significantly delayed (possibly dues to long queues but the subject should not be penalized);
+ // in this case, if the activation handler is still registered, remove it and update the books.
+ loadBalancerData.putActivation(activationId, {
actorSystem.scheduler.scheduleOnce(timeout) {
processCompletion(Left(activationId), transid, forced = true)
}
- ActivationEntry(activationId, namespaceId, invokerName, Instant.now(Clock.systemUTC()), Promise[Either[ActivationId, WhiskActivation]]())
+ ActivationEntry(activationId, namespaceId, invokerName, Promise[Either[ActivationId, WhiskActivation]]())
})
-
- // add the entry to our maps, for bookkeeping
- activationByInvoker.getOrElseUpdate(invokerName, new TrieSet[ActivationEntry]).put(entry, {})
- activationByNamespaceId.getOrElseUpdate(namespaceId, new TrieSet[ActivationEntry]).put(entry, {})
- entry
}
/**
* When invoker health detects a new invoker has come up, this callback is called.
*/
private def clearInvokerState(invokerName: String) = {
- val actSet = activationByInvoker.getOrElseUpdate(invokerName, new TrieSet[ActivationEntry])
+ val actSet = loadBalancerData.activationsByInvoker(invokerName)
actSet.keySet map {
- case actEntry @ ActivationEntry(activationId, namespaceId, invokerIndex, _, promise) =>
+ case actEntry @ ActivationEntry(activationId, namespaceId, invokerIndex, promise) =>
promise.tryFailure(new LoadBalancerException(s"Invoker $invokerIndex restarted"))
- activationById.remove(activationId)
- activationByNamespaceId.get(namespaceId) map { _.remove(actEntry) }
+ loadBalancerData.removeActivation(actEntry)
}
actSet.clear()
}
- /** Make a new immutable map so caller cannot mess up the state */
- private def getActiveCountByInvoker(): Map[String, Int] = activationByInvoker.toMap mapValues { _.size }
-
/**
* Creates or updates a health test action by updating the entity store.
* This method is intended for use on startup.
@@ -300,7 +278,7 @@ class LoadBalancerService(
invokers.flatMap { invokers =>
LoadBalancerService.schedule(
invokers,
- activationByInvoker.mapValues(_.size),
+ loadBalancerData.activationCountByInvoker,
config.loadbalancerInvokerBusyThreshold,
hash) match {
case Some(invoker) => Future.successful(invoker)
diff --git a/tests/src/test/scala/whisk/core/loadBalancer/test/LoadBalancerDataTests.scala b/tests/src/test/scala/whisk/core/loadBalancer/test/LoadBalancerDataTests.scala
new file mode 100644
index 0000000..e314abd
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/loadBalancer/test/LoadBalancerDataTests.scala
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.loadBalancer.test
+
+import org.scalatest.{FlatSpec, Matchers}
+import whisk.core.entity.{ActivationId, UUID, WhiskActivation}
+import whisk.core.loadBalancer.{ActivationEntry, LoadBalancerData}
+
+import scala.concurrent.{Promise}
+
+class LoadBalancerDataTests extends FlatSpec with Matchers {
+
+ val activationIdPromise = Promise[Either[ActivationId, WhiskActivation]]()
+ val firstEntry: ActivationEntry = ActivationEntry(ActivationId(), UUID(), "invoker0", activationIdPromise)
+ val secondEntry: ActivationEntry = ActivationEntry(ActivationId(), UUID(), "invoker1", activationIdPromise)
+
+ behavior of "LoadBalancerData"
+
+ it should "return the number of activations for a namespace" in {
+
+ val loadBalancerData = new LoadBalancerData()
+ loadBalancerData.putActivation(firstEntry.id, firstEntry)
+
+ val result = loadBalancerData.activationCountByNamespace
+ result shouldBe Map(firstEntry.namespaceId -> 1)
+ loadBalancerData.activationCountByInvoker("invoker0") shouldBe 1
+ loadBalancerData.activationById(firstEntry.id) shouldBe Some(firstEntry)
+ }
+
+ it should "return the number of activations for each invoker" in {
+
+ val loadBalancerData = new LoadBalancerData()
+ loadBalancerData.putActivation(firstEntry.id, firstEntry)
+ loadBalancerData.putActivation(secondEntry.id, secondEntry)
+
+ val result = loadBalancerData.activationCountByInvoker
+
+ result shouldBe Map(firstEntry.invokerName -> 1, secondEntry.invokerName -> 1)
+
+ loadBalancerData.activationCountByInvoker("invoker0") shouldBe 1
+ loadBalancerData.activationCountByInvoker("invoker1") shouldBe 1
+ loadBalancerData.activationById(firstEntry.id) shouldBe Some(firstEntry)
+ loadBalancerData.activationById(secondEntry.id) shouldBe Some(secondEntry)
+ }
+
+ it should "remove activations from all 3 maps" in {
+
+ val loadBalancerData = new LoadBalancerData()
+ loadBalancerData.putActivation(firstEntry.id, firstEntry)
+ loadBalancerData.putActivation(secondEntry.id, secondEntry)
+
+ loadBalancerData.removeActivation(firstEntry)
+ loadBalancerData.removeActivation(secondEntry)
+
+ val activationsByInvoker = loadBalancerData.activationCountByInvoker
+ val activationsByNamespace = loadBalancerData.activationCountByNamespace
+
+ activationsByInvoker.values.sum shouldBe 0
+ activationsByNamespace.values.sum shouldBe 0
+ loadBalancerData.activationById(firstEntry.id) shouldBe None
+ loadBalancerData.activationById(secondEntry.id) shouldBe None
+ }
+
+ it should "remove activations from all 3 maps by activation id" in {
+
+ val loadBalancerData = new LoadBalancerData()
+ loadBalancerData.putActivation(firstEntry.id, firstEntry)
+
+ loadBalancerData.removeActivation(firstEntry.id)
+
+ val activationsByInvoker = loadBalancerData.activationCountByInvoker
+ val activationsByNamespace = loadBalancerData.activationCountByNamespace
+
+ activationsByInvoker.values.sum shouldBe 0
+ activationsByNamespace.values.sum shouldBe 0
+ loadBalancerData.activationById(firstEntry.id) shouldBe None
+
+ }
+
+ it should "return None if the entry doesn't exist when we remove it" in {
+ val loadBalancerData = new LoadBalancerData()
+ loadBalancerData.removeActivation(firstEntry) shouldBe None
+ }
+
+ it should "respond with different values accordingly" in {
+
+ val entry = ActivationEntry(ActivationId(), UUID(), "invoker1", activationIdPromise)
+ val entrySameInvokerAndNamespace = entry.copy(id = ActivationId())
+ val entrySameInvoker = entry.copy(id = ActivationId(), namespaceId = UUID())
+
+ val loadBalancerData = new LoadBalancerData()
+ loadBalancerData.putActivation(entry.id, entry)
+ loadBalancerData.activationCountByNamespace(entry.namespaceId) shouldBe 1
+ loadBalancerData.activationCountByInvoker(entry.invokerName) shouldBe 1
+
+ loadBalancerData.putActivation(entrySameInvokerAndNamespace.id, entrySameInvokerAndNamespace)
+ loadBalancerData.activationCountByNamespace(entry.namespaceId) shouldBe 2
+ loadBalancerData.activationCountByInvoker(entry.invokerName) shouldBe 2
+
+ loadBalancerData.putActivation(entrySameInvoker.id, entrySameInvoker)
+ loadBalancerData.activationCountByNamespace(entry.namespaceId) shouldBe 2
+ loadBalancerData.activationCountByInvoker(entry.invokerName) shouldBe 3
+
+ loadBalancerData.removeActivation(entrySameInvokerAndNamespace)
+ loadBalancerData.activationCountByNamespace(entry.namespaceId) shouldBe 1
+ loadBalancerData.activationCountByInvoker(entry.invokerName) shouldBe 2
+
+ // removing non existing entry doesn't mess up
+ loadBalancerData.removeActivation(entrySameInvokerAndNamespace)
+ loadBalancerData.activationCountByNamespace(entry.namespaceId) shouldBe 1
+ loadBalancerData.activationCountByInvoker(entry.invokerName) shouldBe 2
+
+ }
+
+ it should "not add the same entry more then once" in {
+
+ val loadBalancerData = new LoadBalancerData()
+
+ loadBalancerData.putActivation(firstEntry.id, firstEntry)
+ loadBalancerData.activationCountByInvoker shouldBe Map(firstEntry.invokerName -> 1)
+ loadBalancerData.activationCountByNamespace shouldBe Map(firstEntry.namespaceId -> 1)
+
+ loadBalancerData.putActivation(firstEntry.id, firstEntry)
+ loadBalancerData.activationCountByInvoker shouldBe Map(firstEntry.invokerName -> 1)
+ loadBalancerData.activationCountByNamespace shouldBe Map(firstEntry.namespaceId -> 1)
+ }
+
+ it should "not evaluate the given block if an entry already exists" in {
+
+ val loadBalancerData = new LoadBalancerData()
+ var called = 0
+
+ val entry = loadBalancerData.putActivation(firstEntry.id, {
+ called += 1
+ firstEntry
+ })
+
+ called shouldBe 1
+
+ // entry already exists, should not evaluate the block
+ val entryAfterSecond = loadBalancerData.putActivation(firstEntry.id, {
+ called += 1
+ firstEntry
+ })
+
+ called shouldBe 1
+ entry shouldBe entryAfterSecond
+ }
+
+ it should "not evaluate the given block even if an entry is different (but has the same id)" in {
+
+ val loadBalancerData = new LoadBalancerData()
+ var called = 0
+ val entrySameId = secondEntry.copy(id = firstEntry.id)
+
+ val entry = loadBalancerData.putActivation(firstEntry.id, {
+ called += 1
+ firstEntry
+ })
+
+ called shouldBe 1
+ loadBalancerData.activationCountByInvoker shouldBe Map(firstEntry.invokerName -> 1)
+ loadBalancerData.activationCountByNamespace shouldBe Map(firstEntry.namespaceId -> 1)
+
+ // entry already exists, should not evaluate the block and change the state
+ val entryAfterSecond = loadBalancerData.putActivation(entrySameId.id, {
+ called += 1
+ entrySameId
+ })
+
+ called shouldBe 1
+ entry shouldBe entryAfterSecond
+ loadBalancerData.activationCountByInvoker shouldBe Map(firstEntry.invokerName -> 1)
+ loadBalancerData.activationCountByNamespace shouldBe Map(firstEntry.namespaceId -> 1)
+ }
+
+}
--
To stop receiving notification emails like this one, please contact
['"commits@openwhisk.apache.org" <co...@openwhisk.apache.org>'].