felix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cziege...@apache.org
Subject svn commit: r1628132 - in /felix/trunk/eventadmin/impl: ./ src/main/java/org/apache/felix/eventadmin/impl/handler/ src/main/java/org/apache/felix/eventadmin/impl/tasks/
Date Mon, 29 Sep 2014 07:55:52 GMT
Author: cziegeler
Date: Mon Sep 29 07:55:52 2014
New Revision: 1628132

URL: http://svn.apache.org/r1628132
Log:
FELIX-4638 - Less locking on event handler timing. Apply patch from Bob Paulin

Added:
    felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/BlacklistLatch.java
  (with props)
    felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/HandlerTask.java
  (with props)
Removed:
    felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/Rendezvous.java
Modified:
    felix/trunk/eventadmin/impl/changelog.txt
    felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/handler/EventHandlerProxy.java
    felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/SyncDeliverTasks.java

Modified: felix/trunk/eventadmin/impl/changelog.txt
URL: http://svn.apache.org/viewvc/felix/trunk/eventadmin/impl/changelog.txt?rev=1628132&r1=1628131&r2=1628132&view=diff
==============================================================================
--- felix/trunk/eventadmin/impl/changelog.txt (original)
+++ felix/trunk/eventadmin/impl/changelog.txt Mon Sep 29 07:55:52 2014
@@ -2,6 +2,7 @@ Changes from 1.4.2 to 1.4.4
 ---------------------------
 ** Improvement
     * [FELIX-4623] - Add test for thread based ordering
+    * [FELIX-4638] - Less locking on event handler timing
 
 
 Changes from 1.4.0 to 1.4.2

Modified: felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/handler/EventHandlerProxy.java
URL: http://svn.apache.org/viewvc/felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/handler/EventHandlerProxy.java?rev=1628132&r1=1628131&r2=1628132&view=diff
==============================================================================
--- felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/handler/EventHandlerProxy.java
(original)
+++ felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/handler/EventHandlerProxy.java
Mon Sep 29 07:55:52 2014
@@ -431,12 +431,15 @@ public class EventHandlerProxy {
      */
     public void blackListHandler()
     {
-        LogWrapper.getLogger().log(
-                        LogWrapper.LOG_WARNING,
-                        "Blacklisting ServiceReference [" + this.reference + " | Bundle("
-                                        + this.reference.getBundle() + ")] due to timeout!");
-        this.blacklisted = true;
-        // we can free the handler now.
-        this.release();
+    	if(!this.blacklisted)
+    	{
+	        LogWrapper.getLogger().log(
+	                        LogWrapper.LOG_WARNING,
+	                        "Blacklisting ServiceReference [" + this.reference + " | Bundle("
+	                                        + this.reference.getBundle() + ")] due to timeout!");
+	        this.blacklisted = true;
+	        // we can free the handler now.
+	        this.release();
+    	}
     }
 }

