<commit_before>//
// CounterProxy.scala -- Scala trait CounterProxyManager, and classes RemoteCounterProxy and RemoteCounterMembersProxy
// Project PorcE
//
// Created by jthywiss on Aug 15, 2017.
//
// Copyright (c) 2017 The University of Texas at Austin. All rights reserved.
//
// Use and redistribution of this file is governed by the license terms in
// the LICENSE file found in the project's top-level directory and also found at
// URL: http://orc.csres.utexas.edu/license.shtml .
//

package orc.run.porce.distrib

import java.util.concurrent.atomic.AtomicInteger

import orc.Schedulable
import orc.run.porce.runtime.Counter

/** A DOrcExecution mix-in to create and communicate among proxied counters.
  *
  * @see CounterProxyManager#RemoteCounterProxy
  * @see CounterProxyManager#RemoteCounterMembersProxy
  * @author jthywiss
  */
trait CounterProxyManager {
  execution: DOrcExecution =>

  type CounterProxyId = RemoteRefId

  /** Proxy for a counter the resides on a remote dOrc node.
    * RemoteCounterProxy is created locally when a token has been migrated from
    * another node. When all local counter members halt, the remote
    * counter will be notified.
    *
    * This counter can maintain multiple parent tokens.
    *
    * @author jthywiss
    */
  class RemoteCounterProxy private[CounterProxyManager] (
      val remoteProxyId: CounterProxyManager#CounterProxyId,
      onHaltFunc: (Int) => Unit,
      onDiscorporateFunc: (Int) => Unit,
      onResurrectFunc: () => Unit)
    extends Counter(0) {
    // Proxy counters start "floating" in a non-halted, but 0 count situation.
    val parentTokens = new AtomicInteger(0)

    override def onHalt(): Unit = {
      //Logger.entering(getClass.getName, "onHalt")
      val n = parentTokens.getAndSet(0)
      assert(n > 0)
      if (isDiscorporated) {
        onDiscorporateFunc(n)
      } else {
        onHaltFunc(n)
      }
    }

    override def onResurrect(): Unit = {
      //Logger.entering(getClass.getName, "onResurrect")
      onResurrectFunc()
    }

    override def toString: String = f"${getClass.getName}(remoteProxyId=$remoteProxyId%#x)"

    /** Take a parent token and add it to this proxy.
     *
     *  The parent token will be handled through it's halting by the proxy. The
     *  caller to this method gets a token on this proxy.
     *
     *  This call will never require a newToken from the parent. Instead, the
     *  caller must provide that token.
     */
    final def takeParentToken(): Unit = {
      val n = incrementAndGet()
      logChange(s"takeParentToken to $n")
      parentTokens.getAndIncrement()
    }
  }

  /** Proxy for one or more remote members of a counter (tokens and counters).
    * RemoteCounterMembersProxy is created when a token is migrated to another node.
    * As the migrated token continues to execute, this proxy may come to represent
    * multiple tokens, and/or sub-counters.  When all remote members halt, this
    * proxy will remove itself from the parent.
    *
    * @author jthywiss
    */
  final class RemoteCounterMembersProxy private[CounterProxyManager] (
      val thisProxyId: CounterProxyManager#CounterProxyId,
      val enclosingCounter: Counter) {
    // TODO: SAFETY: Are all calls into methods on this object guaranteed to be from the same receiver thread.
    private var alive = true
    private var discorporated = false

    /** Remote counter members all halted, so notify parent that we've halted. */
    def notifyParentOfHalt(n: Int): Unit = synchronized {
      Logger.entering(getClass.getName, s"halt $n")
      if (alive) {
        alive = false
        (0 until n).foreach { (_) =>
          enclosingCounter.haltToken()
        }
      }
    }

    /** Remote counter members all halted, but there was discorporated members, so so notify parent that we've discorporated. */
    def notifyParentOfDiscorporate(n: Int): Unit = synchronized {
      Logger.entering(getClass.getName, s"discorporate $n")
      if (!discorporated && alive) {
        discorporated = true
        (0 until n).foreach { (_) =>
          enclosingCounter.discorporateToken()
        }
      } else {
        throw new IllegalStateException(s"RemoteCounterMembersProxy.notifyParentOfDiscorporate when alive=$alive, discorporated=$discorporated")
      }
    }

    /** Remote counter had only discorporated members, but now has a new member, so notify parent that we've resurrected. */
    def notifyParentOfResurrect(): Unit = synchronized {
      //Logger.entering(getClass.getName, "resurrect")
      if (discorporated && alive) {
        discorporated = false
        enclosingCounter.newToken()
      } else {
        throw new IllegalStateException(s"RemoteCounterMembersProxy.notifyParentOfResurrect when alive=$alive, discorporated=$discorporated")
      }
    }

  }

