[hornetq-commits] JBoss hornetq SVN: r9464 - in trunk/tests/src/org/hornetq/tests: soak/journal and 1 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Mon Jul 26 01:03:59 EDT 2010
Author: clebert.suconic at jboss.com
Date: 2010-07-26 01:03:59 -0400 (Mon, 26 Jul 2010)
New Revision: 9464
Added:
trunk/tests/src/org/hornetq/tests/soak/journal/
trunk/tests/src/org/hornetq/tests/soak/journal/SoakJournal.java
Modified:
trunk/tests/src/org/hornetq/tests/util/RandomUtil.java
Log:
Adding a soak test for the journal
Added: trunk/tests/src/org/hornetq/tests/soak/journal/SoakJournal.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/soak/journal/SoakJournal.java (rev 0)
+++ trunk/tests/src/org/hornetq/tests/soak/journal/SoakJournal.java 2010-07-26 05:03:59 UTC (rev 9464)
@@ -0,0 +1,339 @@
+/*
+ * 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.soak.journal;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import org.hornetq.core.config.impl.FileConfiguration;
+import org.hornetq.core.journal.IOAsyncTask;
+import org.hornetq.core.journal.impl.AIOSequentialFileFactory;
+import org.hornetq.core.journal.impl.JournalImpl;
+import org.hornetq.core.persistence.impl.journal.OperationContextImpl;
+import org.hornetq.tests.util.RandomUtil;
+import org.hornetq.tests.util.ServiceTestBase;
+import org.hornetq.utils.HornetQThreadFactory;
+import org.hornetq.utils.OrderedExecutorFactory;
+import org.hornetq.utils.SimpleIDGenerator;
+import org.hornetq.utils.concurrent.LinkedBlockingDeque;
+
+/**
+ * A SoakJournal
+ *
+ * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
+ *
+ *
+ */
+public class SoakJournal extends ServiceTestBase
+{
+
+ public static SimpleIDGenerator idGen = new SimpleIDGenerator(1);
+
+ private volatile boolean running;
+
+ private JournalImpl journal;
+
+ ThreadFactory tFactory = new HornetQThreadFactory("SoakTest" + System.identityHashCode(this),
+ false,
+ SoakJournal.class.getClassLoader());
+
+ private ExecutorService threadPool = Executors.newFixedThreadPool(20, tFactory);
+
+ OrderedExecutorFactory executorFactory = new OrderedExecutorFactory(threadPool);
+
+ public void setUp() throws Exception
+ {
+ super.setUp();
+
+ File dir = new File(getTemporaryDir());
+ dir.mkdirs();
+
+ FileConfiguration fileConf = new FileConfiguration();
+
+ fileConf.setJournalDirectory(getTemporaryDir());
+
+ fileConf.setCreateJournalDir(true);
+
+ fileConf.setCreateBindingsDir(true);
+
+ fileConf.start();
+
+ fileConf.setJournalMinFiles(10);
+
+ journal = new JournalImpl(fileConf.getJournalFileSize(),
+ fileConf.getJournalMinFiles(),
+ fileConf.getJournalCompactMinFiles(),
+ fileConf.getJournalCompactPercentage(),
+ new AIOSequentialFileFactory(fileConf.getJournalDirectory()),
+ "hornetq-data",
+ "hq",
+ fileConf.getJournalMaxIO_NIO());
+
+ journal.start();
+ journal.loadInternalOnly();
+
+ }
+
+ public void tearDown() throws Exception
+ {
+ journal.stop();
+ }
+
+ public void testAppend() throws Exception
+ {
+ running = true;
+ SlowAppenderNoTX t1 = new SlowAppenderNoTX();
+
+ int NTHREADS = 5;
+
+ FastAppenderTx appenders[] = new FastAppenderTx[NTHREADS];
+ FastUpdateTx updaters[] = new FastUpdateTx[NTHREADS];
+
+ for (int i = 0; i < NTHREADS; i++)
+ {
+ appenders[i] = new FastAppenderTx();
+ updaters[i] = new FastUpdateTx(appenders[i].queue);
+ }
+
+ t1.start();
+
+ for (int i = 0; i < NTHREADS; i++)
+ {
+ appenders[i].start();
+ updaters[i].start();
+ }
+
+ Thread.sleep(3600000000l);
+
+ running = false;
+
+ for (Thread t : appenders)
+ {
+ t.join();
+ }
+
+ for (Thread t : updaters)
+ {
+ t.join();
+ }
+
+ t1.join();
+ }
+
+ private byte[] generateRecord()
+ {
+ int size = RandomUtil.randomPositiveInt() % 10000;
+ if (size == 0)
+ {
+ size = 10000;
+ }
+ return RandomUtil.randomBytes(size);
+ }
+
+ class FastAppenderTx extends Thread
+ {
+ LinkedBlockingDeque<Long> queue = new LinkedBlockingDeque<Long>();
+
+ OperationContextImpl ctx = new OperationContextImpl(executorFactory.getExecutor());
+
+ public void run()
+ {
+ try
+ {
+
+ while (running)
+ {
+ int txSize = RandomUtil.randomMax(1000);
+
+ long txID = idGen.generateID();
+
+ final ArrayList<Long> ids = new ArrayList<Long>();
+
+ for (int i = 0; i < txSize; i++)
+ {
+ long id = idGen.generateID();
+ ids.add(id);
+ journal.appendAddRecordTransactional(txID, id, (byte)0, generateRecord());
+ Thread.sleep(1);
+ }
+ journal.appendCommitRecord(txID, true, ctx);
+ ctx.executeOnCompletion(new IOAsyncTask()
+ {
+
+ public void onError(int errorCode, String errorMessage)
+ {
+ }
+
+ public void done()
+ {
+ for (Long id : ids)
+ {
+ queue.add(id);
+ }
+ }
+ });
+ }
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ System.exit(-1);
+ }
+ }
+ }
+
+ class FastUpdateTx extends Thread
+ {
+ final LinkedBlockingDeque<Long> queue;
+
+ OperationContextImpl ctx = new OperationContextImpl(executorFactory.getExecutor());
+
+ public FastUpdateTx(LinkedBlockingDeque<Long> queue)
+ {
+ this.queue = queue;
+ }
+
+ public void run()
+ {
+ try
+ {
+ int txSize = RandomUtil.randomMax(1000);
+ int txCount = 0;
+ long ids[] = new long[txSize];
+
+ long txID = idGen.generateID();
+
+ while (running)
+ {
+
+ long id = queue.poll(60, TimeUnit.MINUTES);
+ ids[txCount] = id;
+ journal.appendUpdateRecordTransactional(txID, id, (byte)0, generateRecord());
+ Thread.sleep(1);
+ if (++txCount == txSize)
+ {
+ journal.appendCommitRecord(txID, true, ctx);
+ ctx.executeOnCompletion(new DeleteTask(ids));
+ txCount = 0;
+ txSize = RandomUtil.randomMax(1000);
+ txID = idGen.generateID();
+ ids = new long[txSize];
+ }
+ }
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ System.exit(-1);
+ }
+ }
+ }
+
+ class DeleteTask implements IOAsyncTask
+ {
+ final long ids[];
+
+ DeleteTask(long ids[])
+ {
+ this.ids = ids;
+ }
+
+ public void done()
+ {
+ try
+ {
+ for (int i = 0; i < ids.length; i++)
+ {
+ journal.appendDeleteRecord(ids[i], false);
+ }
+ }
+ catch (Exception e)
+ {
+ System.err.println("Can't delete id");
+ e.printStackTrace();
+ }
+ }
+
+ public void onError(int errorCode, String errorMessage)
+ {
+ }
+
+ }
+
+ /** Adds stuff to the journal, but it will take a long time to remove them.
+ * This will cause cleanup and compacting to happen more often
+ */
+ class SlowAppenderNoTX extends Thread
+ {
+ public void run()
+ {
+ try
+ {
+ while (running)
+ {
+ long ids[] = new long[1000];
+ // Append
+ for (int i = 0; running & i < 1000; i++)
+ {
+ ids[i] = idGen.generateID();
+ journal.appendAddRecord(ids[i], (byte)1, generateRecord(), true);
+ Thread.sleep(300);
+ }
+ // Update
+ for (int i = 0; running & i < 1000; i++)
+ {
+ ids[i] = idGen.generateID();
+ journal.appendUpdateRecord(ids[i], (byte)1, generateRecord(), true);
+ Thread.sleep(300);
+ }
+ // Delete
+ for (int i = 0; running & i < 1000; i++)
+ {
+ ids[i] = idGen.generateID();
+ journal.appendUpdateRecord(ids[i], (byte)1, generateRecord(), true);
+ Thread.sleep(300);
+ }
+ }
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ System.exit(-1);
+ }
+ }
+ }
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ // Public --------------------------------------------------------
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
Modified: trunk/tests/src/org/hornetq/tests/util/RandomUtil.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/util/RandomUtil.java 2010-07-23 08:13:58 UTC (rev 9463)
+++ trunk/tests/src/org/hornetq/tests/util/RandomUtil.java 2010-07-26 05:03:59 UTC (rev 9464)
@@ -71,6 +71,18 @@
{
return Math.abs(RandomUtil.randomInt());
}
+
+ public static int randomMax(int max)
+ {
+ int value = randomPositiveInt() % max;
+
+ if (value == 0)
+ {
+ value = max;
+ }
+
+ return value;
+ }
public static int randomPort()
{
More information about the hornetq-commits
mailing list