<commit_before>package akka.dispatch.verification

import akka.actor.ActorCell
import akka.actor.Cell
import akka.actor.ActorSystem
import akka.actor.ActorRef
import akka.actor.ActorPath
import akka.pattern.PromiseActorRef
import akka.actor.Actor
import akka.actor.PoisonPill
import akka.actor.Props
import akka.actor.Cancellable
import akka.actor.RootActorPath
import akka.actor.Address
import akka.cluster.VectorClock
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.Semaphore
import java.io.Closeable

import com.typesafe.config.Config

import akka.dispatch.Envelope
import akka.dispatch.MessageQueue
import akka.dispatch.MessageDispatcher

import scala.collection.concurrent.TrieMap
import scala.collection.mutable.Queue
import scala.collection.mutable.Stack
import scala.collection.mutable.HashMap
import scala.collection.mutable.HashSet

import scala.util.Random
import scala.util.control.Breaks

import java.nio.charset.StandardCharsets

import org.slf4j.LoggerFactory,
       ch.qos.logback.classic.Level,
       ch.qos.logback.classic.Logger


// Wrap cancellable so we get notified about cancellation
class WrappedCancellable (c: Cancellable, rcv: ActorRef, msg: Any) extends Cancellable {
  val instrumenter = Instrumenter()
  def cancel(): Boolean = {
    val success = c.cancel
    instrumenter.cancelTimer(this, rcv, msg, success)
    return success
  }

  def isCancelled: Boolean = c.isCancelled
}

class InstrumenterCheckpoint(
  val actorMappings : HashMap[String, ActorRef],
  val seenActors : HashSet[(ActorSystem, Any)],
  val allowedEvents: HashSet[(ActorCell, Envelope)],
  val dispatchers : HashMap[ActorRef, MessageDispatcher],
  val vectorClocks : HashMap[String, VectorClock],
  val sys: ActorSystem,
  var cancellableToTimer : HashMap[Cancellable, Tuple2[String, Any]],
  var ongoingCancellableTasks : HashSet[Cancellable],
  var timerToCancellable : HashMap[Tuple2[String,Any], Cancellable],
  // TODO(cs): add the random associated with this actor system
  val applicationCheckpoint: Any
) {}

class Instrumenter {
  // Provides the application a hook to compute after each shutdown
  type ShutdownCallback = () => Unit
  // Checkpoints the application state (outside the actor system)
  type CheckpointCallback = () => Any
  // Restores the application state.
  type RestoreCheckpointCallback = (Any) => Unit

  var scheduler : Scheduler = new NullScheduler
  var tellEnqueue : TellEnqueue = new TellEnqueueSemaphore
  
  val dispatchers = new HashMap[ActorRef, MessageDispatcher]
  
  val allowedEvents = new HashSet[(ActorCell, Envelope)]  
  
  val seenActors = new HashSet[(ActorSystem, Any)]
  val actorMappings = new HashMap[String, ActorRef]
  
  // Track the executing context (i.e., source of events)
  var currentActor = ""
  var inActor = false
  var counter = 0   
  // Whether we're currently dispatching.
  var started = new AtomicBoolean(false)
  // Whether to ignore (pass-through) all events
  var _passThrough = new AtomicBoolean(false)
  // Whether to stop dispatch at the next afterMessageReceive()
  var stopDispatch = new AtomicBoolean(false)
  // If an actor blocks while it is `receive`ing, we need to continue making
  // progress by finding a new message to schedule. While that actor is
  // blocked, we should not deliver any messages to it. This data structure
  // tracks which actors are currently blocked.
  // For a detailed design doc see:
  // https://docs.google.com/document/d/1RnCDOQFLa2prliF5y5VDNcdGDmEeOlgUcXSNk7abpSo
  var blockedActors = Set[String]()
  // Mapping from temp actors (used for `ask`) to the name of the actor that created them.
  // For a detailed design doc see:
  // https://docs.google.com/document/d/1_LUceHvQoamlBtNNbqA4CxBH-zvUKlZhnSjLwTc16q4
  var tempToParent = new HashMap[ActorPath, String]
  // After an answer has been sent but before it has been scheduled for
  // delivery, we store the temp actor (receipient) here to mark it as
  // pending.
  var askAnswerNotYetScheduled = new HashSet[PromiseActorRef]
  // Set to true an when external thread signals that it wants to send
  // messages in a thread-safe manner (rather than having its sends delayed)
  var sendingKnownExternalMessages = new AtomicBoolean(false)
  var shutdownCallback : ShutdownCallback = () => {}
  var checkpointCallback : CheckpointCallback = () => {null}
  var restoreCheckpointCallback : RestoreCheckpointCallback = (a: Any) => {}
  var registeredCancellableTasks = new HashSet[Cancellable]
  // Mark which of the timers are scheduleOnce vs schedule [ongoing]
  var ongoingCancellableTasks = new HashSet[Cancellable]
  // Track which cancellables correspond to which messages
  var cancellableToTimer = new HashMap[Cancellable, Tuple2[String, Any]]
  // And vice versa
  var timerToCancellable = new HashMap[Tuple2[String,Any], Cancellable]

