Losing objects in a multithreaded environment

[rankorn]

I am using a TreeCache in a multithreaded environment where 3 JVMs put objects into the cache at the same time (something like 10000 objects each during a 30 seconds time period into the same node).

The problem is that I am losing objects, meaning that when I check the number of objects at the end of that period I get a number less then 30000.

I have tried all kind of configurations - SYNC / ASYNC, with/withput a queue, with different sizes and intervals but still I get the same problem.

Is that something that have to do with the isolation level? what about the transaction manager?

Any help is welcome!
Ran.

[ben.wang]

Ran,

If you are willing to provide your test code, I will be happy to take a look at the problem you mentioned.

Thanks,

-Ben

[rankorn]

Here is my test Java code:

import java.io.*;
import javax.transaction.*;

import org.jboss.cache.*;

public class MultithreadedClient {

 private int numOfThreads;
 private int time;
 private int randomNum;
 private TreeCache cache;

 public MultithreadedClient(int numOfThreads, int time) {
 this.numOfThreads = numOfThreads;
 this.time = time;
 randomNum = (int)(Math.random()*10000);

 initCache();
 Thread t = null;
 int wait = (int)(1000f*time/numOfThreads);
 for (int i = 0; i < numOfThreads; i++) {
 t = new Thread(new CacheThread());
 t.start();
 try {
 t.join();

 Thread.sleep(wait);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }

 waitForKey();

 int total = 0;
 try {
 Node node = cache.get("/aNode");
 total = node.getDataKeys().size();
 } catch (Exception e) {
 e.printStackTrace();
 }

 System.out.println("Finished. Total num of objects: "+total);
 }

 public static void main(String[] args) {
 MultithreadedClient client = new MultithreadedClient(10000, 30);
 }

 private void initCache() {
 try {
 cache = new TreeCache();
 PropertyConfigurator config = new PropertyConfigurator();
 config.configure(cache, "./replSync-service.xml");
 cache.start();

 System.out.println("Cache initialized.");
 } catch (Exception e) {
 e.printStackTrace();
 }

 }

 private static void waitForKey() {
 System.out.println("Press enter to continue.");
 try {
 System.in.read();
 } catch (IOException e) {
 e.printStackTrace();
 }

 }
 private class CacheThread implements Runnable {

 public void run() {
 String tName = Thread.currentThread().getName();
 UserTransaction tx=null;
 try {
 cache.put("/aNode/"+randomNum, tName+"-"+randomNum, new int[] { 1, 2, 3 });
 } catch (Exception e) {
 try {
 tx.rollback();
 } catch (Exception e1) {}
 e.printStackTrace();
 }

 }

 }
}

And this is my cache configuration file (even though I have tried it with all kinds of configurations):

<?xml version="1.0" encoding="UTF-8"?>

<!-- ===================================================================== -->
<!-- -->
<!-- Sample TreeCache Service Configuration -->
<!-- -->
<!-- ===================================================================== -->

<server>

 <classpath codebase="./lib" archives="jboss-cache.jar, jgroups.jar"/>


 <!-- ==================================================================== -->
 <!-- Defines TreeCache configuration -->
 <!-- ==================================================================== -->

 <mbean code="org.jboss.cache.TreeCache"
 name="jboss.cache:service=TreeCache">

 <depends>jboss:service=Naming</depends>
 <depends>jboss:service=TransactionManager</depends>

 <!--
 Configure the TransactionManager
 -->
 <attribute name="TransactionManagerLookupClass">org.jboss.cache.DummyTransactionManagerLookup</attribute>

 <!--
 Isolation level : SERIALIZABLE
 REPEATABLE_READ (default)
 READ_COMMITTED
 READ_UNCOMMITTED
 NONE
 -->
 <attribute name="IsolationLevel">REPEATABLE_READ</attribute>

 <!--
 Valid modes are LOCAL, REPL_SYNC and REPL_SYNC
 -->
 <attribute name="CacheMode">REPL_SYNC</attribute>

 <!--
 Just used for async repl: use a replication queue
 -->
 <attribute name="UseReplQueue">false</attribute>

 <!--
 Replication interval for replication queue (in ms)
 -->
 <attribute name="ReplQueueInterval">30000</attribute>

