openwhisk-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From markusthoem...@apache.org
Subject [incubator-openwhisk] branch master updated: Actor state safety improvements. (#2802)
Date Wed, 08 Nov 2017 16:23:29 GMT
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 b3fb4b7  Actor state safety improvements. (#2802)
b3fb4b7 is described below

commit b3fb4b78f1e120fac1c9891ae587d9157f1b4b7b
Author: Brendan McAdams <brendan@bytes.codes>
AuthorDate: Wed Nov 8 08:23:26 2017 -0800

    Actor state safety improvements. (#2802)
    
    * Move MessageConsumer internal `outstandingMessages` Queue to immutable
    - There's a risk with a mutable collection in actor state of leaking
      outside the actor (and then mutated outside) if sent in
      a message accidentally
    - Instead, a mutable var (always protected by actor isolation) pointing
      to an immutable collection provides better actor safety as a
      best practice
    
    * Move InvokerPool Actor internal instance tracking collections to immutable from mutable
      - As these collection items were already marked as risky if closed over, this cleanup
        should obviate any risk of that leakage (including being messaged out of the stack)
      - Change to immutable maps required a small refactor of `PingMessage` receive branch
    
    * Minor style fix to use foreach instead of map where unit was being returned in the map
    
    * Obvious cleanup/optimization of my last revision of PingMessage branch
    
    * In ActivationFinisher Actor, replace mutable state collection with immutable
    
    * ContainerPool Actor immutability cleanup
      - migrate mutable free/busy/prewarmed pools to immutable maps
      - should prevent accidental state leakage via closures and messages
    
    * Spacing fix for style guidelines
    
    * Remove unnecessary usage of "new" to instantiate a case class
---
 .../whisk/core/connector/MessageConsumer.scala     | 20 +++++++---
 .../core/controller/actions/PrimitiveActions.scala |  9 ++---
 .../core/loadBalancer/InvokerSupervision.scala     | 45 +++++++++++++---------
 .../whisk/core/containerpool/ContainerPool.scala   | 45 ++++++++++++----------
 4 files changed, 69 insertions(+), 50 deletions(-)

diff --git a/common/scala/src/main/scala/whisk/core/connector/MessageConsumer.scala b/common/scala/src/main/scala/whisk/core/connector/MessageConsumer.scala
index 31fcea0..5197767 100644
--- a/common/scala/src/main/scala/whisk/core/connector/MessageConsumer.scala
+++ b/common/scala/src/main/scala/whisk/core/connector/MessageConsumer.scala
@@ -18,14 +18,12 @@
 package whisk.core.connector
 
 import scala.annotation.tailrec
-import scala.collection.mutable
+import scala.collection.immutable
 import scala.concurrent.Future
 import scala.concurrent.blocking
 import scala.concurrent.duration._
 import scala.util.Failure
-
 import org.apache.kafka.clients.consumer.CommitFailedException
-
 import akka.actor.FSM
 import akka.pattern.pipe
 import whisk.common.Logging
@@ -111,7 +109,13 @@ class MessageFeed(description: String,
     consumer.maxPeek <= maxPipelineDepth,
     "consumer may not yield more messages per peek than permitted by max depth")
 
-  private val outstandingMessages = mutable.Queue[(String, Int, Long, Array[Byte])]()
+  // Immutable Queue
+  // although on the surface it seems to make sense to use an immutable variable with a mutable
Queue,
+  // Akka Actor state defies the usual "prefer immutable" guideline in Scala, esp. w/ Collections.
+  // If, for some reason, this Queue was mutable and is accidentally leaked in say an Akka
message,
+  // another Actor or recipient would be able to mutate the internal state of this Actor.
+  // Best practice dictates a mutable variable pointing at an immutable collection for this
reason
+  private var outstandingMessages = immutable.Queue.empty[(String, Int, Long, Array[Byte])]
   private var handlerCapacity = maximumHandlerCapacity
 
   private implicit val tid = TransactionId.dispatcher
@@ -137,7 +141,7 @@ class MessageFeed(description: String,
       stay
 
     case Event(FillCompleted(messages), _) =>
-      outstandingMessages.enqueue(messages: _*)
+      outstandingMessages = outstandingMessages ++ messages
       sendOutstandingMessages()
 
       if (shouldFillQueue()) {
@@ -202,7 +206,11 @@ class MessageFeed(description: String,
   private def sendOutstandingMessages(): Unit = {
     val occupancy = outstandingMessages.size
     if (occupancy > 0 && handlerCapacity > 0) {
-      val (topic, partition, offset, bytes) = outstandingMessages.dequeue()
+      // Easiest way with an immutable queue to cleanly dequeue
+      // Head is the first elemeent of the queue, desugared w/ an assignment pattern
+      // Tail is everything but the first element, thus mutating the collection variable
+      val (topic, partition, offset, bytes) = outstandingMessages.head
+      outstandingMessages = outstandingMessages.tail
 
       if (logHandoff) logging.info(this, s"processing $topic[$partition][$offset] ($occupancy/$handlerCapacity)")
       handler(bytes)
diff --git a/core/controller/src/main/scala/whisk/core/controller/actions/PrimitiveActions.scala
b/core/controller/src/main/scala/whisk/core/controller/actions/PrimitiveActions.scala
index f159192..2a9fa1f 100644
--- a/core/controller/src/main/scala/whisk/core/controller/actions/PrimitiveActions.scala
+++ b/core/controller/src/main/scala/whisk/core/controller/actions/PrimitiveActions.scala
@@ -17,7 +17,6 @@
 
 package whisk.core.controller.actions
 
-import scala.collection.mutable.Buffer
 import scala.concurrent.ExecutionContext
 import scala.concurrent.Future
 import scala.concurrent.Promise
@@ -258,7 +257,7 @@ protected[actions] object ActivationFinisher {
     // when the future completes, self-destruct
     promise.future.andThen { case _ => shutdown() }
 
-    val preemptiveMsgs: Buffer[Cancellable] = Buffer.empty
+    var preemptiveMsgs = Vector.empty[Cancellable]
 
     def receive = {
       case ActivationFinisher.Finish(activation) =>
@@ -267,13 +266,13 @@ protected[actions] object ActivationFinisher {
       case msg @ Scheduler.WorkOnceNow =>
         // try up to three times when pre-emptying the schedule
         fastPollPeriods.foreach { s =>
-          preemptiveMsgs += context.system.scheduler.scheduleOnce(s, poller, msg)
+          preemptiveMsgs = preemptiveMsgs :+ context.system.scheduler.scheduleOnce(s, poller,
msg)
         }
     }
 
     def shutdown(): Unit = {
       preemptiveMsgs.foreach(_.cancel())
-      preemptiveMsgs.clear()
+      preemptiveMsgs = Vector.empty
       context.stop(poller)
       context.stop(self)
     }
@@ -281,7 +280,7 @@ protected[actions] object ActivationFinisher {
     override def postStop() = {
       logging.info(this, "finisher shutdown")
       preemptiveMsgs.foreach(_.cancel())
-      preemptiveMsgs.clear()
+      preemptiveMsgs = Vector.empty
       context.stop(poller)
     }
   }
diff --git a/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
b/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
index 13517de..d188ce4 100644
--- a/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
+++ b/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
@@ -19,7 +19,7 @@ package whisk.core.loadBalancer
 
 import java.nio.charset.StandardCharsets
 
-import scala.collection.mutable
+import scala.collection.immutable
 import scala.concurrent.Future
 import scala.concurrent.duration._
 import scala.util.Failure
@@ -82,33 +82,24 @@ class InvokerPool(childFactory: (ActorRefFactory, InstanceId) => ActorRef,
   implicit val timeout = Timeout(5.seconds)
   implicit val ec = context.dispatcher
 
-  // State of the actor. It's important not to close over these
-  // references directly, so they don't escape the Actor.
-  val instanceToRef = mutable.Map[InstanceId, ActorRef]()
-  val refToInstance = mutable.Map[ActorRef, InstanceId]()
+  // State of the actor. Mutable vars with immutable collections prevents closures or messages
+  // from leaking the state for external mutation
+  var instanceToRef = immutable.Map.empty[InstanceId, ActorRef]
+  var refToInstance = immutable.Map.empty[ActorRef, InstanceId]
   var status = IndexedSeq[(InstanceId, InvokerState)]()
 
   def receive = {
     case p: PingMessage =>
-      val invoker = instanceToRef.getOrElseUpdate(p.instance, {
-        logging.info(this, s"registered a new invoker: invoker${p.instance.toInt}")(TransactionId.invokerHealth)
+      val invoker = instanceToRef.getOrElse(p.instance, registerInvoker(p.instance))
+      instanceToRef = instanceToRef.updated(p.instance, invoker)
 
-        status = padToIndexed(status, p.instance.toInt + 1, i => (InstanceId(i), Offline))
-
-        val ref = childFactory(context, p.instance)
-        ref ! SubscribeTransitionCallBack(self) // register for state change events
-
-        refToInstance.update(ref, p.instance)
-        ref
-      })
       invoker.forward(p)
 
     case GetStatus => sender() ! status
 
-    case msg: InvocationFinishedMessage => {
+    case msg: InvocationFinishedMessage =>
       // Forward message to invoker, if InvokerActor exists
-      instanceToRef.get(msg.invokerInstance).map(_.forward(msg))
-    }
+      instanceToRef.get(msg.invokerInstance).foreach(_.forward(msg))
 
     case CurrentState(invoker, currentState: InvokerState) =>
       refToInstance.get(invoker).foreach { instance =>
@@ -159,6 +150,22 @@ class InvokerPool(childFactory: (ActorRefFactory, InstanceId) => ActorRef,
 
   /** Pads a list to a given length using the given function to compute entries */
   def padToIndexed[A](list: IndexedSeq[A], n: Int, f: (Int) => A) = list ++ (list.size
until n).map(f)
+
+  // Register a new invoker
+  def registerInvoker(instanceId: InstanceId): ActorRef = {
+    logging.info(this, s"registered a new invoker: invoker${instanceId.toInt}")(TransactionId.invokerHealth)
+
+    status = padToIndexed(status, instanceId.toInt + 1, i => (InstanceId(i), Offline))
+
+    val ref = childFactory(context, instanceId)
+
+    ref ! SubscribeTransitionCallBack(self) // register for state change events
+
+    refToInstance = refToInstance.updated(ref, instanceId)
+
+    ref
+  }
+
 }
 
 object InvokerPool {
@@ -179,7 +186,7 @@ object InvokerPool {
     new WhiskAction(
       namespace = healthActionIdentity.namespace.toPath,
       name = EntityName(s"invokerHealthTestAction${i.toInt}"),
-      exec = new CodeExecAsString(manifest, """function main(params) { return params; }""",
None))
+      exec = CodeExecAsString(manifest, """function main(params) { return params; }""", None))
   }
 }
 
diff --git a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
index 22b2b12..b02f528 100644
--- a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
+++ b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
@@ -17,7 +17,7 @@
 
 package whisk.core.containerpool
 
-import scala.collection.mutable
+import scala.collection.immutable
 
 import akka.actor.Actor
 import akka.actor.ActorRef
@@ -67,9 +67,9 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
     extends Actor {
   implicit val logging = new AkkaLogging(context.system.log)
 
-  val freePool = mutable.Map[ActorRef, ContainerData]()
-  val busyPool = mutable.Map[ActorRef, ContainerData]()
-  val prewarmedPool = mutable.Map[ActorRef, ContainerData]()
+  var freePool = immutable.Map.empty[ActorRef, ContainerData]
+  var busyPool = immutable.Map.empty[ActorRef, ContainerData]
+  var prewarmedPool = immutable.Map.empty[ActorRef, ContainerData]
 
   prewarmConfig.foreach { config =>
     logging.info(this, s"pre-warming ${config.count} ${config.exec.kind} containers")
@@ -84,7 +84,7 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
       val container = if (busyPool.size < maxActiveContainers) {
         // Schedule a job to a warm container
         ContainerPool
-          .schedule(r.action, r.msg.user.namespace, freePool.toMap)
+          .schedule(r.action, r.msg.user.namespace, freePool)
           .orElse {
             if (busyPool.size + freePool.size < maxPoolSize) {
               takePrewarmContainer(r.action).orElse {
@@ -94,7 +94,7 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
           }
           .orElse {
             // Remove a container and create a new one for the given job
-            ContainerPool.remove(r.action, r.msg.user.namespace, freePool.toMap).map { toDelete
=>
+            ContainerPool.remove(r.action, r.msg.user.namespace, freePool).map { toDelete
=>
               removeContainer(toDelete)
               takePrewarmContainer(r.action).getOrElse {
                 createContainer()
@@ -105,8 +105,8 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
 
       container match {
         case Some((actor, data)) =>
-          busyPool.update(actor, data)
-          freePool.remove(actor)
+          busyPool = busyPool + (actor -> data)
+          freePool = freePool - actor
           actor ! r // forwards the run request to the container
         case None =>
           logging.error(this, "Rescheduling Run message, too many message in the pool")(r.msg.transid)
@@ -115,26 +115,31 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
 
     // Container is free to take more work
     case NeedWork(data: WarmedData) =>
-      freePool.update(sender(), data)
-      busyPool.remove(sender()).foreach(_ => feed ! MessageFeed.Processed)
+      freePool = freePool + (sender() -> data)
+      busyPool.get(sender()).foreach { _ =>
+        busyPool = busyPool - sender()
+        feed ! MessageFeed.Processed
+      }
 
     // Container is prewarmed and ready to take work
     case NeedWork(data: PreWarmedData) =>
-      prewarmedPool.update(sender(), data)
+      prewarmedPool = prewarmedPool + (sender() -> data)
 
     // Container got removed
     case ContainerRemoved =>
-      freePool.remove(sender())
-      busyPool.remove(sender()).foreach(_ => feed ! MessageFeed.Processed)
+      freePool = freePool - sender()
+      busyPool.get(sender()).foreach { _ =>
+        busyPool = busyPool - sender()
+        feed ! MessageFeed.Processed
+      }
   }
 
   /** Creates a new container and updates state accordingly. */
   def createContainer(): (ActorRef, ContainerData) = {
     val ref = childFactory(context)
     val data = NoData()
-    freePool.update(ref, data)
-
-    (ref, data)
+    freePool = freePool + (ref -> data)
+    ref -> data
   }
 
   /** Creates a new prewarmed container */
@@ -160,8 +165,8 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
         .map {
           case (ref, data) =>
             // Move the container to the usual pool
-            freePool.update(ref, data)
-            prewarmedPool.remove(ref)
+            freePool = freePool + (ref -> data)
+            prewarmedPool = prewarmedPool - ref
             // Create a new prewarm container
             prewarmContainer(config.exec, config.memoryLimit)
 
@@ -172,8 +177,8 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
   /** Removes a container and updates state accordingly. */
   def removeContainer(toDelete: ActorRef) = {
     toDelete ! Remove
-    freePool.remove(toDelete)
-    busyPool.remove(toDelete)
+    freePool = freePool - toDelete
+    busyPool = busyPool - toDelete
   }
 }
 

-- 
To stop receiving notification emails like this one, please contact
['"commits@openwhisk.apache.org" <commits@openwhisk.apache.org>'].

Mime
View raw message