  protected val proxiedCountersById = new java.util.concurrent.ConcurrentHashMap[Counter, CounterProxyId]
  protected val proxiedCounters = new java.util.concurrent.ConcurrentHashMap[CounterProxyId, RemoteCounterProxy]
  protected val proxiedCounterMembers = new java.util.concurrent.ConcurrentHashMap[CounterProxyId, RemoteCounterMembersProxy]

  def makeProxyWithinCounter(enclosingCounter: Counter): CounterProxyId = {
    val counterProxyId = enclosingCounter match {
      case cp: RemoteCounterProxy => cp.remoteProxyId
      case _ => proxiedCountersById.computeIfAbsent(enclosingCounter, (_) => freshRemoteRefId())
    }
    proxiedCounterMembers.computeIfAbsent(counterProxyId, (_) => {
      val rmtCounterMbrProxy = new RemoteCounterMembersProxy(counterProxyId, enclosingCounter)
      Logger.finer(f"Created proxy for $enclosingCounter with id $counterProxyId")
      rmtCounterMbrProxy
    })

    /* Token: enclosingCounter value is correct as is, since we're swapping this "token" with its proxy. */

    counterProxyId
  }

  def makeProxyCounterFor(counterProxyId: CounterProxyId, origin: PeerLocation): RemoteCounterProxy = {
    def newProxyCounter =
      proxiedCounters.computeIfAbsent(counterProxyId, (_) =>
        new RemoteCounterProxy(counterProxyId,
          (n) => execution.sendHalt(origin, counterProxyId)(n),
          (n) => sendDiscorporate(origin, counterProxyId)(n),
          () => sendResurrect(origin, counterProxyId)()))
    newProxyCounter

    // FIXME: Reenable this optimization. It's probably important. However it will require figuring out both how to
    // make sure things are safe (in terms of message orderings) and how to handle the case where a proxy is passed
    // here when we actually have the counter. This may require the sender knowing if this will happen.
    /*<commit_msg>PorcE: Allow members proxy to halt more than once.<commit_after>//
// CounterProxy.scala -- Scala trait CounterProxyManager, and classes RemoteCounterProxy and RemoteCounterMembersProxy
// Project PorcE
//
// Created by jthywiss on Aug 15, 2017.
//
// Copyright (c) 2017 The University of Texas at Austin. All rights reserved.
//
// Use and redistribution of this file is governed by the license terms in
// the LICENSE file found in the project's top-level directory and also found at
// URL: http://orc.csres.utexas.edu/license.shtml .
//

package orc.run.porce.distrib

import java.util.concurrent.atomic.AtomicInteger

import orc.Schedulable
import orc.run.porce.runtime.Counter

/** A DOrcExecution mix-in to create and communicate among proxied counters.
  *
  * @see CounterProxyManager#RemoteCounterProxy
  * @see CounterProxyManager#RemoteCounterMembersProxy
  * @author jthywiss
  */
trait CounterProxyManager {
  execution: DOrcExecution =>

  type CounterProxyId = RemoteRefId

  /** Proxy for a counter the resides on a remote dOrc node.
    * RemoteCounterProxy is created locally when a token has been migrated from
    * another node. When all local counter members halt, the remote
    * counter will be notified.
    *
    * This counter can maintain multiple parent tokens.
    *
    * @author jthywiss
    */
  class RemoteCounterProxy private[CounterProxyManager] (
      val remoteProxyId: CounterProxyManager#CounterProxyId,
      onHaltFunc: (Int) => Unit,
      onDiscorporateFunc: (Int) => Unit,
      onResurrectFunc: () => Unit)
    extends Counter(0) {
    // Proxy counters start "floating" in a non-halted, but 0 count situation.
    val parentTokens = new AtomicInteger(0)

    override def onHalt(): Unit = {
      //Logger.entering(getClass.getName, "onHalt")
      val n = parentTokens.getAndSet(0)
      logChange(s"onHalt from $n (at ${get()})")
      assert(n > 0)
      if (isDiscorporated) {
        onDiscorporateFunc(n)
      } else {
        onHaltFunc(n)
      }
    }

    override def onResurrect(): Unit = {
      //Logger.entering(getClass.getName, "onResurrect")
      onResurrectFunc()
    }

    override def toString: String = f"${getClass.getName}(remoteProxyId=$remoteProxyId%#x)"

    /** Take a parent token and add it to this proxy.
     *
     *  The parent token will be handled through it's halting by the proxy. The
     *  caller to this method gets a token on this proxy.
     *
     *  This call will never require a newToken from the parent. Instead, the
     *  caller must provide that token.
     */
    final def takeParentToken(): Unit = {
      val n = incrementAndGet()
      logChange(s"takeParentToken to $n")
      parentTokens.getAndIncrement()
    }
  }

  /** Proxy for one or more remote members of a counter (tokens and counters).
    * RemoteCounterMembersProxy is created when a token is migrated to another node.
    * As the migrated token continues to execute, this proxy may come to represent
    * multiple tokens, and/or sub-counters.  When all remote members halt, this
    * proxy will remove itself from the parent.
    * 
    * This proxy can be reused and holds no state about whether it is no longer 
    * needed.
    *
    * @author jthywiss, amp
    */
  final class RemoteCounterMembersProxy private[CounterProxyManager] (
      val thisProxyId: CounterProxyManager#CounterProxyId,
      val enclosingCounter: Counter) {
    // TODO: SAFETY: Are all calls into methods on this object guaranteed to be from the same receiver thread.
    //private var parentTokens = 0
    // TODO: MEMORYLEAK: The lack of any local information on if this members proxy is done will make removing proxies from the maps very hard.

    /** Remote counter members all halted, so notify parent that we've halted. */
    def notifyParentOfHalt(n: Int): Unit = synchronized {
      Logger.entering(getClass.getName, s"halt $n")
      //assert(parentTokens >= n)
      //parentTokens -= n
      (0 until n).foreach { (_) =>
        enclosingCounter.haltToken()
      }
    }

    /** Remote counter members all halted, but there was discorporated members, so so notify parent that we've discorporated. */
    def notifyParentOfDiscorporate(n: Int): Unit = synchronized {
      Logger.entering(getClass.getName, s"discorporate $n")
      //assert(parentTokens >= n)
      //parentTokens -= n
      (0 until n).foreach { (_) =>
        enclosingCounter.discorporateToken()
      }
    }
    
    def takeParentToken() = {
      Logger.entering(getClass.getName, s"takeParentToken")
      //parentTokens += 1      
    }

    /** Remote counter had only discorporated members, but now has a new member, so notify parent that we've resurrected. */
    def notifyParentOfResurrect(): Unit = synchronized {
      //Logger.entering(getClass.getName, "resurrect")
      enclosingCounter.newToken()
    }

  }

  protected val proxiedCountersById = new java.util.concurrent.ConcurrentHashMap[Counter, CounterProxyId]
  protected val proxiedCounters = new java.util.concurrent.ConcurrentHashMap[CounterProxyId, RemoteCounterProxy]
  protected val proxiedCounterMembers = new java.util.concurrent.ConcurrentHashMap[CounterProxyId, RemoteCounterMembersProxy]

  def makeProxyWithinCounter(enclosingCounter: Counter): CounterProxyId = {
    val counterProxyId = enclosingCounter match {
      case cp: RemoteCounterProxy => cp.remoteProxyId
      case _ => proxiedCountersById.computeIfAbsent(enclosingCounter, (_) => freshRemoteRefId())
    }
    proxiedCounterMembers.computeIfAbsent(counterProxyId, (_) => {
      val rmtCounterMbrProxy = new RemoteCounterMembersProxy(counterProxyId, enclosingCounter)
      Logger.finer(f"Created proxy for $enclosingCounter with id $counterProxyId")
      rmtCounterMbrProxy
    })
    
    /* Token: enclosingCounter value is correct as is, since we're swapping this "token" with its proxy. */

    counterProxyId
  }

  def makeProxyCounterFor(counterProxyId: CounterProxyId, origin: PeerLocation): RemoteCounterProxy = {
    def newProxyCounter =
      proxiedCounters.computeIfAbsent(counterProxyId, (_) =>
        new RemoteCounterProxy(counterProxyId,
          (n) => execution.sendHalt(origin, counterProxyId)(n),
          (n) => sendDiscorporate(origin, counterProxyId)(n),
          () => sendResurrect(origin, counterProxyId)()))
    newProxyCounter

    // FIXME: Reenable this optimization. It's probably important. However it will require figuring out both how to
    // make sure things are safe (in terms of message orderings) and how to handle the case where a proxy is passed
    // here when we actually have the counter. This may require the sender knowing if this will happen.
    /*
{'fraction_non_alphanumeric': 0.08531478989083296, 'fraction_numerical': 0.0026917900403768506, 'mean_word_length': 3.293739967897271, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 35, 'https://': 0, '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': '7385229', 'n_tokens_mistral': 3838, 'n_tokens_neox': 3657, 'n_words': 1214}