[hornetq-commits] JBoss hornetq SVN: r10254 - in branches/Branch_2_2_EAP: native/src and 3 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Fri Feb 25 17:23:59 EST 2011
Author: clebert.suconic at jboss.com
Date: 2011-02-25 17:23:58 -0500 (Fri, 25 Feb 2011)
New Revision: 10254
Added:
branches/Branch_2_2_EAP/src/main/org/hornetq/core/asyncio/impl/HornetQFileLock.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/AIOFileLockNodeManager.java
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/server/impl/FileLockTest.java
Modified:
branches/Branch_2_2_EAP/native/bin/libHornetQAIO64.so
branches/Branch_2_2_EAP/native/src/JNI_AsynchronousFileImpl.cpp
branches/Branch_2_2_EAP/native/src/Version.h
branches/Branch_2_2_EAP/src/main/org/hornetq/core/asyncio/impl/AsynchronousFileImpl.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/FileLockNodeManager.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
Log:
JBPAPP-5956 - Workaround for gfs2 bug with fctl (adding a native call for flock)
Modified: branches/Branch_2_2_EAP/native/bin/libHornetQAIO64.so
===================================================================
(Binary files differ)
Modified: branches/Branch_2_2_EAP/native/src/JNI_AsynchronousFileImpl.cpp
===================================================================
--- branches/Branch_2_2_EAP/native/src/JNI_AsynchronousFileImpl.cpp 2011-02-25 11:40:04 UTC (rev 10253)
+++ branches/Branch_2_2_EAP/native/src/JNI_AsynchronousFileImpl.cpp 2011-02-25 22:23:58 UTC (rev 10254)
@@ -18,8 +18,8 @@
#include <fcntl.h>
#include <string>
#include <time.h>
+#include <sys/file.h>
-
#include "org_hornetq_core_asyncio_impl_AsynchronousFileImpl.h"
@@ -38,9 +38,48 @@
return (AIOController *) env->GetDirectBufferAddress(controllerAddress);
}
+/* Inaccessible static: log */
+/* Inaccessible static: totalMaxIO */
+/* Inaccessible static: loaded */
+/* Inaccessible static: EXPECTED_NATIVE_VERSION */
+/*
+ * Class: org_hornetq_core_asyncio_impl_AsynchronousFileImpl
+ * Method: openFile
+ * Signature: (Ljava/lang/String;)I
+ */
+JNIEXPORT jint JNICALL Java_org_hornetq_core_asyncio_impl_AsynchronousFileImpl_openFile
+ (JNIEnv * env , jclass , jstring jstrFileName)
+{
+ std::string fileName = convertJavaString(env, jstrFileName);
+ return open(fileName.data(), O_RDWR | O_CREAT, 0666);
+}
/*
+ * Class: org_hornetq_core_asyncio_impl_AsynchronousFileImpl
+ * Method: closeFile
+ * Signature: (I)V
+ */
+JNIEXPORT void JNICALL Java_org_hornetq_core_asyncio_impl_AsynchronousFileImpl_closeFile
+ (JNIEnv * , jclass , jint handle)
+{
+ close(handle);
+}
+
+/*
+ * Class: org_hornetq_core_asyncio_impl_AsynchronousFileImpl
+ * Method: flock
+ * Signature: (I)Z
+ */
+JNIEXPORT jboolean JNICALL Java_org_hornetq_core_asyncio_impl_AsynchronousFileImpl_flock
+ (JNIEnv * , jclass , jint handle)
+{
+ return flock(handle, LOCK_EX | LOCK_NB) == 0;
+}
+
+
+
+/*
* Class: org_jboss_jaio_libaioimpl_LibAIOController
* Method: init
* Signature: (Ljava/lang/String;Ljava/lang/Class;)J
Modified: branches/Branch_2_2_EAP/native/src/Version.h
===================================================================
--- branches/Branch_2_2_EAP/native/src/Version.h 2011-02-25 11:40:04 UTC (rev 10253)
+++ branches/Branch_2_2_EAP/native/src/Version.h 2011-02-25 22:23:58 UTC (rev 10254)
@@ -3,6 +3,6 @@
// This definition needs to match org.hornetq.core.asyncio.impl.AsynchronousFileImpl.EXPECTED_NATIVE_VERSION
// Or else the native module won't be loaded because of version mismatches
-#define _VERSION_NATIVE_AIO 30
+#define _VERSION_NATIVE_AIO 31
#endif
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/asyncio/impl/AsynchronousFileImpl.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/asyncio/impl/AsynchronousFileImpl.java 2011-02-25 11:40:04 UTC (rev 10253)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/asyncio/impl/AsynchronousFileImpl.java 2011-02-25 22:23:58 UTC (rev 10254)
@@ -14,6 +14,7 @@
package org.hornetq.core.asyncio.impl;
import java.nio.ByteBuffer;
+import java.nio.channels.FileLock;
import java.util.PriorityQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Semaphore;
@@ -51,7 +52,7 @@
/** This definition needs to match Version.h on the native sources.
Or else the native module won't be loaded because of version mismatches */
- private static int EXPECTED_NATIVE_VERSION = 30;
+ private static int EXPECTED_NATIVE_VERSION = 31;
/** Used to determine the next writing sequence */
private final AtomicLong nextWritingSequence = new AtomicLong(0);
@@ -615,8 +616,29 @@
// completely done, or we might get beautiful GPFs
pollerLatch.await();
}
+
+ public static FileLock lock(int handle)
+ {
+ if (flock(handle))
+ {
+ return new HornetQFileLock(handle);
+ }
+ else
+ {
+ return null;
+ }
+ }
// Native ----------------------------------------------------------------------------
+
+
+ // Functions used for locking files .....
+ public static native int openFile(String fileName);
+
+ public static native void closeFile(int handle);
+
+ private static native boolean flock(int handle);
+ // Functions used for locking files ^^^^^^^^
private static native void resetBuffer(ByteBuffer directByteBuffer, int size);
Added: branches/Branch_2_2_EAP/src/main/org/hornetq/core/asyncio/impl/HornetQFileLock.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/asyncio/impl/HornetQFileLock.java (rev 0)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/asyncio/impl/HornetQFileLock.java 2011-02-25 22:23:58 UTC (rev 10254)
@@ -0,0 +1,76 @@
+/*
+ * Copyright 2010 Red Hat, Inc.
+ * Red Hat 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.hornetq.core.asyncio.impl;
+
+import java.io.IOException;
+import java.nio.channels.FileLock;
+
+/**
+ * A HornetQFileLock
+ *
+ * @author clebertsuconic
+ *
+ *
+ */
+public class HornetQFileLock extends FileLock
+{
+
+ private final int handle;
+
+ /**
+ * @param channel
+ * @param position
+ * @param size
+ * @param shared
+ */
+ protected HornetQFileLock(final int handle)
+ {
+ super(null, 0, 0, false);
+ this.handle = handle;
+ }
+
+ @Override
+ public boolean isValid()
+ {
+ return true;
+ }
+
+ /* (non-Javadoc)
+ * @see java.nio.channels.FileLock#release()
+ */
+ @Override
+ public void release() throws IOException
+ {
+ AsynchronousFileImpl.closeFile(handle);
+ }
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ // Public --------------------------------------------------------
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
Added: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/AIOFileLockNodeManager.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/AIOFileLockNodeManager.java (rev 0)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/AIOFileLockNodeManager.java 2011-02-25 22:23:58 UTC (rev 10254)
@@ -0,0 +1,131 @@
+/*
+ * Copyright 2010 Red Hat, Inc.
+ * Red Hat 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.hornetq.core.server.impl;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.channels.FileLock;
+
+import org.hornetq.core.asyncio.impl.AsynchronousFileImpl;
+
+/**
+ * This is using the HornetQ Libaio Native to perform calls to flock on Linux system
+ * At the current version of RHEL there's a bug on GFS2 and because of that fctl is not functional
+ * what will cause issues on Failover over Shared Storage.
+ *
+ * This will provide an alternative to perform locks through our native module until fctl is fixed on Linux.
+ *
+ * https://bugzilla.redhat.com/show_bug.cgi?id=678585
+ *
+ * @author clebertsuconic
+ *
+ *
+ */
+public class AIOFileLockNodeManager extends FileLockNodeManager
+{
+
+ /**
+ * @param directory
+ */
+ public AIOFileLockNodeManager(final String directory)
+ {
+ super(directory);
+ }
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ // Public --------------------------------------------------------
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+ @Override
+ protected FileLock tryLock(final int lockPos) throws Exception
+ {
+ File file = newFileForRegionLock(lockPos);
+
+ int handle = AsynchronousFileImpl.openFile(file.getAbsolutePath());
+
+ if (handle < 0)
+ {
+ throw new IOException("couldn't open file " + file.getAbsolutePath());
+ }
+
+ FileLock lock = AsynchronousFileImpl.lock(handle);
+
+ if (lock == null)
+ {
+ AsynchronousFileImpl.closeFile(handle);
+ }
+
+ return lock;
+
+ }
+
+ @Override
+ protected FileLock lock(final int liveLockPos) throws IOException
+ {
+ File file = newFileForRegionLock(liveLockPos);
+
+ int handle = AsynchronousFileImpl.openFile(file.getAbsolutePath());
+
+ if (handle < 0)
+ {
+ throw new IOException("couldn't open file " + file.getAbsolutePath());
+ }
+
+ while (!interrupted)
+ {
+ FileLock lockFile = AsynchronousFileImpl.lock(handle);
+ if (lockFile != null)
+ {
+ return lockFile;
+ }
+ else
+ {
+ try
+ {
+ Thread.sleep(500);
+ }
+ catch (InterruptedException e)
+ {
+ return null;
+ }
+ }
+ }
+
+ return null;
+ }
+
+ /**
+ * @param liveLockPos
+ * @return
+ */
+ protected File newFileForRegionLock(final int liveLockPos)
+ {
+ File file = newFile("server." + liveLockPos + ".lock");
+ return file;
+ }
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/FileLockNodeManager.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/FileLockNodeManager.java 2011-02-25 11:40:04 UTC (rev 10253)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/FileLockNodeManager.java 2011-02-25 22:23:58 UTC (rev 10254)
@@ -13,12 +13,6 @@
package org.hornetq.core.server.impl;
-import org.hornetq.api.core.SimpleString;
-import org.hornetq.core.logging.Logger;
-import org.hornetq.core.server.NodeManager;
-import org.hornetq.utils.UUID;
-import org.hornetq.utils.UUIDGenerator;
-
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
@@ -26,6 +20,12 @@
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
+import org.hornetq.api.core.SimpleString;
+import org.hornetq.core.logging.Logger;
+import org.hornetq.core.server.NodeManager;
+import org.hornetq.utils.UUID;
+import org.hornetq.utils.UUIDGenerator;
+
/**
* @author <a href="mailto:andy.taylor at jboss.com">Andy Taylor</a>
* Date: Oct 13, 2010
@@ -35,13 +35,13 @@
{
private static final Logger log = Logger.getLogger(FileLockNodeManager.class);
- private static final String SERVER_LOCK_NAME = "server.lock";
+ protected static final String SERVER_LOCK_NAME = "server.lock";
private static final String ACCESS_MODE = "rw";
- private static final int LIVE_LOCK_POS = 1;
+ private static final int LIVE_LOCK_POS = 1;
- private static final int BACKUP_LOCK_POS = 2;
+ private static final int BACKUP_LOCK_POS = 2;
private static final int LOCK_LENGTH = 1;
@@ -63,34 +63,34 @@
private final String directory;
- boolean interrupted = false;
+ protected boolean interrupted = false;
-
public FileLockNodeManager(final String directory)
{
this.directory = directory;
}
+ @Override
public void start() throws Exception
{
- if(isStarted())
+ if (isStarted())
{
return;
}
- File file = new File(directory, SERVER_LOCK_NAME);
+ File file = newFile(FileLockNodeManager.SERVER_LOCK_NAME);
boolean fileCreated = false;
if (!file.exists())
{
fileCreated = file.createNewFile();
- if(!fileCreated)
+ if (!fileCreated)
{
throw new IllegalStateException("Unable to create server lock file");
}
}
- RandomAccessFile raFile = new RandomAccessFile(file, ACCESS_MODE);
+ RandomAccessFile raFile = new RandomAccessFile(file, FileLockNodeManager.ACCESS_MODE);
channel = raFile.getChannel();
@@ -98,9 +98,9 @@
{
ByteBuffer id = ByteBuffer.allocateDirect(3);
byte[] bytes = new byte[3];
- bytes[0] = FIRST_TIME_START;
- bytes[1] = FIRST_TIME_START;
- bytes[2] = FIRST_TIME_START;
+ bytes[0] = FileLockNodeManager.FIRST_TIME_START;
+ bytes[1] = FileLockNodeManager.FIRST_TIME_START;
+ bytes[2] = FileLockNodeManager.FIRST_TIME_START;
id.put(bytes, 0, 3);
id.position(0);
channel.write(id, 0);
@@ -112,6 +112,7 @@
super.start();
}
+ @Override
public void stop() throws Exception
{
channel.close();
@@ -122,14 +123,15 @@
@Override
public boolean isAwaitingFailback() throws Exception
{
- return getState() == FAILINGBACK;
+ return getState() == FileLockNodeManager.FAILINGBACK;
}
+ @Override
public boolean isBackupLive() throws Exception
{
FileLock liveAttemptLock;
- liveAttemptLock = channel.tryLock(LIVE_LOCK_POS, LOCK_LENGTH, false);
- if(liveAttemptLock == null)
+ liveAttemptLock = tryLock(FileLockNodeManager.LIVE_LOCK_POS);
+ if (liveAttemptLock == null)
{
return true;
}
@@ -139,7 +141,13 @@
return false;
}
}
+
+ public boolean isLiveLocked()
+ {
+ return liveLock != null;
+ }
+
@Override
public void interrupt()
{
@@ -152,66 +160,69 @@
releaseBackupLock();
}
-
+ @Override
public void awaitLiveNode() throws Exception
{
do
{
byte state = getState();
- while (state == NOT_STARTED || state == FIRST_TIME_START)
+ while (state == FileLockNodeManager.NOT_STARTED || state == FileLockNodeManager.FIRST_TIME_START)
{
- log.debug("awaiting live node startup state='" + state + "'");
+ FileLockNodeManager.log.debug("awaiting live node startup state='" + state + "'");
Thread.sleep(2000);
state = getState();
}
- liveLock = lock(LIVE_LOCK_POS, 1);
+ liveLock = lock(FileLockNodeManager.LIVE_LOCK_POS);
state = getState();
- if (state == PAUSED)
+ if (state == FileLockNodeManager.PAUSED)
{
liveLock.release();
- log.debug("awaiting live node restarting");
+ FileLockNodeManager.log.debug("awaiting live node restarting");
Thread.sleep(2000);
}
- else if (state == FAILINGBACK)
+ else if (state == FileLockNodeManager.FAILINGBACK)
{
liveLock.release();
- log.debug("awaiting live node failing back");
+ FileLockNodeManager.log.debug("awaiting live node failing back");
Thread.sleep(2000);
}
- else if (state == LIVE)
+ else if (state == FileLockNodeManager.LIVE)
{
- log.debug("acquired live node lock state = " + (char)state);
+ FileLockNodeManager.log.debug("acquired live node lock state = " + (char)state);
break;
}
}
while (true);
}
+ @Override
public void startBackup() throws Exception
{
- log.info("Waiting to become backup node");
+ FileLockNodeManager.log.info("Waiting to become backup node");
- backupLock = lock(BACKUP_LOCK_POS, LOCK_LENGTH);
- log.info("** got backup lock");
+ backupLock = lock(FileLockNodeManager.BACKUP_LOCK_POS);
+ FileLockNodeManager.log.info("** got backup lock");
readNodeId();
}
+ @Override
public void startLiveNode() throws Exception
{
setFailingBack();
- log.info("Waiting to obtain live lock");
+ FileLockNodeManager.log.info("Waiting to obtain live lock");
- liveLock = lock(LIVE_LOCK_POS, LOCK_LENGTH);
+ liveLock = lock(FileLockNodeManager.LIVE_LOCK_POS);
- log.info("Live Server Obtained live lock");
+ FileLockNodeManager.log.info("Live Server Obtained live lock");
setLive();
}
+ @Override
public void pauseLiveServer() throws Exception
{
setPaused();
@@ -221,20 +232,36 @@
}
}
+ @Override
public void crashLiveServer() throws Exception
{
- liveLock.release();
+ if (liveLock != null)
+ {
+ liveLock.release();
+ liveLock = null;
+ }
}
+ @Override
public void stopBackup() throws Exception
{
- backupLock.release();
+ if (backupLock != null)
+ {
+ backupLock.release();
+ backupLock = null;
+ }
+
}
+ public String getDirectory()
+ {
+ return directory;
+ }
+
private void setLive() throws Exception
{
ByteBuffer bb = ByteBuffer.allocateDirect(1);
- bb.put(LIVE);
+ bb.put(FileLockNodeManager.LIVE);
bb.position(0);
channel.write(bb, 0);
channel.force(true);
@@ -243,7 +270,7 @@
private void setFailingBack() throws Exception
{
ByteBuffer bb = ByteBuffer.allocateDirect(1);
- bb.put(FAILINGBACK);
+ bb.put(FileLockNodeManager.FAILINGBACK);
bb.position(0);
channel.write(bb, 0);
channel.force(true);
@@ -252,7 +279,7 @@
private void setPaused() throws Exception
{
ByteBuffer bb = ByteBuffer.allocateDirect(1);
- bb.put(PAUSED);
+ bb.put(FileLockNodeManager.PAUSED);
bb.position(0);
channel.write(bb, 0);
channel.force(true);
@@ -265,10 +292,12 @@
read = channel.read(bb, 0);
if (read <= 0)
{
- return NOT_STARTED;
+ return FileLockNodeManager.NOT_STARTED;
}
else
+ {
return bb.get(0);
+ }
}
private void releaseBackupLock() throws Exception
@@ -276,6 +305,7 @@
if (backupLock != null)
{
backupLock.release();
+ backupLock = null;
}
}
@@ -283,7 +313,7 @@
{
ByteBuffer id = ByteBuffer.allocateDirect(16);
int read = channel.read(id, 3);
- if(read != 16)
+ if (read != 16)
{
uuid = UUIDGenerator.getInstance().generateUUID();
nodeID = new SimpleString(uuid.toString());
@@ -306,7 +336,7 @@
{
ByteBuffer id = ByteBuffer.allocateDirect(16);
int read = channel.read(id, 3);
- if(read != 16)
+ if (read != 16)
{
throw new IllegalStateException("live server did not write id to file");
}
@@ -320,39 +350,77 @@
}
}
- private FileLock lock(int liveLockPos, int i) throws IOException
+ /**
+ * @return
+ */
+ protected File newFile(final String fileName)
{
- try
+ File file = new File(directory, fileName);
+ return file;
+ }
+
+ protected FileLock tryLock(final int lockPos) throws Exception
+ {
+ return channel.tryLock(lockPos, LOCK_LENGTH, false);
+ }
+
+
+ protected FileLock lock(final int liveLockPos) throws IOException
+ {
+ while (!interrupted)
{
- return channel.lock(liveLockPos, i, false);
+ FileLock lock = null;
+ try
+ {
+ lock = channel.tryLock(liveLockPos, 1, false);
+ }
+ catch (java.nio.channels.OverlappingFileLockException ex)
+ {
+ // This just means that another object on the same JVM is holding the lock
+ }
+
+ if (lock == null)
+ {
+ try
+ {
+ Thread.sleep(500);
+ }
+ catch (InterruptedException e)
+ {
+ return null;
+ }
+ }
+ else
+ {
+ return lock;
+ }
}
- catch (IOException e)
+
+ // todo this is here because sometimes channel.lock throws a resource deadlock exception but trylock works,
+ // need to investigate further and review
+ FileLock lock;
+ do
{
- //todo this is here because sometimes channel.lock throws a resource deadlock exception but trylock works, need to investigate further and review
- FileLock lock;
- do
+ lock = channel.tryLock(liveLockPos, 1, false);
+ if (lock == null)
{
- lock = channel.tryLock(liveLockPos, i, false);
- if (lock == null)
+ try
{
- try
- {
- Thread.sleep(500);
- }
- catch (InterruptedException e1)
- {
- //
- }
+ Thread.sleep(500);
}
- if (interrupted)
+ catch (InterruptedException e1)
{
- interrupted = false;
- throw new IOException(new InterruptedException());
+ //
}
}
- while(lock == null);
- return lock;
+ if (interrupted)
+ {
+ interrupted = false;
+ throw new IOException(new InterruptedException());
+ }
}
+ while (lock == null);
+ return lock;
}
+
}
-
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-02-25 11:40:04 UTC (rev 10253)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-02-25 22:23:58 UTC (rev 10254)
@@ -38,6 +38,7 @@
import org.hornetq.api.core.HornetQException;
import org.hornetq.api.core.Pair;
import org.hornetq.api.core.SimpleString;
+import org.hornetq.core.asyncio.impl.AsynchronousFileImpl;
import org.hornetq.core.client.impl.ClientSessionFactoryImpl;
import org.hornetq.core.config.BridgeConfiguration;
import org.hornetq.core.config.Configuration;
@@ -87,6 +88,7 @@
import org.hornetq.core.server.Bindable;
import org.hornetq.core.server.Divert;
import org.hornetq.core.server.HornetQServer;
+import org.hornetq.core.server.JournalType;
import org.hornetq.core.server.MemoryManager;
import org.hornetq.core.server.NodeManager;
import org.hornetq.core.server.Queue;
@@ -289,7 +291,14 @@
*/
protected NodeManager createNodeManager(final String directory)
{
- return new FileLockNodeManager(directory);
+ if (configuration.getJournalType() == JournalType.ASYNCIO && AsynchronousFileImpl.isLoaded())
+ {
+ return new AIOFileLockNodeManager(directory);
+ }
+ else
+ {
+ return new FileLockNodeManager(directory);
+ }
}
private class NoSharedStoreLiveActivation implements Activation
Added: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/server/impl/FileLockTest.java
===================================================================
--- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/server/impl/FileLockTest.java (rev 0)
+++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/server/impl/FileLockTest.java 2011-02-25 22:23:58 UTC (rev 10254)
@@ -0,0 +1,100 @@
+/*
+ * Copyright 2010 Red Hat, Inc.
+ * Red Hat 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.hornetq.tests.unit.core.server.impl;
+
+import java.io.File;
+
+import org.hornetq.core.asyncio.impl.AsynchronousFileImpl;
+import org.hornetq.core.server.impl.AIOFileLockNodeManager;
+import org.hornetq.core.server.impl.FileLockNodeManager;
+import org.hornetq.tests.util.UnitTestCase;
+
+/**
+ * A FileLockTest
+ *
+ * @author clebertsuconic
+ *
+ *
+ */
+public class FileLockTest extends UnitTestCase
+{
+
+ protected void setUp() throws Exception
+ {
+ super.setUp();
+ File file = new File(getTestDir());
+ file.mkdirs();
+ }
+
+
+ public void testNIOLock() throws Exception
+ {
+ doTestLock(new FileLockNodeManager(getTestDir()), new FileLockNodeManager(getTestDir()));
+
+ }
+
+ public void testAIOLock() throws Exception
+ {
+ if (AsynchronousFileImpl.isLoaded())
+ {
+ doTestLock(new AIOFileLockNodeManager(getTestDir()), new AIOFileLockNodeManager(getTestDir()));
+ }
+
+ }
+
+ public void doTestLock(final FileLockNodeManager lockManager1, final FileLockNodeManager lockManager2) throws Exception
+ {
+ lockManager1.start();
+ lockManager2.start();
+
+ lockManager1.startLiveNode();
+
+ Thread t = new Thread()
+ {
+ @Override
+ public void run()
+ {
+ try
+ {
+ lockManager2.startLiveNode();
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ }
+ }
+ };
+
+ t.start();
+
+ assertTrue(lockManager1.isLiveLocked());
+ Thread.sleep(500);
+ assertFalse(lockManager2.isLiveLocked());
+
+ lockManager1.crashLiveServer();
+
+ t.join();
+
+ assertFalse(lockManager1.isLiveLocked());
+ assertTrue(lockManager2.isLiveLocked());
+
+ lockManager2.crashLiveServer();
+
+ lockManager1.stop();
+ lockManager2.stop();
+
+
+ }
+
+}
More information about the hornetq-commits
mailing list