//设置package包名称以及导入依赖的类
package com.bwsw.tstreams.velocity

import java.net.InetSocketAddress
import java.util.UUID
import java.util.concurrent.locks.ReentrantLock

import com.bwsw.tstreams.agents.consumer.Offsets.Oldest
import com.bwsw.tstreams.agents.consumer.subscriber.{Callback, SubscribingConsumer}
import com.bwsw.tstreams.agents.consumer.{ConsumerOptions, SubscriberCoordinationOptions}


object SubscriberRunner {
  def main(args: Array[String]) {
    import Common._
    val consumerOptions = new ConsumerOptions[String](transactionsPreload = 10, dataPreload = 7, arrayByteToStringConverter, RoundRobinPolicyCreator.getRoundRobinPolicy(stream, List(0)), Oldest, LocalGeneratorCreator.getGen(), useLastOffset = true)

    val lock = new ReentrantLock()
    var cnt = 0
    var timeNow = System.currentTimeMillis()
    val callback = new Callback[String] {
      override def onEvent(subscriber: SubscribingConsumer[String], partition: Int, transactionUuid: UUID): Unit = {
        lock.lock()
        if (cnt % 1000 == 0) {
          val time = System.currentTimeMillis()
          val diff = time - timeNow
          println(s"subscriber_time = $diff; cnt=$cnt")
          timeNow = time
        }
        cnt += 1
        lock.unlock()
      }
    }

    val subscribeConsumer = new SubscribingConsumer[String](
      name = "test_consumer",
      stream = stream,
      options = consumerOptions,
      subscriberCoordinationOptions =
        new SubscriberCoordinationOptions(agentAddress = "t-streams-4.z1.netpoint-dc.com:8588",
          zkRootPath = "/velocity",
          zkHosts = List(new InetSocketAddress(zkHost, 2181)),
          zkSessionTimeout = 7,
          zkConnectionTimeout = 7),
      callBack = callback,
      persistentQueuePath = "persistent_queue_path")
    subscribeConsumer.start()
  }
} 

//设置package包名称以及导入依赖的类
package tech.artemisia.core

import java.util.concurrent.locks.{ReentrantLock, Lock}
import scala.collection.mutable



object TestEnv extends Env {

  override val osUtil: OSUtil = new TestOsUtil

  class TestOsUtil extends OSUtil {

    private[this] val systemVariableMap: mutable.Map[String,String] = mutable.Map("foo" -> "bar")
    private[this] val systemPropertiesMap: mutable.Map[String,String] = mutable.Map("user.home" -> System.getProperty("user.home"))
    private[this] val lock :Lock = new ReentrantLock()

    override def getSystemVariable(variable: String): Option[String] = {
      systemVariableMap get variable
    }

    def getSystemProperties(variable: String): Option[String] = {
      systemPropertiesMap get variable
    }

    def withSysVar(map: scala.collection.Map[String,String])(body : => Unit): Unit = {
      try {
        lock.lock()
        systemVariableMap.clear()
        systemVariableMap ++= map.toSeq
        body
      }
      catch {
        case e: Throwable => throw e
      }
      finally {
        lock.unlock()
      }
    }
  }

} 

//设置package包名称以及导入依赖的类
package lazyvar2

import scala.collection.mutable.ArrayBuffer
import java.util.concurrent.locks.ReentrantLock
import java.util.concurrent.atomic.AtomicLong

object LazyVar {
  var n = new AtomicLong(0L)
  
  def apply[T](v : T) = new LazyVar[T](v)
}

class LazyVar[T](var value : T) {
  val order = LazyVar.n.getAndIncrement 
  var buffer = ArrayBuffer[T => T]()
  val lock = new ReentrantLock()
  
