[jboss-cvs] JBoss Messaging SVN: r3793 - projects/jaio/trunk/jaio/java/tests/org/jboss/jaio/libaioimpl/test.

[jboss-cvs-commits at lists.jboss.org]

Author: clebert.suconic at jboss.com
Date: 2008-02-25 12:38:14 -0500 (Mon, 25 Feb 2008)
New Revision: 3793

Modified:
   projects/jaio/trunk/jaio/java/tests/org/jboss/jaio/libaioimpl/test/MultiThreadTests.java
Log:
Adding Semaphore to the test

Modified: projects/jaio/trunk/jaio/java/tests/org/jboss/jaio/libaioimpl/test/MultiThreadTests.java
===================================================================
--- projects/jaio/trunk/jaio/java/tests/org/jboss/jaio/libaioimpl/test/MultiThreadTests.java	2008-02-25 16:36:48 UTC (rev 3792)
+++ projects/jaio/trunk/jaio/java/tests/org/jboss/jaio/libaioimpl/test/MultiThreadTests.java	2008-02-25 17:38:14 UTC (rev 3793)
@@ -6,9 +6,11 @@
 import java.util.ArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.jboss.jaio.api.AIOCallback;
+import org.jboss.jaio.api.logging.*;
 import org.jboss.jaio.libaioimpl.JLibAIO;
 
 
@@ -18,17 +20,21 @@
 public class MultiThreadTests extends TestCase
 {
 
+    static Logger log = LogFactory.getFactory().createLogger(MultiThreadTests.class.getName());
+    
     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 = 10000;
+    static final int NUMBER_OF_LINES = 1000;
     
 //    Executor exec
     
     static ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
+    
+    static Semaphore semaphore = new Semaphore(1, false);
 
     
     static class ExecClass implements Runnable
@@ -58,16 +64,33 @@
                 callback.onError(-1, e.toString());
                 e.printStackTrace();
             }
+            finally
+            {
+                try { semaphore.release(); } catch (Exception ignored){}
+            }
         }
         
     }
 
     
     
-    private synchronized static void addData(JLibAIO aio, ByteBuffer buffer, AIOCallback callback)
+    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));
+        //executor.execute(new ExecClass(aio, buffer, callback));
+        
+        semaphore.acquire();
+        try
+        {
+            aio.write(getNewPosition()*SIZE, SIZE, buffer, callback);
+        }
+        finally
+        {
+            semaphore.release();
+        }
+        
+        
+        
     }
     
     
@@ -94,11 +117,12 @@
     
     private void executeTest(boolean sync) throws Throwable
     {
+        log.info(sync?"Sync test:":"Async test");
         JLibAIO jlibAIO = new JLibAIO();
         jlibAIO.open(FILE_NAME, 21000);
-        System.out.println("Preallocating file");
+        log.debug("Preallocating file");
         jlibAIO.preAllocate(1, NUMBER_OF_THREADS * SIZE * NUMBER_OF_LINES);
-        System.out.println("Done Preallocating file");
+        log.debug("Done Preallocating file");
         
         CountDownLatch latchStart = new CountDownLatch (NUMBER_OF_THREADS + 1);
         CountDownLatch latchEnd = new CountDownLatch (NUMBER_OF_THREADS);
@@ -116,13 +140,13 @@
         
         long startTime = System.currentTimeMillis();
         
-        System.out.println("Every started");
+        log.debug("Every started");
         
         latchEnd.await();
         
         long endTime = System.currentTimeMillis();
         
-        System.out.println("Records/Second = " + (NUMBER_OF_THREADS * NUMBER_OF_LINES * 1000 / (endTime - startTime)));
+        log.debug((sync?"Sync result:":"Async result:") + " Records/Second = " + (NUMBER_OF_THREADS * NUMBER_OF_LINES * 1000 / (endTime - startTime)));
 
         
         for (ThreadProducer producer: list)
@@ -197,7 +221,7 @@
                 {
                     ByteBuffer buffer = libaio.newBuffer(SIZE);
 
-                    //System.out.println("Thread " + Thread.currentThread().getName() + " Adding a line");
+                    //log.debug("Thread " + Thread.currentThread().getName() + " Adding a line");
                     addString ("Thread name=" + Thread.currentThread().getName() + ";" + i + "\n", buffer);
                     for (int local = buffer.position(); local < buffer.capacity() - 1; local++)
                     {
@@ -210,14 +234,19 @@
                     list.add(callback);
                     addData(libaio, buffer,callback);
                     //Thread.yield();
-                    if (sync) latchFinishThread.await();
+                    if (sync)
+                    {
+                        //log.debug("Waiting to write");
+                        latchFinishThread.await();
+                        //log.debug("done");
+                    }
                     //libaio.write(getNewPosition() * SIZE, SIZE, buffer, callback);
                 }
                 if (!sync) latchFinishThread.await();
                 
                 long endtime = System.currentTimeMillis();
                 
-                System.out.println(Thread.currentThread().getName() + " Rec/Sec= " + (NUMBER_OF_LINES * 1000 / (endtime-startTime)));
+                log.debug(Thread.currentThread().getName() + " Rec/Sec= " + (NUMBER_OF_LINES * 1000 / (endtime-startTime)));
                 
                 latchEnd.countDown();