5:55 p.m.
Author: manik.surtani(a)jboss.com
Date: 2008-07-29 18:55:16 -0400 (Tue, 29 Jul 2008)
New Revision: 6422
Modified:
core/trunk/src/main/java/org/jboss/cache/loader/TcpDelegatingCacheLoader.java
core/trunk/src/main/java/org/jboss/cache/loader/tcp/TcpCacheServer.java
Log:
Added more debug, sync on loader and not on stream
Modified: core/trunk/src/main/java/org/jboss/cache/loader/TcpDelegatingCacheLoader.java
===================================================================
---
core/trunk/src/main/java/org/jboss/cache/loader/TcpDelegatingCacheLoader.java 2008-07-29
21:40:01 UTC (rev 6421)
+++
core/trunk/src/main/java/org/jboss/cache/loader/TcpDelegatingCacheLoader.java 2008-07-29
22:55:16 UTC (rev 6422)
@@ -42,11 +42,12 @@
*/
public class TcpDelegatingCacheLoader extends AbstractCacheLoader
{
- private Socket sock;
+ volatile private Socket sock;
private TcpDelegatingCacheLoaderConfig config;
- ObjectInputStream in;
- ObjectOutputStream out;
+ volatile ObjectInputStream in;
+ volatile ObjectOutputStream out;
private static final Log log = LogFactory.getLog(TcpDelegatingCacheLoader.class);
+ private static final boolean trace = log.isTraceEnabled();
private static Method GET_CHILDREN_METHOD, GET_METHOD, PUT_KEY_METHOD,
PUT_DATA_METHOD, REMOVE_KEY_METHOD, REMOVE_METHOD, PUT_MODS_METHOD, EXISTS_METHOD,
REMOVE_DATA_METHOD;
static
@@ -106,7 +107,10 @@
{
try
{
- return m.invoke(this, params);
+ if (trace) log.trace("About to invoke operation " + m);
+ Object rv = m.invoke(this, params);
+ if (trace) log.trace("Completed invocation of " + m);
+ return rv;
}
catch (IllegalAccessException e)
{
@@ -137,7 +141,8 @@
throw new CacheException("Problems invoking method call!", e);
}
}
- } while (System.currentTimeMillis() < endTime);
+ }
+ while (System.currentTimeMillis() < endTime);
throw new CacheException("Unable to communicate with TCPCacheServer(" +
config.getHost() + ":" + config.getPort() + ") after " +
config.getTimeout() + " millis, with reconnects every " +
config.getReconnectWaitTime() + " millis.");
}
@@ -149,6 +154,7 @@
return (Set<?>) invokeWithRetries(GET_CHILDREN_METHOD, fqn);
}
+ @SuppressWarnings("unchecked")
public Map<Object, Object> get(Fqn name) throws Exception
{
return (Map<Object, Object>) invokeWithRetries(GET_METHOD, name);
@@ -195,7 +201,7 @@
protected Set<?> _getChildrenNames(Fqn fqn) throws Exception
{
Set cn;
- synchronized (out)
+ synchronized (this)
{
out.reset();
out.writeByte(TcpCacheOperations.GET_CHILDREN_NAMES);
@@ -214,9 +220,10 @@
return cn;
}
+ @SuppressWarnings("unchecked")
protected Map<Object, Object> _get(Fqn name) throws Exception
{
- synchronized (out)
+ synchronized (this)
{
out.reset();
@@ -234,7 +241,7 @@
protected boolean _exists(Fqn name) throws Exception
{
- synchronized (out)
+ synchronized (this)
{
out.reset();
@@ -252,7 +259,7 @@
protected Object _put(Fqn name, Object key, Object value) throws Exception
{
- synchronized (out)
+ synchronized (this)
{
out.reset();
@@ -272,7 +279,7 @@
protected void _put(Fqn name, Map<Object, Object> attributes) throws Exception
{
- synchronized (out)
+ synchronized (this)
{
out.reset();
@@ -290,7 +297,7 @@
protected void _put(List<Modification> modifications) throws Exception
{
- synchronized (out)
+ synchronized (this)
{
out.reset();
@@ -312,7 +319,7 @@
protected Object _remove(Fqn fqn, Object key) throws Exception
{
- synchronized (out)
+ synchronized (this)
{
out.reset();
@@ -331,7 +338,7 @@
protected void _remove(Fqn fqn) throws Exception
{
- synchronized (out)
+ synchronized (this)
{
out.reset();
@@ -348,7 +355,7 @@
protected void _removeData(Fqn fqn) throws Exception
{
- synchronized (out)
+ synchronized (this)
{
out.reset();
Modified: core/trunk/src/main/java/org/jboss/cache/loader/tcp/TcpCacheServer.java
===================================================================
--- core/trunk/src/main/java/org/jboss/cache/loader/tcp/TcpCacheServer.java 2008-07-29
21:40:01 UTC (rev 6421)
+++ core/trunk/src/main/java/org/jboss/cache/loader/tcp/TcpCacheServer.java 2008-07-29
22:55:16 UTC (rev 6422)
@@ -51,7 +51,8 @@
* whether or not to start the server thread as a daemon. Should be false if started
from the command line, true if started as an MBean.
*/
boolean daemon = true;
- static final Log log = LogFactory.getLog(TcpCacheServer.class);
+ private static final Log log = LogFactory.getLog(TcpCacheServer.class);
+ private final static boolean trace = log.isTraceEnabled();
public String getBindAddress()
@@ -251,7 +252,7 @@
public void start()
{
- t = new Thread(this, "TcpCacheServer.Connection");
+ t = new Thread(this, "TcpCacheServer.Connection(" + sock.getPort() +
")");
t.setDaemon(true);
t.start();
}
@@ -296,10 +297,11 @@
NodeSPI n;
boolean flag;
- while (t != null && Thread.currentThread().equals(t))
+ while (t != null && Thread.currentThread().equals(t) &&
t.isAlive())
{
try
{
+ if (trace) log.trace("Reading next byte");
op = input.readByte();
}
catch (IOException e)
@@ -311,6 +313,7 @@
try
{
+ if (trace) log.trace("Resetting output");
output.reset();
switch (op)
{
@@ -328,7 +331,7 @@
break;
case TcpCacheOperations.GET:
fqn = (Fqn) input.readObject();
- n = (NodeSPI) c.getRoot().getChild(fqn);
+ n = c.getNode(fqn);
if (n == null)
{
// node doesn't exist - return null
@@ -428,6 +431,7 @@
log.error("Operation " + op + " unknown");
break;
}
+ if (trace) log.trace("Flushing stream");
output.flush();
}
catch (Exception e)