Peter Firmstone wrote:
Ok,
The unicast socket didn't have a timeout set, the harness was stuck on
.join() waiting for the socket to close.
The tests still failing but it's no longer hanging.
Cheers,
Peter.
I set a timeout on the socket, the test fails and exits, but now I've
got a stale processes. I've fixed some concurrency / visibility issues
in the patches attached.
I'm not going to be able to play with this again until next weekend,
does anyone else have time to look into it? It's almost as if the
client and server are attempting to communicate on different ports or
something?
Thanks, Peter.
This looks like a timing or Socket issue, there no data (test output):
[java] java.rmi.UnmarshalException: exception unmarshalling
response; nested exception is:
[java] java.io.IOException: I/O error reading from mux
connection: java.io.EOFException
[java] at
net.jini.jeri.BasicInvocationHandler.invokeRemoteMethodOnce(BasicInvocationHandler.java:847)
[java] at
net.jini.jeri.BasicInvocationHandler.invokeRemoteMethod(BasicInvocationHandler.java:659)
[java] at
net.jini.jeri.BasicInvocationHandler.invoke(BasicInvocationHandler.java:528)
[java] at
com.sun.jini.test.services.lookupsimulator.$Proxy0.getLocator(Unknown
Source)
[java] at
com.sun.jini.test.services.lookupsimulator.LookupSimulatorProxy.getLocator(LookupSimulatorProxy.java:134)
[java] at
com.sun.jini.test.share.BaseQATest.startLookup(BaseQATest.java:1722)
[java] at
com.sun.jini.test.share.BaseQATest.startInitLookups(BaseQATest.java:1538)
[java] at
com.sun.jini.test.share.BaseQATest.setup(BaseQATest.java:1063)
[java] at
com.sun.jini.test.spec.lookupdiscovery.AbstractBaseTest.setup(AbstractBaseTest.java:77)
[java] at
com.sun.jini.test.spec.lookupdiscovery.Discovered.setup(Discovered.java:69)
[java] at
com.sun.jini.qa.harness.MasterTest.doTest(MasterTest.java:219)
[java] at
com.sun.jini.qa.harness.MasterTest.main(MasterTest.java:144)
[java] Caused by: java.io.IOException: I/O error reading from mux
connection: java.io.EOFException
[java] at
com.sun.jini.jeri.internal.mux.Session$MuxInputStream.read(Session.java:874)
[java] at
net.jini.jeri.connection.ConnectionManager$Outbound$Input.read(ConnectionManager.java:550)
[java] at
net.jini.jeri.BasicObjectEndpoint.executeCall(BasicObjectEndpoint.java:410)
[java] at
net.jini.jeri.BasicInvocationHandler.invokeRemoteMethodOnce(BasicInvocationHandler.java:806)
[java] ... 11 more
[java] Caused by: java.io.EOFException
[java] at
com.sun.jini.jeri.internal.mux.StreamConnectionIO$Reader.run(StreamConnectionIO.java:264)
[java] at
com.sun.jini.thread.ThreadPool$Task.run(ThreadPool.java:140)
[java] at
com.sun.jini.thread.ThreadPool$Worker.run(ThreadPool.java:174)
[java] at java.lang.Thread.run(Thread.java:662)
[java]
[java] TIME: 9:39:51 PM
[java]
[java] Test process was destroyed and returned code 1
[java]
com/sun/jini/test/spec/lookupdiscovery/MulticastMonitorStopReplace.td
[java] Test Failed: Unexpected Exception in setup:
java.rmi.UnmarshalException: exception unmarshalling response; nested
exception is:
[java] java.io.IOException: I/O error reading from mux
connection: java.io.EOFException
The remaining phantom process:
$ jstack 5001
2012-11-25 21:41:19
Full thread dump Java HotSpot(TM) Server VM (20.5-b03 mixed mode):
"Attach Listener" daemon prio=3 tid=0x00196400 nid=0x1e waiting on
condition [0x00000000]
java.lang.Thread.State: RUNNABLE
"(JSK) Idle" daemon prio=3 tid=0x00394800 nid=0x1d waiting on condition
[0xb5c7f000]
java.lang.Thread.State: TIMED_WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0xe7939b30> (a
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at
java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:196)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2116)
at com.sun.jini.thread.ThreadPool$Worker.run(ThreadPool.java:191)
at java.lang.Thread.run(Thread.java:662)
"(JSK) Idle" daemon prio=3 tid=0x006e9800 nid=0x1c waiting on condition
[0xb5d7f000]
java.lang.Thread.State: TIMED_WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0xe7900070> (a
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at
java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:196)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2116)
at com.sun.jini.thread.ThreadPool$Worker.run(ThreadPool.java:191)
at java.lang.Thread.run(Thread.java:662)
"(JSK) Idle" daemon prio=3 tid=0x00675000 nid=0x1b waiting on condition
[0xb5e7f000]
java.lang.Thread.State: TIMED_WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0xe7900070> (a
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at
java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:196)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2116)
at com.sun.jini.thread.ThreadPool$Worker.run(ThreadPool.java:191)
at java.lang.Thread.run(Thread.java:662)
"(JSK) Idle" daemon prio=3 tid=0x00195800 nid=0x1a waiting on condition
[0xb5f7f000]
java.lang.Thread.State: TIMED_WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0xe7900070> (a
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at
java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:196)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2116)
at com.sun.jini.thread.ThreadPool$Worker.run(ThreadPool.java:191)
at java.lang.Thread.run(Thread.java:662)
"(JSK) Idle" daemon prio=3 tid=0x006fd800 nid=0x19 waiting on condition
[0xb607f000]
java.lang.Thread.State: TIMED_WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0xe7939b30> (a
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at
java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:196)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2116)
at com.sun.jini.thread.ThreadPool$Worker.run(ThreadPool.java:191)
at java.lang.Thread.run(Thread.java:662)
"(JSK) Idle" daemon prio=3 tid=0x003ed400 nid=0x18 waiting on condition
[0xb617f000]
java.lang.Thread.State: TIMED_WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0xe7939b30> (a
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at
java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:196)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2116)
at com.sun.jini.thread.ThreadPool$Worker.run(ThreadPool.java:191)
at java.lang.Thread.run(Thread.java:662)
"(JSK) KeepAlive" prio=3 tid=0x00417000 nid=0x17 waiting on condition
[0xb627f000]
java.lang.Thread.State: TIMED_WAITING (sleeping)
at java.lang.Thread.sleep(Native Method)
at
com.sun.jini.jeri.internal.runtime.JvmLifeSupport$2.run(JvmLifeSupport.java:130)
at java.lang.Thread.run(Thread.java:662)
"(JSK) Reaper" daemon prio=3 tid=0x00593c00 nid=0x16 in Object.wait()
[0xb637f000]
java.lang.Thread.State: WAITING (on object monitor)
at java.lang.Object.wait(Native Method)
- waiting on <0xe7904308> (a java.lang.ref.ReferenceQueue$Lock)
at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:118)
- locked <0xe7904308> (a java.lang.ref.ReferenceQueue$Lock)
at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:134)
at
com.sun.jini.jeri.internal.runtime.ImplRefManager$Reaper.run(ImplRefManager.java:426)
at java.lang.Thread.run(Thread.java:662)
"(JSK)
TcpServerEndpoint.LH[ServerSocket[addr=0.0.0.0/0.0.0.0,port=0,localport=34300]]
accept loop" daemon prio=3 tid=0x00593400 nid=0x15 runnable [0xb647f000]
java.lang.Thread.State: RUNNABLE
at java.net.PlainSocketImpl.socketAccept(Native Method)
at java.net.PlainSocketImpl.accept(PlainSocketImpl.java:408)
- locked <0xe7905620> (a java.net.SocksSocketImpl)
at java.net.ServerSocket.implAccept(ServerSocket.java:462)
at java.net.ServerSocket.accept(ServerSocket.java:430)
at
net.jini.jeri.tcp.TcpServerEndpoint$LH.executeAcceptLoop(TcpServerEndpoint.java:798)
at
net.jini.jeri.tcp.TcpServerEndpoint$LH.access$400(TcpServerEndpoint.java:736)
at
net.jini.jeri.tcp.TcpServerEndpoint$LH$1.run(TcpServerEndpoint.java:768)
at com.sun.jini.thread.ThreadPool$Task.run(ThreadPool.java:140)
at com.sun.jini.thread.ThreadPool$Worker.run(ThreadPool.java:174)
at java.lang.Thread.run(Thread.java:662)
"RMI Scheduler(0)" daemon prio=3 tid=0x00410000 nid=0x14 waiting on
condition [0xb657f000]
java.lang.Thread.State: TIMED_WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0xbb8824b8> (a
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at
java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:196)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2025)
at java.util.concurrent.DelayQueue.take(DelayQueue.java:164)
at
java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:609)
at
java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:602)
at
java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:947)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
at java.lang.Thread.run(Thread.java:662)
"DestroyJavaVM" prio=3 tid=0x00031800 nid=0x2 waiting on condition
[0x00000000]
java.lang.Thread.State: RUNNABLE
"Thread-1" prio=3 tid=0x00589000 nid=0x12 waiting on condition [0xb677f000]
java.lang.Thread.State: TIMED_WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0xe7920a90> (a
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at
java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:196)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2025)
at java.util.concurrent.DelayQueue.take(DelayQueue.java:164)
at
java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:609)
at
java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:602)
at
java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:947)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
at java.lang.Thread.run(Thread.java:662)
"GC Daemon" daemon prio=3 tid=0x00606400 nid=0x11 in Object.wait()
[0xb687f000]
java.lang.Thread.State: TIMED_WAITING (on object monitor)
at java.lang.Object.wait(Native Method)
- waiting on <0xbb808038> (a sun.misc.GC$LatencyLock)
at sun.misc.GC$Daemon.run(GC.java:100)
- locked <0xbb808038> (a sun.misc.GC$LatencyLock)
"RMI Reaper" prio=3 tid=0x002e0c00 nid=0x10 in Object.wait() [0xb697f000]
java.lang.Thread.State: WAITING (on object monitor)
at java.lang.Object.wait(Native Method)
- waiting on <0xbb800100> (a java.lang.ref.ReferenceQueue$Lock)
at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:118)
- locked <0xbb800100> (a java.lang.ref.ReferenceQueue$Lock)
at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:134)
at sun.rmi.transport.ObjectTable$Reaper.run(ObjectTable.java:333)
at java.lang.Thread.run(Thread.java:662)
"RMI TCP Accept-0" daemon prio=3 tid=0x003c0c00 nid=0xf runnable
[0xb6a7f000]
java.lang.Thread.State: RUNNABLE
at java.net.PlainSocketImpl.socketAccept(Native Method)
at java.net.PlainSocketImpl.accept(PlainSocketImpl.java:408)
- locked <0xbb8001a8> (a java.net.SocksSocketImpl)
at java.net.ServerSocket.implAccept(ServerSocket.java:462)
at java.net.ServerSocket.accept(ServerSocket.java:430)
at
sun.rmi.transport.tcp.TCPTransport$AcceptLoop.executeAcceptLoop(TCPTransport.java:369)
at
sun.rmi.transport.tcp.TCPTransport$AcceptLoop.run(TCPTransport.java:341)
at java.lang.Thread.run(Thread.java:662)
"Low Memory Detector" daemon prio=3 tid=0x00123c00 nid=0xd runnable
[0x00000000]
java.lang.Thread.State: RUNNABLE
"C2 CompilerThread1" daemon prio=3 tid=0x00121000 nid=0xc waiting on
condition [0x00000000]
java.lang.Thread.State: RUNNABLE
"C2 CompilerThread0" daemon prio=3 tid=0x0011e800 nid=0xb waiting on
condition [0x00000000]
java.lang.Thread.State: RUNNABLE
"Signal Dispatcher" daemon prio=3 tid=0x0011d000 nid=0xa runnable
[0x00000000]
java.lang.Thread.State: RUNNABLE
"Finalizer" daemon prio=3 tid=0x0010cc00 nid=0x9 in Object.wait()
[0xb707f000]
java.lang.Thread.State: WAITING (on object monitor)
at java.lang.Object.wait(Native Method)
- waiting on <0xbb804230> (a java.lang.ref.ReferenceQueue$Lock)
at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:118)
- locked <0xbb804230> (a java.lang.ref.ReferenceQueue$Lock)
at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:134)
at java.lang.ref.Finalizer$FinalizerThread.run(Finalizer.java:159)
"Reference Handler" daemon prio=3 tid=0x00107c00 nid=0x8 in
Object.wait() [0xb717f000]
java.lang.Thread.State: WAITING (on object monitor)
at java.lang.Object.wait(Native Method)
- waiting on <0xbb800358> (a java.lang.ref.Reference$Lock)
at java.lang.Object.wait(Object.java:485)
at java.lang.ref.Reference$ReferenceHandler.run(Reference.java:116)
- locked <0xbb800358> (a java.lang.ref.Reference$Lock)
"VM Thread" prio=3 tid=0x00104000 nid=0x7 runnable
"GC task thread#0 (ParallelGC)" prio=3 tid=0x00039400 nid=0x3 runnable
"GC task thread#1 (ParallelGC)" prio=3 tid=0x0003a800 nid=0x4 runnable
"GC task thread#2 (ParallelGC)" prio=3 tid=0x0003bc00 nid=0x5 runnable
"GC task thread#3 (ParallelGC)" prio=3 tid=0x0003d000 nid=0x6 runnable
"VM Periodic Task Thread" prio=3 tid=0x0012e000 nid=0xe waiting on condition
JNI global references: 1449
$
# This patch file was generated by NetBeans IDE
# Following Index: paths are relative to: /opt/src/River_Fixed_2nd_Try/peterConcurrentPolicy/src
# This patch can be applied using context Tools: Patch action on respective folder.
# It uses platform neutral UTF-8 encoding and \n newlines.
# Above lines and this line are ignored by the patching process.
Index: com/sun/jini/thread/NewThreadAction.java
--- com/sun/jini/thread/NewThreadAction.java Base (BASE)
+++ com/sun/jini/thread/NewThreadAction.java Locally Modified (Based On LOCAL)
@@ -41,14 +41,14 @@
*
* @author Sun Microsystems, Inc.
**/
-public final class NewThreadAction implements PrivilegedAction {
+public final class NewThreadAction implements PrivilegedAction<Thread> {
static final String NAME_PREFIX = "(JSK) ";
/** cached reference to the system (root) thread group */
- static final ThreadGroup systemThreadGroup = (ThreadGroup)
- AccessController.doPrivileged(new PrivilegedAction() {
- public Object run() {
+ static final ThreadGroup systemThreadGroup =
+ AccessController.doPrivileged(new PrivilegedAction<ThreadGroup>() {
+ public ThreadGroup run() {
ThreadGroup group = Thread.currentThread().getThreadGroup();
ThreadGroup parent;
while ((parent = group.getParent()) != null) {
@@ -63,9 +63,9 @@
* may execute user code, so that the security policy for threads in
* the system thread group will not apply
*/
- static final ThreadGroup userThreadGroup = (ThreadGroup)
- AccessController.doPrivileged(new PrivilegedAction() {
- public Object run() {
+ static final ThreadGroup userThreadGroup =
+ AccessController.doPrivileged(new PrivilegedAction<ThreadGroup>() {
+ public ThreadGroup run() {
return new ThreadGroup(systemThreadGroup,
NAME_PREFIX + "Runtime");
}
@@ -124,7 +124,7 @@
runnable, name, daemon);
}
- public Object run() {
+ public Thread run() {
SecurityManager sm = System.getSecurityManager();
if (sm != null) {
sm.checkPermission(getClassLoaderPermission);
Index: com/sun/jini/thread/ThreadPool.java
--- com/sun/jini/thread/ThreadPool.java Base (BASE)
+++ com/sun/jini/thread/ThreadPool.java Locally Modified (Based On LOCAL)
@@ -20,7 +20,12 @@
import com.sun.jini.action.GetLongAction;
import java.security.AccessController;
-import java.util.LinkedList;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -70,14 +75,15 @@
/** thread group that this pool's threads execute in */
private final ThreadGroup threadGroup;
- /** lock guarding all mutable instance state (below) */
- private final Object lock = new Object();
+ /** Lock used to wake idle threads and synchronize writes to idleThreads */
+ private final Lock lock;
+ private final Condition wakeup;
/** threads definitely available to take new tasks */
- private int idleThreads = 0;
+ private volatile int idleThreads;
/** queues of tasks to execute */
- private final LinkedList queue = new LinkedList();
+ private final Queue<Runnable> queue;
/**
* Creates a new thread group that executes tasks in threads of
@@ -85,21 +91,30 @@
*/
ThreadPool(ThreadGroup threadGroup) {
this.threadGroup = threadGroup;
+ idleThreads = 0;
+ queue = new ConcurrentLinkedQueue<Runnable>(); //Non blocking queue.
+ lock = new ReentrantLock();
+ wakeup = lock.newCondition();
}
+ // This method must not block - Executor
public void execute(Runnable runnable, String name) {
- Task task = new Task(runnable, name);
- synchronized (lock) {
- if (queue.size() < idleThreads) {
- queue.addLast(task);
- lock.notify();
- return;
- }
- }
- Thread t = (Thread) AccessController.doPrivileged(
+ Runnable task = new Task(runnable, name);
+ if (idleThreads < 3){ // create a new thread, non blocking approximate
+ Thread t = AccessController.doPrivileged(
new NewThreadAction(threadGroup, new Worker(task), name, true));
t.start();
+ } else {
+ boolean accepted = queue.offer(task); //non blocking.
+ if (accepted) {
+ wakeup.signal(); // non blocking.
+ } else { // Should never happen.
+ Thread t = AccessController.doPrivileged(
+ new NewThreadAction(threadGroup, new Worker(task), name, true));
+ t.start();
}
+ }
+ }
public void execute(Runnable command) {
execute(command, "com.sun.jini.thread.ThreadPool");
@@ -108,65 +123,80 @@
/**
* Task simply encapsulates a task's Runnable object with its name.
*/
- private static class Task {
+ private static class Task implements Runnable{
- final Runnable runnable;
- final String name;
+ private final Runnable runnable;
+ private final String name;
Task(Runnable runnable, String name) {
this.runnable = runnable;
this.name = name;
}
+
+ public void run(){
+ try {
+ runnable.run();
+ } catch (Exception t) { // Don't catch Error
+ logger.log(Level.WARNING, "uncaught exception", t);
+ if (t instanceof RuntimeException){
+ if (t instanceof SecurityException){
+ // ignore it will be logged.
+ } else {
+ // Ignorance of RuntimeException is generally bad, bail out.
+ throw (RuntimeException) t;
}
+ }
+ }
+ }
+ public String toString(){
+ return name;
+ }
+ }
+
/**
* Worker executes an initial task, and then it executes tasks from the
* queue, passing away if ever idle for more than the idle timeout value.
*/
private class Worker implements Runnable {
- private Task first;
+ private volatile Runnable first;
- Worker(Task first) {
+ Worker(Runnable first) {
this.first = first;
}
public void run() {
- Task task = first;
- first = null;
-
- while (true) {
- try {
- task.runnable.run();
- } catch (Throwable t) {
- logger.log(Level.WARNING, "uncaught exception", t);
- }
+ Runnable task = first;
+ first = null; // For garbage collection.
+ task.run();
+ Thread thread = Thread.currentThread();
+ while (!thread.isInterrupted()) {
/*
* REMIND: What if the task changed this thread's
* priority? or context class loader?
*/
-
- synchronized (lock) {
- if (queue.isEmpty()) {
- Thread.currentThread().setName(
- NewThreadAction.NAME_PREFIX + "Idle");
- idleThreads++;
+ for ( task = queue.poll(); task != null; task = queue.poll()){
+ // Keep executing while tasks are available.
+ thread.setName(NewThreadAction.NAME_PREFIX + task);
+ task.run();
+ }
+ // queue is empty;
+ thread.setName(NewThreadAction.NAME_PREFIX + "Idle");
+ lock.lock();
try {
- lock.wait(idleTimeout);
- } catch (InterruptedException e) {
- // ignore interrupts at this level
+ idleThreads++;
+ wakeup.await(idleTimeout, TimeUnit.MILLISECONDS);// releases lock and obtains when woken.
+ // Allow thread to expire if queue empty after waking.
+ if (queue.peek() == null) thread.interrupt();
+ } catch (InterruptedException ex) {
+ // Interrupt thread, another thread can pick up tasks.
+ thread.interrupt();
} finally {
idleThreads--;
+ lock.unlock();
}
- if (queue.isEmpty()) {
- break; // timed out
}
}
- task = (Task) queue.removeFirst();
- Thread.currentThread().setName(
- NewThreadAction.NAME_PREFIX + task.name);
}
- };
}
- }
-}
# This patch file was generated by NetBeans IDE
# Following Index: paths are relative to: /opt/src/River_Fixed_2nd_Try/peterConcurrentPolicy/qa
# This patch can be applied using context Tools: Patch action on respective folder.
# It uses platform neutral UTF-8 encoding and \n newlines.
# Above lines and this line are ignored by the patching process.
Index: src/com/sun/jini/qa/harness/QAConfig.java
--- src/com/sun/jini/qa/harness/QAConfig.java Base (BASE)
+++ src/com/sun/jini/qa/harness/QAConfig.java Locally Modified (Based On LOCAL)
@@ -48,9 +48,11 @@
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.List;
import java.util.logging.Logger;
import java.util.logging.Level;
import java.util.MissingResourceException;
@@ -198,102 +200,94 @@
* defined which are referenced by the configuration.
*/
private static MethodConstraints testLocatorConstraints;
+ private static final long serialVersionUID = 1L;
/** A unique string which is used to generate unique group names. */
- private String uniqueString;
+ private volatile String uniqueString;
- /** the name of the configurator class */
- private String configuratorClassName = null;
-
/** the configuration override providers */
- private ArrayList overrideProviders = new ArrayList();
+ private volatile List<OverrideProvider> overrideProviders = Collections.synchronizedList(new ArrayList<OverrideProvider>());
/** the failure analyzers */
- private ArrayList failureAnalyzers = new ArrayList();
+ private final List<FailureAnalyzer> failureAnalyzers = Collections.synchronizedList(new ArrayList<FailureAnalyzer>());
- /** The number of configurations to cycle through */
- private int configurationCount = -1;
-
/** Properties obtained from the user configuration file. */
- private Properties configProps;
+ private volatile Properties configProps;
/** properties obtained from the configuration group file */
- private Properties configSetProps = new Properties();
+ private volatile Properties configSetProps = new Properties();
/** Properties obtained from qaDefaults.properties. */
- private Properties defaultProps = new Properties();
+ private volatile Properties defaultProps = new Properties();
/** The test <code>Configuration</code> object, not serializable*/
- private transient Configuration configuration;
+ private volatile transient Configuration configuration;
/** The TestDescription */
- private TestDescription td;
+ private volatile TestDescription td;
/** Dynamic properties set on-the-fly by tests. */
- private Properties dynamicProps = new Properties();
+ private volatile Properties dynamicProps = new Properties();
/** Override properties, for cmd lines with non-default install props */
- private Properties propertyOverrides = null;
+ private volatile Properties propertyOverrides = null;
/** The original command line arguments. */
- private String[] args;
+ private final String[] args;
/** The configuration group tags for the test run */
- private String[] configTags;
+ private volatile String[] configTags;
/** The currently active configuration group tag */
- private String currentTag;
+ private volatile String currentTag;
/** The key name to track */
- private String trackKey;
+ private volatile String trackKey;
/** The set of hosts participating in this test run */
- private ArrayList hostList = new ArrayList();
+ private final List<String> hostList = Collections.synchronizedList(new ArrayList<String>());
/** The host index memory for the uniformrandom selection policy */
- private ArrayList selectedIndexes;
+ private volatile List<Integer> selectedIndexes ;
- /** Flag controlling use of IP address or host names during resolution */
- private boolean useAddress = false;
-
/** The resolver */
- private Resolver resolver;
+ private volatile Resolver resolver;
/** the jar file containing the tests */
- private String testJar = null;
+ private volatile String testJar = null;
/** the harness jar file name */
- private String harnessJar = null;
+ private volatile String harnessJar = null;
/** the test class loader */
- private transient ClassLoader testLoader;
+ private volatile transient ClassLoader testLoader;
/** the search list, usually empty */
- private String[] searchList = new String[0];
+ private volatile String[] searchList = new String[0];
/** lock object for suspend/resume run */
private transient Object runLock = new Object();
/** suspended state (for UI) */
- private boolean testSuspended = false;
+ private volatile boolean testSuspended = false;
/**
* The host selection index for the selection policy.
* It's intiialized to 1 in case the roundrobin policy
* is being used, which begins with the first slave.
*/
- private int hostIndex = 1;
+ private volatile int hostIndex = 1;
/** The rerun pass counter */
- private int passCount;
+ private volatile int passCount;
- private boolean callAutot = false;
+ private volatile boolean callAutot = false;
- private int testTotal = 0;
+ private volatile int testTotal = 0;
- private int testIndex = 0;
+ private volatile int testIndex = 0;
- private String resumeMessage;
+ private volatile String resumeMessage;
/**
* Static accessor for the config instance.
@@ -386,7 +380,7 @@
*/
public QAConfig(String[] args) throws TestException {
this.args = args;
- resolver = new Resolver(this);
+ resolver = new Resolver(this); // Unsafe publication
configProps = loadProperties(args[0]);
if (configProps.size() == 0) {
throw new TestException("Config file"
@@ -456,12 +450,14 @@
if (hostNames != null) {
StringTokenizer tok = new StringTokenizer(hostNames, "|");
if (tok.countTokens() > 1) {
+ synchronized (hostList){
while (tok.hasMoreTokens()) {
hostList.add(tok.nextToken());
}
}
}
}
+ }
/**
* Set the token <gethost> to have the value of the local host
@@ -521,7 +517,6 @@
{
stream.defaultReadObject();
harnessConfig = this;
- runLock = new Object();
}
/**
@@ -2113,6 +2108,7 @@
*/
boolean isMaster() {
boolean isMaster = true;
+ synchronized (hostList){
if (hostList.size() > 0) {
try {
InetAddress thisAddr = InetAddress.getLocalHost(); //XXX multinic??
@@ -2123,6 +2119,7 @@
logger.log(Level.SEVERE, "Unexpected exception", e);
}
}
+ }
return isMaster;
}
@@ -2269,7 +2266,7 @@
* <li>the overrides
* </ul>
* The last source which returns a non-null value is used, so
- * the search is done from lowest to highest precedence. Configration
+ * the search is done from lowest to highest precedence. Configuration
* definitions may be self-referential, causing a substitution
* to be performed from lower precedence definitions. For example
* assume the default property file had the following definition:
@@ -2284,7 +2281,7 @@
* would be <code>foo bar</code>.
* <p>
* Before the parameter value is returned, any '$' substitutions
- * are performed (recursively), and any occurances of the
+ * are performed (recursively), and any occurrences of the
* token <code><gethost></code> are replaced with the name of
* the host on which this code is running. If a parsing error
* is encountered during symbol resolution, an error message
@@ -2484,7 +2481,7 @@
* or an exception is thrown while instantiating the provider
*/
private void registerOverrideProviders() throws TestException {
- overrideProviders = new ArrayList();
+ overrideProviders = Collections.synchronizedList(new ArrayList<OverrideProvider>());
String[] providers =
parseString(getStringConfigVal("testOverrideProviders", null),
", \t");
@@ -2661,7 +2658,7 @@
initUniqueString(); // so each test uses a new unique value
this.td = td;
dynamicProps = new Properties();
- selectedIndexes = new ArrayList();
+ selectedIndexes = Collections.synchronizedList(new ArrayList<Integer>());
trackKey = getParameterString("track");
propertyOverrides = null; // paranoid, probably unnecessary
resolver.setToken("config", currentTag);
@@ -2730,7 +2727,7 @@
*
* @return the host list
*/
- public ArrayList getHostList() {
+ public List getHostList() {
return hostList;
}
@@ -2893,9 +2890,12 @@
private int nextIndex(boolean includeMaster) {
ArrayList available = new ArrayList();
int i = (includeMaster ? 0 : 1);
+ Integer selected;
+ List selectedInd = selectedIndexes; // copy reference to ensure sync on correct object.
+ synchronized (selectedInd){
while (i < hostList.size()) {
Integer iInt = new Integer(i++);
- if (! selectedIndexes.contains(iInt)) {
+ if (! selectedInd.contains(iInt)) {
available.add(iInt);
}
}
@@ -2903,12 +2903,13 @@
throw new IllegalStateException("Can't find an available index");
}
int randIndex = new Random().nextInt(available.size());
- Integer selected = (Integer) available.get(randIndex);
- selectedIndexes.add(selected);
+ selected = (Integer) available.get(randIndex);
+ selectedInd.add(selected);
int maxSize = (includeMaster ? hostList.size() : (hostList.size() - 1));
- if (selectedIndexes.size() == maxSize) {
- selectedIndexes.clear();
+ if (selectedInd.size() >= maxSize) {
+ selectedInd.clear();
}
+ }
return selected.intValue();
}
@@ -2973,6 +2974,7 @@
try {
runLock.wait();
} catch (InterruptedException ignore) {
+ Thread.currentThread().interrupt(); //Ignore and restore.
}
}
}
Index: src/com/sun/jini/qa/harness/QAConfiguration.java
--- src/com/sun/jini/qa/harness/QAConfiguration.java Base (BASE)
+++ src/com/sun/jini/qa/harness/QAConfiguration.java Locally Modified (Based On LOCAL)
@@ -34,8 +34,8 @@
*/
public class QAConfiguration implements Configuration {
- private Configuration testConfiguration;
- private Configuration defaultConfiguration;
+ private final Configuration testConfiguration;
+ private final Configuration defaultConfiguration;
public QAConfiguration(String[] options, QAConfig config)
throws ConfigurationException
Index: src/com/sun/jini/qa/harness/Resolver.java
--- src/com/sun/jini/qa/harness/Resolver.java Base (BASE)
+++ src/com/sun/jini/qa/harness/Resolver.java Locally Modified (Based On LOCAL)
@@ -55,8 +55,8 @@
class Resolver implements Serializable {
private static Logger logger = Logger.getLogger("com.sun.jini.qa.harness");
- private QAConfig config;
- private HashMap tokenMap = new HashMap();
+ private final QAConfig config;
+ private final HashMap tokenMap = new HashMap();
/**
* Construct an instance of a <code>Resolver</code>.
@@ -72,8 +72,10 @@
token = "<" + token + ">";
}
logger.log(Level.FINEST, "setting token " + token + " to " + value);
+ synchronized (tokenMap){
tokenMap.put(token, value);
}
+ }
/**
* Resolves a value in the input string by splitting it into words and
@@ -419,6 +421,7 @@
if (str == null) {
return null;
}
+ synchronized (tokenMap){
Iterator it = tokenMap.keySet().iterator();
while (it.hasNext()) {
String token = (String) it.next();
@@ -430,6 +433,7 @@
str = newString;
}
}
+ }
while (true) {
String newString = handleFileToken(str);
if (newString == str) {
Index: src/com/sun/jini/qa/harness/SlaveHarness.java
--- src/com/sun/jini/qa/harness/SlaveHarness.java Base (BASE)
+++ src/com/sun/jini/qa/harness/SlaveHarness.java Locally Modified (Based On LOCAL)
@@ -32,6 +32,7 @@
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
+import java.util.List;
import java.util.logging.Handler;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -136,7 +137,7 @@
* @throws TestException on connection timeout
*/
static void connect() throws TestException {
- ArrayList hostList = QAConfig.getConfig().getHostList();
+ List hostList = QAConfig.getConfig().getHostList();
if (hostList.size() < 2) {
return;
}
Index: src/com/sun/jini/qa/harness/SlaveTest.java
--- src/com/sun/jini/qa/harness/SlaveTest.java Base (BASE)
+++ src/com/sun/jini/qa/harness/SlaveTest.java Locally Modified (Based On LOCAL)
@@ -26,6 +26,7 @@
import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.logging.Logger;
import java.util.logging.Level;
@@ -123,7 +124,7 @@
* @param request the <code>SlaveRequest</code> to send
*/
public static void broadcast(SlaveRequest request) {
- ArrayList hostList = QAConfig.getConfig().getHostList();
+ List hostList = QAConfig.getConfig().getHostList();
if (hostList.size() < 2) {
return;
}
@@ -147,11 +148,11 @@
* zero or negative, this method returns immediately
*/
public static void waitForSlaveDeath(int timeout) {
- ArrayList hostList = QAConfig.getConfig().getHostList();
+ List hostList = QAConfig.getConfig().getHostList();
if (hostList.size() < 2) {
return;
}
- ArrayList slaveList = new ArrayList();
+ List slaveList = new ArrayList();
for (int i = 1; i < hostList.size(); i++) {
slaveList.add(hostList.get(i));
}
Index: src/com/sun/jini/test/share/BaseQATest.java
--- src/com/sun/jini/test/share/BaseQATest.java Base (BASE)
+++ src/com/sun/jini/test/share/BaseQATest.java Locally Modified (Based On LOCAL)
@@ -2474,36 +2474,35 @@
* simulated lookup services.
*/
protected void verifyAnnouncementsSent() {
- logger.log(Level.FINE,
- " number of announcements to wait for -- "
- +minNAnnouncements);
- logger.log(Level.FINE,
- " number of intervals to wait through -- "
- +nIntervalsToWait);
+ logger.log(Level.FINE, " number of announcements to wait for -- {0}",
+ minNAnnouncements);
+ logger.log(Level.FINE, " number of intervals to wait through -- {0}",
+ nIntervalsToWait);
+ Set discoProtSim; // Devensive copy to avoid sync while iterating.
synchronized (genMap){
- Iterator iter = genMap.keySet().iterator();
+ Set set = genMap.keySet();
+ discoProtSim = new HashSet(set.size());
+ discoProtSim.addAll(set);
+ }
+ Iterator iter = discoProtSim.iterator();
for(int i=0;iter.hasNext();i++) {
DiscoveryProtocolSimulator curGen =
(DiscoveryProtocolSimulator)iter.next();
logger.log(Level.FINE,
- " gen "+i
- +" - waiting ... announcements so far -- "
- +curGen.getNAnnouncementsSent());
+ " gen {0} - waiting ... announcements so far -- {1}",
+ new Object[]{i, curGen.getNAnnouncementsSent()});
for(int j=0; ((j<nIntervalsToWait)
&&(curGen.getNAnnouncementsSent()< minNAnnouncements));j++)
{
DiscoveryServiceUtil.delayMS(announceInterval);
logger.log(Level.FINE,
- " gen "+i
- +" - waiting ... announcements so far -- "
- +curGen.getNAnnouncementsSent());
+ " gen {0} - waiting ... announcements so far -- {1}",
+ new Object[]{i, curGen.getNAnnouncementsSent()});
}//end loop
logger.log(Level.FINE,
- " gen "+i
- +" - wait complete ... announcements -- "
- +curGen.getNAnnouncementsSent());
+ " gen {0} - wait complete ... announcements -- {1}",
+ new Object[]{i, curGen.getNAnnouncementsSent()});
}//end loop
- }
}//end verifyAnnouncementsSent
/** This method replaces, with the given set of groups, the current
Index: src/com/sun/jini/test/share/DiscoveryProtocolSimulator.java
--- src/com/sun/jini/test/share/DiscoveryProtocolSimulator.java Base (BASE)
+++ src/com/sun/jini/test/share/DiscoveryProtocolSimulator.java Locally Modified (Based On LOCAL)
@@ -154,6 +154,9 @@
/** Default timeout to set on sockets used for unicast discovery */
private static final int DEFAULT_SOCKET_TIMEOUT = 1*60*1000;
+ /** Socket timeout for multicast request receive() */
+ private static final int SOCKET_TIMEOUT = 5*60*1000;
+
private static Logger logger = Logger.getLogger("com.sun.jini.qa.harness");
private volatile InetAddress thisInetAddress = null;
@@ -485,17 +488,17 @@
}
/** True if thread has been interrupted */
- private boolean interrupted = false;
+ private volatile boolean interrupted = false;
/* This is a workaround for Thread.interrupt not working on
* MulticastSocket.receive on all platforms.
*/
- public synchronized void interrupt() {
+ public void interrupt() {
interrupted = true;
socket.close();
}
- public synchronized boolean isInterrupted() {
+ public boolean isInterrupted() {
return interrupted;
}
@@ -575,7 +578,7 @@
try {
listen = new ServerSocket(Constants.discoveryPort);
} catch (IOException ex){
- logger.log(Level.FINE, "Failed to bind to default port: " + Constants.discoveryPort, ex);
+ logger.log(Level.FINEST, "Unable to bind to default port, selecting ephemeral: " + Constants.discoveryPort, ex);
}
}
if (listen == null) {
@@ -583,30 +586,33 @@
listen = new ServerSocket(port);
} catch (BindException ex){
ex.fillInStackTrace();
- throw new IOException("Failed to bind to default port: " + port, ex);
+ throw new IOException("Failed to bind to port: " + port, ex);
}
}
this.listen = listen;
this.port = listen.getLocalPort();
+ listen.setSoTimeout(unicastTimeout);
}
/** True if thread has been interrupted */
- private boolean interrupted = false;
+ private volatile boolean interrupted = false;
/* This is a workaround for Thread.interrupt not working on
* ServerSocket.accept on all platforms. ServerSocket.close
* can't be used as a workaround, because it also doesn't work
* on all platforms.
*/
- public synchronized void interrupt() {
+ public void interrupt() {
interrupted = true;
try {
- (new Socket(InetAddress.getLocalHost(), port)).close();
- } catch (IOException e) {
+ listen.close(); // Try correct method first.
+ } catch (IOException ex){/*ignore*/}
+ try {
+ (new Socket(InetAddress.getLocalHost(), port)).close(); // Try workaround
+ } catch (IOException e) {/*ignore*/}
}
- }
- public synchronized boolean isInterrupted() {
+ public boolean isInterrupted() {
return interrupted;
}
@@ -644,17 +650,17 @@
/** Multicast socket to send packets on */
private final MulticastSocket socket;
- private boolean interrupted = false;
+ private volatile boolean interrupted = false;
/* This is a workaround for Thread.interrupt not working due
* to the logging system sometimes throwing away InterruptedIOException
*/
- public synchronized void interrupt() {
+ public void interrupt() {
interrupted = true;
super.interrupt();
}
- public synchronized boolean isInterrupted() {
+ public boolean isInterrupted() {
return interrupted;
}
@@ -676,15 +682,17 @@
}
}
- public synchronized void run() {
+ public void run() { // synchronization prevented interruption - removed, run only ever run by this thread anyway.
if (multicastInterfaces != null && multicastInterfaces.length == 0)
{
return;
}
try {
while (!isInterrupted() && announce(memberGroups)) {
+ synchronized (this){
wait(multicastAnnouncementInterval);
}
+ }
} catch (InterruptedException e) {
}
// disable this to allow simulation of disappearance of multicast announcements
Index: src/com/sun/jini/test/spec/discoveryservice/event/MulticastMonitorStop.td
--- src/com/sun/jini/test/spec/discoveryservice/event/MulticastMonitorStop.td Base (BASE)
+++ src/com/sun/jini/test/spec/discoveryservice/event/MulticastMonitorStop.td Locally Modified (Based On LOCAL)
@@ -2,3 +2,7 @@
testCategories=discoveryservice,discoveryservice_spec,discoveryserviceevent
include0=../discoveryservice.properties
include1=simReg6_0regInfo3_0.properties
+#testjvmargs=-Xdebug,\
+#-Xrunjdwp:transport=dt_socket+,address=8000+,server=y+,suspend=y,\
+#-Djava.security.debug=access:failure,\
+#${testjvmargs}
\ No newline at end of file
Index: src/com/sun/jini/test/spec/lookupdiscovery/AbstractBaseTest.java
--- src/com/sun/jini/test/spec/lookupdiscovery/AbstractBaseTest.java Base (BASE)
+++ src/com/sun/jini/test/spec/lookupdiscovery/AbstractBaseTest.java Locally Modified (Based On LOCAL)
@@ -35,6 +35,7 @@
import java.io.IOException;
import java.rmi.RemoteException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
/**
@@ -57,9 +58,9 @@
*/
abstract public class AbstractBaseTest extends BaseQATest {
- protected LookupDiscovery lookupDiscovery = null;
- protected ArrayList lookupDiscoveryList = new ArrayList(1);
- protected LookupListener mainListener = null;
+ protected volatile LookupDiscovery lookupDiscovery = null;
+ protected final List lookupDiscoveryList = Collections.synchronizedList(new ArrayList(1));
+ protected volatile LookupListener mainListener = null;
/** Performs actions necessary to prepare for execution of the
* current test as follows:
Index: src/com/sun/jini/test/spec/lookupdiscovery/Discovered.java
--- src/com/sun/jini/test/spec/lookupdiscovery/Discovered.java Base (BASE)
+++ src/com/sun/jini/test/spec/lookupdiscovery/Discovered.java Locally Modified (Based On LOCAL)
@@ -56,10 +56,10 @@
*/
public class Discovered extends AbstractBaseTest {
- protected List locGroupsList = new ArrayList(1);
- protected LookupDiscovery ldToUse = null;
- protected LookupListener listenerToUse = null;
- protected String[] groupsToDiscover = DiscoveryGroupManagement.NO_GROUPS;
+ protected volatile List locGroupsList = new ArrayList(1);
+ protected volatile LookupDiscovery ldToUse = null;
+ protected volatile LookupListener listenerToUse = null;
+ protected volatile String[] groupsToDiscover = DiscoveryGroupManagement.NO_GROUPS;
/** Performs actions necessary to prepare for execution of the
* current test (refer to the description of this method in the
Index: src/com/sun/jini/test/spec/lookupdiscovery/MulticastMonitorStopReplace.java
--- src/com/sun/jini/test/spec/lookupdiscovery/MulticastMonitorStopReplace.java Base (BASE)
+++ src/com/sun/jini/test/spec/lookupdiscovery/MulticastMonitorStopReplace.java Locally Modified (Based On LOCAL)
@@ -97,16 +97,17 @@
* examines the contents of those maps. So we don't want those
* maps to change until setLookupsToDiscover returns.
*/
- synchronized(mainListener) {
+ LookupListener listener = mainListener;
+ synchronized(listener) {
stopAnnouncements();
/* Replace current groups with new groups to cause discards */
List locGroupsPairList = replaceMemberGroups();
/* Set the expected discard event info */
- mainListener.setLookupsToDiscover
+ listener.setLookupsToDiscover
(locGroupsPairList,
toGroupsArray(initLookupsToStart));
}//end sync(mainListener)
- waitForDiscard(mainListener);
+ waitForDiscard(listener);
}//end run
}//end class MulticastMonitorStopReplace
Index: src/com/sun/jini/test/spec/lookupdiscovery/MulticastMonitorStopReplace.td
--- src/com/sun/jini/test/spec/lookupdiscovery/MulticastMonitorStopReplace.td Base (BASE)
+++ src/com/sun/jini/test/spec/lookupdiscovery/MulticastMonitorStopReplace.td Locally Modified (Based On LOCAL)
@@ -2,3 +2,6 @@
testCategories=lookupdiscovery,lookupdiscovery_spec,speclookupdiscovery
include0=lookupdiscovery.properties
include1=com/sun/jini/test/share/simReg3_0.properties
+testjvmargs=-Xdebug,\
+-Xrunjdwp:transport=dt_socket+,address=8000+,server=y+,suspend=y,\
+${testjvmargs}
\ No newline at end of file
