peter-lawrey
commented
10 years ago Once you have a reference to an off heap entry it works much the same as if you have an on heap object. The entry may have come from a thread safe collection but that doesn't make the entry also inherently thread safe. To make your code thread safe you need to add entry/record level locking. See IntValue or JavaBeanInterface for examples. One you have added a lock field, you need to busyLock () before accessing it and finally unlock () it. On 11/09/2014 8:04 PM, "andycrutchlow" notifications@github.com wrote:
When concurrent threads reading/writing to DIFFERENT map entries through standard DataValueClass generated getter/setters (marshalling to corresponding UNSAFE methods) and if these map entries phyically reside in the SAME SEGMENT then there is possibility of corruption of data. Problem seems since version 3.0.2 collections i.e. I cannot reliably reproduce on that version..but I can pretty consistantly reproduce on latest 3.2.1 collections and 6.4.6 lang on two completely different platforms i.e. ubuntu 12.04 AMD Athlon(tm) 64 X2 Dual Core Processor 3600+ × 2 and on MacOSX 10.7.5 2.2GHz Core i7 using the same test program below : NOTE : ignore the junit test result .. its the logging that indicates the actual problem.
package com.nimrod.test;
import static org.junit.Assert.*;
import java.io.File; import java.io.IOException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.ThreadPoolExecutor;
import net.openhft.collections.SharedHashMap; import net.openhft.collections.SharedHashMapBuilder; import net.openhft.lang.model.DataValueClasses;
import org.junit.After; import org.junit.Before; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory;
public class TestSharedHashMap { final static Logger logger = LoggerFactory.getLogger(TestSharedHashMap.class); SharedHashMap shm; TestDataValue data1 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data2 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data3 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data4 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data5 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data6 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data7 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data8 = DataValueClasses.newDirectReference(TestDataValue.class);
ThreadPoolExecutor threads = (ThreadPoolExecutor)Executors.newFixedThreadPool(2);
final CountDownLatch endGate = new CountDownLatch(2);
@Before public void setUp() throws Exception { try { shm = new SharedHashMapBuilder() .generatedValueType(true) .entries(2048) .entrySize(256) .file(new File("/run/shm/TestOpenHFT")).kClass(String.class).vClass(TestDataValue.class).create();
//For older versions // shm = new SharedHashMapBuilder() // .generatedValueType(true) // //.entries(2048) // .entrySize(1024) // .create(new File("/run/shm/TestOpenHFT"),String.class,TestDataValue.class);
//Map entries are loaded into various segments in mappeddatastore shm.acquireUsing("1111.2222.A0", data1); shm.acquireUsing("1111.2222.B0", data2); shm.acquireUsing("1111.2222.A1", data3); shm.acquireUsing("1111.2222.B1", data4); shm.acquireUsing("1111.2222.A2", data5); shm.acquireUsing("1111.2222.B2", data6); //data7 typically points to data in same segment as data1 points to i.e. segmentnum=0 //To see segments i added line //LOG.info("key="+key+" segmentnum="+segmentNum); //in method : V lookupUsing(K key, V value, boolean create) in class VanillaSharedHashMap shm.acquireUsing("1111.2222.A3", data7); shm.acquireUsing("1111.2222.B3", data8); } catch(Exception e) { endGate.countDown(); endGate.countDown(); e.printStackTrace(); }}
@After public void tearDown() { try { endGate.await(); shm.close(); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } }
@Test public void test() { //Simple case just to prove all is OK data1.setLongString("AAAAAAAAAAAAAAAAAAAA"); assertEquals("AAAAAAAAAAAAAAAAAAAA", data1.getLongString()); data7.setLongString("BBBBBBBBBBBBBBBBBBBB"); assertEquals("BBBBBBBBBBBBBBBBBBBB", data7.getLongString());
logger.info("Start threads to test concurrent read/write in same segment .. this should usually fail"); //..change data7 to another e.g. data2 and should work threads.execute(new TestTask(data1,"AAAAAAAAAAAAAAAAAAAA")); threads.execute(new TestTask(data7,"BBBBBBBBBBBBBBBBBBBB"));}
class TestTask implements Runnable { TestDataValue data; String msg; TestTask(TestDataValue data, String msg) { this.data = data; this.msg = msg; } public void run() { for(int i=0;i<100;i++) { data.setLongString(msg); logger.info("test "+i+" "+data.getLongString()); try { assertEquals(msg, data.getLongString()); } catch (AssertionError e) { logger.error("AssertionError",e); break; } } endGate.countDown(); }
}
}
— Reply to this email directly or view it on GitHub https://github.com/OpenHFT/HugeCollections/issues/43.
andycrutchlow
RobAustin
danielshaya
When concurrent threads reading/writing to DIFFERENT map entries through standard DataValueClass generated getter/setters (marshalling to corresponding UNSAFE methods) and if these map entries phyically reside in the SAME SEGMENT then there is possibility of corruption of data. Problem seems since version 3.0.2 collections i.e. I cannot reliably reproduce on that version..but I can pretty consistantly reproduce on latest 3.2.1 collections and 6.4.6 lang on two completely different platforms i.e. ubuntu 12.04 AMD Athlon(tm) 64 X2 Dual Core Processor 3600+ × 2 and on MacOSX 10.7.5 2.2GHz Core i7 using the same test program below : NOTE : ignore the junit test result .. its the logging that indicates the actual problem.
package com.nimrod.test;
import static org.junit.Assert.*;
import java.io.File; import java.io.IOException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.ThreadPoolExecutor;
import net.openhft.collections.SharedHashMap; import net.openhft.collections.SharedHashMapBuilder; import net.openhft.lang.model.DataValueClasses;
import org.junit.After; import org.junit.Before; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory;
public class TestSharedHashMap { final static Logger logger = LoggerFactory.getLogger(TestSharedHashMap.class); SharedHashMap<String, TestDataValue> shm; TestDataValue data1 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data2 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data3 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data4 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data5 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data6 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data7 = DataValueClasses.newDirectReference(TestDataValue.class); TestDataValue data8 = DataValueClasses.newDirectReference(TestDataValue.class);
//For older versions // shm = new SharedHashMapBuilder() // .generatedValueType(true) // //.entries(2048) // .entrySize(1024) // .create(new File("/run/shm/TestOpenHFT"),String.class,TestDataValue.class);
}