You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@river.apache.org by pe...@apache.org on 2011/06/19 02:43:42 UTC
svn commit: r1137269 [1/2] - in
/river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime:
BasicExportTable.java Binding.java DgcRequestDispatcher.java
ImplRefManager.java JvmLifeSupport.java Lease.java ObjectTable.java
SequenceEntry.java Target.java
Author: peter_firmstone
Date: Sun Jun 19 00:43:42 2011
New Revision: 1137269
URL: http://svn.apache.org/viewvc?rev=1137269&view=rev
Log:
River-142 Slightly different to the original patch, this commit fixes delayed garbage collection synchronization issues by processing expired leases immediately, without locking the entire object table. Lease has been changed to be responsible for expiry, notification and processing (on the garbage collection thread), synchronized internally. A Lease in the object table must now be replaced once it expires and cannot be renewed, it is removed from the table after it is marked expired, to prevent garbage collection of potentially active leases. Internal classes have been separated from ObjectTable and BasicExportTable to encapsulate or simplify synchronization and locking. Target is now more faithful to Exporter.unexport's documented behaviour and interrupts dispatched method calls when force is true when possible.
I wasn't able to create a test to simulate the original failure condition, to do so requires a large number of leases to be processed (to create a time window to process garbage collection of leases after releasing the table lock) and proper timing of dirty calls, garbage collection and clean calls. The new code processes the lease immediately and isn't subject to the time window.
Added:
river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Binding.java (with props)
river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/DgcRequestDispatcher.java (with props)
river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/JvmLifeSupport.java (with props)
river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Lease.java (with props)
river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/SequenceEntry.java (with props)
river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Target.java (with props)
Modified:
river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/BasicExportTable.java
river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/ImplRefManager.java
river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/ObjectTable.java
Modified: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/BasicExportTable.java
URL: http://svn.apache.org/viewvc/river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/BasicExportTable.java?rev=1137269&r1=1137268&r2=1137269&view=diff
==============================================================================
--- river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/BasicExportTable.java (original)
+++ river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/BasicExportTable.java Sun Jun 19 00:43:42 2011
@@ -19,15 +19,13 @@
package com.sun.jini.jeri.internal.runtime;
import java.io.IOException;
-import java.io.InterruptedIOException;
import java.rmi.Remote;
import java.rmi.server.ExportException;
-import java.rmi.server.Unreferenced;
-import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.Collections;
import java.util.List;
-import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import net.jini.id.Uuid;
import net.jini.jeri.Endpoint;
import net.jini.jeri.InvocationDispatcher;
@@ -35,8 +33,6 @@ import net.jini.jeri.RequestDispatcher;
import net.jini.jeri.ServerEndpoint;
import net.jini.jeri.ServerEndpoint.ListenCookie;
import net.jini.jeri.ServerEndpoint.ListenEndpoint;
-import net.jini.jeri.ServerEndpoint.ListenHandle;
-import net.jini.security.Security;
/**
* An ObjectTable front end for exporting remote objects with a
@@ -49,23 +45,16 @@ import net.jini.security.Security;
**/
public final class BasicExportTable {
- /**
- * listen pool marker value to signal that a listen operation on a
- * ListenEndpoint is currently being started by another thread
- **/
- private static final Object PENDING = new Object();
-
/** underlying object table */
private final ObjectTable objectTable = new ObjectTable();
- /** guards listenPool and all Binding.exportsInProgress fields */
- private final Object lock = new Object();
-
/**
* pool of endpoints that we're listening on:
- * maps SameClassKey(ServerEndpoint.ListenEndpoint) to Binding
+ * maps SameClassKey(ServerEndpoint.ListenEndpoint) to Binding.
+ * A binding removes itself from the listen pool.
**/
- private final Map listenPool = new HashMap();
+ private final ConcurrentMap<SameClassKey,Binding> listenPool =
+ new ConcurrentHashMap<SameClassKey,Binding>(128);// 128 to reduce map resizing
/**
* Creates a new instance.
@@ -84,7 +73,7 @@ public final class BasicExportTable {
throws ExportException
{
List bindings = null;
- ObjectTable.Target target = null;
+ Target target = null;
Endpoint endpoint;
try {
LC listenContext = new LC();
@@ -101,7 +90,7 @@ public final class BasicExportTable {
new RequestDispatcher[bindings.size()];
for (int i = 0; i < requestDispatchers.length; i++) {
requestDispatchers[i] =
- ((Binding) bindings.get(i)).requestDispatcher;
+ ((Binding) bindings.get(i)).getRequestDispatcher();
}
target = objectTable.export(
impl, requestDispatchers, allowDGC, keepAlive, id);
@@ -115,9 +104,7 @@ public final class BasicExportTable {
*/
for (int i = 0; i < bindings.size(); i++) {
Binding binding = (Binding) bindings.get(i);
- synchronized (lock) {
- binding.exportsInProgress--;
- }
+ binding.decrementExportInProgress();
/*
* If export wasn't successful, check to see if
* binding can be released.
@@ -142,10 +129,10 @@ public final class BasicExportTable {
**/
public static final class Entry {
private final List bindings;
- private final ObjectTable.Target target;
+ private final Target target;
private final Endpoint endpoint;
- Entry(List bindings, ObjectTable.Target target, Endpoint endpoint) {
+ Entry(List bindings, Target target, Endpoint endpoint) {
this.bindings = bindings;
this.target = target;
this.endpoint = endpoint;
@@ -189,45 +176,50 @@ public final class BasicExportTable {
private Binding getBinding(ListenEndpoint listenEndpoint)
throws IOException
{
- Object key = new SameClassKey(listenEndpoint);
+ SameClassKey key = new SameClassKey(listenEndpoint);
Binding binding = null;
- synchronized (lock) {
- do {
- Object value = listenPool.get(key);
- if (value instanceof Binding) {
- binding = (Binding) value;
- binding.exportsInProgress++;
- return binding;
- } else if (value == PENDING) {
- try {
- lock.wait();
- } catch (InterruptedException e) {
- throw new InterruptedIOException();
- }
- continue;
- } else {
- assert value == null;
- listenPool.put(key, PENDING);
- break;
- }
- } while (true);
- }
- try {
- // start listen operation without holding global lock
- binding = new Binding(listenEndpoint);
- } finally {
- synchronized (lock) {
- assert listenPool.get(key) == PENDING;
- if (binding != null) {
- listenPool.put(key, binding);
- binding.exportsInProgress++;
- } else {
- listenPool.remove(key);
- }
- lock.notifyAll();
- }
- }
- return binding;
+ // This while loop ensures that a binding has it's exportInProgress
+ // field incremented and the binding was not closed prior.
+ // Once the exportInProgress field is incremented, the binding will stay active.
+ // It is still possible for activation to be unsuccessful, resulting
+ // in an IOException.
+ // The reason for this while loop, is Binding's remove themselves from
+ // the listenPool if inactive, a binding may be removed from the
+ // listenPool by another thread without the current threads knowledge.
+ // This will only happen while the binding has no Exports in progress.
+ // Thus the increment calls are checked to be active;
+ while (binding == null){
+ binding = listenPool.get(key);
+ if ( binding == null){
+ binding = new Binding(listenEndpoint,objectTable, listenPool);
+ Binding existed = listenPool.putIfAbsent(key, binding);
+ if (existed != null){
+ binding = existed;
+ boolean active = binding.incrementExportInProgress();
+ if (!active){
+ binding = null;
+ }
+ continue;
+ } else {
+ boolean active = binding.activate();
+ if (!active){
+ binding = null;
+ }
+ continue;
+ }
+ } else {
+ // Although unlikely the binding could become inactive
+ // after retrieval, since the operation of getting and checking is not atomic.
+ // If inactive, the binding has removed itself from the listenPool.
+ boolean active = binding.incrementExportInProgress();
+ if (!active) {
+ binding = null;
+ // This binding will have removed itself from listenPool.
+ }
+ }
+ }
+ binding.activate(); //Prevent a thread returning normally with an inactive object
+ return binding;
}
/**
@@ -235,12 +227,13 @@ public final class BasicExportTable {
* and gets the corresponding bindings using the listen pool.
**/
private class LC implements ServerEndpoint.ListenContext {
- private boolean done = false;
- private final List bindings = new ArrayList();
+ private volatile boolean done = false;
+ private final List<Binding> bindings =
+ Collections.synchronizedList(new ArrayList<Binding>());
LC() { }
- public synchronized ListenCookie addListenEndpoint(
+ public ListenCookie addListenEndpoint(
ListenEndpoint listenEndpoint)
throws IOException
{
@@ -252,75 +245,13 @@ public final class BasicExportTable {
listenEndpoint.checkPermissions();
Binding binding = getBinding(listenEndpoint);
- bindings.add(binding);
- return binding.listenHandle.getCookie();
+ bindings.add(binding);
+ return binding.getListenHandle().getCookie();
}
- synchronized List getFinalBindings() {
+ private List getFinalBindings() {
done = true;
return bindings;
}
}
-
- /**
- * A bound ListenEndpoint and the associated ListenHandle and
- * RequestDispatcher.
- **/
- private class Binding {
- private final ListenEndpoint listenEndpoint;
- final RequestDispatcher requestDispatcher;
- final ListenHandle listenHandle;
-
- int exportsInProgress = 0; // guarded by outer "lock"
-
- /**
- * Creates a binding for the specified ListenEndpoint by
- * attempting to listen on it.
- **/
- Binding(final ListenEndpoint listenEndpoint) throws IOException {
- this.listenEndpoint = listenEndpoint;
- requestDispatcher =
- objectTable.createRequestDispatcher(new Unreferenced() {
- public void unreferenced() { checkReferenced(); }
- });
- try {
- /*
- * We don't want this (potentially) shared listen
- * operation to inherit the access control context of
- * the current callers arbitrarily (their permissions
- * were already checked by the ListenContext, and the
- * ObjectTable will take care of checking permissions
- * per requests against the appropriate callers'
- * access control context).
- */
- listenHandle = (ListenHandle)
- Security.doPrivileged(new PrivilegedExceptionAction() {
- public Object run() throws IOException {
- return listenEndpoint.listen(requestDispatcher);
- }
- });
- } catch (java.security.PrivilegedActionException e) {
- throw (IOException) e.getException();
- }
- }
-
- /**
- * Checks whether there are any objects currently exported to
- * this binding's RequestDispatcher or if there are any
- * exports in progress for this binding; if there are neither,
- * this binding is removed from the listen pool and its listen
- * operation is closed.
- **/
- void checkReferenced() {
- synchronized (lock) {
- if (exportsInProgress > 0 ||
- objectTable.isReferenced(requestDispatcher))
- {
- return;
- }
- listenPool.remove(new SameClassKey(listenEndpoint));
- }
- listenHandle.close();
- }
- }
}
Added: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Binding.java
URL: http://svn.apache.org/viewvc/river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Binding.java?rev=1137269&view=auto
==============================================================================
--- river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Binding.java (added)
+++ river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Binding.java Sun Jun 19 00:43:42 2011
@@ -0,0 +1,118 @@
+/*
+ * 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 com.sun.jini.jeri.internal.runtime;
+
+import java.io.IOException;
+import java.rmi.server.Unreferenced;
+import java.security.PrivilegedActionException;
+import java.security.PrivilegedExceptionAction;
+import java.util.concurrent.ConcurrentMap;
+import net.jini.jeri.RequestDispatcher;
+import net.jini.jeri.ServerEndpoint.ListenEndpoint;
+import net.jini.jeri.ServerEndpoint.ListenHandle;
+import net.jini.security.Security;
+
+/**
+ * A bound ListenEndpoint and the associated ListenHandle and
+ * RequestDispatcher.
+ *
+ * @since 2.2.0
+ * @author Peter Firmstone.
+ */
+class Binding {
+
+ private final ListenEndpoint listenEndpoint;
+ private final ObjectTable table;
+ private final ConcurrentMap listenPool;
+ private RequestDispatcher requestDispatcher;
+ private ListenHandle listenHandle;
+ boolean activated;
+ boolean closed;
+ // Changed to start at 1, so export in progress starts with construction.
+ private int exportsInProgress = 1;
+
+ Binding(final ListenEndpoint listenEndpoint, ObjectTable table,
+ ConcurrentMap listenPool) throws IOException {
+ super();
+ this.table = table;
+ this.listenPool = listenPool;
+ this.listenEndpoint = listenEndpoint;
+ activated = false;
+ closed = false;
+ }
+
+ synchronized boolean incrementExportInProgress() {
+ if (closed) {
+ return false;
+ }
+ exportsInProgress++;
+ return true;
+ }
+
+ synchronized void decrementExportInProgress() {
+ if (closed) {
+ throw new IllegalStateException("Cannot decrement closed Binding");
+ }
+ exportsInProgress--;
+ }
+
+ synchronized boolean activate() throws IOException {
+ if (closed) {
+ return false;
+ }
+ if (activated) {
+ return true;
+ }
+ requestDispatcher = table.createRequestDispatcher(new Unreferenced() {
+
+ public void unreferenced() {
+ checkReferenced();
+ }
+ });
+ try {
+ listenHandle = (ListenHandle) Security.doPrivileged(new PrivilegedExceptionAction() {
+
+ public Object run() throws IOException {
+ return listenEndpoint.listen(requestDispatcher);
+ }
+ });
+ } catch (PrivilegedActionException e) {
+ throw (IOException) e.getException();
+ }
+ activated = true;
+ return true;
+ }
+
+ synchronized void checkReferenced() {
+ if (exportsInProgress > 0 || table.isReferenced(requestDispatcher)) {
+ return;
+ }
+ listenPool.remove(new SameClassKey(listenEndpoint), this);
+ listenHandle.close();
+ closed = true;
+ }
+
+ synchronized RequestDispatcher getRequestDispatcher() {
+ return requestDispatcher;
+ }
+
+ synchronized ListenHandle getListenHandle() {
+ return listenHandle;
+ }
+}
Propchange: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Binding.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/DgcRequestDispatcher.java
URL: http://svn.apache.org/viewvc/river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/DgcRequestDispatcher.java?rev=1137269&view=auto
==============================================================================
--- river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/DgcRequestDispatcher.java (added)
+++ river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/DgcRequestDispatcher.java Sun Jun 19 00:43:42 2011
@@ -0,0 +1,252 @@
+/*
+ * 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 com.sun.jini.jeri.internal.runtime;
+
+import com.sun.jini.jeri.internal.runtime.ObjectTable.NoSuchObject;
+import com.sun.jini.logging.Levels;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.io.OutputStream;
+import java.lang.reflect.Method;
+import java.rmi.server.ExportException;
+import java.rmi.server.Unreferenced;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import net.jini.core.constraint.InvocationConstraints;
+import net.jini.export.ServerContext;
+import net.jini.id.Uuid;
+import net.jini.id.UuidFactory;
+import net.jini.io.MarshalInputStream;
+import net.jini.io.UnsupportedConstraintException;
+import net.jini.jeri.BasicInvocationDispatcher;
+import net.jini.jeri.InboundRequest;
+import net.jini.jeri.InvocationDispatcher;
+import net.jini.jeri.RequestDispatcher;
+import net.jini.jeri.ServerCapabilities;
+
+/**
+ *
+ * @author peter
+ */
+public class DgcRequestDispatcher implements RequestDispatcher {
+ private static final Logger logger =
+ Logger.getLogger("net.jini.jeri.BasicJeriExporter");
+
+ private static final Collection<Method> dgcDispatcherMethods =
+ new ArrayList<Method>(2);
+ static {
+ Method[] methods = DgcServer.class.getMethods();
+ for (int i = 0; i < methods.length; i++) {
+ final Method m = methods[i];
+ AccessController.doPrivileged(new PrivilegedAction() {
+ public Object run() {
+ m.setAccessible(true);
+ return null;
+ }
+ });
+ dgcDispatcherMethods.add(m);
+ }
+ }
+
+ private static final ServerCapabilities dgcServerCapabilities =
+ new ServerCapabilities() {
+ public InvocationConstraints checkConstraints(
+ InvocationConstraints constraints)
+ throws UnsupportedConstraintException
+ {
+ assert constraints.equals(InvocationConstraints.EMPTY);
+ return InvocationConstraints.EMPTY;
+ }
+ };
+
+ private final Unreferenced unrefCallback;
+ private final ObjectTable table;
+ private final ConcurrentMap<Uuid,Target> idTable =
+ new ConcurrentHashMap<Uuid,Target>();
+ private final AtomicInteger dgcEnabledCount = new AtomicInteger(); // guarded by idTable lock
+
+ private final InvocationDispatcher dgcDispatcher;
+ private final DgcServer dgcServer;
+
+ DgcRequestDispatcher(Unreferenced unrefCallback, ObjectTable table ) {
+ this.unrefCallback = unrefCallback;
+ this.table = table;
+ try {
+ dgcDispatcher =
+ new BasicInvocationDispatcher(
+ dgcDispatcherMethods, dgcServerCapabilities,
+ null, null, this.getClass().getClassLoader())
+ {
+ protected ObjectInputStream createMarshalInputStream(
+ Object impl,
+ InboundRequest request,
+ boolean integrity,
+ Collection context)
+ throws IOException
+ {
+ ClassLoader loader = getClassLoader();
+ return new MarshalInputStream(
+ request.getRequestInputStream(),
+ loader, integrity, loader,
+ Collections.unmodifiableCollection(context));
+ // useStreamCodebases() not invoked
+ }
+ };
+ } catch (ExportException e) {
+ throw new AssertionError();
+ }
+ this.dgcServer = table.getDgcServer(this);
+ }
+
+ boolean forTable(ObjectTable table) {
+ return this.table == table;
+ }
+
+ boolean isReferenced() {
+ return !idTable.isEmpty();
+ }
+
+ Target get(Uuid id) {
+ return idTable.get(id);
+ }
+
+ void put(Target target) throws ExportException {
+ Uuid id = target.getObjectIdentifier();
+ if (id.equals(Jeri.DGC_ID)) {
+ throw new ExportException(
+ "object identifier reserved for DGC");
+ }
+ Target exists = idTable.putIfAbsent(id, target);
+ if (exists != null){
+ throw new ExportException(
+ "object identifier already in use");
+ }
+ if (target.getEnableDGC()) {
+ dgcEnabledCount.incrementAndGet();
+ }
+ }
+
+ void remove(Target target, boolean gc) {
+ Uuid id = target.getObjectIdentifier();
+ boolean removed = idTable.remove(id, target);
+ if (target.getEnableDGC() && removed) {
+ int count = dgcEnabledCount.decrementAndGet();
+ assert count >= 0;
+ }
+
+ if (gc && idTable.isEmpty()) {
+ /*
+ * We have to be careful to make this callback without holding
+ * the lock for idTable, because the callback implementation
+ * will likely be code that calls this object's isReferenced
+ * method in its own synchronized block.
+ *
+ * This also means it is possible (although unlikely) for the
+ * idtable to become non empty before making this call.
+ */
+ unrefCallback.unreferenced();
+ }
+ }
+
+ private boolean hasDgcEnabledTargets() {
+ return dgcEnabledCount.get() > 0;
+ }
+
+ public void dispatch(InboundRequest request) {
+ try {
+ InputStream in = request.getRequestInputStream();
+ Uuid id = UuidFactory.read(in);
+ Target target = null;
+ if (logger.isLoggable(Level.FINEST)) {
+ logger.log(Level.FINEST, "id={0}", id);
+ }
+
+ try {
+ /*
+ * The DGC object identifier is hardwired here,
+ * rather than install it in idTable; this
+ * eliminates the need to worry about not counting
+ * the DGC server as an exported object in the
+ * table, and it doesn't need all of the machinery
+ * that Target provides.
+ */
+ if (id.equals(Jeri.DGC_ID)) {
+ dispatchDgcRequest(request);
+ return;
+ }
+
+ target = get(id);
+ if (target == null) {
+ logger.log(Level.FINEST, "id not in table");
+ throw new NoSuchObject();
+ }
+ target.dispatch(request);
+ } catch (NoSuchObject e) {
+ in.close();
+ OutputStream out = request.getResponseOutputStream();
+ out.write(Jeri.NO_SUCH_OBJECT);
+ out.close();
+
+ if (logger.isLoggable(Levels.FAILED)) {
+ logger.log(Levels.FAILED, "no such object: {0}", id);
+ }
+ }
+ } catch (IOException e) {
+ request.abort();
+
+ if (logger.isLoggable(Levels.FAILED)) {
+ logger.log(Levels.FAILED,
+ "I/O exception dispatching request", e);
+ }
+ }
+ }
+
+ private void dispatchDgcRequest(final InboundRequest request)
+ throws IOException, NoSuchObject
+ {
+ if (!hasDgcEnabledTargets()) {
+ logger.log(Level.FINEST, "no DGC-enabled targets");
+ throw new NoSuchObject();
+ }
+
+ OutputStream out = request.getResponseOutputStream();
+ out.write(Jeri.OBJECT_HERE);
+
+ final Collection context = new ArrayList(5);
+ request.populateContext(context);
+
+ ServerContext.doWithServerContext(new Runnable() {
+ public void run() {
+ dgcDispatcher.dispatch(dgcServer, request, context);
+ }
+ }, Collections.unmodifiableCollection(context));
+
+ }
+}
Propchange: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/DgcRequestDispatcher.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/ImplRefManager.java
URL: http://svn.apache.org/viewvc/river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/ImplRefManager.java?rev=1137269&r1=1137268&r2=1137269&view=diff
==============================================================================
--- river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/ImplRefManager.java (original)
+++ river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/ImplRefManager.java Sun Jun 19 00:43:42 2011
@@ -18,7 +18,7 @@
package com.sun.jini.jeri.internal.runtime;
-import com.sun.jini.jeri.internal.runtime.ObjectTable.Target;
+import com.sun.jini.jeri.internal.runtime.Target;
import com.sun.jini.thread.NewThreadAction;
import java.lang.ref.Reference;
import java.lang.ref.ReferenceQueue;
@@ -69,7 +69,7 @@ final class ImplRefManager {
private final Object lock = new Object();
/** maps WeakKey(impl) to ImplRef(WeakKey(impl)) */
- private final Map weakImplTable = new HashMap();
+ private final Map<Reference,ImplRef> weakImplTable = new HashMap<Reference,ImplRef>();
/** thread to process garbage collected impls */
private Thread reaper = null;
@@ -99,7 +99,7 @@ final class ImplRefManager {
*/
Reference lookupKey = new WeakKey(impl, reapQueue);
synchronized (lock) {
- ImplRef implRef = (ImplRef) weakImplTable.get(lookupKey);
+ ImplRef implRef = weakImplTable.get(lookupKey);
if (implRef == null) {
implRef = new ImplRef(lookupKey);
weakImplTable.put(lookupKey, implRef);
Added: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/JvmLifeSupport.java
URL: http://svn.apache.org/viewvc/river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/JvmLifeSupport.java?rev=1137269&view=auto
==============================================================================
--- river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/JvmLifeSupport.java (added)
+++ river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/JvmLifeSupport.java Sun Jun 19 00:43:42 2011
@@ -0,0 +1,144 @@
+/*
+ * 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 com.sun.jini.jeri.internal.runtime;
+
+import com.sun.jini.thread.NewThreadAction;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * This class maintains a thread if necessary, in the wait state, to prevent
+ * the jvm shutting down while remote objects hold strong references in the
+ * DGC.
+ *
+ * This implementation is more about scalability than performance, when a jvm
+ * only has a small number of remote objects exported, blocking is not likely
+ * to cause a performance problem, access to keepAliveCount blocked.
+ *
+ * However with very large numbers of remote objects exported, in a dynamic
+ * environment, blocking is unlikely to be an issue, in this case the read locks
+ * will remain uncontended as the blocking write lock is only required as
+ * the number of exported object approach zero.
+ *
+ * If the thread is interrupted, it will pass away, regardless of the number of
+ * objects exported.
+ *
+ * @since 2.2.0
+ * @author Peter Firmstone
+ */
+public class JvmLifeSupport {
+ /** lock guarding keepAliveCount and keeper */
+ private final ReadWriteLock rwl;
+ private final Lock rl;
+ private final Lock wl;
+
+ /** number of objects exported with keepAlive == true */
+ private final AtomicInteger keepAliveCount;
+
+ /** thread to keep VM alive while keepAliveCount > 0 */
+ private volatile Thread keeper;
+
+ JvmLifeSupport(){
+ rwl = new ReentrantReadWriteLock();
+ rl = rwl.readLock();
+ wl = rwl.writeLock();
+ keepAliveCount = new AtomicInteger();
+ keeper = null;
+ }
+
+ /**
+ * Increments the count of objects exported with keepAlive true,
+ * starting a non-daemon thread if necessary.
+ *
+ * The old implementation contained in ObjectTable, used synchronization
+ * on a single lock for incrementing and decrementing to judge when an
+ * idle thread should be created or interrupted.
+ *
+ **/
+ void incrementKeepAliveCount() {
+ int value;
+ rl.lock();
+ try {
+ value = keepAliveCount.getAndIncrement();
+ } finally {
+ rl.unlock();
+ }
+ if (value < 3){
+ check();
+ }
+ }
+
+ /**
+ * Decrements the count of objects exported with keepAlive true,
+ * stopping the non-daemon thread if decremented to zero.
+ **/
+ void decrementKeepAliveCount() {
+ int value;
+ rl.lock();
+ try {
+ value = keepAliveCount.decrementAndGet();
+ } finally {
+ rl.unlock();
+ }
+ if (value < 3){
+ check();
+ }
+ }
+
+ private void check(){
+ wl.lock();
+ try {
+ int count = keepAliveCount.get();
+ if (count == 0){
+ assert keeper != null;
+ AccessController.doPrivileged(new PrivilegedAction() {
+ public Object run() {
+ keeper.interrupt();
+ return null;
+ }
+ });
+ keeper = null;
+ } else if ( count > 0){
+ if (keeper == null) {
+ // This thread keeps the jvm alive, while remote objects
+ // exist and all local processes have completed.
+ keeper = (Thread) AccessController.doPrivileged(
+ new NewThreadAction(new Runnable() {
+ public void run() {
+ try {
+ while (!Thread.currentThread().isInterrupted()) {
+ Thread.sleep(Long.MAX_VALUE);
+ }
+ } catch (InterruptedException e) {
+ // pass away if interrupted
+ }
+ }
+ }, "KeepAlive", false));
+ keeper.start();
+ }
+ }
+ } finally {
+ wl.unlock();
+ }
+ }
+}
Propchange: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/JvmLifeSupport.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Lease.java
URL: http://svn.apache.org/viewvc/river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Lease.java?rev=1137269&view=auto
==============================================================================
--- river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Lease.java (added)
+++ river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Lease.java Sun Jun 19 00:43:42 2011
@@ -0,0 +1,98 @@
+/*
+ * 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 com.sun.jini.jeri.internal.runtime;
+
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+import net.jini.id.Uuid;
+
+/**
+ *
+ * @since 2.2.0
+ * @author Peter Firmstone.
+ */
+class Lease {
+
+ private final Uuid clientID;
+ private final Set<Target> notifySet = new HashSet<Target>(3);
+ // guarded?
+ private long expiration;
+ // guarded by leaseTable lock
+ /* Once all targets have been removed, Lease is locked for removal.
+ */
+ private volatile boolean lockForRemoval;
+
+ Lease(Uuid clientID, long duration) {
+ super();
+ this.clientID = clientID;
+ expiration = System.currentTimeMillis() + duration;
+ lockForRemoval = false;
+ }
+
+ Uuid getClientID() {
+ return clientID;
+ }
+
+ boolean renew(long duration) {
+ synchronized (this) {
+ if (lockForRemoval) {
+ return false;
+ }
+ long newExpiration = System.currentTimeMillis() + duration;
+ if (newExpiration > expiration) {
+ expiration = newExpiration;
+ }
+ return true;
+ }
+ }
+
+ boolean notifyIfExpired(long now) {
+ boolean expired = false;
+ synchronized (this) {
+ expired = expiration < now;
+ if (expired) {
+ lockForRemoval = true;
+ Iterator<Target> i = notifySet.iterator();
+ while (i.hasNext()) {
+ Target t = i.next();
+ t.leaseExpired(clientID);
+ i.remove();
+ }
+ }
+ }
+ return expired;
+ }
+
+ void remove(Target target) {
+ synchronized (this) {
+ notifySet.remove(target);
+ }
+ }
+
+ boolean add(Target target) {
+ synchronized (this) {
+ if (lockForRemoval) {
+ return false;
+ }
+ notifySet.add(target);
+ return true;
+ }
+ }
+}
Propchange: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/Lease.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/ObjectTable.java
URL: http://svn.apache.org/viewvc/river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/ObjectTable.java?rev=1137269&r1=1137268&r2=1137269&view=diff
==============================================================================
--- river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/ObjectTable.java (original)
+++ river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/ObjectTable.java Sun Jun 19 00:43:42 2011
@@ -19,41 +19,19 @@
package com.sun.jini.jeri.internal.runtime;
import com.sun.jini.jeri.internal.runtime.ImplRefManager.ImplRef;
-import com.sun.jini.logging.Levels;
import com.sun.jini.thread.NewThreadAction;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.ObjectInputStream;
-import java.io.OutputStream;
-import java.lang.reflect.Method;
import java.rmi.Remote;
import java.rmi.server.ExportException;
import java.rmi.server.Unreferenced;
import java.security.AccessController;
-import java.security.PrivilegedAction;
-import java.security.PrivilegedExceptionAction;
-import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
-import java.util.Map;
-import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.logging.Level;
import java.util.logging.Logger;
-import net.jini.export.ServerContext;
import net.jini.id.Uuid;
-import net.jini.id.UuidFactory;
-import net.jini.io.MarshalInputStream;
-import net.jini.io.UnsupportedConstraintException;
-import net.jini.jeri.BasicInvocationDispatcher;
-import net.jini.jeri.InvocationDispatcher;
-import net.jini.jeri.InboundRequest;
import net.jini.jeri.RequestDispatcher;
-import net.jini.jeri.ServerCapabilities;
-import net.jini.core.constraint.InvocationConstraints;
import net.jini.security.Security;
import net.jini.security.SecurityContext;
@@ -67,66 +45,48 @@ final class ObjectTable {
private static final Logger logger =
Logger.getLogger("net.jini.jeri.BasicJeriExporter");
- private static final Collection dgcDispatcherMethods = new ArrayList(2);
- static {
- Method[] methods = DgcServer.class.getMethods();
- for (int i = 0; i < methods.length; i++) {
- final Method m = methods[i];
- AccessController.doPrivileged(new PrivilegedAction() {
- public Object run() {
- m.setAccessible(true);
- return null;
- }
- });
- dgcDispatcherMethods.add(m);
- }
- }
-
- private static final ServerCapabilities dgcServerCapabilities =
- new ServerCapabilities() {
- public InvocationConstraints checkConstraints(
- InvocationConstraints constraints)
- throws UnsupportedConstraintException
- {
- assert constraints.equals(InvocationConstraints.EMPTY);
- return InvocationConstraints.EMPTY;
- }
- };
-
/**
* lock to serialize request dispatcher reservation per export, so
* that a partial export will not cause another export to fail
* unnecessarily
**/
- private final Object requestDispatchersLock = new Object();
+ private final Object requestDispatchersLock;
/** table of references to impls exported with DGC */
- private final ImplRefManager implRefManager = new ImplRefManager();
-
- /** lock guarding keepAliveCount and keeper */
- private final Object keepAliveLock = new Object();
+ private final ImplRefManager implRefManager;
/** number of objects exported with keepAlive == true */
- private int keepAliveCount = 0;
-
- /** thread to keep VM alive while keepAliveCount > 0 */
- private Thread keeper = null;
+ private final JvmLifeSupport keepAliveCount;
/** maps client ID to Lease (lock guards leaseChecker too) */
- private final Map leaseTable = new HashMap();
+ private final ConcurrentMap<Uuid,Lease> leaseTable;
/** thread to check for expired leases */
- private Thread leaseChecker = null;
-
- ObjectTable() { }
+ private Thread leaseChecker;
+
+ /** thread guard */
+ private Boolean running;
+
+ ObjectTable() {
+ requestDispatchersLock = new Object();
+ implRefManager = new ImplRefManager();
+ keepAliveCount = new JvmLifeSupport();
+ leaseTable = new ConcurrentHashMap<Uuid,Lease>(256);//Plenty of capacity to reduce resizing.
+ leaseChecker = null;
+ running = Boolean.FALSE;
+ }
RequestDispatcher createRequestDispatcher(Unreferenced unrefCallback) {
- return new RD(unrefCallback);
+ return new DgcRequestDispatcher(unrefCallback, this);
}
boolean isReferenced(RequestDispatcher requestDispatcher) {
return getRD(requestDispatcher).isReferenced();
}
+
+ DgcServer getDgcServer(DgcRequestDispatcher dgdRD){
+ return new DgcServerImpl(dgdRD);
+ }
Target export(Remote impl,
RequestDispatcher[] requestDispatchers,
@@ -135,21 +95,31 @@ final class ObjectTable {
Uuid id)
throws ExportException
{
- RD[] rds = new RD[requestDispatchers.length];
+ DgcRequestDispatcher[] rds = new DgcRequestDispatcher[requestDispatchers.length];
for (int i = 0; i < requestDispatchers.length; i++) {
rds[i] = getRD(requestDispatchers[i]);
}
-
- return new Target(impl, id, rds, allowDGC, keepAlive);
+ SecurityContext securityContext = Security.getContext();
+ ClassLoader ccl = Thread.currentThread().getContextClassLoader();
+ Target t = null;
+ t = new Target(id, rds, allowDGC, keepAlive,this,
+ securityContext, ccl, keepAliveCount);
+ synchronized (requestDispatchersLock){
+ t.procRequestDispatchers();
+ }
+ ImplRef implRef = implRefManager.getImplRef(impl, t);
+ t.setImplRef(implRef);
+ t.setExported();
+ return t;
}
- private RD getRD(RequestDispatcher requestDispatcher) {
+ private DgcRequestDispatcher getRD(RequestDispatcher requestDispatcher) {
/*
* The following cast will throw a ClassCastException if we were
* passed a RequestDispatcher that was not returned by this class's
* createRequestDispatcher method:
*/
- RD rd = (RD) requestDispatcher;
+ DgcRequestDispatcher rd = (DgcRequestDispatcher) requestDispatcher;
if (!rd.forTable(this)) {
throw new IllegalArgumentException(
"request dispatcher for different object table");
@@ -157,813 +127,245 @@ final class ObjectTable {
return rd;
}
- /**
- * Increments the count of objects exported with keepAlive true,
- * starting a non-daemon thread if necessary.
- **/
- private void incrementKeepAliveCount() {
- synchronized (keepAliveLock) {
- keepAliveCount++;
-
- if (keeper == null) {
- keeper = (Thread) AccessController.doPrivileged(
- new NewThreadAction(new Runnable() {
- public void run() {
- try {
- while (true) {
- Thread.sleep(Long.MAX_VALUE);
- }
- } catch (InterruptedException e) {
- // pass away if interrupted
- }
- }
- }, "KeepAlive", false));
- keeper.start();
- }
- }
- }
-
- /**
- * Decrements the count of objects exported with keepAlive true,
- * stopping the non-daemon thread if decremented to zero.
- **/
- private void decrementKeepAliveCount() {
- synchronized (keepAliveLock) {
- keepAliveCount--;
-
- if (keepAliveCount == 0) {
- assert keeper != null;
- AccessController.doPrivileged(new PrivilegedAction() {
- public Object run() {
- keeper.interrupt();
- return null;
- }
- });
- keeper = null;
- }
- }
- }
-
- /**
- * A Target is returned by the export method to represent the object
- * exported to this ObjectTable. It can be used to unexport the
- * exported object.
- */
- final class Target {
-
- private final ImplRef implRef;
- private final Uuid id;
- private final RD[] requestDispatchers;
- private final boolean allowDGC;
- private final boolean keepAlive;
- private final SecurityContext securityContext;
- private final ClassLoader ccl;
-
- /** lock guarding all mutable instance state (below) */
- private final Object lock = new Object();
- private InvocationDispatcher invocationDispatcher;
- private boolean exported = false;
- private int callsInProgress = 0;
- private final Set referencedSet;
- private final Map sequenceTable;
-
- Target(Remote impl,
- Uuid id,
- RD[] requestDispatchers,
- boolean allowDGC,
- boolean keepAlive)
- throws ExportException
- {
- this.id = id;
- this.requestDispatchers = requestDispatchers;
- this.allowDGC = allowDGC;
- this.keepAlive = keepAlive;
-
- securityContext = Security.getContext();
- ccl = Thread.currentThread().getContextClassLoader();
-
- synchronized (requestDispatchersLock) {
- boolean success = false;
- int i = 0;
- try {
- for (i = 0; i < requestDispatchers.length; i++) {
- requestDispatchers[i].put(this);
- }
- success = true;
- } finally {
- if (!success) {
- for (int j = 0; j < i; j++) {
- requestDispatchers[i].remove(this, false);
- }
- }
- }
- }
-
- implRef = implRefManager.getImplRef(impl, this);
-
- if (allowDGC) {
- referencedSet = new HashSet(3);
- sequenceTable = new HashMap(3);
- } else {
- referencedSet = null;
- sequenceTable = null;
- }
-
- if (keepAlive) {
- incrementKeepAliveCount();
- }
-
- synchronized (lock) {
- exported = true;
- }
- }
-
- void setInvocationDispatcher(InvocationDispatcher id) {
- assert id != null;
- synchronized (lock) {
- assert invocationDispatcher == null;
- invocationDispatcher = id;
- }
- }
-
- boolean unexport(boolean force) {
- synchronized (lock) {
- if (!exported) {
- return true;
- }
- if (!force && callsInProgress > 0) {
- return false;
- }
- exported = false;
-
- if (keepAlive && callsInProgress == 0) {
- decrementKeepAliveCount();
- }
-
- if (allowDGC) {
- if (!referencedSet.isEmpty()) {
- for (Iterator i = referencedSet.iterator();
- i.hasNext();)
- {
- Uuid clientID = (Uuid) i.next();
- unregisterTarget(this, clientID);
- }
- referencedSet.clear();
- }
- sequenceTable.clear();
- }
- }
-
- implRef.release(this);
-
- for (int i = 0; i < requestDispatchers.length; i++) {
- requestDispatchers[i].remove(this, false);
- }
- return true;
- }
-
- void collect() {
- synchronized (lock) {
- if (!exported) {
- return;
- }
-
- if (logger.isLoggable(Level.FINE)) {
- logger.log(Level.FINE,
- "garbage collection of object with id {0}", id);
- }
-
- exported = false;
-
- if (keepAlive && callsInProgress == 0) {
- decrementKeepAliveCount();
- }
-
- if (allowDGC) {
- assert referencedSet.isEmpty();
- sequenceTable.clear();
- }
- }
-
- for (int i = 0; i < requestDispatchers.length; i++) {
- requestDispatchers[i].remove(this, true);
- }
- }
-
- Uuid getObjectIdentifier() {
- return id;
- }
-
- // used by ImplRef for invoking Unreferenced.unreferenced
- boolean getEnableDGC() {
- return allowDGC;
- }
-
- // used by ImplRef for invoking Unreferenced.unreferenced
- SecurityContext getSecurityContext() {
- return securityContext;
- }
-
- // used by ImplRef for invoking Unreferenced.unreferenced
- ClassLoader getContextClassLoader() {
- return ccl;
- }
-
- void referenced(Uuid clientID, long sequenceNum) {
- if (!allowDGC) {
- return; // ignore if DGC not enabled for this object
- }
-
- synchronized (lock) {
- if (!exported) {
- return;
- }
-
- if (logger.isLoggable(Level.FINEST)) {
- logger.log(Level.FINEST,
- "this={0}, clientID={1}, sequenceNum={2}",
- new Object[] {
- this, clientID, new Long(sequenceNum)
- });
- }
-
- /*
- * Check current sequence number against the last
- * recorded sequence number for the client. If the
- * current value is lower, then this is a "late dirty
- * call", which should not be processed. Otherwise,
- * update the last recorded sequence number.
- */
- SequenceEntry entry =
- (SequenceEntry) sequenceTable.get(clientID);
- if (entry == null) {
- // no record: must assume this is not a late dirty call
- entry = new SequenceEntry(sequenceNum);
- sequenceTable.put(clientID, entry);
- } else if (sequenceNum < entry.sequenceNum) {
- return; // late dirty call: ignore
- } else {
- entry.sequenceNum = sequenceNum;
- }
-
- if (!referencedSet.contains(clientID)) {
- if (referencedSet.isEmpty()) {
- Remote impl = implRef.getImpl();
- if (impl == null) {
- return; // too late if impl was collected
- }
- implRef.pin(this);
- }
- referencedSet.add(clientID);
-
- registerTarget(this, clientID);
- }
- }
- }
-
- void unreferenced(Uuid clientID, long sequenceNum, boolean strong) {
- if (!allowDGC) {
- return; // ignore if DGC not enabled for this object
- }
-
- synchronized (lock) {
- if (!exported) {
- return;
- }
-
- if (logger.isLoggable(Level.FINEST)) {
- logger.log(Level.FINEST,
- "this={0}, clientID={1}, sequenceNum={2}, strong={3}",
- new Object[] {
- this, clientID, new Long(sequenceNum),
- Boolean.valueOf(strong)
- });
- }
-
- /*
- * Check current sequence number against the last
- * recorded sequence number for the client. If the
- * current value is lower, then this is a "late clean
- * call", which should not be processed. Otherwise:
- * if this is for a strong clean call, then update the
- * last recorded sequence number; if no strong clean
- * call has been processed for this client, discard
- * its sequence number record.
- */
- SequenceEntry entry =
- (SequenceEntry) sequenceTable.get(clientID);
- if (entry == null) {
- // no record: must assume this is not a late clean call
- if (strong) {
- entry = new SequenceEntry(sequenceNum);
- sequenceTable.put(clientID, entry);
- entry.keep = true;
- }
- } else if (sequenceNum < entry.sequenceNum) {
- return; // late clean call: ignore
- } else if (strong) {
- entry.sequenceNum = sequenceNum;
- entry.keep = true; // strong clean: retain sequence number
- } else if (!entry.keep) {
- sequenceTable.remove(clientID);
- }
-
- unregisterTarget(this, clientID);
-
- if (referencedSet.remove(clientID) &&
- referencedSet.isEmpty())
- {
- implRef.unpin(this);
- }
- }
- }
-
- void leaseExpired(Uuid clientID) {
- assert allowDGC;
-
- synchronized (lock) {
- if (!exported) {
- return;
- }
-
- if (logger.isLoggable(Level.FINEST)) {
- logger.log(Level.FINEST,
- "this={0}, clientID={1}",
- new Object[] { this, clientID });
- }
-
- SequenceEntry entry =
- (SequenceEntry) sequenceTable.get(clientID);
- if (entry != null && !entry.keep) {
- /*
- * REMIND: We could be removing the sequence number
- * for a more recent lease, thus allowing a "late
- * clean call" to be inappropriately processed?
- * (See 4848840 Comments.)
- */
- sequenceTable.remove(clientID);
- }
-
- if (referencedSet.remove(clientID) &&
- referencedSet.isEmpty())
- {
- implRef.unpin(this);
- }
- }
- }
-
- void dispatch(InboundRequest request)
- throws IOException, NoSuchObject
- {
- InvocationDispatcher id;
- synchronized (lock) {
- if (!exported || invocationDispatcher == null) {
- if (logger.isLoggable(Level.FINEST)) {
- logger.log(Level.FINEST,
- "this={0}, not exported", this);
- }
- throw new NoSuchObject();
- }
- id = invocationDispatcher; // save for reference outside lock
- callsInProgress++;
- }
- try {
- Remote impl = implRef.getImpl();
- if (impl == null) {
- if (logger.isLoggable(Level.FINEST)) {
- logger.log(Level.FINEST,
- "this={0}, garbage collected", this);
- }
- throw new NoSuchObject();
- }
-
- dispatch(request, id, impl);
-
- } finally {
- synchronized (lock) {
- assert callsInProgress > 0;
- callsInProgress--;
-
- if (keepAlive && !exported && callsInProgress == 0) {
- decrementKeepAliveCount();
- }
- }
- }
- }
-
- private void dispatch(final InboundRequest request,
- final InvocationDispatcher id,
- final Remote impl)
- throws IOException, NoSuchObject
- {
- Thread t = Thread.currentThread();
- ClassLoader savedCcl = t.getContextClassLoader();
- try {
- if (ccl != savedCcl) {
- t.setContextClassLoader(ccl);
- }
- AccessController.doPrivileged(securityContext.wrap(
- new PrivilegedExceptionAction() {
- public Object run() throws IOException {
- dispatch0(request, id, impl);
- return null;
- }
- }), securityContext.getAccessControlContext());
-
- } catch (java.security.PrivilegedActionException e) {
- throw (IOException) e.getException();
- } finally {
- if (ccl != savedCcl || savedCcl != t.getContextClassLoader()) {
- t.setContextClassLoader(savedCcl);
- }
- }
- }
-
- private void dispatch0(final InboundRequest request,
- final InvocationDispatcher id,
- final Remote impl)
- throws IOException
- {
- request.checkPermissions();
-
- OutputStream out = request.getResponseOutputStream();
- out.write(Jeri.OBJECT_HERE);
-
- final Collection context = new ArrayList(5);
- request.populateContext(context);
-
- ServerContext.doWithServerContext(new Runnable() {
- public void run() {
- id.dispatch(impl, request, context);
- }
- }, Collections.unmodifiableCollection(context));
- }
-
- public String toString() { // for logging
- return "Target@" + Integer.toHexString(hashCode()) +
- "[" + id + "]";
- }
- }
-
- private static final class SequenceEntry {
- long sequenceNum;
- boolean keep;
-
- SequenceEntry(long sequenceNum) {
- this.sequenceNum = sequenceNum;
- }
- }
-
void registerTarget(Target target, Uuid clientID) {
- synchronized (leaseTable) {
- Lease lease = (Lease) leaseTable.get(clientID);
- if (lease == null) {
- target.leaseExpired(clientID);
- } else {
- synchronized (lease.notifySet) {
- lease.notifySet.add(target);
- }
- }
- }
+ Lease lease = leaseTable.get(clientID);
+ if (lease == null) {
+ target.leaseExpired(clientID);
+ } else {
+ boolean added = lease.add(target);
+ if ( added == false){
+ // lease has been locked because it has expired
+ // prior to removal
+ target.leaseExpired(clientID);
+ }
+ }
}
void unregisterTarget(Target target, Uuid clientID) {
- synchronized (leaseTable) {
- Lease lease = (Lease) leaseTable.get(clientID);
- if (lease != null) {
- synchronized (lease.notifySet) {
- lease.notifySet.remove(target);
- }
- }
- }
- }
-
- /**
- * RequestDispatcher implementation.
- **/
- private class RD implements RequestDispatcher {
-
- private final Unreferenced unrefCallback;
-
- private final Map idTable = new HashMap();
- private int dgcEnabledCount = 0; // guarded by idTable lock
-
- private final InvocationDispatcher dgcDispatcher;
- private final DgcServerImpl dgcServerImpl;
-
- RD(Unreferenced unrefCallback) {
- this.unrefCallback = unrefCallback;
- try {
- dgcDispatcher =
- new BasicInvocationDispatcher(
- dgcDispatcherMethods, dgcServerCapabilities,
- null, null, this.getClass().getClassLoader())
- {
- protected ObjectInputStream createMarshalInputStream(
- Object impl,
- InboundRequest request,
- boolean integrity,
- Collection context)
- throws IOException
- {
- ClassLoader loader = getClassLoader();
- return new MarshalInputStream(
- request.getRequestInputStream(),
- loader, integrity, loader,
- Collections.unmodifiableCollection(context));
- // useStreamCodebases() not invoked
- }
- };
- } catch (ExportException e) {
- throw new AssertionError();
- }
- dgcServerImpl = new DgcServerImpl();
- }
-
- boolean forTable(ObjectTable table) {
- return ObjectTable.this == table;
- }
-
- boolean isReferenced() {
- synchronized (idTable) {
- return !idTable.isEmpty();
- }
- }
-
- Target get(Uuid id) {
- synchronized (idTable) {
- return (Target) idTable.get(id);
- }
- }
-
- void put(Target target) throws ExportException {
- synchronized (idTable) {
- Uuid id = target.getObjectIdentifier();
- if (id.equals(Jeri.DGC_ID)) {
- throw new ExportException(
- "object identifier reserved for DGC");
- }
- if (idTable.containsKey(id)) {
- throw new ExportException(
- "object identifier already in use");
- }
- idTable.put(id, target);
- if (target.getEnableDGC()) {
- dgcEnabledCount++;
- }
- }
- }
-
- void remove(Target target, boolean gc) {
- boolean empty = false;
- synchronized (idTable) {
- Uuid id = target.getObjectIdentifier();
- assert idTable.get(id) == target;
- idTable.remove(id);
- if (target.getEnableDGC()) {
- dgcEnabledCount--;
- assert dgcEnabledCount >= 0;
- }
-
- if (idTable.isEmpty()) {
- empty = true;
- }
- }
-
- if (gc && empty) {
- /*
- * We have to be careful to make this callback without holding
- * the lock for idTable, because the callback implementation
- * will likely be code that calls this object's isReferenced
- * method in its own synchronized block.
- */
- unrefCallback.unreferenced();
- }
- }
-
- private boolean hasDgcEnabledTargets() {
- synchronized (idTable) {
- return dgcEnabledCount > 0;
- }
- }
-
- public void dispatch(InboundRequest request) {
- try {
- InputStream in = request.getRequestInputStream();
- Uuid id = UuidFactory.read(in);
-
- if (logger.isLoggable(Level.FINEST)) {
- logger.log(Level.FINEST, "id={0}", id);
- }
-
- try {
- /*
- * The DGC object identifier is hardwired here,
- * rather than install it in idTable; this
- * eliminates the need to worry about not counting
- * the DGC server as an exported object in the
- * table, and it doesn't need all of the machinery
- * that Target provides.
- */
- if (id.equals(Jeri.DGC_ID)) {
- dispatchDgcRequest(request);
- return;
- }
-
- Target target = (Target) get(id);
- if (target == null) {
- logger.log(Level.FINEST, "id not in table");
- throw new NoSuchObject();
- }
- target.dispatch(request);
-
- } catch (NoSuchObject e) {
- in.close();
- OutputStream out = request.getResponseOutputStream();
- out.write(Jeri.NO_SUCH_OBJECT);
- out.close();
-
- if (logger.isLoggable(Levels.FAILED)) {
- logger.log(Levels.FAILED, "no such object: {0}", id);
- }
- }
- } catch (IOException e) {
- request.abort();
-
- if (logger.isLoggable(Levels.FAILED)) {
- logger.log(Levels.FAILED,
- "I/O exception dispatching request", e);
- }
- }
- }
-
- private void dispatchDgcRequest(final InboundRequest request)
- throws IOException, NoSuchObject
- {
- if (!hasDgcEnabledTargets()) {
- logger.log(Level.FINEST, "no DGC-enabled targets");
- throw new NoSuchObject();
- }
-
- OutputStream out = request.getResponseOutputStream();
- out.write(Jeri.OBJECT_HERE);
-
- final Collection context = new ArrayList(5);
- request.populateContext(context);
-
- ServerContext.doWithServerContext(new Runnable() {
- public void run() {
- dgcDispatcher.dispatch(dgcServerImpl, request, context);
- }
- }, Collections.unmodifiableCollection(context));
- }
-
- private class DgcServerImpl implements DgcServer {
-
- public long dirty(Uuid clientID,
- long sequenceNum,
- Uuid[] ids)
- {
- if (logger.isLoggable(Level.FINEST)) {
- logger.log(Level.FINEST,
- "clientID={0}, sequenceNum={1}, ids={2}",
- new Object[] {
- clientID, new Long(sequenceNum), Arrays.asList(ids)
- });
- }
-
- long duration = Jeri.leaseValue;
-
- synchronized (leaseTable) {
- Lease lease = (Lease) leaseTable.get(clientID);
- if (lease == null) {
- leaseTable.put(clientID,
- new Lease(clientID, duration));
- if (leaseChecker == null) {
- leaseChecker =
- (Thread) AccessController.doPrivileged(
- new NewThreadAction(new LeaseChecker(),
- "DGC Lease Checker", true));
- leaseChecker.start();
- }
- } else {
- lease.renew(duration);
- }
- }
-
- for (int i = 0; i < ids.length; i++) {
- Target target = get(ids[i]);
- if (target != null) {
- target.referenced(clientID, sequenceNum);
- }
- }
-
- return duration;
- }
-
- public void clean(Uuid clientID,
- long sequenceNum,
- Uuid[] ids,
- boolean strong)
- {
- if (logger.isLoggable(Level.FINEST)) {
- logger.log(Level.FINEST,
- "clientID={0}, sequenceNum={1}, ids={2}, strong={3}",
- new Object[] {
- clientID, new Long(sequenceNum),
- Arrays.asList(ids), Boolean.valueOf(strong)
- });
- }
-
- for (int i = 0; i < ids.length; i++) {
- Target target = get(ids[i]);
- if (target != null) {
- target.unreferenced(clientID, sequenceNum, strong);
- }
- }
- }
- }
+ Lease lease = leaseTable.get(clientID);
+ if (lease != null) {
+ lease.remove(target);
+ }
+ }
+
+
+
+ private class DgcServerImpl implements DgcServer {
+ private final DgcRequestDispatcher dgcRequestDispatcher;
+
+ DgcServerImpl(DgcRequestDispatcher dgcRequestDispatcher){
+ this.dgcRequestDispatcher = dgcRequestDispatcher;
+ }
+
+ public long dirty(Uuid clientID,
+ long sequenceNum,
+ Uuid[] ids)
+ {
+ if (logger.isLoggable(Level.FINEST)) {
+ logger.log(Level.FINEST,
+ "clientID={0}, sequenceNum={1}, ids={2}",
+ new Object[] {
+ clientID, new Long(sequenceNum), Arrays.asList(ids)
+ });
+ }
+
+ long duration = Jeri.leaseValue;
+
+ Lease lease = leaseTable.get(clientID);
+ if (lease == null) {
+ lease = new Lease(clientID, duration);
+ Lease existed = leaseTable.putIfAbsent(clientID,lease);
+ if (existed != null){
+ assert clientID.equals(existed.getClientID());
+ boolean renewed = existed.renew(duration);
+ if (!renewed){
+ /* The probability of getting here is low,
+ * it indicates a lease with an extremely short
+ * expiry and a very small lease table.
+ */
+ if (logger.isLoggable(Level.WARNING)) {
+ logger.log(Level.WARNING,
+ "Problem with lease table, try a longer " +
+ "lease duration clientID={0}, " +
+ "sequenceNum={1}, ids={2}",
+ new Object[] {
+ clientID, new Long(sequenceNum), Arrays.asList(ids)
+ });
+ }
+ }
+ }
+ } else {
+ assert clientID.equals(lease.getClientID());
+ boolean renewed = lease.renew(duration);
+ if (!renewed){
+ // Indicates an expired lease in the table. A lease
+ // always becomes expired prior to removal, it is
+ // never removed prior to expiry, in case it is
+ // renewed by another thread, which would risk a renewed
+ // lease being removed from the table.
+ // An expired lease must be replaced.
+ leaseTable.remove(clientID, lease); // Another thread could remove it first.
+ lease = new Lease(clientID, duration);
+ Lease existed = leaseTable.putIfAbsent(clientID, lease);
+ if (existed != null){
+ lease = existed;
+ assert clientID.equals(lease.getClientID());
+ renewed = lease.renew(duration);
+ if (!renewed){
+ /* The probability of getting here is low,
+ * it indicates a lease of extremely short
+ * duration and a very small lease table.
+ */
+ if (logger.isLoggable(Level.WARNING)) {
+ logger.log(Level.WARNING,
+ "Problem with lease table, try a longer " +
+ "lease duration clientID={0}, " +
+ "sequenceNum={1}, ids={2}",
+ new Object[] {
+ clientID, new Long(sequenceNum), Arrays.asList(ids)
+ });
+ }
+ }
+ }
+ }
+ }
+ /* FIXED: River-142:
+ * In the server-side DGC implementation's thread that check's for
+ * lease expirations
+ * (com.sun.jini.jeri.internal.runtime.ObjectTable.LeaseChecker.run),
+ * it checks for them while synchronized on the overall lease table,
+ * but it delays notifying the expired leases' individual registered
+ * Targets about the expirations until after it has released the
+ * lease table lock. This approach was taken from the
+ * JRMP implementation, which is that way because of the fix
+ * for 4118056 (a previous deadlock bug-- but now, I'm thinking
+ * that the JRMP implementation has this bug too).
+ *
+ * The problem seems to be that after releasing the lease table
+ * lock, it is possible for another lease renewal/request to
+ * come in (from the same DGC client and for the same remote object)
+ * that would then be invalidated by the subsequent Target
+ * notification made by the lease expiration check thread-- and
+ * thus the client's lease renewal (for that remote object) will
+ * be forgotten. It would appear that the synchronization approach
+ * here needs to be reconsidered.
+ *
+ * ( Comments note: )
+ * In addition to the basic problem of the expired-then-renewed
+ * client being removed from the referenced set, there is also
+ * the problem of the sequence table entry being forgotten-- which
+ * prevents detection of a "late clean call".
+ * Normally, late clean calls are not a problem because sequence
+ * numbers are retained while the client is in the referenced set
+ * (and there is no such thing as a "strong dirty").
+ * But in this case, with the following order of events on
+ * the server side:
+ *
+ * 1. dirty, seqNo=2
+ * 2. (lease expiration)
+ * 3. clean, seqNo=1
+ *
+ * The primary bug here is that the first two events will leave
+ * the client missing from the referenced set. But the secondary
+ * bug is that even if that's fixed, with the sequence number
+ * forgotten, the third event (the "late clean call") will still
+ * cause the client to be removed from the referenced set.
+ *
+ * FIX:
+ * This issue has been fixed by making the Lease responsible for
+ * it's own state, which is protected internally by synchronization.
+ * The lease checker passes the time to the Lease, which checks
+ * itself and notifies the Targets in the event of expiry.
+ *
+ * Because the notification is not delayed, the client id and
+ * sequence number will not be not be removed by the LeaseChecker
+ * thread after being updated by the second dirty call (when
+ * the lock was cleared as described).
+ *
+ * The client id and sequence number are added to the Target sequence
+ * table by the second dirty call, after the Lease removes
+ * them immediately upon expiry being invoked by the LeaseChecker.
+ *
+ * Then because the late clean call sequence number is less than the
+ * second dirty call and exists, it is correctly recognised.
+ */
+ synchronized (running){
+ if (!running) {
+ leaseChecker =
+ (Thread) AccessController.doPrivileged(
+ new NewThreadAction(new LeaseChecker(),
+ "DGC Lease Checker", true));
+ leaseChecker.start();
+ }
+ }
+ for (int i = 0; i < ids.length; i++) {
+ Target target = dgcRequestDispatcher.get(ids[i]);
+ if (target != null) {
+ target.referenced(clientID, sequenceNum);
+ }
+ }
+ return duration;
+ }
+
+ public void clean(Uuid clientID,
+ long sequenceNum,
+ Uuid[] ids,
+ boolean strong)
+ {
+ if (logger.isLoggable(Level.FINEST)) {
+ logger.log(Level.FINEST,
+ "clientID={0}, sequenceNum={1}, ids={2}, strong={3}",
+ new Object[] {
+ clientID, new Long(sequenceNum),
+ Arrays.asList(ids), Boolean.valueOf(strong)
+ });
+ }
+
+ for (int i = 0; i < ids.length; i++) {
+ Target target = dgcRequestDispatcher.get(ids[i]);
+ if (target != null) {
+ target.unreferenced(clientID, sequenceNum, strong);
+ }
+ }
+ }
}
private class LeaseChecker implements Runnable {
public void run() {
boolean done = false;
- do {
- try {
- Thread.sleep(Jeri.leaseCheckInterval);
- } catch (InterruptedException e) {
- // REMIND: shouldn't happen, OK to ignore?
- }
-
- long now = System.currentTimeMillis();
-
- Collection expiredLeases = new ArrayList();
-
- synchronized (leaseTable) {
- for (Iterator i = leaseTable.values().iterator();
- i.hasNext();)
- {
- Lease lease = (Lease) i.next();
- if (lease.hasExpired(now)) {
- expiredLeases.add(lease);
- i.remove();
- }
- }
-
- if (leaseTable.isEmpty()) {
- leaseChecker = null;
- done = true;
- }
- }
-
- if (expiredLeases.isEmpty()) {
- continue;
- }
-
- for (Iterator i = expiredLeases.iterator(); i.hasNext();) {
- Lease lease = (Lease) i.next();
- if (lease.notifySet.isEmpty()) {
- continue;
- }
-
- for (Iterator i2 = lease.notifySet.iterator();
- i2.hasNext();)
- {
- Target target = (Target) i2.next();
- target.leaseExpired(lease.getClientID());
- }
- }
- } while (!done);
- }
- }
-
- private static class Lease {
-
- private final Uuid clientID;
- final Set notifySet = new HashSet(3); // guarded?
- private long expiration; // guarded by leaseTable lock
-
- Lease(Uuid clientID, long duration) {
- this.clientID = clientID;
- expiration = System.currentTimeMillis() + duration;
- }
-
- Uuid getClientID() {
- return clientID;
- }
-
- void renew(long duration) {
- long newExpiration = System.currentTimeMillis() + duration;
- if (newExpiration > expiration) {
- expiration = newExpiration;
- }
- }
-
- boolean hasExpired(long now) {
- return expiration < now;
+ try {
+ do {
+ Thread.sleep(Jeri.leaseCheckInterval);
+ long now = System.currentTimeMillis();
+ for (Iterator i = leaseTable.values().iterator();
+ i.hasNext();)
+ {
+ Lease lease = (Lease) i.next();
+ boolean expired = lease.notifyIfExpired(now);
+ if (expired) {
+ i.remove();
+ }
+ }
+ if (leaseTable.isEmpty()) {
+ done = true;
+ }
+ } while (!done);
+ } catch (InterruptedException e) {
+ // REMIND: shouldn't happen, OK to ignore?
+ // No, restore the interrupted status
+ Thread.currentThread().interrupt();
+ } finally {
+ // This is always executed and returns the lease checker
+ // to the non running state, such that if the application
+ // has not exited, another thread will be started eventually.
+ synchronized (running){
+ leaseChecker = null;
+ running = Boolean.FALSE;
+ }
+ }
}
}
- private static class NoSuchObject extends Exception { }
+ static class NoSuchObject extends Exception { }
}
Added: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/SequenceEntry.java
URL: http://svn.apache.org/viewvc/river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/SequenceEntry.java?rev=1137269&view=auto
==============================================================================
--- river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/SequenceEntry.java (added)
+++ river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/SequenceEntry.java Sun Jun 19 00:43:42 2011
@@ -0,0 +1,41 @@
+package com.sun.jini.jeri.internal.runtime;
+
+final class SequenceEntry {
+
+ private volatile long sequenceNum;
+ private volatile boolean keep;
+
+ SequenceEntry(long sequenceNum) {
+ super();
+ this.sequenceNum = sequenceNum;
+ keep = false;
+ }
+
+ SequenceEntry(long sequenceNum, boolean strong){
+ this.sequenceNum = sequenceNum;
+ this.keep = strong;
+ }
+
+ /**
+ * If the passed in sequence number is greater than the current number,
+ * it is updated.
+ * If
+ * @param seqNum - passed in sequence number.
+ * @param strong - strong clean call is kept in the event of an update.
+ * @return true if the sequence number is updated.
+ */
+ boolean update(long seqNum, boolean strong){
+ synchronized (this){
+ if (seqNum > sequenceNum){
+ sequenceNum = seqNum;
+ if (strong) keep = true;
+ return true;
+ }
+ return false;
+ }
+ }
+
+ boolean keep(){
+ return keep;
+ }
+}
Propchange: river/jtsk/trunk/src/com/sun/jini/jeri/internal/runtime/SequenceEntry.java
------------------------------------------------------------------------------
svn:eol-style = native