[jboss-cvs] JBoss Messaging SVN: r3824 - in trunk/tests: src/org/jboss/messaging/core and 4 other directories.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Wed Feb 27 12:15:28 EST 2008
Author: clebert.suconic at jboss.com
Date: 2008-02-27 12:15:28 -0500 (Wed, 27 Feb 2008)
New Revision: 3824
Added:
trunk/tests/src/org/jboss/messaging/core/asyncio/
trunk/tests/src/org/jboss/messaging/core/asyncio/impl/
trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/
trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/integration/
trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/integration/MultiThreadWriteNativeTest.java
trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/integration/SingleThreadWriteNativeTest.java
Modified:
trunk/tests/build.xml
Log:
Adding native tests
Modified: trunk/tests/build.xml
===================================================================
--- trunk/tests/build.xml 2008-02-27 17:14:55 UTC (rev 3823)
+++ trunk/tests/build.xml 2008-02-27 17:15:28 UTC (rev 3824)
@@ -96,6 +96,7 @@
<property name="source.tests.java" value="${tests.root}/src"/>
<property name="source.tests.stylesheets" value="${source.tests.java}/stylesheets"/>
<property name="tests.output" value="${tests.root}/output"/>
+ <property name="build.native" value="../native/src/.libs"/>
<property name="build.tests.classes" value="${tests.output}/classes"/>
<property name="build.tests.lib" value="${tests.output}/lib"/>
<property name="build.tests.reports" value="${tests.output}/reports"/>
@@ -475,6 +476,7 @@
showoutput="${junit.showoutput}"
timeout="${junit.timeout}">
+ <sysproperty key="java.library.path" value="${build.native}" />
<sysproperty key="jbm.remoting.disable.invm" value="${disable.invm}" />
<sysproperty key="remote" value="false"/>
<sysproperty key="module.output" value="${tests.output}"/>
@@ -511,6 +513,7 @@
<exclude name="**/jms/ConnectionConsumerTest.class"/>
<include name="**/messaging/util/**/${test-mask}.class"/>
+ <exclude name="**/*NativeTest.class"/>
<exclude name="**/jms/MemLeakTest.class"/>
<exclude name="**/jms/RemotingConnectionConfigurationTest.class"/>
<exclude name="**/jms/stress/**"/>
@@ -526,6 +529,57 @@
</junit>
</target>
+ <target name="native-tests" depends="tests-jar, prepare-testdirs, clear-test-logs"
+ description="Runs all available tests an in-VM configuration">
+
+ <echo message=""/>
+ <echo message="Running nativeTests tests, fork=${junit.fork}, junit.batchtest.fork=${junit.batchtest.fork}, disable.invm=${disable.invm}"/>
+ <echo message=""/>
+
+ <junit printsummary="${junit.printsummary}"
+ fork="on"
+ forkMode="once"
+ includeantruntime="${junit.includeantruntime}"
+ haltonerror="${junit.haltonerror}"
+ haltonfailure="${junit.haltonfailure}"
+ showoutput="${junit.showoutput}"
+ timeout="${junit.timeout}">
+
+ <sysproperty key="java.library.path" value="${build.native}" />
+ <sysproperty key="jbm.remoting.disable.invm" value="${disable.invm}" />
+ <sysproperty key="remote" value="false"/>
+ <sysproperty key="module.output" value="${tests.output}"/>
+ <sysproperty key="test.bind.address" value="${test.bind.address}"/>
+ <sysproperty key="jgroups.bind_addr" value="${test.bind.address}"/>
+ <sysproperty key="test.bind.address" value="${test.bind.address}"/>
+ <sysproperty key="jboss.messaging.groupname" value="${jboss.messaging.groupname}"/>
+ <sysproperty key="jboss.messaging.datachanneludpport" value="${jboss.messaging.datachanneludpport}"/>
+ <sysproperty key="jboss.messaging.controlchanneludpport" value="${jboss.messaging.controlchanneludpport}"/>
+ <sysproperty key="jboss.messaging.datachanneludpaddress" value="${jboss.messaging.datachanneludpaddress}"/>
+ <sysproperty key="jboss.messaging.controlchanneludpaddress" value="${jboss.messaging.controlchanneludpaddress}"/>
+ <sysproperty key="jboss.messaging.ipttl" value="${jboss.messaging.ipttl}"/>
+ <sysproperty key="test.database" value="${functional.tests.database}"/>
+ <sysproperty key="test.logfile.suffix" value="invm"/>
+ <sysproperty key="build.lib" value="${build.lib}"/>
+ <sysproperty key="objectstore.dir" value="${objectstore.dir}"/>
+ <jvmarg value="-Xmx512M"/>
+ <!--<jvmarg line="-Xmx512M -Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5005"/>-->
+ <!--<jvmarg value="-ea"/>-->
+ <classpath refid="test.execution.classpath"/>
+ <formatter type="xml" usefile="${junit.formatter.usefile}"/>
+ <batchtest todir="${junit.batchtest.todir}"
+ haltonfailure="${junit.batchtest.haltonfailure}"
+ haltonerror="${junit.batchtest.haltonerror}">
+ <formatter type="plain" usefile="${junit.formatter.usefile}"/>
+ <fileset dir="${build.tests.classes}">
+ <include name="**/*NativeTest.class"/>
+ </fileset>
+ </batchtest>
+ </junit>
+ </target>
+
+
+
<target name="invm-thirdparty-tests" depends="tests-jar, prepare-testdirs, clear-test-logs"
description="Runs all available thirdparty tests an in-VM configuration">
Added: trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/integration/MultiThreadWriteNativeTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/integration/MultiThreadWriteNativeTest.java (rev 0)
+++ trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/integration/MultiThreadWriteNativeTest.java 2008-02-27 17:15:28 UTC (rev 3824)
@@ -0,0 +1,308 @@
+/*
+ * JBoss, the OpenSource J2EE webOS
+ *
+ * Distributable under LGPL license.
+ * See terms of license at gnu.org.
+ */
+
+package org.jboss.messaging.core.asyncio.impl.test.integration;
+
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import junit.framework.TestCase;
+
+import org.jboss.messaging.core.asyncio.AIOCallback;
+import org.jboss.messaging.core.asyncio.impl.JlibAIO;
+import org.jboss.messaging.core.logging.Logger;
+
+// you need to define java.library.path=${project-root}/native/src/.libs
+public class MultiThreadWriteNativeTest extends TestCase
+{
+
+ static Logger log = Logger.getLogger(MultiThreadWriteNativeTest.class);
+
+ static AtomicInteger position = new AtomicInteger(0);
+
+ String FILE_NAME="/tmp/libaio.log";
+
+ static final int SIZE = 1024;
+ static final int NUMBER_OF_THREADS = 40;
+ static final int NUMBER_OF_LINES = 5000;
+
+// Executor exec
+
+ static Executor executor = Executors.newSingleThreadExecutor();
+
+ static Semaphore semaphore = new Semaphore(1, false);
+
+
+ static class ExecClass implements Runnable
+ {
+
+ JlibAIO aio;
+ ByteBuffer buffer;
+ AIOCallback callback;
+
+
+ public ExecClass(JlibAIO aio, ByteBuffer buffer, AIOCallback callback)
+ {
+ this.aio = aio;
+ this.buffer = buffer;
+ this.callback = callback;
+ }
+
+ public void run()
+ {
+ try
+ {
+ aio.write(getNewPosition()*SIZE, SIZE, buffer, callback);
+
+ }
+ catch (Exception e)
+ {
+ callback.onError(-1, e.toString());
+ e.printStackTrace();
+ }
+ finally
+ {
+ try { semaphore.release(); } catch (Exception ignored){}
+ }
+ }
+
+ }
+
+
+
+ private static void addData(JlibAIO aio, ByteBuffer buffer, AIOCallback callback) throws Exception
+ {
+ //aio.write(getNewPosition()*SIZE, SIZE, buffer, callback);
+ executor.execute(new ExecClass(aio, buffer, callback));
+
+ //semaphore.acquire();
+ //try
+ //{
+ //aio.write(getNewPosition()*SIZE, SIZE, buffer, callback);
+ //}
+ //finally
+ //{
+ // semaphore.release();
+ //}
+
+
+
+ }
+
+
+
+
+ protected void setUp() throws Exception
+ {
+ super.setUp();
+ File file = new File(FILE_NAME);
+ file.delete();
+
+ position.set(0);
+ }
+
+ public void testMultipleASynchronousWrites() throws Throwable
+ {
+ executeTest(false);
+ }
+
+ public void testMultipleSynchronousWrites() throws Throwable
+ {
+ executeTest(true);
+ }
+
+ private void executeTest(boolean sync) throws Throwable
+ {
+ log.info(sync?"Sync test:":"Async test");
+ JlibAIO jlibAIO = new JlibAIO();
+ jlibAIO.open(FILE_NAME, 21000);
+ log.debug("Preallocating file");
+ jlibAIO.preAllocate(NUMBER_OF_THREADS, SIZE * NUMBER_OF_LINES);
+ log.debug("Done Preallocating file");
+
+ CountDownLatch latchStart = new CountDownLatch (NUMBER_OF_THREADS + 1);
+
+ ArrayList<ThreadProducer> list = new ArrayList<ThreadProducer>(NUMBER_OF_THREADS);
+ for(int i=0;i<NUMBER_OF_THREADS;i++)
+ {
+ ThreadProducer producer = new ThreadProducer("Thread " + i, latchStart, jlibAIO, sync);
+ list.add(producer);
+ producer.start();
+ }
+
+ latchStart.countDown();
+ latchStart.await();
+
+
+ long startTime = System.currentTimeMillis();
+
+
+
+ for (ThreadProducer producer: list)
+ {
+ producer.join();
+ if (producer.failed != null)
+ {
+ throw producer.failed;
+ }
+ }
+ long endTime = System.currentTimeMillis();
+
+ log.debug((sync?"Sync result:":"Async result:") + " Records/Second = " + (NUMBER_OF_THREADS * NUMBER_OF_LINES * 1000 / (endTime - startTime)) + " total time = " + (endTime - startTime) + " total number of records = " + (NUMBER_OF_THREADS * NUMBER_OF_LINES));
+
+ jlibAIO.close();
+
+ }
+
+
+
+
+ private static int getNewPosition()
+ {
+ return position.addAndGet(1);
+ }
+
+ static class ThreadProducer extends Thread
+ {
+
+ Throwable failed = null;
+ CountDownLatch latchStart;
+ boolean sync;
+ JlibAIO libaio;
+
+ public ThreadProducer(String name, CountDownLatch latchStart, JlibAIO libaio, boolean sync)
+ {
+ super(name);
+ this.latchStart = latchStart;
+ this.libaio = libaio;
+ this.sync = sync;
+ }
+
+ public void run()
+ {
+ super.run();
+
+
+ try
+ {
+
+ ByteBuffer buffer = libaio.newBuffer(SIZE);
+
+ // I'm aways reusing the same buffer, as I don't want any noise from malloc on the measurement
+ // Encoding buffer
+ addString ("Thread name=" + Thread.currentThread().getName() + ";" + "\n", buffer);
+ for (int local = buffer.position(); local < buffer.capacity() - 1; local++)
+ {
+ buffer.put((byte)' ');
+ }
+ buffer.put((byte)'\n');
+
+
+ latchStart.countDown();
+ latchStart.await();
+
+ long startTime = System.currentTimeMillis();
+
+
+ CountDownLatch latchFinishThread = null;
+
+ if (!sync) latchFinishThread = new CountDownLatch(NUMBER_OF_LINES);
+
+ LinkedList<LocalCallback> list = new LinkedList<LocalCallback>();
+
+ for (int i=0;i<NUMBER_OF_LINES;i++)
+ {
+
+ if (sync) latchFinishThread = new CountDownLatch(1);
+ LocalCallback callback = new LocalCallback(latchFinishThread, buffer, libaio);
+ if (!sync) list.add(callback);
+ addData(libaio, buffer,callback);
+ if (sync)
+ {
+ latchFinishThread.await();
+ assertTrue(callback.doneCalled);
+ assertFalse(callback.errorCalled);
+ }
+ }
+ if (!sync) latchFinishThread.await();
+ for (LocalCallback callback: list)
+ {
+ assertTrue (callback.doneCalled);
+ assertFalse (callback.errorCalled);
+ }
+
+ long endtime = System.currentTimeMillis();
+
+ log.debug(Thread.currentThread().getName() + " Rec/Sec= " + (NUMBER_OF_LINES * 1000 / (endtime-startTime)) + " total time = " + (endtime-startTime) + " number of lines=" + NUMBER_OF_LINES);
+
+ libaio.destroyBuffer(buffer);
+
+
+ for (LocalCallback callback: list)
+ {
+ assertTrue (callback.doneCalled);
+ assertFalse (callback.errorCalled);
+ }
+
+ }
+ catch (Throwable e)
+ {
+ e.printStackTrace();
+ failed = e;
+ }
+
+ }
+ }
+
+ private static void addString(String str, ByteBuffer buffer)
+ {
+ byte bytes[] = str.getBytes();
+ //buffer.putInt(bytes.length);
+ buffer.put(bytes);
+ //CharBuffer charBuffer = CharBuffer.wrap(str);
+ //UTF_8_ENCODER.encode(charBuffer, buffer, true);
+
+ }
+
+ static class LocalCallback implements AIOCallback
+ {
+ boolean doneCalled = false;
+ boolean errorCalled = false;
+ CountDownLatch latchDone;
+ ByteBuffer releaseMe;
+ JlibAIO libaio;
+
+ public LocalCallback(CountDownLatch latchDone, ByteBuffer releaseMe, JlibAIO libaio)
+ {
+ this.latchDone = latchDone;
+ this.releaseMe = releaseMe;
+ this.libaio = libaio;
+ }
+
+ public void done()
+ {
+ doneCalled=true;
+ latchDone.countDown();
+ //libaio.destroyBuffer(releaseMe);
+ }
+
+ public void onError(int errorCode, String errorMessage)
+ {
+ errorCalled=true;
+ latchDone.countDown();
+ libaio.destroyBuffer(releaseMe);
+ }
+
+ }
+}
Property changes on: trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/integration/MultiThreadWriteNativeTest.java
___________________________________________________________________
Name: svn:keywords
+ Id LastChangedDate Author Revision
Added: trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/integration/SingleThreadWriteNativeTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/integration/SingleThreadWriteNativeTest.java (rev 0)
+++ trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/integration/SingleThreadWriteNativeTest.java 2008-02-27 17:15:28 UTC (rev 3824)
@@ -0,0 +1,409 @@
+/*
+ * JBoss, the OpenSource J2EE webOS
+ *
+ * Distributable under LGPL license.
+ * See terms of license at gnu.org.
+ */
+
+package org.jboss.messaging.core.asyncio.impl.test.integration;
+
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.nio.CharBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetEncoder;
+import java.util.ArrayList;
+import java.util.concurrent.CountDownLatch;
+
+import org.jboss.messaging.core.asyncio.AIOCallback;
+import org.jboss.messaging.core.asyncio.impl.JlibAIO;
+
+import junit.framework.TestCase;
+
+//you need to define java.library.path=${project-root}/native/src/.libs
+public class SingleThreadWriteNativeTest extends TestCase
+{
+ private static CharsetEncoder UTF_8_ENCODER = Charset.forName("UTF-8").newEncoder();
+
+
+ byte commonBuffer[] = null;
+
+ String FILE_NAME="/tmp/libaio.log";
+
+
+ @Override
+ protected void setUp() throws Exception
+ {
+ super.setUp();
+ LocalAIO.staticDone = 0;
+ File file = new File(FILE_NAME);
+ file.delete();
+ }
+
+ private void encodeBufer(ByteBuffer buffer)
+ {
+ buffer.clear();
+ int size = buffer.limit();
+ for (int i=0;i<size-1;i++)
+ {
+ buffer.put((byte)('a' + (i%20)));
+ }
+
+ buffer.put((byte)'\n');
+
+ }
+
+ public void testAddBeyongSimultaneousLimit() throws Exception
+ {
+ asyncData(150000,1024,100);
+ }
+
+ public void testAddAsyncData() throws Exception
+ {
+ asyncData(150000,1024,20000);
+ }
+
+ public void testRead() throws Exception
+ {
+
+
+
+
+ final JlibAIO controller = new JlibAIO();
+ try
+ {
+
+ final int NUMBER_LINES = 300;
+ final int SIZE = 1024;
+
+ controller.open(FILE_NAME, 10);
+
+ ByteBuffer buffer = controller.newBuffer(SIZE);
+
+
+ for (int i=0; i<NUMBER_LINES; i++)
+ {
+ buffer.clear();
+ addString ("Str value " + i + "\n", buffer);
+ for (int j=buffer.position(); j<buffer.capacity()-1;j++)
+ {
+ buffer.put((byte)' ');
+ }
+ buffer.put((byte)'\n');
+
+
+ CountDownLatch latch = new CountDownLatch(1);
+ LocalAIO aio = new LocalAIO(latch);
+ controller.write(i * SIZE, SIZE, buffer, aio);
+ latch.await();
+ assertFalse(aio.errorCalled);
+ assertTrue(aio.doneCalled);
+ }
+
+ ByteBuffer newBuffer = ByteBuffer.allocateDirect(SIZE);
+
+ for (int i=0; i<NUMBER_LINES; i++)
+ {
+ newBuffer.clear();
+ addString ("Str value " + i + "\n", newBuffer);
+ for (int j=newBuffer.position(); j<newBuffer.capacity()-1;j++)
+ {
+ newBuffer.put((byte)' ');
+ }
+ newBuffer.put((byte)'\n');
+
+
+ CountDownLatch latch = new CountDownLatch(1);
+ LocalAIO aio = new LocalAIO(latch);
+ controller.read(i * SIZE, SIZE, buffer, aio);
+ latch.await();
+ assertFalse(aio.errorCalled);
+ assertTrue(aio.doneCalled);
+
+ byte bytesRead[] = new byte[SIZE];
+ byte bytesCompare[] = new byte[SIZE];
+
+ newBuffer.rewind();
+ newBuffer.get(bytesCompare);
+ buffer.rewind();
+ buffer.get(bytesRead);
+
+ for (int count=0;count<SIZE;count++)
+ {
+ assertEquals("byte position " + count + " differs on line " + i, bytesCompare[count], bytesRead[count]);
+ }
+
+
+ //byte[] byteCompare = new byte[SIZE];
+ //byte[] byteRead = new byte[SIZE];
+
+ assertTrue(buffer.equals(newBuffer));
+ }
+
+ controller.destroyBuffer(buffer);
+ }
+ finally
+ {
+ try { controller.close(); } catch (Throwable ignored){}
+
+ }
+
+ }
+
+ private void asyncData(int numberOfLines, int size, int aioLimit) throws Exception
+ {
+ final JlibAIO controller = new JlibAIO();
+ controller.open(FILE_NAME, aioLimit);
+
+ try
+ {
+ System.out.println("++testDirectDataNoPage"); System.out.flush();
+ CountDownLatch latchDone = new CountDownLatch(numberOfLines);
+
+ ByteBuffer block = controller.newBuffer(size);
+ encodeBufer(block);
+
+ preAlloc(controller, numberOfLines * size);
+
+ ArrayList<LocalAIO> list = new ArrayList<LocalAIO>();
+
+ for (int i=0; i<numberOfLines; i++)
+ {
+ list.add(new LocalAIO(latchDone));
+ }
+
+
+ long valueInitial = System.currentTimeMillis();
+
+ System.out.println("Adding data");
+
+ long lastTime = System.currentTimeMillis();
+ int counter = 0;
+ for (LocalAIO tmp: list)
+ {
+ controller.write(counter * size, size, block, tmp);
+ if (++counter % 5000 == 0)
+ {
+ System.out.println(5000*1000/(System.currentTimeMillis()-lastTime) + " rec/sec (Async)");
+ lastTime = System.currentTimeMillis();
+ }
+
+ }
+
+ System.out.println("Data added " + (System.currentTimeMillis() - valueInitial));
+
+
+ System.out.println("Finished append " + (System.currentTimeMillis() - valueInitial) + " received = " + LocalAIO.staticDone);
+ System.out.println("Flush now");
+ System.out.println("Received " + LocalAIO.staticDone);
+ long timeTotal = System.currentTimeMillis() - valueInitial;
+
+ System.out.println("Asynchronous time = " + timeTotal + " for " + numberOfLines + " registers " + " size each line = " + size + " Records/Sec=" + (numberOfLines*1000/timeTotal) + " (Assynchronous)");
+
+ latchDone.await();
+
+ timeTotal = System.currentTimeMillis() - valueInitial;
+ System.out.println("After completions time = " + timeTotal + " for " + numberOfLines + " registers " + " size each line = " + size + " Records/Sec=" + (numberOfLines*1000/timeTotal) + " (Assynchronous)");
+
+ for (LocalAIO tmp: list)
+ {
+ assertEquals(1, tmp.timesDoneCalled);
+ assertTrue(tmp.doneCalled);
+ assertFalse(tmp.errorCalled);
+ }
+
+ controller.destroyBuffer(block);
+
+ controller.close();
+ }
+ finally
+ {
+ try {controller.close();} catch (Exception ignored){}
+ }
+
+
+ }
+
+ public void testDirectSynchronous() throws Exception
+ {
+ try
+ {
+ System.out.println("++testDirectDataNoPage"); System.out.flush();
+ final int NUMBER_LINES = 10000;
+ final int SIZE = 1024;
+ //final int SIZE = 512;
+
+ final JlibAIO controller = new JlibAIO();
+ controller.open(FILE_NAME, 2000);
+
+ ByteBuffer block = controller.newBuffer(SIZE);
+ encodeBufer(block);
+
+ preAlloc(controller, NUMBER_LINES * SIZE);
+
+ long valueInitial = System.currentTimeMillis();
+
+ System.out.println("Adding data");
+
+ long lastTime = System.currentTimeMillis();
+ int counter = 0;
+
+ for (int i=0; i<NUMBER_LINES; i++)
+ {
+ CountDownLatch latchDone = new CountDownLatch(1);
+ LocalAIO aioBlock = new LocalAIO(latchDone);
+ controller.write(i*512, 512, block, aioBlock);
+ latchDone.await();
+ assertTrue(aioBlock.doneCalled);
+ assertFalse(aioBlock.errorCalled);
+ if (++counter % 500 == 0)
+ {
+ System.out.println(500*1000/(System.currentTimeMillis()-lastTime) + " rec/sec (Synchronous)");
+ lastTime = System.currentTimeMillis();
+ }
+ }
+
+ System.out.println("Data added " + (System.currentTimeMillis() - valueInitial));
+
+
+ System.out.println("Finished append " + (System.currentTimeMillis() - valueInitial) + " received = " + LocalAIO.staticDone);
+ System.out.println("Flush now");
+ System.out.println("Received " + LocalAIO.staticDone);
+
+ long timeTotal = System.currentTimeMillis() - valueInitial;
+ System.out.println("Flushed " + timeTotal);
+ System.out.println("time = " + timeTotal + " for " + NUMBER_LINES + " registers " + " size each line = " + SIZE + " Records/Sec=" + (NUMBER_LINES*1000/timeTotal) + " Synchronous");
+
+ controller.destroyBuffer(block);
+ controller.close();
+ }
+ catch (Exception e)
+ {
+ System.out.println("Received " + LocalAIO.staticDone + " before it failed");
+ throw e;
+ }
+
+ }
+
+ private void preAlloc(JlibAIO controller, long size)
+ {
+ System.out.println("Pre allocating"); System.out.flush();
+ long startPreAllocate = System.currentTimeMillis();
+ controller.preAllocate(1, size);
+ long endPreAllocate = System.currentTimeMillis() - startPreAllocate;
+ if (endPreAllocate != 0) System.out.println("PreAllocated the file in " + endPreAllocate + " seconds, What means " + (size/endPreAllocate) + " bytes per millisecond");
+ }
+
+
+ public void testInvalidWrite() throws Exception
+ {
+ final JlibAIO controller = new JlibAIO();
+ controller.open(FILE_NAME, 2000);
+
+ try
+ {
+
+ final int SIZE=512;
+
+ ByteBuffer block = controller.newBuffer(SIZE);
+ encodeBufer(block);
+
+ preAlloc(controller, 1000 * 512);
+
+
+ CountDownLatch latchDone = new CountDownLatch(1);
+
+ LocalAIO aioBlock = new LocalAIO(latchDone);
+ controller.write(11, 512, block, aioBlock);
+
+ latchDone.await();
+
+ assertTrue (aioBlock.errorCalled);
+ assertFalse(aioBlock.doneCalled);
+
+ controller.destroyBuffer(block);
+ }
+ catch (Exception e)
+ {
+ System.out.println("Received " + LocalAIO.staticDone + " before it failed");
+ throw e;
+ }
+ finally
+ {
+ controller.close();
+ }
+
+ }
+
+ public void testInvalidAlloc() throws Exception
+ {
+ JlibAIO controller = new JlibAIO();
+ try
+ {
+ // You don't need to open the file to alloc it
+ ByteBuffer buffer = controller.newBuffer(300);
+ fail ("Exception expected");
+ }
+ catch (Exception ignored)
+ {
+ }
+
+ }
+
+ private static class LocalAIO implements AIOCallback
+ {
+
+ CountDownLatch latch;
+
+ public LocalAIO(CountDownLatch latch)
+ {
+ this.latch = latch;
+ }
+
+ boolean doneCalled = false;
+ boolean errorCalled = false;
+ int timesDoneCalled = 0;
+ static int staticDone = 0;
+ public void decode(int length, ByteBuffer buffer)
+ {
+ // TODO Auto-generated method stub
+
+ }
+
+ public void done()
+ {
+ //System.out.println("Received Done"); System.out.flush();
+ doneCalled = true;
+ timesDoneCalled++;
+ staticDone++;
+ if (latch != null)
+ {
+ latch.countDown();
+ }
+
+ }
+
+ public void onError(int errorCode, String errorMessage)
+ {
+ errorCalled = true;
+ if (latch != null)
+ {
+ // even thought an error happened, we need to inform the latch, or the test won't finish
+ latch.countDown();
+ }
+ System.out.println("Received an Error - " + errorCode + " message=" + errorMessage);
+
+ }
+
+ }
+
+ private void addString(String str, ByteBuffer buffer)
+ {
+ CharBuffer charBuffer = CharBuffer.wrap(str);
+ UTF_8_ENCODER.encode(charBuffer, buffer, true);
+
+ }
+
+
+
+}
Property changes on: trunk/tests/src/org/jboss/messaging/core/asyncio/impl/test/integration/SingleThreadWriteNativeTest.java
___________________________________________________________________
Name: svn:keywords
+ Id LastChangedDate Author Revision
More information about the jboss-cvs-commits
mailing list