Null element in the Vector

[roman.mandeleil1]

I am chasing after a strange issue in pojo cache:
There is a Vector that I am using it as a queue
one Thread is pushing things into the tail and
the other poping the head of the Vector.

The Vector is being aspectize with the following
configuration (pojocache-aop.xml):

<?xml version="1.0" encoding="UTF-8"?>
<!--
 This is a variant of jboss-aop.xml.
-->
<aop>
 <!-- If a POJO has a Replicable annotation, it will be asepectized. -->

 <!--
 Supports inheritance and polymorphism. It can either be a concrete class
 or an interface. All sub-classes or interface implementors will be

instrumeneted.
 -->
 <prepare expr="field(* $instanceof

{@org.jboss.cache.pojo.annotation.Replicable}->*)" />

 <!-- Work around that ensures annotated classes which do not access

fields are instrumented -->
 <introduction expr="class($instanceof

{@org.jboss.cache.pojo.annotation.Replicable})"/>

 <!-- Array support -->
 <!-- Comment entire section to disable -->
 <arrayreplacement expr="class($instanceof

{@org.jboss.cache.pojo.annotation.Replicable})"/>
 <interceptor name="pojocache-array"

class="org.jboss.cache.pojo.interceptors.dynamic.ArrayInterceptor"/>
 <introduction expr="class($instanceof

{@org.jboss.cache.pojo.annotation.Replicable})">
 <interfaces>org.jboss.cache.pojo.impl.ArrayInterceptable</interfaces>
 </introduction>
 <arraybind name="pojocache-array" type="READ_WRITE">
 <interceptor-ref name="pojocache-array"/>
 </arraybind>

</aop>

I have tried different cache configurations, the last one is like this:

Configuration config = new Configuration();
 config.setTransactionManagerLookupClass( JBossTransactionManagerLookup.class.getName() );
 config.setIsolationLevel(IsolationLevel.SERIALIZABLE);
 config.setCacheMode(CacheMode.REPL_SYNC);
 config.setLockAcquisitionTimeout(30000);
 config.setClusterName("DefaultPartition");
 config.setConcurrencyLevel(5000);
 config.setNodeLockingScheme(NodeLockingScheme.PESSIMISTIC);

The vector is placed inside the pojo cache on JBoss 5.1.0 GA.
And then I see the following problem happens from time to time:
I see null elements apears inside the vector when I have cirtanly
no code that puts it there and by inspecting the logs I see
that it happens when there is concurrent activity on the Vector.

Before I am going to open a bug on it, I would like to know if
there is someone else that has similar expirience and maybe know
about any work around to prevent this behavior.


Regards
Roman

[roman.mandeleil1]

Here is the output of simple concurrency test

Thread: [44] pushing
Thread: [45] poping
Thread: [45]
[null],
Thread: [45] poping
Thread: [44]
[23],
Thread: [44] pushing
Thread: [44]
[23], [23],
Thread: [45]
[null],
Thread: [44] pushing
Thread: [44]
[23], [23], [23],
Thread: [44] pushing
Thread: [44]
[23], [23], [23], [23],
Thread: [44] pushing
Thread: [44]
[23], [23], [23], [23], [23],
Thread: [44] pushing
Thread: [44]
[23], [23], [23], [23], [23], [23],
Thread: [45] poping
Thread: [44] pushing
Thread: [45]
[23], [23], [23], [23], [null],
Thread: [45] poping
Thread: [45]
[23], [23], [23], [null], [null], [23],
Thread: [45] poping
Thread: [44]
[null], [null], [null], [null], [null], [23], [23],
Thread: [44] pushing
Thread: [45]
[23], [23], [null], [null], [23], [null],
Thread: [45] poping
Thread: [45]
[23], [null], [null], [23], [null], [null], [23],

Here is the code of the test:

package org.galtstreet.cache.test;

import java.util.Properties;

import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.resource.NotSupportedException;
import javax.transaction.SystemException;
import javax.transaction.UserTransaction;

import junit.framework.TestCase;

