core/common/src/main/scala/com/webank/wedatasphere/linkis/common/listener/ListenerEventBus.scala - linkis - Git at Google

 /*
  * Copyright 2019 WeBank
  *
  * Licensed 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 com.webank.wedatasphere.linkis.common.listener

 import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}
 import java.util.concurrent.{ArrayBlockingQueue, CopyOnWriteArrayList, Future, LinkedBlockingQueue, TimeoutException}

 import com.webank.wedatasphere.linkis.common.collection.BlockingLoopArray
 import com.webank.wedatasphere.linkis.common.utils.{ByteTimeUtils, Logging, Utils}
 import org.apache.commons.lang.time.DateFormatUtils

 import scala.util.control.NonFatal

 /**
   * Created by enjoyyin on 2017/6/6.
   */
 private[wedatasphere] trait ListenerBus[L <: EventListener, E <: Event] extends Logging {
   val self = this

   private val listeners = new CopyOnWriteArrayList[L]

   /**
     * Add a listener to listen events. This method is thread-safe and can be called in any thread.
     */
   final def addListener(listener: L): Unit = {
     listeners.add(listener)
     info(toString + " add a new listener => " + listener.getClass)
   }

   /**
     * Remove a listener and it won't receive any events. This method is thread-safe and can be called
     * in any thread.
     */
   final def removeListener(listener: L): Unit = {
     listeners.remove(listener)
   }

   /**
     * Post the event to all registered listeners. The `postToAll` caller should guarantee calling
     * `postToAll` in the same thread for all events.
     */
   final def postToAll(event: E): Unit = {
     // JavaConverters can create a JIterableWrapper if we use asScala.
     // However, this method will be called frequently. To avoid the wrapper cost, here we use
     // Java Iterator directly.
     val iter = listeners.iterator
     while (iter.hasNext) {
       val listener = iter.next()
       Utils.tryCatch{
         doPostEvent(listener, event)
       }{
         case NonFatal(e) =>
           Utils.tryAndError(listener.onEventError(event, e))
         case t: Throwable => throw t
       }
     }
   }

   /**
     * Post an event to the specified listener. `onPostEvent` is guaranteed to be called in the same
     * thread for all listeners.
     */
   protected def doPostEvent(listener: L, event: E): Unit

 }
 private[wedatasphere] abstract class ListenerEventBus[L <: EventListener, E <: Event]
       (val eventQueueCapacity: Int, name: String)
       (listenerConsumerThreadSize: Int = 5, listenerThreadMaxFreeTime: Long = ByteTimeUtils.timeStringAsMs("2m"))
   extends ListenerBus[L, E] with Logging {

 //  protected val listenerConsumerThreadSize: Int = 5
 //  protected val listenerThreadMaxFreeTime: Long = ByteTimeUtils.timeStringAsMs("2m")

   private lazy val eventQueue = new ArrayBlockingQueue[E](eventQueueCapacity)
   protected val executorService = Utils.newCachedThreadPool(listenerConsumerThreadSize + 2, name + "-Consumer-ThreadPool", true)
   private val eventDealThreads = Array.tabulate(listenerConsumerThreadSize)(new ListenerEventThread(_))
   private val started = new AtomicBoolean(false)
   private val stopped = new AtomicBoolean(false)

   private var listenerThread: Future[_] = _

   /**
     * Start sending events to attached listeners.
     *
     * This first sends out all buffered events posted before this listener bus has started, then
     * listens for any additional events asynchronously while the listener bus is still running.
     * This should only be called once.
     *
     */
   def start(): Unit = {
     if (started.compareAndSet(false, true)) {
       listenerThread = executorService.submit(new Runnable {
         override def run(): Unit =
           while (!stopped.get) {
             val event = Utils.tryCatch(eventQueue.take()){
               case t: InterruptedException => info(s"stopped $name thread.")
                 return
             }
             while(!eventDealThreads.exists(_.putEvent(event)) && !stopped.get) Utils.tryAndError(Thread.sleep(1))
           }
       })
     } else {
       throw new IllegalStateException(s"$name already started!")
     }
   }

   protected val dropEvent: DropEvent = new IgnoreDropEvent

   def post(event: E): Unit = {
     if (stopped.get || executorService.isTerminated || (listenerThread.isDone && started.get())) {
       dropEvent.onBusStopped(event)
     } else if(!eventQueue.offer(event)) {
       dropEvent.onDropEvent(event)
     }
   }

   /**
     * For testing only. Wait until there are no more events in the queue, or until the specified
     * time has elapsed. Throw `TimeoutException` if the specified time elapsed before the queue
     * emptied.
     * Exposed for testing.
     */
   @throws(classOf[TimeoutException])
   def waitUntilEmpty(timeoutMillis: Long): Unit = {
     val finishTime = System.currentTimeMillis + timeoutMillis
     while (!queueIsEmpty) {
       if (System.currentTimeMillis > finishTime) {
         throw new TimeoutException(
           s"The event queue is not empty after $timeoutMillis milliseconds")
       }
       /* Sleep rather than using wait/notify, because this is used only for testing and
        * wait/notify add overhead in the general case. */
       Thread.sleep(10)
     }
   }

   /**
     * For testing only. Return whether the listener daemon thread is still alive.
     * Exposed for testing.
     */
   def listenerThreadIsAlive: Boolean = !listenerThread.isDone

   /**
     * Return whether the event queue is empty.
     *
     * The use of synchronized here guarantees that all events that once belonged to this queue
     * have already been processed by all attached listeners, if this returns true.
     */
   private def queueIsEmpty: Boolean = synchronized { eventQueue.isEmpty && !eventDealThreads.exists(_.isRunning) }

   /**
     * Stop the listener bus. It will wait until the queued events have been processed, but drop the
     * new events after stopping.
     */
   def stop(): Unit = {
     if (!started.get()) {
       throw new IllegalStateException(s"Attempted to stop $name that has not yet started!")
     }
     if (stopped.compareAndSet(false, true)) {
       // Call eventLock.release() so that listenerThread will poll `null` from `eventQueue` and know
       // `stop` is called.
       info(s"try to stop $name thread.")
       //      eventLock.release()
       listenerThread.cancel(true)
       eventDealThreads.foreach(_.shutdown())
     } else {
       // Keep quiet
     }
   }


   override val toString: String = name + "-ListenerBus"

   trait DropEvent {
     def onDropEvent(event: E): Unit
     def onBusStopped(event: E): Unit
   }

   class IgnoreDropEvent extends DropEvent {
     private val droppedEventsCounter = new AtomicLong(0L)
     @volatile private var lastReportTimestamp = 0L
     private val logDroppedEvent = new AtomicBoolean(false)
     private val logStoppedEvent = new AtomicBoolean(false)
     executorService.submit(new Runnable {
       override def run(): Unit = while(true) {
         val droppedEvents = droppedEventsCounter.get
         if (droppedEvents > 0) {
           // Don't log too frequently
           if (System.currentTimeMillis() - lastReportTimestamp >= 60 * 1000) {
             // There may be multiple threads trying to decrease droppedEventsCounter.
             // Use "compareAndSet" to make sure only one thread can win.
             // And if another thread is increasing droppedEventsCounter, "compareAndSet" will fail and
             // then that thread will update it.
             if (droppedEventsCounter.compareAndSet(droppedEvents, 0)) {
               val prevLastReportTimestamp = lastReportTimestamp
               lastReportTimestamp = System.currentTimeMillis()
               warn(s"Dropped $droppedEvents ListenerEvents since " +
                 DateFormatUtils.format(prevLastReportTimestamp, "yyyy-MM-dd HH:mm:ss"))
             }
           }
         }
         Utils.tryQuietly(Thread.sleep(600000))
       }
     })
     /**
       * If the event queue exceeds its capacity, the new events will be dropped. The subclasses will be
       * notified with the dropped events.
       *
       * Note: `onDropEvent` can be called in any thread.
       */
     def onDropEvent(event: E): Unit = {
       droppedEventsCounter.incrementAndGet()
       if (logDroppedEvent.compareAndSet(false, true)) {
         // Only log the following message once to avoid duplicated annoying logs.
         error("Dropping ListenerEvent because no remaining room in event queue. " +
           "This likely means one of the Listeners is too slow and cannot keep up with " +
           "the rate at which tasks are being started by the scheduler.")
       }
     }
     override def onBusStopped(event: E): Unit = {
       droppedEventsCounter.incrementAndGet()
       if (logStoppedEvent.compareAndSet(false, true)) {
         error(s"$name has already stopped! Dropping event $event.")
       }
     }
   }

   protected class ListenerEventThread(index: Int) extends Runnable {
     private var future: Option[Future[_]] = None
     private var continue = true
     private var event: Option[E] = None
     private var lastEventDealData = 0l

     def releaseFreeThread(): Unit = if(listenerThreadMaxFreeTime > 0 && future.isDefined && event.isEmpty && lastEventDealData > 0 &&
       System.currentTimeMillis() - lastEventDealData >= listenerThreadMaxFreeTime) synchronized {
       if(lastEventDealData == 0 && future.isEmpty) return
       lastEventDealData = 0l
       continue = false
       future.foreach(_.cancel(true))
       future = None
     }
     def isRunning: Boolean = event.isDefined

     def putEvent(event: E): Boolean = if(this.event.isDefined) false else synchronized {
       if(this.event.isDefined) false
       else {
         lastEventDealData = System.currentTimeMillis()
         this.event = Some(event)
         if(future.isEmpty) future = Some(executorService.submit(this))
         else notify()
         true
       }
     }
     override def run(): Unit = {
       val threadName = Thread.currentThread().getName
       val currentThreadName = s"$name-Thread-$index"
       Thread.currentThread().setName(currentThreadName)
       info(s"$currentThreadName begin.")
       def threadRelease(): Unit = {
         info(s"$currentThreadName released.")
         Thread.currentThread().setName(threadName)
       }
       while(continue) {
         synchronized {
           while(event.isEmpty) Utils.tryQuietly(wait(), _ => {
             threadRelease()
             return})
         }
         Utils.tryFinally(event.foreach(postToAll)) (synchronized {
           lastEventDealData = System.currentTimeMillis()
           event = None
         })
       }
       threadRelease()
     }
     def shutdown(): Unit = {
       continue = false
       future.foreach(_.cancel(true))
     }
   }
 }
	/*
	* Copyright 2019 WeBank
	*
	* Licensed 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 com.webank.wedatasphere.linkis.common.listener

	import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}
	import java.util.concurrent.{ArrayBlockingQueue, CopyOnWriteArrayList, Future, LinkedBlockingQueue, TimeoutException}

	import com.webank.wedatasphere.linkis.common.collection.BlockingLoopArray
	import com.webank.wedatasphere.linkis.common.utils.{ByteTimeUtils, Logging, Utils}
	import org.apache.commons.lang.time.DateFormatUtils

	import scala.util.control.NonFatal

	/**
	* Created by enjoyyin on 2017/6/6.
	*/
	private[wedatasphere] trait ListenerBus[L <: EventListener, E <: Event] extends Logging {
	val self = this

	private val listeners = new CopyOnWriteArrayList[L]

	/**
	* Add a listener to listen events. This method is thread-safe and can be called in any thread.
	*/
	final def addListener(listener: L): Unit = {
	listeners.add(listener)
	info(toString + " add a new listener => " + listener.getClass)
	}

	/**
	* Remove a listener and it won't receive any events. This method is thread-safe and can be called
	* in any thread.
	*/
	final def removeListener(listener: L): Unit = {
	listeners.remove(listener)
	}

	/**
	* Post the event to all registered listeners. The `postToAll` caller should guarantee calling
	* `postToAll` in the same thread for all events.
	*/
	final def postToAll(event: E): Unit = {
	// JavaConverters can create a JIterableWrapper if we use asScala.
	// However, this method will be called frequently. To avoid the wrapper cost, here we use
	// Java Iterator directly.
	val iter = listeners.iterator
	while (iter.hasNext) {
	val listener = iter.next()
	Utils.tryCatch{
	doPostEvent(listener, event)
	}{
	case NonFatal(e) =>
	Utils.tryAndError(listener.onEventError(event, e))
	case t: Throwable => throw t
	}
	}
	}

	/**
	* Post an event to the specified listener. `onPostEvent` is guaranteed to be called in the same
	* thread for all listeners.
	*/
	protected def doPostEvent(listener: L, event: E): Unit

	}
	private[wedatasphere] abstract class ListenerEventBus[L <: EventListener, E <: Event]
	(val eventQueueCapacity: Int, name: String)
	(listenerConsumerThreadSize: Int = 5, listenerThreadMaxFreeTime: Long = ByteTimeUtils.timeStringAsMs("2m"))
	extends ListenerBus[L, E] with Logging {

	// protected val listenerConsumerThreadSize: Int = 5
	// protected val listenerThreadMaxFreeTime: Long = ByteTimeUtils.timeStringAsMs("2m")

	private lazy val eventQueue = new ArrayBlockingQueue[E](eventQueueCapacity)
	protected val executorService = Utils.newCachedThreadPool(listenerConsumerThreadSize + 2, name + "-Consumer-ThreadPool", true)
	private val eventDealThreads = Array.tabulate(listenerConsumerThreadSize)(new ListenerEventThread(_))
	private val started = new AtomicBoolean(false)
	private val stopped = new AtomicBoolean(false)

	private var listenerThread: Future[_] = _

	/**
	* Start sending events to attached listeners.
	*
	* This first sends out all buffered events posted before this listener bus has started, then
	* listens for any additional events asynchronously while the listener bus is still running.
	* This should only be called once.
	*
	*/
	def start(): Unit = {
	if (started.compareAndSet(false, true)) {
	listenerThread = executorService.submit(new Runnable {
	override def run(): Unit =
	while (!stopped.get) {
	val event = Utils.tryCatch(eventQueue.take()){
	case t: InterruptedException => info(s"stopped $name thread.")
	return
	}
	while(!eventDealThreads.exists(_.putEvent(event)) && !stopped.get) Utils.tryAndError(Thread.sleep(1))
	}
	})
	} else {
	throw new IllegalStateException(s"$name already started!")
	}
	}

	protected val dropEvent: DropEvent = new IgnoreDropEvent

	def post(event: E): Unit = {
	if (stopped.get \|\| executorService.isTerminated \|\| (listenerThread.isDone && started.get())) {
	dropEvent.onBusStopped(event)
	} else if(!eventQueue.offer(event)) {
	dropEvent.onDropEvent(event)
	}
	}

	/**
	* For testing only. Wait until there are no more events in the queue, or until the specified
	* time has elapsed. Throw `TimeoutException` if the specified time elapsed before the queue
	* emptied.
	* Exposed for testing.
	*/
	@throws(classOf[TimeoutException])
	def waitUntilEmpty(timeoutMillis: Long): Unit = {
	val finishTime = System.currentTimeMillis + timeoutMillis
	while (!queueIsEmpty) {
	if (System.currentTimeMillis > finishTime) {
	throw new TimeoutException(
	s"The event queue is not empty after $timeoutMillis milliseconds")
	}
	/* Sleep rather than using wait/notify, because this is used only for testing and
	* wait/notify add overhead in the general case. */
	Thread.sleep(10)
	}
	}

	/**
	* For testing only. Return whether the listener daemon thread is still alive.
	* Exposed for testing.
	*/
	def listenerThreadIsAlive: Boolean = !listenerThread.isDone

	/**
	* Return whether the event queue is empty.
	*
	* The use of synchronized here guarantees that all events that once belonged to this queue
	* have already been processed by all attached listeners, if this returns true.
	*/
	private def queueIsEmpty: Boolean = synchronized { eventQueue.isEmpty && !eventDealThreads.exists(_.isRunning) }

	/**
	* Stop the listener bus. It will wait until the queued events have been processed, but drop the
	* new events after stopping.
	*/
	def stop(): Unit = {
	if (!started.get()) {
	throw new IllegalStateException(s"Attempted to stop $name that has not yet started!")
	}
	if (stopped.compareAndSet(false, true)) {
	// Call eventLock.release() so that listenerThread will poll `null` from `eventQueue` and know
	// `stop` is called.
	info(s"try to stop $name thread.")
	// eventLock.release()
	listenerThread.cancel(true)
	eventDealThreads.foreach(_.shutdown())
	} else {
	// Keep quiet
	}
	}


	override val toString: String = name + "-ListenerBus"

	trait DropEvent {
	def onDropEvent(event: E): Unit
	def onBusStopped(event: E): Unit
	}

	class IgnoreDropEvent extends DropEvent {
	private val droppedEventsCounter = new AtomicLong(0L)
	@volatile private var lastReportTimestamp = 0L
	private val logDroppedEvent = new AtomicBoolean(false)
	private val logStoppedEvent = new AtomicBoolean(false)
	executorService.submit(new Runnable {
	override def run(): Unit = while(true) {
	val droppedEvents = droppedEventsCounter.get
	if (droppedEvents > 0) {
	// Don't log too frequently
	if (System.currentTimeMillis() - lastReportTimestamp >= 60 * 1000) {
	// There may be multiple threads trying to decrease droppedEventsCounter.
	// Use "compareAndSet" to make sure only one thread can win.
	// And if another thread is increasing droppedEventsCounter, "compareAndSet" will fail and
	// then that thread will update it.
	if (droppedEventsCounter.compareAndSet(droppedEvents, 0)) {
	val prevLastReportTimestamp = lastReportTimestamp
	lastReportTimestamp = System.currentTimeMillis()
	warn(s"Dropped $droppedEvents ListenerEvents since " +
	DateFormatUtils.format(prevLastReportTimestamp, "yyyy-MM-dd HH:mm:ss"))
	}
	}
	}
	Utils.tryQuietly(Thread.sleep(600000))
	}
	})
	/**
	* If the event queue exceeds its capacity, the new events will be dropped. The subclasses will be
	* notified with the dropped events.
	*
	* Note: `onDropEvent` can be called in any thread.
	*/
	def onDropEvent(event: E): Unit = {
	droppedEventsCounter.incrementAndGet()
	if (logDroppedEvent.compareAndSet(false, true)) {
	// Only log the following message once to avoid duplicated annoying logs.
	error("Dropping ListenerEvent because no remaining room in event queue. " +
	"This likely means one of the Listeners is too slow and cannot keep up with " +
	"the rate at which tasks are being started by the scheduler.")
	}
	}
	override def onBusStopped(event: E): Unit = {
	droppedEventsCounter.incrementAndGet()
	if (logStoppedEvent.compareAndSet(false, true)) {
	error(s"$name has already stopped! Dropping event $event.")
	}
	}
	}

	protected class ListenerEventThread(index: Int) extends Runnable {
	private var future: Option[Future[_]] = None
	private var continue = true
	private var event: Option[E] = None
	private var lastEventDealData = 0l

	def releaseFreeThread(): Unit = if(listenerThreadMaxFreeTime > 0 && future.isDefined && event.isEmpty && lastEventDealData > 0 &&
	System.currentTimeMillis() - lastEventDealData >= listenerThreadMaxFreeTime) synchronized {
	if(lastEventDealData == 0 && future.isEmpty) return
	lastEventDealData = 0l
	continue = false
	future.foreach(_.cancel(true))
	future = None
	}
	def isRunning: Boolean = event.isDefined

	def putEvent(event: E): Boolean = if(this.event.isDefined) false else synchronized {
	if(this.event.isDefined) false
	else {
	lastEventDealData = System.currentTimeMillis()
	this.event = Some(event)
	if(future.isEmpty) future = Some(executorService.submit(this))
	else notify()
	true
	}
	}
	override def run(): Unit = {
	val threadName = Thread.currentThread().getName
	val currentThreadName = s"$name-Thread-$index"
	Thread.currentThread().setName(currentThreadName)
	info(s"$currentThreadName begin.")
	def threadRelease(): Unit = {
	info(s"$currentThreadName released.")
	Thread.currentThread().setName(threadName)
	}
	while(continue) {
	synchronized {
	while(event.isEmpty) Utils.tryQuietly(wait(), _ => {
	threadRelease()
	return})
	}
	Utils.tryFinally(event.foreach(postToAll)) (synchronized {
	lastEventDealData = System.currentTimeMillis()
	event = None
	})
	}
	threadRelease()
	}
	def shutdown(): Unit = {
	continue = false
	future.foreach(_.cancel(true))
	}
	}
	}