001: /*
002: * All content copyright (c) 2003-2006 Terracotta, Inc., except as may otherwise be noted in a separate copyright notice. All rights reserved.
003: */
004: package com.tctest;
005:
006: import EDU.oswego.cs.dl.util.concurrent.BrokenBarrierException;
007: import EDU.oswego.cs.dl.util.concurrent.CyclicBarrier;
008:
009: import com.tc.exception.TCRuntimeException;
010: import com.tc.memorydatastore.client.MemoryDataStoreClient;
011: import com.tc.memorydatastore.message.TCByteArrayKeyValuePair;
012: import com.tc.memorydatastore.server.MemoryDataStoreServer;
013: import com.tc.util.Assert;
014:
015: import java.util.Collection;
016: import java.util.Iterator;
017:
018: import junit.framework.TestCase;
019:
020: public class MemoryDataMapTest extends TestCase {
021: private MemoryDataStoreServer server = null;
022:
023: protected void setUp() throws Exception {
024: super .setUp();
025: server = MemoryDataStoreServer.createInstance(0);
026: server.start();
027: System.err.println("Server started on port "
028: + server.getListenPort());
029: }
030:
031: protected void tearDown() throws Exception {
032: super .tearDown();
033: server.shutdown();
034: server = null;
035: Thread.sleep(5000);
036: }
037:
038: public void testPerf() throws Exception {
039: int numOfPut = 5000;
040: MemoryDataStoreClient client = new MemoryDataStoreClient(
041: "storePerf", "localhost", server.getListenPort());
042: for (int i = 0; i < numOfPut; i++) {
043: String key = "key" + i;
044: String value = "value" + i;
045: client.put(key.getBytes(), value.getBytes());
046: if (i % 100 == 0) {
047: Thread.sleep(1);
048: }
049: if (i % 100000 == 0) {
050: System.err.println("Send out: " + i + " requests");
051: }
052: }
053: long start = System.currentTimeMillis();
054: Object o = client.get("key10".getBytes());
055: Assert.assertNotNull(o);
056: long end = System.currentTimeMillis();
057: System.err.println("Time to get 1 item: " + (end - start)
058: + "ms");
059:
060: byte[] bytes = new byte[] { 107, 101 };
061: start = System.currentTimeMillis();
062: Collection allValues = client.getAll(bytes);
063: end = System.currentTimeMillis();
064: Assert.assertEquals(numOfPut, allValues.size());
065: System.err.println("Time to get " + numOfPut + " items: "
066: + (end - start) + "ms");
067:
068: client.close();
069: }
070:
071: public void testBasic() throws Exception {
072: CyclicBarrier barrier = new CyclicBarrier(16);
073:
074: MemoryDataStoreClient client1 = new MemoryDataStoreClient(
075: "store1", "localhost", server.getListenPort());
076: MemoryDataStoreClient client2 = new MemoryDataStoreClient(
077: "store2", "localhost", server.getListenPort());
078:
079: Runnable[] putClients = new Runnable[15];
080: putClients[0] = new TestPutClient(barrier, client1, "key1"
081: .getBytes(), "value1".getBytes());
082: putClients[1] = new TestPutClient(barrier, client1, "key2"
083: .getBytes(), "value2".getBytes());
084: putClients[2] = new TestPutClient(barrier, client1, "key3"
085: .getBytes(), "value3".getBytes());
086: putClients[3] = new TestPutClient(barrier, client1, "key4"
087: .getBytes(), "value4".getBytes());
088: putClients[4] = new TestPutClient(barrier, client1, "key5"
089: .getBytes(), "value5".getBytes());
090: putClients[5] = new TestPutClient(barrier, client1, "key6"
091: .getBytes(), "value6".getBytes());
092: putClients[6] = new TestPutClient(barrier, client1, "key7"
093: .getBytes(), "value7".getBytes());
094: putClients[7] = new TestPutClient(barrier, client1, "key8"
095: .getBytes(), "value8".getBytes());
096: putClients[8] = new TestPutClient(barrier, client1, "key9"
097: .getBytes(), "value9".getBytes());
098: putClients[9] = new TestPutClient(barrier, client1, "key10"
099: .getBytes(), "value10".getBytes());
100: putClients[10] = new TestPutClient(barrier, client2, "key1"
101: .getBytes(), "value1".getBytes());
102: putClients[11] = new TestPutClient(barrier, client2, "key2"
103: .getBytes(), "value2".getBytes());
104: putClients[12] = new TestPutClient(barrier, client2, "key3"
105: .getBytes(), "value3".getBytes());
106: putClients[13] = new TestPutClient(barrier, client2, "key4"
107: .getBytes(), "value4".getBytes());
108: putClients[14] = new TestPutClient(barrier, client2, "key5"
109: .getBytes(), "value5".getBytes());
110:
111: runAllClients(putClients);
112:
113: barrier.barrier();
114:
115: Runnable[] getClients = new Runnable[17];
116: barrier = new CyclicBarrier(18);
117:
118: getClients[0] = new TestGetClient(barrier, client1, "key1"
119: .getBytes(), "value1".getBytes());
120: getClients[1] = new TestGetClient(barrier, client1, "key2"
121: .getBytes(), "value2".getBytes());
122: getClients[2] = new TestGetClient(barrier, client1, "key3"
123: .getBytes(), "value3".getBytes());
124: getClients[3] = new TestGetClient(barrier, client1, "key4"
125: .getBytes(), "value4".getBytes());
126: getClients[4] = new TestGetClient(barrier, client1, "key5"
127: .getBytes(), "value5".getBytes());
128: getClients[5] = new TestGetClient(barrier, client1, "key6"
129: .getBytes(), "value6".getBytes());
130: getClients[6] = new TestGetClient(barrier, client1, "key7"
131: .getBytes(), "value7".getBytes());
132: getClients[7] = new TestGetClient(barrier, client1, "key8"
133: .getBytes(), "value8".getBytes());
134: getClients[8] = new TestGetClient(barrier, client1, "key9"
135: .getBytes(), "value9".getBytes());
136: getClients[9] = new TestGetClient(barrier, client1, "key10"
137: .getBytes(), "value10".getBytes());
138: getClients[10] = new TestGetClient(barrier, client2, "key1"
139: .getBytes(), "value1".getBytes());
140: getClients[11] = new TestGetClient(barrier, client2, "key2"
141: .getBytes(), "value2".getBytes());
142: getClients[12] = new TestGetClient(barrier, client2, "key3"
143: .getBytes(), "value3".getBytes());
144: getClients[13] = new TestGetClient(barrier, client2, "key4"
145: .getBytes(), "value4".getBytes());
146: getClients[14] = new TestGetClient(barrier, client2, "key5"
147: .getBytes(), "value5".getBytes());
148: getClients[15] = new TestGetClient(barrier, client2, "key6"
149: .getBytes(), null);
150: getClients[16] = new TestGetClient(barrier, client1, "key11"
151: .getBytes(), null);
152:
153: runAllClients(getClients);
154:
155: barrier.barrier();
156:
157: client1.close();
158: client2.close();
159:
160: MemoryDataStoreClient client = new MemoryDataStoreClient(
161: "store2", "localhost", server.getListenPort());
162: byte[] bytes = new byte[] { 107, 101 };
163: Collection allValues = client.getAll(bytes);
164: System.err.println("Getting all of [107 101]:");
165: printByteArrayCollection(allValues);
166:
167: client = new MemoryDataStoreClient("store1", "localhost",
168: server.getListenPort());
169: client.remove("key1".getBytes());
170:
171: Object o = client.get("key1".getBytes());
172: Assert.assertNull(o);
173:
174: bytes = new byte[] { 107, 101 };
175: client.removeAll(bytes);
176:
177: o = client.get("key5".getBytes());
178: Assert.assertNull(o);
179:
180: }
181:
182: private void printByteArrayCollection(Collection allValues) {
183: int size = allValues.size();
184:
185: System.err.println("No of values: " + size);
186: for (Iterator i = allValues.iterator(); i.hasNext();) {
187: TCByteArrayKeyValuePair keyValuePair = (TCByteArrayKeyValuePair) i
188: .next();
189:
190: byte[] key = keyValuePair.getKey();
191: byte[] value = keyValuePair.getValue();
192:
193: System.err.print("key: [");
194: for (int j = 0; j < key.length; j++) {
195: System.err.print(key[j]);
196: System.err.print(" ");
197: }
198: System.err.print("], value: [");
199: for (int j = 0; j < value.length; j++) {
200: System.err.print(value[j]);
201: System.err.print(" ");
202: }
203: System.err.println("]");
204: }
205: }
206:
207: private static void runAllClients(Runnable[] runnableClients) {
208: Thread[] allClients = new Thread[runnableClients.length];
209: for (int i = 0; i < runnableClients.length; i++) {
210: allClients[i] = new Thread(runnableClients[i]);
211: allClients[i].start();
212: }
213: }
214:
215: private static class TestGetClient implements Runnable {
216: private final CyclicBarrier barrier;
217: private final byte[] key;
218: private final byte[] expectedValue;
219: private final MemoryDataStoreClient client;
220:
221: public TestGetClient(CyclicBarrier barrier,
222: MemoryDataStoreClient client, byte[] key,
223: byte[] expectedValue) {
224: this .barrier = barrier;
225: this .key = key;
226: this .expectedValue = expectedValue;
227: this .client = client;
228: }
229:
230: public void run() {
231: byte[] value = client.get(key);
232: Assert.assertEquals(expectedValue, value);
233:
234: try {
235: barrier.barrier();
236: } catch (InterruptedException e) {
237: throw new TCRuntimeException(e);
238: } catch (BrokenBarrierException e) {
239: throw new TCRuntimeException(e);
240: }
241: System.err.println("Finish running get client");
242: }
243: }
244:
245: private static class TestPutClient implements Runnable {
246: private final CyclicBarrier barrier;
247: private final byte[] key;
248: private final byte[] value;
249: private final MemoryDataStoreClient client;
250:
251: public TestPutClient(CyclicBarrier barrier,
252: MemoryDataStoreClient client, byte[] key, byte[] value) {
253: this .barrier = barrier;
254: this .key = key;
255: this .value = value;
256: this .client = client;
257: }
258:
259: public void run() {
260: client.put(key, value);
261:
262: try {
263: barrier.barrier();
264: } catch (InterruptedException e) {
265: throw new TCRuntimeException(e);
266: } catch (BrokenBarrierException e) {
267: throw new TCRuntimeException(e);
268: }
269: }
270: }
271:
272: }
|