import org.galtstreet.exchangeboard.trying.WrappedQueue;
import org.jboss.cache.config.Configuration.CacheMode;
import org.jboss.cache.factories.UnitTestConfigurationFactory;
import org.jboss.cache.lock.IsolationLevel;
import org.jboss.cache.pojo.PojoCache;
import org.jboss.cache.pojo.PojoCacheFactory;


public class SimpleConcurrencyTest extends TestCase{



 PojoCache cache1;

 protected void setUp() throws Exception {

 Properties prop = new Properties();
 prop.put(Context.INITIAL_CONTEXT_FACTORY, "org.jboss.cache.transaction.DummyContextFactory");
 boolean toStart = false;



 cache1 = PojoCacheFactory.createCache(UnitTestConfigurationFactory.createConfiguration(CacheMode.REPL_SYNC), toStart);
 cache1.getCache().getConfiguration().setSyncCommitPhase(true);

 cache1.getCache().getConfiguration().setLockParentForChildInsertRemove(true);
 cache1.getCache().getConfiguration().setIsolationLevel(IsolationLevel.SERIALIZABLE);
 cache1.start();
 }


 UserTransaction getTransaction() throws SystemException, NotSupportedException, NamingException
 {
 Properties prop = new Properties();
 prop.put(Context.INITIAL_CONTEXT_FACTORY,
 "org.jboss.cache.transaction.DummyContextFactory");
 return (UserTransaction) new InitialContext(prop).lookup("UserTransaction");
 }


 public void testWrappedQueue(){

 WrappedQueue queue = new WrappedQueue();

 cache1.attach("queue-1", queue);


 ProducerThread pt = new ProducerThread(queue);
 pt.start();


 ConsumerThread ct = new ConsumerThread(queue);
 ct.start();

 sleeping(120000);

 }



 class ConsumerThread extends Thread{

 WrappedQueue queue = null;

 public ConsumerThread(WrappedQueue queue) {
 this.queue = queue;
 }

 @Override
 public void run() {

 while (true){

 UserTransaction ut = null;

 try {
 ut = getTransaction();
 ut.begin();
 }catch (Throwable th){
 th.printStackTrace();
 }

 System.out.println("Thread: [" + this.getId() + "] poping ");
 queue.popInteger();

 System.out.println("Thread: [" + this.getId() + "] \n" + queue.toString() );
 sleeping(500);


 try {
 ut.commit();
 }catch (Throwable th){
 th.printStackTrace();
 }
 }
 }
 }

 class ProducerThread extends Thread{

 WrappedQueue queue = null;

 public ProducerThread(WrappedQueue queue) {
 this.queue = queue;
 }

 @Override
 public void run() {

 while (true){

 UserTransaction ut = null;

 try {
 ut = getTransaction();
 ut.begin();
 }catch (Throwable th){
 th.printStackTrace();
 }

 System.out.println("Thread: [" + this.getId() + "] pushing ");
 queue.pushInteger(23);
 System.out.println("Thread: [" + this.getId() + "] \n" + queue.toString() );
 sleeping(100);


 try {
 ut.commit();
 }catch (Throwable th){
 th.printStackTrace();
 }

 }
 }
 }


 private void sleeping(int millisec){
 try {
 Thread.sleep(millisec);
 } catch (InterruptedException e) {
 // TODO Auto-generated catch block
 e.printStackTrace();
 }
 }


 protected void tearDown() throws Exception {
 cache1.stop();
 cache2.stop();

 }



}

package org.galtstreet.exchangeboard.trying;

import java.util.ArrayList;

import org.jboss.cache.pojo.annotation.Replicable;


@Replicable
public class WrappedQueue {



 private ArrayList<Integer> queueData = new ArrayList<Integer>();


 public void pushInteger(Integer tmp){
 queueData.add(queueData.size(), tmp);
 }

 public Integer popInteger(){
 return queueData.remove(0);
 }

 public boolean isEmpty(){
 return queueData.isEmpty();
 }

 public String toString(){
 return queueData.toString();
 }
}