 <!--
 Max number of elements which trigger replication
 -->
 <attribute name="ReplQueueMaxElements">100000</attribute>

 <!-- Name of cluster. Needs to be the same for all clusters, in order
 to find each other
 -->
 <attribute name="ClusterName">TreeCache-Cluster</attribute>

 <!-- JGroups protocol stack properties. Can also be a URL,
 e.g. file:/home/bela/default.xml
 <attribute name="ClusterProperties"></attribute>
 -->

 <attribute name="ClusterConfig">
 <config>
 <!-- UDP: if you have a multihomed machine,
 set the bind_addr attribute to the appropriate NIC IP address, e.g bind_addr="192.168.0.2"
 -->
 <!-- UDP: On Windows machines, because of the media sense feature
 being broken with multicast (even after disabling media sense)
 set the loopback attribute to true -->
 <UDP mcast_addr="228.1.2.3" mcast_port="45566"
 ip_ttl="64" ip_mcast="true"
 mcast_send_buf_size="150000" mcast_recv_buf_size="80000"
 ucast_send_buf_size="150000" ucast_recv_buf_size="80000"
 loopback="false"/>
 <PING timeout="2000" num_initial_members="3"
 up_thread="false" down_thread="false"/>
 <MERGE2 min_interval="10000" max_interval="20000"/>
 <!-- <FD shun="true" up_thread="true" down_thread="true" />-->
 <FD_SOCK/>
 <VERIFY_SUSPECT timeout="1500"
 up_thread="false" down_thread="false"/>
 <pbcast.NAKACK gc_lag="50" retransmit_timeout="600,1200,2400,4800"
 max_xmit_size="8192" up_thread="false" down_thread="false"/>
 <UNICAST timeout="600,1200,2400" window_size="100" min_threshold="10"
 down_thread="false"/>
 <pbcast.STABLE desired_avg_gossip="20000"
 up_thread="false" down_thread="false"/>
 <FRAG frag_size="8192"
 down_thread="false" up_thread="false"/>
 <pbcast.GMS join_timeout="5000" join_retry_timeout="2000"
 shun="true" print_local_addr="true"/>
 <pbcast.STATE_TRANSFER up_thread="true" down_thread="true"/>
 </config>
 </attribute>

 <!--
 Max number of entries in the cache. If this is exceeded, the
 eviction policy will kick some entries out in order to make
 more room
 -->
 <attribute name="MaxCapacity">100000</attribute>

 <!--
 Whether or not to fetch state on joining a cluster
 -->
 <attribute name="FetchStateOnStartup">true</attribute>

 <!--
 The max amount of time (in milliseconds) we wait until the
 initial state (ie. the contents of the cache) are retrieved from
 existing members in a clustered environment
 -->
 <attribute name="InitialStateRetrievalTimeout">5000</attribute>

 <!--
 Number of milliseconds to wait until all responses for a
 synchronous call have been received.
 -->
 <attribute name="SyncReplTimeout">10000</attribute>

 <!-- Max number of milliseconds to wait for a lock acquisition -->
 <attribute name="LockAcquisitionTimeout">15000</attribute>

 <!-- Max number of milliseconds we hold a lock (not currently
 implemented) -->
 <attribute name="LockLeaseTimeout">60000</attribute>

 <!-- Name of the eviction policy class. Not supported now. -->
 <attribute name="EvictionPolicyClass"></attribute>
 </mbean>


 <!-- Uncomment to get a graphical view of the TreeCache MBean above -->
 <!-- <mbean code="org.jboss.cache.TreeCacheView" name="jboss.cache:service=TreeCacheView">-->
 <!-- <depends>jboss.cache:service=TreeCache</depends>-->
 <!-- <attribute name="CacheService">jboss.cache:service=TreeCache</attribute>-->
 <!-- </mbean>-->


</server>

When I run this from 3 different JVMs on the same machine, I get that the total number of objects is always less than 30000 (depends on the config file...)

Thanks
Ran.

[ben.wang]

Ran,

I haven't had time to run it yet. But let me get back to you later this week.

Thanks,

-Ben

[ben.wang]

Ran,

Can you do me a favor by re-writing your test code into Junit test? You can put the package name as: org.jboss.test.cache.test.replicated.

I can then incorporate this test into test suite as well.

Thanks,

-Ben