Added: felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/BlacklistLatch.java
URL: http://svn.apache.org/viewvc/felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/BlacklistLatch.java?rev=1628132&view=auto
==============================================================================
--- felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/BlacklistLatch.java
(added)
+++ felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/BlacklistLatch.java
Mon Sep 29 07:55:52 2014
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.felix.eventadmin.impl.tasks;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.felix.eventadmin.impl.util.LogWrapper;
+
+/**
+ *
+ * A latch that checks handlers for blacklisting on an interval.
+ *
+ */
+public class BlacklistLatch {
+
+	private final Semaphore internalSemaphore;
+
+	private final int count;
+
+	private final long timeout;
+
+	private final List<HandlerTask> handlerTasks;
+
+	/**
+	 * @param count Number of handlers that must call countdown
+	 * @param timeout Timeout in Milliseconds to check for blacklisting handlers
+	 */
+	public BlacklistLatch(final int count, final long timeout)
+	{
+		this.handlerTasks = new ArrayList<HandlerTask>(count);
+		this.count = count;
+		this.timeout = timeout;
+		internalSemaphore = new Semaphore(count);
+		internalSemaphore.drainPermits();
+	}
+
+	/**
+	 *
+	 * Count down the number of handlers blocking event completion.
+	 *
+	 */
+	public void countDown()
+	{
+		internalSemaphore.release();
+	}
+
+	/**
+	 *
+	 * Adds a handler task to the timeout based blackout checking.
+	 *
+	 * @param task
+	 */
+	public void addToBlacklistCheck(final HandlerTask task)
+	{
+		this.handlerTasks.add(task);
+	}
+
+	/**
+	 *
+	 * Causes current thread to wait until each handler has called countDown.
+	 * Checks on timeout interval to determine if a handler needs blacklisting.
+	 *
+	 */
+	public void awaitAndBlacklistCheck()
+	{
+		try
+        {
+        	while(!internalSemaphore.tryAcquire(this.count, this.timeout, TimeUnit.MILLISECONDS))
+            {
+            	final Iterator<HandlerTask> handlerTaskIt = handlerTasks.iterator();
+            	while(handlerTaskIt.hasNext())
+            	{
+            		HandlerTask currentTask = handlerTaskIt.next();
+            		currentTask.checkForBlacklist();
+            	}
+            }
+        }
+        catch (final InterruptedException e)
+        {
+        	LogWrapper.getLogger().log(
+                    LogWrapper.LOG_WARNING,
+                    "Event Task Processing Interrupted. Events may not be recieved in proper
order.");
+        }
+	}
+
+
+}

Propchange: felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/BlacklistLatch.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/BlacklistLatch.java
------------------------------------------------------------------------------
    svn:keywords = author date id revision rev url

Propchange: felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/BlacklistLatch.java
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Added: felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/HandlerTask.java
URL: http://svn.apache.org/viewvc/felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/HandlerTask.java?rev=1628132&view=auto
==============================================================================
--- felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/HandlerTask.java
(added)
+++ felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/HandlerTask.java
Mon Sep 29 07:55:52 2014
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.felix.eventadmin.impl.tasks;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadMXBean;
+
+import org.apache.felix.eventadmin.impl.handler.EventHandlerProxy;
+import org.osgi.service.event.Event;
+
+/**
+ * A task that processes an event handler
+ *
+ */
+public class HandlerTask implements Runnable
+{
+	private final EventHandlerProxy task;
+
+	private final Event event;
+
+	private final long timeout;
+
+	private final BlacklistLatch handlerLatch;
+
+	private volatile long threadId;
+
+	private volatile long startTime;
+
+	private volatile long endTime;
+
+	/**
+	 *
+	 *
+	 * @param task Proxy to the event handler
+	 * @param event The event to send to the handler
+	 * @param timeout Timeout for handler blacklisting
+	 * @param handlerLatch The latch used to ensure events fire in proper order
+	 */
+	public HandlerTask(final EventHandlerProxy task, final Event event, final long timeout,
final BlacklistLatch handlerLatch)
+	{
+		this.task = task;
+		this.event = event;
+		this.timeout = timeout;
+		this.handlerLatch = handlerLatch;
+		this.threadId = -1l;
+		this.startTime = -1l;
+		this.endTime = -1l;
+	}
+
+	/**
+	 *
+	 * Perform timing based on thread CPU time with clock time fall back.
+	 *
+	 * @return
+	 */
+	public long getTimeInMillis()
+    {
+        ThreadMXBean bean = ManagementFactory.getThreadMXBean();
+        return bean.isThreadCpuTimeEnabled() ?
+            bean.getThreadCpuTime(threadId)/1000000 : System.currentTimeMillis();
+    }
+
+	/**
+	 * Run Hander Event
+	 */
+    @Override
+    public void run()
+    {
+        try
+        {
+        	threadId = Thread.currentThread().getId();
+            startTime = getTimeInMillis();
+            // execute the task
+            task.sendEvent(event);
+            endTime = getTimeInMillis();
+            checkForBlacklist();
+        }
+        finally
+        {
+        	handlerLatch.countDown();
+        }
+    }
+
+    public void runWithoutBlacklistTiming()
+    {
+    	task.sendEvent(event);
+    	handlerLatch.countDown();
+    }
+
+    /**
+     * This method defines if a timeout handling should be used for the
+     * task.
+     */
+    public boolean useTimeout()
+    {
+        // we only check the proxy if a timeout is configured
+        if ( this.timeout > 0)
+        {
+            return task.useTimeout();
+        }
+        return false;
+    }
+
+    /**
+     * Check to see if we need to blacklist this handler
+     *
+     */
+    public void checkForBlacklist()
+    {
+    	if(useTimeout() && getTaskTime() > this.timeout)
+		{
+			task.blackListHandler();
+		}
+    }
+
+    /**
+     *
+     * Determine the amount of time spent running this task
+     *
+     * @return
+     */
+    public long getTaskTime()
+    {
+    	if(threadId < 0l || startTime < 0l)
+    	{
+    		return 0l;
+    	}
+    	else if(endTime < 0l)
+    	{
+    		return getTimeInMillis() - startTime;
+    	}
+    	return endTime - startTime;
+
+    }
+
+}

Propchange: felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/HandlerTask.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/HandlerTask.java
------------------------------------------------------------------------------
    svn:keywords = author date id revision rev url

Propchange: felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/HandlerTask.java
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Modified: felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/SyncDeliverTasks.java
URL: http://svn.apache.org/viewvc/felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/SyncDeliverTasks.java?rev=1628132&r1=1628131&r2=1628132&view=diff
==============================================================================
--- felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/SyncDeliverTasks.java
(original)
+++ felix/trunk/eventadmin/impl/src/main/java/org/apache/felix/eventadmin/impl/tasks/SyncDeliverTasks.java
Mon Sep 29 07:55:52 2014
@@ -20,7 +20,6 @@ package org.apache.felix.eventadmin.impl
 
 import java.util.Collection;
 import java.util.Iterator;
-import java.util.concurrent.TimeoutException;
 
 import org.apache.felix.eventadmin.impl.handler.EventHandlerProxy;
 import org.osgi.service.event.Event;
@@ -80,21 +79,6 @@ public class SyncDeliverTasks
     }
 
     /**
-     * This method defines if a timeout handling should be used for the
-     * task.
-     * @param tasks The event handler dispatch tasks to execute
-     */
-    private boolean useTimeout(final EventHandlerProxy proxy)
-    {
-        // we only check the proxy if a timeout is configured
-        if ( this.timeout > 0)
-        {
-            return proxy.useTimeout();
-        }
-        return false;
-    }
-
-    /**
      * This blocks an unrelated thread used to send a synchronous event until the
      * event is send (or a timeout occurs).
      *
@@ -107,69 +91,34 @@ public class SyncDeliverTasks
         final SyncThread syncThread = sleepingThread instanceof SyncThread ? (SyncThread)sleepingThread
: null;
 
         final Iterator<EventHandlerProxy> i = tasks.iterator();
+        final BlacklistLatch handlerLatch = new BlacklistLatch(tasks.size(), this.timeout/2);
+        
         while ( i.hasNext() )
         {
             final EventHandlerProxy task = i.next();
+            HandlerTask handlerTask = new HandlerTask(task, event, this.timeout, handlerLatch);
 //            if ( !filterAsyncUnordered || task.isAsyncOrderedDelivery() )
 //            {
-                if ( !useTimeout(task) )
+                if( !handlerTask.useTimeout() )
                 {
-                    // no timeout, we can directly execute
-                    task.sendEvent(event);
+                	handlerTask.runWithoutBlacklistTiming();
                 }
-                else if ( syncThread != null )
+            	else if ( syncThread != null  )
                 {
                     // if this is a cascaded event, we directly use this thread
                     // otherwise we could end up in a starvation
-                    final long startTime = System.currentTimeMillis();
-                    task.sendEvent(event);
-                    if ( System.currentTimeMillis() - startTime > this.timeout )
-                    {
-                        task.blackListHandler();
-                    }
+                    handlerTask.run();
                 }
                 else
                 {
-                    final Rendezvous startBarrier = new Rendezvous();
-                    final Rendezvous timerBarrier = new Rendezvous();
-                    this.pool.executeTask(new Runnable()
-                    {
-                        @Override
-                        public void run()
-                        {
-                            try
-                            {
-                                // notify the outer thread to start the timer
-                                startBarrier.waitForRendezvous();
-                                // execute the task
-                                task.sendEvent(event);
-                                // stop the timer
-                                timerBarrier.waitForRendezvous();
-                            }
-                            catch (final IllegalStateException ise)
-                            {
-                                // this can happen on shutdown, so we ignore it
-                            }
-                        }
-                    });
-                    // we wait for the inner thread to start
-                    startBarrier.waitForRendezvous();
-
-                    // timeout handling
-                    // we sleep for the sleep time
-                    // if someone wakes us up it's the finished inner task
-                    try
-                    {
-                        timerBarrier.waitAttemptForRendezvous(this.timeout);
-                    }
-                    catch (final TimeoutException ie)
-                    {
-                        // if we timed out, we have to blacklist the handler
-                        task.blackListHandler();
-                    }
-
+                	
+                	handlerLatch.addToBlacklistCheck(handlerTask);
+                    this.pool.executeTask(handlerTask);
                 }
+               
 //            }
         }
+        handlerLatch.awaitAndBlacklistCheck();
+        
     }
 }



Mime
View raw message