  val logger = LoggerFactory.getLogger("Instrumenter")

  private[dispatch] def cancelTimer (c: Cancellable, rcv: ActorRef, msg: Any, success: Boolean) = {
    // Need this here since by the time DPORwHeuristics gets here the thing is already canceled
    if (cancellableToTimer contains c) {
      removeCancellable(c)
    }
    if (scheduler != null) {
      scheduler.notify_timer_cancel(rcv, msg)
    }
  }

  // AspectJ runs into initialization problems if a new ActorSystem is created
  // by the constructor. Instead use a getter to create on demand.
  var _actorSystem : ActorSystem = null
  def actorSystem (config:Option[com.typesafe.config.Config]) : ActorSystem = {
    if (_actorSystem == null) {
      config match {
        case Some(c) =>
          _actorSystem = ActorSystem("new-system-" + counter, c)
        case None =>
          _actorSystem = ActorSystem("new-system-" + counter)
      }
      _random = new Random(0)
      counter += 1
    }
    _actorSystem
  }

  def actorSystem () : ActorSystem = {
    return actorSystem(None)
  }

  private[this] var _random = new Random(0)
  def seededRandom() : Random = {
    return _random
  }
 
  
  def await_enqueue() {
    tellEnqueue.await()
  }

  // Whether to ignore (pass-through) all events
  def setPassthrough() {
    _passThrough.set(true)
  }

  def unsetPassthrough() {
    _passThrough.set(false)
  }

  /**
   * When an external thread (one not named .*dispatcher.*) wants to send
   * messages at a known (thread-safe) point, it should invoke this interface.
   *
   * If an external thread sends messages outside of this interface, its
   * messages will not be sent right away; instead they will be passed to
   * scheduler.enqueue_message() to be sent later at known point.
   */
  def sendKnownExternalMessages(sendBlock: () => Any) {
    sendingKnownExternalMessages.set(true)
    sendBlock()
    sendingKnownExternalMessages.set(false)
  }

  /**
   * Rather than having akka assign labels $a, $b, $c, etc to temp actors,
   * assign our own that will be more resilient to divergence in the execution
   * as we replay. See this design doc for more details:
   *   https://docs.google.com/document/d/1rAM8EEy3WnLRhhPROvHmBhAREv0rmihz0Gw0GgF1xC4
   */
  def assignTempPath(tempPath: ActorPath): ActorPath = {
    val callStack = Thread.currentThread().getStackTrace().drop(14) // drop off common prefix
    val min3 = math.min(3, callStack.length)
    val truncated = callStack.take(min3)
    val bytes = truncated.mkString("-").getBytes(StandardCharsets.UTF_8)
    val b64 = java.util.Base64.getEncoder.encodeToString(bytes)
    val path = tempPath / b64
    return path
  }