  def force = {
    if(buffer.size > 0) {
      lock.lock()
      for(f <- buffer)
        value = f(value)
      buffer = ArrayBuffer[T => T]() //clear()
      lock.unlock()
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.citypay.pan.search.util

import java.util.concurrent.locks.{Condition, ReentrantLock}

trait Lockable {

  protected val lock: ReentrantLock = new ReentrantLock(true)
  protected val cond1: Condition = lock.newCondition()

  def lock[X](func: => X): X = {
    lock.lock()
    try {
      func
    } finally {
      lock.unlock()
    }
  }

} 

//设置package包名称以及导入依赖的类
package ppl.delite.runtime.messages

import java.util.concurrent.locks.ReentrantLock

class Future[T] {

  private var notReady: Boolean = true
  private var _result: T = _

  private val lock = new ReentrantLock
  private val cond = lock.newCondition

  def get: T = {
    if (notReady) block
    _result
  }

  private def block {
    val lock = this.lock
    lock.lock
    try {
      while (notReady)
        cond.await
    }
    finally {
      lock.unlock
    }
  }

  def set(result: T) {
    val lock = this.lock
    lock.lock
    try {
      _result = result
      notReady = false
      cond.signalAll
    }
    finally {
      lock.unlock
    }
  }

} 

//设置package包名称以及导入依赖的类
package ppl.delite.runtime.graph.ops

import java.util.concurrent.locks.ReentrantLock
import ppl.delite.runtime.graph.targets.Targets


  private val lock = new ReentrantLock
  private val end = lock.newCondition
  private var notDone: Boolean = true
  private var result: Any = null

  def put(res: Any) {
    lock.lock
    try {
      notDone = false
      result = res
      end.signal
    }
    finally {
      lock.unlock
    }
  }

  def take(): Any = {
    var res: Any = null
    lock.lock
    try {
      while (notDone) end.await
      notDone = true //reset for re-use
      res = result
      result = null
    }
    finally {
      lock.unlock
    }
    res
  }  
}

object EOP_Kernel {

  def apply[T](): T = apply(null.asInstanceOf[T])

  def apply[T](result: T): T = {
    EOP_Global.put(result)
    result
  }

} 

//设置package包名称以及导入依赖的类
package ppl.delite.runtime.codegen.kernels.scala.examples

import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.locks.ReentrantLock



object ExampleForeachHeader {
  def apply(in0: Array[Int], in1: Array[Double], in2: Double) = new ExampleForeachHeader(in0, in1, in2)
}

final class ExampleForeachHeader(in0: Array[Int], in1: Array[Double], in2: Double) {

  //this is the apply method of another (kernel) object: shouldn't be generated
  def kernel_apply(in0: Array[Int], in1: Array[Double], in2: Double): Foreach = {
    new Foreach {
      def in = in0
      def foreach(elem: Int) { in1(elem) = (in1(elem-1) + in1(elem) + in1(elem+1))/3 }
      def sync(idx: Int) = List(in0(idx-1), in0(idx), in0(idx+1))
    }
  }

  abstract class Foreach {
    def in: Array[Int]
    def foreach(elem: Int)
    def sync(idx: Int): List[Any]
  }

  val closure = kernel_apply(in0, in1, in2)
  val lockMap = new ConcurrentHashMap[Any, ReentrantLock]
}

object ExampleForeach {

  def apply(foreach: ExampleForeachHeader) {
    val in = foreach.closure.in
    val size = in.size
    var i = size*2/4 //size*chunkIdx/numChunks
    val end = size*3/4 //size*(chunkIdx+1)/numChunks

    while (i < end) {
      val sync = foreach.closure.sync(i).sortBy(System.identityHashCode(_)) //TODO: optimize locking mechanism
      for (e <- sync) {
        foreach.lockMap.putIfAbsent(e, new ReentrantLock)
        foreach.lockMap.get(e).lock
      }

      foreach.closure.foreach(in(i))

      for (e <- sync.reverse) {
        foreach.lockMap.get(e).unlock
      }

      i += 1
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.peim.service

import java.time.OffsetDateTime
import java.util.concurrent.locks.ReentrantLock

import com.peim.model.Transfer
import com.peim.repository.TransfersRepository
import scaldi.{Injectable, Injector}

import scala.concurrent.duration._
import scala.concurrent.{Await, ExecutionContext}
import scala.util.{Failure, Success, Try}

class TransfersService(implicit inj: Injector, executionContext: ExecutionContext) extends Injectable {

  private val createLock = new ReentrantLock()
  private val approveLock = new ReentrantLock()
  private val transfersRepository = inject[TransfersRepository]

  def createTransfer(transfer: Transfer): Try[Int] = {
    createLock.lock()
    val transferId = try {
      Await.result(transfersRepository.create(transfer), 5.seconds)
    } finally {
      createLock.unlock()
    }
    Try(transferId) match {
      case Success(newId) =>
        approveLock.lock()
        try {
          val createdTransfer = transfer.approved(newId)
          Try(Await.result(transfersRepository.approve(createdTransfer), 5.seconds))
        } finally {
          approveLock.unlock()
        }
      case Failure(error) => Failure(error)
    }
  }

  def rollbackFailedTransfers(): Runnable = new Runnable {
    override def run(): Unit = {
      transfersRepository.rollback(OffsetDateTime.now().minusSeconds(10))
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.jvm

import java.util.concurrent.locks.ReentrantLock

import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.concurrent.Eventually
import org.scalatest.junit.JUnitRunner
import org.scalatest.time.{Millis, Seconds, Span}

import com.twitter.util.{Await, Promise}

class Philosopher {
  val ready = new Promise[Unit]
  private val lock = new ReentrantLock()
  def dine(neighbor: Philosopher): Unit = {
    lock.lockInterruptibly()
    ready.setValue(Unit)
    Await.ready(neighbor.ready)
    neighbor.dine(this)
    lock.unlock()
  }
}

@RunWith(classOf[JUnitRunner])
class ContentionTest extends FunSuite with Eventually {

  implicit override val patienceConfig =
    PatienceConfig(timeout = scaled(Span(15, Seconds)), interval = scaled(Span(5, Millis)))

  test("Deadlocks") {
    val c = new ContentionSnapshot()

    val descartes = new Philosopher()
    val plato = new Philosopher()

    val d = new Thread(new Runnable() {
      def run() { descartes.dine(plato) }
    })
    d.start()

    val p = new Thread(new Runnable() {
      def run() { plato.dine(descartes) }
    })
    p.start()
    Await.all(descartes.ready, plato.ready)

    eventually { assert(c.snap().deadlocks.size == 2) }
    d.interrupt()
    p.interrupt()
    p.join()
    d.join()
    assert(c.snap().deadlocks.size == 0)
  }
} 

//设置package包名称以及导入依赖的类
package org.rebeam.tree.server.util

import java.util.concurrent.locks.ReentrantLock

class Lock {
  private val lock: ReentrantLock = new ReentrantLock()
  def apply[T](w: =>T): T = run(w)
  def run[T](w: =>T): T = {
    lock.lock()
    try {
      return w
    } finally {
      lock.unlock()
    }
  }
}

object Lock {
  def apply() = new Lock()
}