Author: dblevins
Date: Tue Apr 3 00:27:48 2012
New Revision: 1308631
URL: http://svn.apache.org/viewvc?rev=1308631&view=rev
Log:
Porting from trunk:
OPENEJB-1813: EJBd Protocol communicates server/container invocation times to
client
OPENEJB-1814: JMX Management of Multipoint discovery server service
OPENEJB-1815: Ability to restart Multipoint ServerService
OPENEJB-1816: Client ConnectionPoolCreated and ConnectionPoolTimeout events
Added:
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/event/ConnectionPoolCreated.java
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/event/ConnectionPoolTimeout.java
openejb/branches/openejb-3.1.x/server/openejb-server/src/main/java/org/apache/openejb/server/ServerRuntimeException.java
Modified:
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/EJBResponse.java
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/SocketConnectionFactory.java
openejb/branches/openejb-3.1.x/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbRequestHandler.java
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/MultipointDiscoveryAgent.java
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/MultipointServer.java
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/Tracker.java
openejb/branches/openejb-3.1.x/server/openejb-server/src/main/java/org/apache/openejb/server/ServiceManager.java
Modified:
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/EJBResponse.java
URL:
http://svn.apache.org/viewvc/openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/EJBResponse.java?rev=1308631&r1=1308630&r2=1308631&view=diff
==============================================================================
---
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/EJBResponse.java
(original)
+++
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/EJBResponse.java
Tue Apr 3 00:27:48 2012
@@ -25,6 +25,7 @@ public class EJBResponse implements Clus
private transient int responseCode = -1;
private transient Object result;
private transient ServerMetaData server;
+ private transient long[] times = new long[Time.values().length];
public EJBResponse() {
@@ -55,7 +56,7 @@ public class EJBResponse implements Clus
public ServerMetaData getServer() {
return server;
}
-
+
public String toString() {
StringBuffer s = null;
switch (responseCode) {
@@ -89,11 +90,21 @@ public class EJBResponse implements Clus
default:
s = new StringBuffer("UNKNOWN_RESPONSE");
}
- s.append(':').append(result);
+ s.append(",
serverTime=").append(times[Time.TOTAL.ordinal()]).append("ns");
+ s.append(",
containerTime").append(times[Time.CONTAINER.ordinal()]).append("ns");
+ s.append(" : ").append(result);
return s.toString();
}
+ public void start(EJBResponse.Time time) {
+ times[time.ordinal()] = System.nanoTime();
+ }
+
+ public void stop(EJBResponse.Time time) {
+ times[time.ordinal()] = System.nanoTime() - times[time.ordinal()];
+ }
+
public void readExternal(ObjectInput in) throws IOException,
ClassNotFoundException {
byte version = in.readByte(); // future use
@@ -106,6 +117,11 @@ public class EJBResponse implements Clus
responseCode = in.readByte();
result = in.readObject();
+
+ times = new long[in.readByte()];
+ for (int i = 0; i < times.length; i++) {
+ times[i] = in.readLong();
+ }
}
public void writeExternal(ObjectOutput out) throws IOException {
@@ -130,7 +146,22 @@ public class EJBResponse implements Clus
result = new ThrowableArtifact(throwable);
}
}
+
+ start(Time.SERIALIZATION);
out.writeObject(result);
+ stop(Time.SERIALIZATION);
+ stop(Time.TOTAL);
+
+ out.writeByte(times.length);
+ for (int i = 0; i < times.length; i++) {
+ out.writeLong(times[i]);
+ }
}
+ public static enum Time {
+ TOTAL,
+ CONTAINER,
+ SERIALIZATION,
+ DESERIALIZATION
+ }
}
Modified:
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/SocketConnectionFactory.java
URL:
http://svn.apache.org/viewvc/openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/SocketConnectionFactory.java?rev=1308631&r1=1308630&r2=1308631&view=diff
==============================================================================
---
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/SocketConnectionFactory.java
(original)
+++
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/SocketConnectionFactory.java
Tue Apr 3 00:27:48 2012
@@ -309,6 +309,8 @@ public class SocketConnectionFactory imp
for (int i = 0; i < objects.length; i++) {
pool.push(null);
}
+
+ Client.fireEvent(new ConnectionPoolCreated(uri, size, timeout,
timeUnit));
}
public SocketConnection get() throws IOException{
@@ -320,7 +322,10 @@ public class SocketConnectionFactory imp
Thread.interrupted();
}
- throw new ConnectionPoolTimeoutException("No connections available
in pool (size " + size + "). Waited for " + timeout + " seconds for a
connection.");
+ ConnectionPoolTimeoutException exception = new
ConnectionPoolTimeoutException("No connections available in pool (size " + size
+ "). Waited for " + timeout + " " + timeUnit.name().toLowerCase() + " for a
connection.");
+ exception.fillInStackTrace();
+ Client.fireEvent(new ConnectionPoolTimeout(uri, size, timeout,
timeUnit, exception));
+ throw exception;
}
public void put(SocketConnection connection) {
Added:
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/event/ConnectionPoolCreated.java
URL:
http://svn.apache.org/viewvc/openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/event/ConnectionPoolCreated.java?rev=1308631&view=auto
==============================================================================
---
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/event/ConnectionPoolCreated.java
(added)
+++
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/event/ConnectionPoolCreated.java
Tue Apr 3 00:27:48 2012
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.client.event;
+
+import java.net.URI;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @version $Rev$ $Date$
+ */
+@Log(Log.Level.FINE)
+public class ConnectionPoolCreated {
+
+ private final long timeout;
+ private final TimeUnit timeUnit;
+ private final int size;
+ private final URI uri;
+
+ public ConnectionPoolCreated(URI uri, int size, long timeout, TimeUnit
timeUnit) {
+ this.uri = uri;
+ this.size = size;
+ this.timeUnit = timeUnit;
+ this.timeout = timeout;
+ }
+
+ public long getTimeout() {
+ return timeout;
+ }
+
+ public TimeUnit getTimeUnit() {
+ return timeUnit;
+ }
+
+ public int getSize() {
+ return size;
+ }
+
+ public URI getUri() {
+ return uri;
+ }
+
+ @Override
+ public String toString() {
+ return "ConnectionPoolCreated{" +
+ "uri=" + uri +
+ ", size=" + size +
+ ", timeout='" + timeout + " " + timeUnit + "'" +
+ '}';
+ }
+}
Added:
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/event/ConnectionPoolTimeout.java
URL:
http://svn.apache.org/viewvc/openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/event/ConnectionPoolTimeout.java?rev=1308631&view=auto
==============================================================================
---
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/event/ConnectionPoolTimeout.java
(added)
+++
openejb/branches/openejb-3.1.x/server/openejb-client/src/main/java/org/apache/openejb/client/event/ConnectionPoolTimeout.java
Tue Apr 3 00:27:48 2012
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.client.event;
+
+import java.net.URI;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @version $Rev$ $Date$
+ */
+@Log(Log.Level.FINE)
+public class ConnectionPoolTimeout {
+
+ private final URI uri;
+ private final int size;
+ private final long timeout;
+ private final TimeUnit timeUnit;
+ private final Throwable caller;
+
+ public ConnectionPoolTimeout(URI uri, int size, long timeout, TimeUnit
timeUnit, Throwable caller) {
+ this.uri = uri;
+ this.size = size;
+ this.timeout = timeout;
+ this.timeUnit = timeUnit;
+ this.caller = caller;
+ }
+
+ public long getTimeout() {
+ return timeout;
+ }
+
+ public TimeUnit getTimeUnit() {
+ return timeUnit;
+ }
+
+ public int getSize() {
+ return size;
+ }
+
+ public URI getUri() {
+ return uri;
+ }
+
+ @Override
+ public String toString() {
+ return "ConnectionPoolCreated{" +
+ "uri=" + uri +
+ ", size=" + size +
+ ", timeout='" + timeout + " " + timeUnit + "'" +
+ '}';
+ }
+}
Modified:
openejb/branches/openejb-3.1.x/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbRequestHandler.java
URL:
http://svn.apache.org/viewvc/openejb/branches/openejb-3.1.x/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbRequestHandler.java?rev=1308631&r1=1308630&r2=1308631&view=diff
==============================================================================
---
openejb/branches/openejb-3.1.x/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbRequestHandler.java
(original)
+++
openejb/branches/openejb-3.1.x/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbRequestHandler.java
Tue Apr 3 00:27:48 2012
@@ -47,7 +47,7 @@ class EjbRequestHandler {
EjbRequestHandler(EjbDaemon daemon) {
this.daemon = daemon;
-
+
clusterableRequestHandler = newClusterableRequestHandler();
}
@@ -64,6 +64,8 @@ class EjbRequestHandler {
EJBRequest req = new EJBRequest();
EJBResponse res = new EJBResponse();
+ res.start(EJBResponse.Time.TOTAL);
+
try {
req.readExternal(in);
} catch (Throwable t) {
@@ -98,7 +100,11 @@ class EjbRequestHandler {
Thread.currentThread().setContextClassLoader(classLoader);
try {
+ res.start(EJBResponse.Time.DESERIALIZATION);
+
req.getBody().readExternal(in);
+
+ res.stop(EJBResponse.Time.DESERIALIZATION);
} catch (Throwable t) {
replyWithFatalError(out, t, "Error caught during request
processing");
return;
@@ -122,6 +128,7 @@ class EjbRequestHandler {
return;
}
+ res.start(EJBResponse.Time.CONTAINER);
try {
switch (req.getRequestMethod()) {
// Remote interface methods
@@ -192,6 +199,8 @@ class EjbRequestHandler {
break;
}
+ res.stop(EJBResponse.Time.CONTAINER);
+
} catch (org.apache.openejb.InvalidateReferenceException e) {
res.setResponse(ResponseCodes.EJB_SYS_EXCEPTION, new
ThrowableArtifact(e.getRootCause()));
} catch (org.apache.openejb.ApplicationException e) {
@@ -240,7 +249,7 @@ class EjbRequestHandler {
req.getMethodParameters(),
req.getPrimaryKey()
);
-
+
res.setResponse(ResponseCodes.EJB_OK, result);
}
Modified:
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/MultipointDiscoveryAgent.java
URL:
http://svn.apache.org/viewvc/openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/MultipointDiscoveryAgent.java?rev=1308631&r1=1308630&r2=1308631&view=diff
==============================================================================
---
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/MultipointDiscoveryAgent.java
(original)
+++
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/MultipointDiscoveryAgent.java
Tue Apr 3 00:27:48 2012
@@ -16,6 +16,8 @@
*/
package org.apache.openejb.server.discovery;
+import org.apache.openejb.monitoring.Event;
+import org.apache.openejb.monitoring.Managed;
import org.apache.openejb.server.SelfManaging;
import org.apache.openejb.server.ServerService;
import org.apache.openejb.server.ServiceException;
@@ -33,6 +35,7 @@ import java.io.OutputStream;
import java.net.Socket;
import java.net.URI;
import java.util.LinkedHashSet;
+import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.StringTokenizer;
@@ -47,21 +50,30 @@ public class MultipointDiscoveryAgent im
private AtomicBoolean running = new AtomicBoolean(false);
+ @Managed
private String host = "127.0.0.1";
+
+ @Managed
private int port = 4212;
private String initialServers = "";
private long heartRate = 500;
+ @Managed(append = false)
private Tracker tracker;
+
private MultipointServer multipointServer;
+
private boolean debug = true;
private String name;
private String discoveryHost;
private Set<URI> roots;
private Duration reconnectDelay;
+ @Managed
+ private final Event restarts = new Event();
+
public MultipointDiscoveryAgent() {
}
@@ -80,7 +92,7 @@ public class MultipointDiscoveryAgent im
initialServers = options.get("initialServers", initialServers);
heartRate = options.get("heart_rate", heartRate);
discoveryHost = options.get("discoveryHost", host);
- name = options.get("discoveryName", MultipointServer.randomColor());
+ name = name != null ? name : options.get("discoveryName",
MultipointServer.randomColor());
reconnectDelay = options.get("reconnectDelay", new Duration("30
seconds"));
final Set<URI> uris = new LinkedHashSet<URI>();
@@ -88,8 +100,14 @@ public class MultipointDiscoveryAgent im
// Connect the initial set of peer servers
StringTokenizer st = new StringTokenizer(initialServers, ",");
while (st.hasMoreTokens()) {
- URI uri = URI.create("conn://" + st.nextToken().trim());
- uris.add(uri);
+ final String string = st.nextToken().trim();
+ if (string.startsWith("conn://")) {
+ final URI uri = URI.create(string);
+ uris.add(uri);
+ } else {
+ final URI uri = URI.create("conn://" + string);
+ uris.add(uri);
+ }
}
roots = uris;
@@ -111,6 +129,7 @@ public class MultipointDiscoveryAgent im
return host;
}
+ @Override
public String getName() {
return "multipoint";
}
@@ -147,11 +166,13 @@ public class MultipointDiscoveryAgent im
*
* @throws Exception
*/
+ @Managed
public void start() throws ServiceException {
try {
if (running.compareAndSet(false, true)) {
-
+ log.info("MultipointDiscoveryAgent Starting");
multipointServer = new MultipointServer(host, discoveryHost,
port, tracker, name, debug, roots, reconnectDelay).start();
+ log.info("MultipointDiscoveryAgent Started");
this.port = multipointServer.getPort();
@@ -161,13 +182,22 @@ public class MultipointDiscoveryAgent im
}
}
+ @Managed
+ public void restart() throws ServiceException {
+ stop();
+ start();
+ restarts.record();
+ }
+
/**
* stop the channel
*
* @throws Exception
*/
+ @Managed
public void stop() throws ServiceException {
if (running.compareAndSet(true, false)) {
+ log.info("MultipointDiscoveryAgent Stopping");
multipointServer.stop();
}
}
@@ -185,4 +215,95 @@ public class MultipointDiscoveryAgent im
public void setHost(String host) {
this.host = host;
}
+
+ @Managed
+ public URI getURI() {
+ return multipointServer.getMe();
+ }
+
+ @Managed
+ public Set<URI> getRoots() {
+ return multipointServer.getRoots();
+ }
+
+ @Managed
+ public long getRuns() {
+ return multipointServer.getRuns().get();
+ }
+
+ @Managed
+ public String getRunsLatest() {
+ return multipointServer.getRuns().getLatest();
+ }
+
+ @Managed
+ public long getRunsLatestTime() {
+ return multipointServer.getRuns().getLatestTime();
+ }
+
+ @Managed
+ public long getHeartbeats() {
+ return multipointServer.getHeartbeats().get();
+ }
+
+ @Managed
+ public String getHeartbeatsLatest() {
+ return multipointServer.getHeartbeats().getLatest();
+ }
+
+ @Managed
+ public long getHeartbeatsLatestTime() {
+ return multipointServer.getHeartbeats().getLatestTime();
+ }
+
+ @Managed
+ public long getSessionsCreated() {
+ return multipointServer.getSessionsCreated().get();
+ }
+
+ @Managed
+ public String getSessionsCreatedLatest() {
+ return multipointServer.getSessionsCreated().getLatest();
+ }
+
+ @Managed
+ public long getSessionsCreatedLatestTime() {
+ return multipointServer.getSessionsCreated().getLatestTime();
+ }
+
+ @Managed
+ public long getReconnects() {
+ return multipointServer.getReconnects().get();
+ }
+
+ @Managed
+ public String getReconnectsLatest() {
+ return multipointServer.getReconnects().getLatest();
+ }
+
+ @Managed
+ public long getReconnectsLatestTime() {
+ return multipointServer.getReconnects().getLatestTime();
+ }
+
+ @Managed
+ public long getJoined() {
+ return multipointServer.getJoined();
+ }
+
+ @Managed
+ public List<URI> getSessions() {
+ return multipointServer.getSessions();
+ }
+
+ @Managed
+ public List<URI> getConnectionsQueued() {
+ return multipointServer.getConnectionsQueued();
+ }
+
+ @Managed
+ public long getReconnectDelay() {
+ return multipointServer.getReconnectDelay();
+ }
+
}
\ No newline at end of file
Modified:
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/MultipointServer.java
URL:
http://svn.apache.org/viewvc/openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/MultipointServer.java?rev=1308631&r1=1308630&r2=1308631&view=diff
==============================================================================
---
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/MultipointServer.java
(original)
+++
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/MultipointServer.java
Tue Apr 3 00:27:48 2012
@@ -16,6 +16,9 @@
*/
package org.apache.openejb.server.discovery;
+import org.apache.openejb.monitoring.Event;
+import org.apache.openejb.monitoring.Managed;
+import org.apache.openejb.server.ServerRuntimeException;
import org.apache.openejb.util.Duration;
import org.apache.openejb.util.Join;
import org.apache.openejb.util.LogCategory;
@@ -50,20 +53,32 @@ import java.util.Random;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
/**
* @version $Rev$ $Date$
*/
+@Managed
public class MultipointServer {
- private static final Logger log =
Logger.getInstance(LogCategory.OPENEJB_SERVER.createChild("discovery"),
MultipointServer.class);
+ private static final Logger log =
Logger.getInstance(LogCategory.OPENEJB_SERVER.createChild("discovery").createChild("multipoint"),
MultipointServer.class);
private static final URI END_LIST = URI.create("end:list");
private final int port;
- private final Selector selector;
+
private final URI me;
+
private final Set<URI> roots = new LinkedHashSet<URI>();
+ private final Event runs = new Event();
+
+ private final Event heartbeats = new Event();
+
+ private final Event reconnects = new Event();
+ private final Event sessionsCreated = new Event();
+
/**
* Only used for toString to make debugging easier
*/
@@ -73,11 +88,19 @@ public class MultipointServer {
private final LinkedList<URI> connect = new LinkedList<URI>();
private final Map<URI, Session> connections = new HashMap<URI, Session>();
- private boolean debug = true;
private long joined = 0;
+
private long reconnectDelay;
+ private ServerSocketChannel serverChannel;
+
+ private final Selector selector;
+
+ private final Lock lock = new ReentrantLock();
+ private final Condition started = lock.newCondition();
+ private final Condition stopped = lock.newCondition();
+
public MultipointServer(int port, Tracker tracker) throws IOException {
this("localhost", "localhost", port, tracker, randomColor(), true,
Collections.EMPTY_SET, new Duration(30, TimeUnit.SECONDS));
}
@@ -91,7 +114,7 @@ public class MultipointServer {
this.tracker = tracker;
this.name = name;
- this.debug = debug;
+
if (roots != null) {
this.roots.addAll(roots);
}
@@ -109,9 +132,11 @@ public class MultipointServer {
log.debug(format);
- final InetSocketAddress address = new InetSocketAddress(bindHost,
port);
- final ServerSocketChannel serverChannel = ServerSocketChannel.open();
+ selector = Selector.open();
+
+ final InetSocketAddress address = new InetSocketAddress(bindHost,
port);
+ serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false);
final ServerSocket serverSocket = serverChannel.socket();
@@ -124,13 +149,55 @@ public class MultipointServer {
me = URI.create("conn://" + broadcastHost + ":" + this.port);
}
- selector = Selector.open();
-
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
println("Broadcasting");
}
+ public URI getMe() {
+ return me;
+ }
+
+ public Set<URI> getRoots() {
+ return roots;
+ }
+
+ public Event getRuns() {
+ return runs;
+ }
+
+ public Event getHeartbeats() {
+ return heartbeats;
+ }
+
+ public Event getReconnects() {
+ return reconnects;
+ }
+
+ public Event getSessionsCreated() {
+ return sessionsCreated;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public long getJoined() {
+ return joined;
+ }
+
+ public List<URI> getSessions() {
+ return new ArrayList<URI>(connections.keySet());
+ }
+
+ public List<URI> getConnectionsQueued() {
+ return new ArrayList<URI>(connect);
+ }
+
+ public long getReconnectDelay() {
+ return reconnectDelay;
+ }
+
public int getPort() {
return port;
}
@@ -146,6 +213,10 @@ public class MultipointServer {
if (connect.size() > 0) return;
if (System.nanoTime() - joined <= reconnectDelay) return;
+ log.info("MultipointReconnect{initialServers=" + roots.size() + "}");
+
+ reconnects.record();
+
for (URI root : roots) {
connect(root);
}
@@ -154,19 +225,63 @@ public class MultipointServer {
}
public MultipointServer start() {
if (running.compareAndSet(false, true)) {
+
+ String multipointServer = Join.join(".", "MultipointServer", name,
port);
+ log.info("MultipointServer Starting : Thread '" + multipointServer
+ "'");
+
Thread thread = new Thread(new Runnable() {
public void run() {
- _run();
+ signal(started);
+ try {
+ _run();
+ } finally {
+ signal(stopped);
+ }
}
});
- thread.setName(Join.join(".", "MultipointServer", name, port));
+ thread.setName(multipointServer);
thread.start();
+
+ await(started, 10, TimeUnit.SECONDS);
}
return this;
}
+ private void signal(Condition condition) {
+ lock.lock();
+ try {
+ condition.signal();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void await(Condition condition, long time, TimeUnit unit) {
+ lock.lock();
+ try {
+ condition.await(time, unit);
+ } catch (InterruptedException e) {
+ Thread.interrupted();
+ } finally {
+ lock.unlock();
+ }
+ }
+
public void stop() {
running.set(false);
+ try {
+ serverChannel.close();
+ } catch (IOException e) {
+ throw new CloseException(e);
+ } finally {
+ await(stopped, 10, TimeUnit.SECONDS);
+ }
+ }
+
+ public static class CloseException extends RuntimeException {
+ public CloseException(Throwable cause) {
+ super(cause);
+ }
}
public class Session {
@@ -177,8 +292,11 @@ public class MultipointServer {
private final ByteBuffer read = ByteBuffer.allocate(1024);
private final SelectionKey key;
private final List<URI> listed = new ArrayList<URI>();
+ private final long created = System.currentTimeMillis();
private ByteBuffer write;
+
+ @Managed
private State state = State.OPEN;
private URI uri;
public boolean hangup;
@@ -189,6 +307,8 @@ public class MultipointServer {
this.client = uri != null;
this.uri = uri != null ? uri : URI.create("conn://" +
address.getHostName() + ":" + address.getPort());
this.key = channel.register(selector, 0, this);
+ sessionsCreated.record();
+ log.info("Constructing " + this);
}
public Session ops(int ops) {
@@ -196,8 +316,17 @@ public class MultipointServer {
return this;
}
+ public long getCreated() {
+ return created;
+ }
+
public void state(int ops, State state) {
// trace("transition "+state +" "+ops);
+ if (this.state != state) {
+ if (log.isDebugEnabled()) {
+ log.debug(message(state.name()));
+ }
+ }
this.state = state;
if (ops > 0) key.interestOps(ops);
}
@@ -209,8 +338,9 @@ public class MultipointServer {
private void trace(String str) {
// println(message(str));
- if (debug && log.isDebugEnabled()) {
+ if (log.isDebugEnabled()) {
log.debug(message(str));
+// new Exception().fillInStackTrace().printStackTrace();
}
}
@@ -296,6 +426,7 @@ public class MultipointServer {
public String toString() {
return "Session{" +
"uri=" + uri +
+ ", created=" + created +
", state=" + state +
", owner=" + port +
", s=" + (client ? channel.socket().getPort() :
channel.socket().getLocalPort()) +
@@ -320,11 +451,12 @@ public class MultipointServer {
}
private void heartbeat() throws IOException {
+ heartbeats.record();
final Set<String> strings = tracker.getRegisteredServices();
- for (String string : strings) {
- trace(string);
- }
+// for (String string : strings) {
+// trace(string);
+// }
write(strings);
state(SelectionKey.OP_READ | SelectionKey.OP_WRITE,
State.HEARTBEAT);
}
@@ -350,15 +482,26 @@ public class MultipointServer {
// on each iteration of the loop, shrinking it down just a little to a
// account for the execution time of the loop itself.
+ int failed = 0;
while (running.get()) {
+ runs.record();
final long start = System.nanoTime();
try {
selector.select(selectorTimeout);
+ failed = 0;
} catch (IOException ex) {
- ex.printStackTrace();
- break;
+ if (failed++ > 100) {
+ log.fatal("Too many Multipoint Failures. Terminating
service.", ex);
+ return;
+ }
+ log.error("Multipoint Failure.", ex);
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ Thread.interrupted();
+ }
}
final Set keys = selector.selectedKeys();
@@ -462,7 +605,7 @@ public class MultipointServer {
// seen - needs to get maintained as "connected"
// TODO remove from seen
} catch (IOException e) {
- throw new RuntimeException(e);
+ throw new ServerRuntimeException(e);
}
}
}
@@ -493,7 +636,7 @@ public class MultipointServer {
// CLIENTs list last, so at this point we've read
// the server's list and have written ours
- session.trace("DONE WRITING");
+// session.trace("DONE WRITING");
session.state(SelectionKey.OP_READ, State.HEARTBEAT);
@@ -514,7 +657,7 @@ public class MultipointServer {
session.last = System.currentTimeMillis();
- session.trace("send");
+// session.trace("send");
session.state(SelectionKey.OP_READ, State.HEARTBEAT);
@@ -579,7 +722,7 @@ public class MultipointServer {
while ((message = session.read()) != null) {
- session.trace(message);
+// session.trace(message);
final URI uri = URI.create(message);
@@ -647,7 +790,7 @@ public class MultipointServer {
String message = null;
while ((message = session.read()) != null) {
- session.trace(message);
+// session.trace(message);
tracker.processData(message);
}
}
@@ -717,7 +860,6 @@ public class MultipointServer {
private void close(SelectionKey key) {
final Session session = (Session) key.attachment();
- session.state(0, State.CLOSED);
if (session.hangup) {
// This was a duplicate connection and was closed
@@ -725,14 +867,17 @@ public class MultipointServer {
// map as this particular session is not in that
// map -- only the good session that will not be
// closed is in there.
+ log.info("Hungup " + session);
session.trace("hungup");
} else {
+ log.info("Closed " + session);
session.trace("closed");
synchronized (connect) {
connections.remove(session.uri);
}
}
+ session.state(0, State.CLOSED);
hangup(key);
}
@@ -758,6 +903,7 @@ public class MultipointServer {
synchronized (connect) {
if (!connections.containsKey(uri) && !connect.contains(uri)) {
+ log.debug("Queuing{uri=" + uri + "}");
connect.addLast(uri);
}
}
@@ -770,15 +916,36 @@ public class MultipointServer {
// Session duplicate = null;
if (duplicate != null) {
+
+
session.trace("duplicate");
// At this point we know we have two sockets open
- // to the client, one created by them and one created
- // by us. We will both have detected this situation
+ // to the client, this can happen in two different ways
+ //
+ // 1. one created by them and one created by us.
+ // 2. two created by them and none created by us.
+ //
+ // For case #1, we will both have detected this situation
// and know it needs fixing. Only one of us can hangup
+ //
+ // For case #2, the client was likely disconnected and
+ // is calling back.
final Session[] sessions = {session, duplicate};
- Arrays.sort(sessions, new Comparator<Session>() {
+
+ if (!sessions[0].client && !sessions[1].client) {
+ // Case 1 -- Client is calling back
+ Arrays.sort(sessions, new Comparator<Session>() {
+ @Override
+ public int compare(Session a, Session b) {
+ return (int) (b.created - a.created);
+ }
+ });
+ } else {
+ // Case 2 -- We called each other at the same time
+
+ Arrays.sort(sessions, new Comparator<Session>() {
// Goal: Keep the connection with the lowest port number
///
// Low vs high is not very significant. The critical
@@ -811,6 +978,7 @@ public class MultipointServer {
return !a.client ? socket.getPort() :
socket.getLocalPort();
}
});
+ }
session = sessions[0];
duplicate = sessions[1];
Modified:
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/Tracker.java
URL:
http://svn.apache.org/viewvc/openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/Tracker.java?rev=1308631&r1=1308630&r2=1308631&view=diff
==============================================================================
---
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/Tracker.java
(original)
+++
openejb/branches/openejb-3.1.x/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/Tracker.java
Tue Apr 3 00:27:48 2012
@@ -16,10 +16,12 @@
*/
package org.apache.openejb.server.discovery;
+import org.apache.openejb.monitoring.Managed;
import org.apache.openejb.server.DiscoveryListener;
import org.apache.openejb.util.Logger;
import org.apache.openejb.util.LogCategory;
+import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -35,14 +37,21 @@ import java.io.IOException;
/**
* @version $Rev$ $Date$
*/
+@Managed(append = false)
public class Tracker {
private final Logger log;
+ @Managed
private final String group;
private final String groupPrefix;
+
+ @Managed
private final long heartRate;
+
+ @Managed
private final int maxMissedHeartbeats;
+
private final long reconnectDelay;
private final long maxReconnectDelay;
private final int maxReconnectAttempts;
@@ -87,6 +96,16 @@ public class Tracker {
return registeredServices.keySet();
}
+ @Managed
+ public Set<String> getServicesRegistered() {
+ return new HashSet<String>(registeredServices.keySet());
+ }
+
+ @Managed
+ public Set<String> getServicesDiscovered() {
+ return new HashSet<String>(discoveredServices.keySet());
+ }
+
public void registerService(URI serviceUri) throws IOException {
Service service = new Service(serviceUri);
this.registeredServices.put(service.broadcastString, service);
@@ -177,8 +196,8 @@ public class Tracker {
});
private void fireServiceRemovedEvent(final URI uri) {
- if (debug()) {
- log.debug(String.format("Removed Service{uri=%s}", uri));
+ if (log.isInfoEnabled()) {
+ log.info(String.format("Removed Service{uri=%s}", uri));
}
if (discoveryListener != null) {
@@ -198,8 +217,8 @@ public class Tracker {
}
private void fireServiceAddedEvent(final URI uri) {
- if (debug()) {
- log.debug(String.format("Added Service{uri=%s}", uri));
+ if (log.isInfoEnabled()) {
+ log.info(String.format("Added Service{uri=%s}", uri));
}
if (discoveryListener != null) {
@@ -226,8 +245,11 @@ public class Tracker {
}
}
+ @Managed
public class Service {
+ @Managed
private final URI uri;
+ @Managed
private final String broadcastString;
public Service(URI uri) {
@@ -253,11 +275,16 @@ public class Tracker {
private class ServiceVitals {
+ @Managed
private final Service service;
+ @Managed
private long lastHeartBeat;
+ @Managed
private long recoveryTime;
+ @Managed
private int failureCount;
+ @Managed
private boolean dead;
public ServiceVitals(Service service) {
Added:
openejb/branches/openejb-3.1.x/server/openejb-server/src/main/java/org/apache/openejb/server/ServerRuntimeException.java
URL:
http://svn.apache.org/viewvc/openejb/branches/openejb-3.1.x/server/openejb-server/src/main/java/org/apache/openejb/server/ServerRuntimeException.java?rev=1308631&view=auto
==============================================================================
---
openejb/branches/openejb-3.1.x/server/openejb-server/src/main/java/org/apache/openejb/server/ServerRuntimeException.java
(added)
+++
openejb/branches/openejb-3.1.x/server/openejb-server/src/main/java/org/apache/openejb/server/ServerRuntimeException.java
Tue Apr 3 00:27:48 2012
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.server;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class ServerRuntimeException extends RuntimeException {
+ public ServerRuntimeException(final String str) {
+ super(str);
+ }
+
+ public ServerRuntimeException(final String str, final Throwable e) {
+ super(str, e);
+ }
+
+ public ServerRuntimeException(final Exception e) {
+ super(e);
+ }
+}
Modified:
openejb/branches/openejb-3.1.x/server/openejb-server/src/main/java/org/apache/openejb/server/ServiceManager.java
URL:
http://svn.apache.org/viewvc/openejb/branches/openejb-3.1.x/server/openejb-server/src/main/java/org/apache/openejb/server/ServiceManager.java?rev=1308631&r1=1308630&r2=1308631&view=diff
==============================================================================
---
openejb/branches/openejb-3.1.x/server/openejb-server/src/main/java/org/apache/openejb/server/ServiceManager.java
(original)
+++
openejb/branches/openejb-3.1.x/server/openejb-server/src/main/java/org/apache/openejb/server/ServiceManager.java
Tue Apr 3 00:27:48 2012
@@ -20,6 +20,7 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -30,11 +31,16 @@ import org.apache.openejb.assembler.clas
import org.apache.openejb.assembler.classic.ServiceInfo;
import org.apache.openejb.loader.FileUtils;
import org.apache.openejb.loader.SystemInstance;
+import org.apache.openejb.monitoring.ManagedMBean;
+import org.apache.openejb.monitoring.ObjectNameBuilder;
import org.apache.openejb.util.LogCategory;
import org.apache.openejb.util.Logger;
import org.apache.xbean.recipe.ObjectRecipe;
import org.apache.xbean.recipe.Option;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
/**
* @version $Rev$ $Date$
* @org.apache.xbean.XBean element="serviceManager"
@@ -152,6 +158,19 @@ public abstract class ServiceManager {
registry.addDiscoveryAgent(agent);
}
+ MBeanServer server =
ManagementFactory.getPlatformMBeanServer();
+
+ final ObjectNameBuilder jmxName = new
ObjectNameBuilder("openejb");
+ jmxName.set("type", "ServerService");
+ jmxName.set("name", serviceName);
+
+ try {
+ final ObjectName objectName = jmxName.build();
+ server.registerMBean(new ManagedMBean(service),
objectName);
+ } catch (Exception e) {
+ logger.error("Unable to register MBean ", e);
+ }
+
return service;
} catch (Throwable t) {
logger.error("service.instantiation.err", t,
serviceClass.getName(), t.getClass().getName(), t.getMessage());