  /**
   * Two cases:
   *  - In a normal `tell` from actor to actor, we need to mark
   *    tellEnqueue.tell() to let us know that there should later be a
   *    tellEnqueue.enqueue(). [See TellEnqueue class docs].
   *
   *  - If the thread doing the `tell` is external (outside of akka's thread
   *    pool and any of our schedulers), we can't let them send right
   *    now, since that won't be thread-safe. Instead, send it through
   *    scheduler.enqueue_message.
   */
  def tell(receiver: ActorRef, msg: Any, sender: ActorRef) : Boolean = {
    assert(receiver != null)

    if (_passThrough.get()) {
      return true
    }
<commit_msg>Avoid ambiguity in temp actor names: prepend ID<commit_after>package akka.dispatch.verification

import akka.actor.ActorCell
import akka.actor.Cell
import akka.actor.ActorSystem
import akka.actor.ActorRef
import akka.actor.ActorPath
import akka.pattern.PromiseActorRef
import akka.actor.Actor
import akka.actor.PoisonPill
import akka.actor.Props
import akka.actor.Cancellable
import akka.actor.RootActorPath
import akka.actor.Address
import akka.cluster.VectorClock
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.Semaphore
import java.io.Closeable

import com.typesafe.config.Config

import akka.dispatch.Envelope
import akka.dispatch.MessageQueue
import akka.dispatch.MessageDispatcher

import scala.collection.concurrent.TrieMap
import scala.collection.mutable.Queue
import scala.collection.mutable.Stack
import scala.collection.mutable.HashMap
import scala.collection.mutable.HashSet

import scala.concurrent.Future

import scala.util.Random
import scala.util.control.Breaks

import java.nio.charset.StandardCharsets

import org.slf4j.LoggerFactory,
       ch.qos.logback.classic.Level,
       ch.qos.logback.classic.Logger


// Wrap cancellable so we get notified about cancellation
class WrappedCancellable (c: Cancellable, rcv: ActorRef, msg: Any) extends Cancellable {
  val instrumenter = Instrumenter()
  def cancel(): Boolean = {
    val success = c.cancel
    instrumenter.cancelTimer(this, rcv, msg, success)
    return success
  }

  def isCancelled: Boolean = c.isCancelled
}

class InstrumenterCheckpoint(
  val actorMappings : HashMap[String, ActorRef],
  val seenActors : HashSet[(ActorSystem, Any)],
  val allowedEvents: HashSet[(ActorCell, Envelope)],
  val dispatchers : HashMap[ActorRef, MessageDispatcher],
  val vectorClocks : HashMap[String, VectorClock],
  val sys: ActorSystem,
  var cancellableToTimer : HashMap[Cancellable, Tuple2[String, Any]],
  var ongoingCancellableTasks : HashSet[Cancellable],
  var timerToCancellable : HashMap[Tuple2[String,Any], Cancellable],
  // TODO(cs): add the random associated with this actor system
  val applicationCheckpoint: Any
) {}

class Instrumenter {
  // Provides the application a hook to compute after each shutdown
  type ShutdownCallback = () => Unit
  // Checkpoints the application state (outside the actor system)
  type CheckpointCallback = () => Any
  // Restores the application state.
  type RestoreCheckpointCallback = (Any) => Unit

  var scheduler : Scheduler = new NullScheduler
  var tellEnqueue : TellEnqueue = new TellEnqueueSemaphore
  
  val dispatchers = new HashMap[ActorRef, MessageDispatcher]
  
  val allowedEvents = new HashSet[(ActorCell, Envelope)]  
  
  val seenActors = new HashSet[(ActorSystem, Any)]
  val actorMappings = new HashMap[String, ActorRef]
  
  // Track the executing context (i.e., source of events)
  var currentActor = ""
  var inActor = false
  var counter = 0   
  // Whether we're currently dispatching.
  var started = new AtomicBoolean(false)
  // Whether to ignore (pass-through) all events
  var _passThrough = new AtomicBoolean(false)
  // Whether to stop dispatch at the next afterMessageReceive()
  var stopDispatch = new AtomicBoolean(false)
  // If an actor blocks while it is `receive`ing, we need to continue making
  // progress by finding a new message to schedule. While that actor is
  // blocked, we should not deliver any messages to it. This data structure
  // tracks which actors are currently blocked.
  // For a detailed design doc see:
  // https://docs.google.com/document/d/1RnCDOQFLa2prliF5y5VDNcdGDmEeOlgUcXSNk7abpSo
  var blockedActors = Set[String]()
  // Mapping from temp actors (used for `ask`) to the name of the actor that created them.
  // For a detailed design doc see:
  // https://docs.google.com/document/d/1_LUceHvQoamlBtNNbqA4CxBH-zvUKlZhnSjLwTc16q4
  var tempToParent = new HashMap[ActorPath, String]
  // After an answer has been sent but before it has been scheduled for
  // delivery, we store the temp actor (receipient) here to mark it as
  // pending.
  var askAnswerNotYetScheduled = new HashSet[PromiseActorRef]
  // Set to true an when external thread signals that it wants to send
  // messages in a thread-safe manner (rather than having its sends delayed)
  var sendingKnownExternalMessages = new AtomicBoolean(false)
  var shutdownCallback : ShutdownCallback = () => {}
  var checkpointCallback : CheckpointCallback = () => {null}
  var restoreCheckpointCallback : RestoreCheckpointCallback = (a: Any) => {}
  var registeredCancellableTasks = new HashSet[Cancellable]
  // Mark which of the timers are scheduleOnce vs schedule [ongoing]
  var ongoingCancellableTasks = new HashSet[Cancellable]
  // Track which cancellables correspond to which messages
  var cancellableToTimer = new HashMap[Cancellable, Tuple2[String, Any]]
  // And vice versa
  var timerToCancellable = new HashMap[Tuple2[String,Any], Cancellable]

  val logger = LoggerFactory.getLogger("Instrumenter")

  private[dispatch] def cancelTimer (c: Cancellable, rcv: ActorRef, msg: Any, success: Boolean) = {
    // Need this here since by the time DPORwHeuristics gets here the thing is already canceled
    if (cancellableToTimer contains c) {
      removeCancellable(c)
    }
    if (scheduler != null) {
      scheduler.notify_timer_cancel(rcv, msg)
    }
  }

  // AspectJ runs into initialization problems if a new ActorSystem is created
  // by the constructor. Instead use a getter to create on demand.
  var _actorSystem : ActorSystem = null
  def actorSystem (config:Option[com.typesafe.config.Config]) : ActorSystem = {
    if (_actorSystem == null) {
      config match {
        case Some(c) =>
          _actorSystem = ActorSystem("new-system-" + counter, c)
        case None =>
          _actorSystem = ActorSystem("new-system-" + counter)
      }
      _random = new Random(0)
      counter += 1
    }
    _actorSystem
  }

  def actorSystem () : ActorSystem = {
    return actorSystem(None)
  }

  private[this] var _random = new Random(0)
  def seededRandom() : Random = {
    return _random
  }
 
  
  def await_enqueue() {
    tellEnqueue.await()
  }

  // Whether to ignore (pass-through) all events
  def setPassthrough() {
    _passThrough.set(true)
  }

  def unsetPassthrough() {
    _passThrough.set(false)
  }

  /**
   * When an external thread (one not named .*dispatcher.*) wants to send
   * messages at a known (thread-safe) point, it should invoke this interface.
   *
   * If an external thread sends messages outside of this interface, its
   * messages will not be sent right away; instead they will be passed to
   * scheduler.enqueue_message() to be sent later at known point.
   */
  def sendKnownExternalMessages(sendBlock: () => Any) {
    sendingKnownExternalMessages.set(true)
    sendBlock()
    sendingKnownExternalMessages.set(false)
  }

  /**
   * Rather than having akka assign labels $a, $b, $c, etc to temp actors,
   * assign our own that will be more resilient to divergence in the execution
   * as we replay. See this design doc for more details:
   *   https://docs.google.com/document/d/1rAM8EEy3WnLRhhPROvHmBhAREv0rmihz0Gw0GgF1xC4
   */
  def assignTempPath(tempPath: ActorPath): ActorPath = synchronized {
    val callStack = Thread.currentThread().getStackTrace().map(e => e.getMethodName).drop(14) // drop off common prefix
    val min3 = math.min(3, callStack.length)
    var truncated = callStack.take(min3).toList
    if (idToPrependToCallStack != "") {
      truncated = idToPrependToCallStack :: truncated
      idToPrependToCallStack = ""
    }
    val bytes = truncated.mkString("-").getBytes(StandardCharsets.UTF_8)
    val b64 = java.util.Base64.getEncoder.encodeToString(bytes)
    val path = tempPath / b64
    return path
  }

  // In cases where there might be multiple threads with the same callstack
  // invoking `ask`, have them call this with an ID to avoid ambiguity in temp
  // actor names. Serialized with a lock on the Instrumenter object.
  var idToPrependToCallStack = ""

  def serializeAskWithID[E](id: String, ask: () => Future[E]) : Future[E] = synchronized {
    assert(idToPrependToCallStack == "")
    idToPrependToCallStack = id
    return ask()
  }

  /**
   * Two cases:
   *  - In a normal `tell` from actor to actor, we need to mark
   *    tellEnqueue.tell() to let us know that there should later be a
   *    tellEnqueue.enqueue(). [See TellEnqueue class docs].
   *
   *  - If the thread doing the `tell` is external (outside of akka's thread
   *    pool and any of our schedulers), we can't let them send right
   *    now, since that won't be thread-safe. Instead, send it through
   *    scheduler.enqueue_message.
   */
  def tell(receiver: ActorRef, msg: Any, sender: ActorRef) : Boolean = {
    assert(receiver != null)

    if (_passThrough.get()) {
      return true
    }

{'fraction_non_alphanumeric': 0.08805839416058395, 'fraction_numerical': 0.004554744525547445, 'mean_word_length': 3.7611898804559356, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, 'https://': 6, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '520711', 'n_tokens_mistral': 5264, 'n_tokens_neox': 5081, 'n_words': 1621}