| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF 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. |
| */ |
| |
| using System; |
| using System.Collections; |
| using System.Collections.Generic; |
| using System.Text; |
| using System.Threading; |
| |
| namespace Apache.Geode.Client.FwkLib |
| { |
| using Apache.Geode.DUnitFramework; |
| using Apache.Geode.Client.Tests; |
| using Apache.Geode.Client; |
| |
| public class EntryTxTask<TKey, TVal> : ClientTask |
| { |
| #region Private members |
| |
| static protected int BEGIN_TX = 1001; |
| static protected int EXECUTE_TX_OPS = 1002; |
| static protected int EXECUTE_NONTX_OPS = 1003; |
| static protected int COMMIT_TX = 1004; |
| static protected int ROLLBACK_TX = 1005; |
| |
| |
| private IRegion<TKey, TVal> m_region; |
| private static int keyCount = 0; |
| private static int kCount = 0; |
| private int m_MaxKeys; |
| private static Dictionary<TransactionId, object> activeTxns; |
| private static Dictionary<IRegion<TKey, TVal>, Dictionary<TKey, TVal>> OpMap; |
| private object txlock = new object(); |
| private List<TransactionId> keyList ; |
| private Int32 m_create; |
| private Int32 m_update; |
| private Int32 m_destroy; |
| //private Int32 m_invalidate; |
| private Int32 m_cnt; |
| //bool m_isDestroy; |
| bool m_finish; |
| bool m_beginTx; |
| bool m_commitTx; |
| private object CLASS_LOCK = new object(); |
| private const string SerialExc = "isSerialExecution"; |
| private const string ConcurrentExc = "isConcurrentExecution"; |
| private const string WorkTime = "workTime"; |
| |
| #endregion |
| |
| private static String getClientIdString() |
| { |
| return "Cl_Id" + Util.ClientId + "_thr_" + System.Threading.Thread.CurrentThread.ManagedThreadId.ToString(); |
| } |
| |
| public TxInfo begin(IRegion<TKey,TVal> region) |
| { |
| CacheTransactionManager txManager = null; |
| IRegion<TKey, TVal> reg = region; |
| TransactionId id = null; |
| TxInfo txInfo = new TxInfo(); |
| txManager = CacheHelper<TKey, TVal>.DCache.CacheTransactionManager; |
| try |
| { |
| txManager.Begin(); |
| try |
| { |
| kCount++; |
| TKey key = (TKey)(object)(kCount); |
| TVal value = GetValue(); |
| reg[key] = value; |
| Util.Log("The Key is={0} and value={1}", key.ToString(),value.ToString()); |
| } |
| catch (Exception e) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Caught {0}", e); |
| } |
| } |
| catch (IllegalStateException e) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Caught {0}", e); |
| } |
| try |
| { |
| id = txManager.Suspend(); |
| Util.Log("Suspend() complete "); |
| } |
| catch (IllegalStateException e) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Got {0}", e); |
| } |
| Util.Log("BEGIN_TX returning txId = {0}",id); |
| try |
| { |
| txInfo.setTxId(id); |
| } |
| catch (Exception e) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Got this {0}", e); |
| } |
| |
| return txInfo; |
| } |
| |
| public bool commit(TxInfo txInfo) |
| { |
| TransactionId id = (TransactionId)(object)txInfo.getTxId(); |
| bool commited = false; |
| CacheTransactionManager txManager = null; |
| txManager = CacheHelper<TKey, TVal>.DCache.CacheTransactionManager; |
| if (txManager.TryResume(id, TimeSpan.FromMilliseconds(30000))) |
| { |
| try |
| { |
| txManager.Commit(); |
| commited = true; |
| } |
| catch (CommitConflictException ex) |
| { |
| //Expected exception with concurrent transactions. |
| Util.Log("Got expected exception {0}", ex); |
| } |
| catch (TransactionDataRebalancedException ex) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Got {0}", ex); |
| } |
| catch (TransactionDataNodeHasDepartedException ex) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Got {0}", ex); |
| } |
| } |
| else |
| Util.Log("TxId {0} is not suspended in this member with tryResume time limit, cannot commit. Expected, continuing test."); |
| Util.Log("Commited returning {0}",commited); |
| return commited; |
| |
| } |
| |
| public void rollback(TxInfo txInfo) |
| { |
| TransactionId id = (TransactionId)(object)txInfo.getTxId(); |
| CacheTransactionManager txManager = null; |
| bool isRollBack = false; |
| txManager = CacheHelper<TKey, TVal>.DCache.CacheTransactionManager; |
| if (txManager.TryResume(id, TimeSpan.FromMilliseconds(30000))) |
| { |
| txManager.Rollback(); |
| isRollBack = true; |
| } |
| else |
| Util.Log("TxId {0} is not suspended in this member with tryResume time limit, cannot rollback. Expected with concurrent execution, continuing test."); |
| Util.Log("RollbackTx returning {0}",isRollBack); |
| |
| } |
| |
| public void executeTxOps(TxInfo txInfo) |
| { |
| //This will do a resume+doOps+suspend |
| TransactionId id = (TransactionId)(object)txInfo.getTxId(); |
| CacheTransactionManager txManager = null; |
| txManager = CacheHelper<TKey, TVal>.DCache.CacheTransactionManager; |
| bool executedOps = false; |
| int numOfOpsToDo = 5; |
| if (txManager.TryResume(id, TimeSpan.FromMilliseconds(30000))) |
| { |
| try |
| { |
| doEntryOps(numOfOpsToDo); |
| executedOps = true; |
| } |
| catch (TransactionDataNodeHasDepartedException ex) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Caught {0}", ex); |
| } |
| catch (TransactionDataRebalancedException ex) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Caught {0}", ex); |
| } |
| catch (Exception ex) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Caught unexpected exception during doEntryOps task {0}", ex.Message); |
| } |
| finally |
| { |
| id = txManager.Suspend(); |
| Util.Log("Suspend() complete txId is {0}", id); |
| } |
| } |
| Util.Log("EXECUTE_TX returned {0}", executedOps); |
| } |
| |
| public void doEntryOps(int numOfOpsToDo) |
| { |
| int numOpsCompleted = 0; |
| string opcode = null; |
| List<string> opList = new List<string>(); |
| IRegion<TKey, TVal> region = m_region; |
| int size = 0; |
| int create = 0, update = 0, destroy = 0, invalidate = 0, localdestroy = 0, /*localinvalidate = 0,*/ |
| get = 0/*, putall = 0*/; |
| try |
| { |
| size = region.Count; |
| opcode = FwkTest<TKey, TVal>.CurrentTest.GetStringValue("entryOps"); |
| Util.Log("Op code is {0} and numOf opsCompleted is {1} and Regionsize is {2}", opcode, numOpsCompleted, size); |
| if (opcode == null) opcode = "no-opcode"; |
| if (((size < 1) && (opcode != "create")) || (opcode == "create")) |
| { |
| opcode = "create"; |
| addEntry(region); |
| Interlocked.Increment(ref m_create); |
| create++; |
| } |
| else if (opcode == "update") |
| { |
| updateEntry(region); |
| Interlocked.Increment(ref m_update); |
| update++; |
| } |
| else if (opcode == "destroy") |
| { |
| destroyEntry(region, false); |
| Interlocked.Increment(ref m_destroy); |
| destroy++; |
| } |
| else if (opcode == "localDestroy") |
| { |
| destroyEntry(region, true); |
| //Interlocked.Increment(); |
| localdestroy++; |
| } |
| else if (opcode == "get") |
| { |
| getKey(region); |
| get++; |
| } |
| else if (opcode == "invalidate") |
| { |
| invalidateEntry(region,false); |
| invalidate++; |
| } |
| else |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("CacheServer.doEntryOps() Invalid operation " + |
| "specified: {0}", opcode); |
| } |
| numOpsCompleted++; |
| opList.Add(opcode); |
| |
| } |
| catch (Exception ex) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("CacheServer.doEntryOps() Caught unexpected " + |
| "exception during entry '{0}' operation: {1}.", opcode, ex); |
| } |
| Util.Log("DoEntryOP: create = {0}, update = {1}, destroy = {2},get = {3} invalidate={4}, num of Ops ={5}", |
| m_create, m_update, m_destroy, get, invalidate, numOpsCompleted); |
| } |
| |
| protected TKey addEntry(IRegion<TKey, TVal> m_region) |
| { |
| TKey key = GetNewKey(); |
| TVal value = GetValue(); |
| //int beforeSize = 0; |
| bool isSerialExecution = false; |
| Dictionary<TKey, TVal> addMap = new Dictionary<TKey, TVal>(); |
| try |
| { |
| m_region.Add(key, value); |
| addMap[key] = value; |
| OpMap[m_region] = addMap; |
| doEdgeClientValidation(); |
| } |
| catch (EntryExistsException ex) |
| { |
| if (isSerialExecution) |
| { |
| // cannot get this exception; nobody else can have this key |
| throw new Exception(ex.StackTrace); |
| } |
| else |
| { |
| Util.Log("Caught {0} (expected with concurrent execution); continuing with test", ex); |
| } |
| } |
| return key; |
| } |
| |
| protected void updateEntry(IRegion<TKey, TVal> r) |
| { |
| TKey key = GetExistingKey(true,r); |
| TVal value = GetValue(); |
| Dictionary<TKey, TVal> updateMap = new Dictionary<TKey, TVal>(); |
| r[key] = value; |
| updateMap[key] = value; |
| OpMap[m_region] = updateMap; |
| doEdgeClientValidation(); |
| } |
| |
| protected void invalidateEntry(IRegion<TKey, TVal> r, bool isLocalInvalidate) |
| { |
| TKey key = GetExistingKey(true,r); |
| bool containsKey = m_region.GetLocalView().ContainsKey(key); |
| bool containsValueForKey = m_region.GetLocalView().ContainsValueForKey(key); |
| Util.Log("containsKey for " + key + ": " + containsKey); |
| Util.Log("containsValueForKey for " + key + ": " + containsValueForKey); |
| try |
| { |
| if (isLocalInvalidate) |
| { // do a local invalidate |
| m_region.GetLocalView().Invalidate(key); |
| } |
| else |
| { // do a distributed invalidate |
| m_region.Invalidate(key); |
| } |
| } |
| catch (EntryNotFoundException e) |
| { |
| Util.Log("Caught {0} (expected); continuing with test", e); |
| return; |
| |
| } |
| } |
| |
| protected void destroyEntry(IRegion<TKey, TVal> m_region, bool isLocalDestroy) |
| { |
| TKey key = GetExistingKey(true, m_region); |
| bool isSerialExecution = false; |
| try |
| { |
| if (isLocalDestroy) |
| { // do a local invalidate |
| m_region.GetLocalView().Remove(key); |
| } |
| else |
| { // do a distributed invalidate |
| m_region.Remove(key); |
| } |
| } |
| catch (EntryNotFoundException e) |
| { |
| if (isSerialExecution) |
| throw new Exception(e.StackTrace); |
| else |
| { |
| Util.Log("Caught {0} (expected with concurrent execution); continuing with test", e); |
| return; |
| } |
| } |
| } |
| |
| protected void getKey(IRegion<TKey, TVal> aRegion) |
| { |
| TKey key = GetExistingKey(true,aRegion); |
| TVal anObj = default(TVal); |
| try |
| { |
| anObj = aRegion[key]; |
| } |
| catch (Apache.Geode.Client.KeyNotFoundException) |
| { |
| if (!EqualityComparer<TVal>.Default.Equals(anObj, default(TVal))) |
| { |
| throw new Apache.Geode.Client.KeyNotFoundException(); |
| } |
| |
| } |
| } |
| |
| protected TKey GetExistingKey(bool useServerKeys,IRegion<TKey,TVal> region) |
| { |
| TKey key = default(TKey); |
| if (useServerKeys) |
| { |
| int size = region.Count; |
| TKey[] keys = (TKey[])region.Keys; |
| key = keys[Util.Rand(0, size)]; |
| } |
| else |
| { |
| int size = region.GetLocalView().Count; |
| TKey[] keys = (TKey[])region.GetLocalView().Keys; |
| key = keys[Util.Rand(0, size)]; |
| } |
| return key; |
| } |
| |
| protected TKey GetNewKey() |
| { |
| keyCount++; |
| FwkTest<TKey, TVal>.CurrentTest.ResetKey("distinctKeys"); |
| int numKeys = FwkTest<TKey, TVal>.CurrentTest.GetUIntValue("distinctKeys"); |
| // String keybuf = String.Format("Key-{0}-{1}-{2}", Util.PID, Util.ThreadID, keyCount); |
| TKey key = (TKey)(object)(keyCount); |
| // keyCount++; |
| return key; |
| } |
| |
| protected TVal GetValue() |
| { |
| TVal tmpValue = default(TVal); |
| FwkTest<TKey,TVal>.CurrentTest.ResetKey("valueSizes"); |
| int size = FwkTest<TKey, TVal>.CurrentTest.GetUIntValue("valueSizes"); |
| StringBuilder builder = new StringBuilder(); |
| Random random = new Random(); |
| char ch; |
| for (int j = 0; j < size; j++) |
| { |
| ch = Convert.ToChar(Convert.ToInt32(Math.Floor(26 * random.NextDouble() + 65))); |
| builder.Append(ch); |
| } |
| if (typeof(TVal) == typeof(string)) |
| { |
| tmpValue = (TVal)(object)builder.ToString(); |
| } |
| else if (typeof(TVal) == typeof(byte[])) |
| { |
| tmpValue = (TVal)(object)(Encoding.ASCII.GetBytes(builder.ToString())); |
| } |
| return tmpValue; |
| } |
| |
| public void doTransactions(IRegion<TKey,TVal> region)//,Dictionary<TransactionId, object> m_maps) |
| { |
| Util.Log("Inside ResumableTx:doTransactions()"); |
| TransactionId txId = null; |
| //String txIdStr = ""; |
| TxInfo txInfo = null; |
| IRegion<TKey, TVal> reg = region; |
| int action = 0; |
| FwkTest<TKey, TVal>.CurrentTest.ResetKey("minExecutions"); |
| int minExecutions = FwkTest<TKey, TVal>.CurrentTest.GetUIntValue("minExecutions"); |
| int numExecutions = 0; |
| Util.Log("activeTx map count is {0}",activeTxns.Count); |
| if (activeTxns.Count > 0) |
| { |
| try |
| { |
| keyList = new List<TransactionId>(activeTxns.Keys); |
| } |
| catch (Exception e) { FwkTest<TKey, TVal>.CurrentTest.FwkException("Inside keylist exception {0}", e); } |
| } |
| FwkTest<TKey, TVal>.CurrentTest.ResetKey("numThreads"); |
| string NUMTHREADS = "numThreads"; |
| FwkTest<TKey, TVal>.CurrentTest.ResetKey(NUMTHREADS); |
| int numActiveThd = FwkTest<TKey, TVal>.CurrentTest.GetUIntValue(NUMTHREADS); |
| int numActiveTx = activeTxns.Count; |
| try |
| { |
| Util.Log("numAvtiveTx is {0} and no of AvtiveThread = {1} and activemap count= {2}", numActiveTx, numActiveThd,activeTxns.Count); |
| } |
| catch (Exception e) { FwkTest<TKey, TVal>.CurrentTest.FwkException("Caught exception {0}",e); } |
| lock (CLASS_LOCK) |
| { |
| if (numActiveTx < numActiveThd) |
| { |
| action = BEGIN_TX; |
| } |
| else |
| { |
| for (int i = 0; i < keyList.Count; i++) |
| { |
| txId = (TransactionId)(object)keyList[i]; |
| txInfo = (TxInfo)(object)activeTxns[txId]; |
| |
| numExecutions = txInfo.getNumExecutions(); |
| if (numExecutions > minExecutions) |
| { |
| bool isCommit = FwkTest<TKey, TVal>.CurrentTest.GetBoolValue("TxBool"); |
| if (isCommit) |
| { |
| action = COMMIT_TX; |
| } |
| else |
| { |
| action = ROLLBACK_TX; |
| } |
| break; |
| } |
| } |
| if (action == 0) |
| { |
| action = EXECUTE_TX_OPS; |
| Random random = new Random(); |
| int rant = random.Next(0, keyList.Count - 1); |
| txId = (TransactionId)(object)keyList[rant]; |
| txInfo = (TxInfo)(object)activeTxns[txId]; |
| } |
| } |
| } |
| Boolean success = false; |
| try |
| { |
| lock (txlock) |
| { |
| switch (action) |
| { |
| case (1001): |
| txInfo = begin(reg); |
| txId = txInfo.getTxId(); |
| activeTxns[txId] = (Object)txInfo; |
| break; |
| case (1002): |
| executeTxOps(txInfo); |
| txId = txInfo.getTxId(); |
| txInfo = (TxInfo)(object)activeTxns[txId]; |
| if(txInfo != null) |
| { // could happen if tx committed, entry removed |
| txInfo.incrementNumExecutions(); |
| activeTxns[txId] = (Object)txInfo; |
| } |
| break; |
| case (1004): |
| success = commit(txInfo); |
| break; |
| case (1005): |
| rollback(txInfo); |
| txId = txInfo.getTxId(); |
| activeTxns.Remove(txId); |
| break; |
| } |
| } |
| } |
| catch (Exception e) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("The Test Threw this exception {0}", e); |
| } |
| } |
| |
| public void doSerialExecution(IRegion<TKey, TVal> region) |
| { |
| Util.Log("Inside ResumableTX doSerialExecution()"); |
| keyList = new List<TransactionId>(activeTxns.Keys); |
| TxInfo txInfo; |
| TransactionId txId = null; |
| for (int i = 0; i < keyList.Count; i++) |
| { |
| txInfo = (TxInfo)activeTxns[keyList[i]]; |
| txId = (TransactionId)(object)txInfo.getTxId(); |
| //bool executedOps; |
| try |
| { |
| executeTxOps(txInfo); |
| } |
| catch (Exception ex) |
| { |
| Util.Log("Caught {0} while executing tx ops", ex); |
| //executedOps = false; |
| } |
| txInfo.incrementNumExecutions(); |
| activeTxns[txId] = (Object)txInfo; |
| } |
| } |
| |
| public void doEdgeClientValidation() |
| { |
| //Validate the number of expected key/values and no. of destroyed entries. |
| Util.Log("Verifying expected keys/values"); |
| Util.Log("Region count is {0} and local region count is {1}",m_region.Count,m_region.GetLocalView().Count); |
| TKey key; |
| TVal val; |
| bool success = true; |
| lock (OpMap) |
| { |
| foreach (KeyValuePair<IRegion<TKey, TVal>, Dictionary<TKey, TVal>> kvp in OpMap) |
| { |
| IRegion<TKey, TVal> myregion = (IRegion<TKey, TVal>)kvp.Key; |
| Dictionary<TKey, TVal> mapp = (Dictionary<TKey, TVal>)kvp.Value; |
| foreach (KeyValuePair<TKey, TVal> sp in mapp) |
| { |
| key = sp.Key; |
| val = sp.Value; |
| if (!m_region.ContainsKey(key)) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Expected containsKey() to be true for key {0} ,but it was false", key); |
| success = false; |
| } |
| if (val != null && !m_region.ContainsValueForKey(key)) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Expected containsValueForKey() to be true for key {0} ,but it was false", key); |
| success = false; |
| } |
| } |
| } |
| } |
| if (!success) |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("edge client validation failed"); |
| else |
| Util.Log("Done executing doEdgeClientValidation - validation successful"); |
| |
| } |
| |
| public void finishAllActiveTx(Dictionary<TransactionId,object> activeTX,Boolean isCommit) |
| { |
| lock (CLASS_LOCK) |
| { |
| Util.Log("In Finish active map count is {0}",activeTX.Count); |
| keyList = new List<TransactionId>(activeTX.Keys); |
| bool txCompleted = false; |
| for (int i = 0; i < keyList.Count; i++) |
| { |
| TransactionId txId = (TransactionId)(object)keyList[i]; |
| TxInfo txInfo = (TxInfo)(object)activeTX[txId]; |
| try |
| { |
| txCompleted = commit(txInfo); |
| Util.Log("Commit got {0}",txCompleted); |
| } |
| catch (Exception e) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkException("Got this exception {0}", e); |
| } |
| finally |
| { |
| activeTX.Remove(txId); |
| } |
| Util.Log("After commit/update the Active Map is {0}", activeTX.Count); |
| } |
| } |
| |
| } |
| |
| public EntryTxTask(IRegion<TKey, TVal> region, int keyCnt, Dictionary<TransactionId, object> maps,Dictionary<IRegion<TKey,TVal>,Dictionary<TKey,TVal>>omaps, Boolean finish,Boolean isBeginTx,Boolean isCommitTx) |
| : base() |
| { |
| m_region = region; |
| m_MaxKeys = keyCnt; |
| activeTxns = maps; |
| OpMap = omaps; |
| m_create = 0; |
| m_update = 0; |
| m_destroy = 0; |
| //m_invalidate = 0; |
| m_cnt = 0; |
| //m_isDestroy = true; |
| m_finish = finish; |
| m_beginTx = isBeginTx; |
| m_commitTx = isCommitTx; |
| } |
| public override void DoTask(int iters, object data) |
| { |
| Int32 localcnt = m_cnt; |
| Interlocked.Increment(ref m_cnt); |
| int offset = Util.Rand(m_MaxKeys); |
| int count = offset; |
| //TKey key = default(TKey); |
| //TVal value = default(TVal); |
| FwkTest<TKey, TVal>.CurrentTest.ResetKey(SerialExc); |
| bool isSerialExecution = FwkTest<TKey, TVal>.CurrentTest.GetBoolValue(SerialExc); |
| |
| FwkTest<TKey, TVal>.CurrentTest.ResetKey(ConcurrentExc); |
| bool isConcExecution = FwkTest<TKey, TVal>.CurrentTest.GetBoolValue(ConcurrentExc); |
| int Thrd = FwkTest<TKey, TVal>.CurrentTest.GetUIntValue("numThreads"); |
| Util.Log("EntryTask::DoTask: starting {0} iterations. and isSerialExecution is {1}", iters,isSerialExecution); |
| while (Running && (iters-- != 0)) |
| { |
| if (isConcExecution && !m_finish) |
| { |
| doTransactions(m_region); |
| } |
| if (isSerialExecution && !m_finish) |
| { |
| lock (CLASS_LOCK) |
| { |
| doSerialExecution(m_region); |
| } |
| } |
| if (m_finish && !m_commitTx) |
| { |
| finishAllActiveTx(activeTxns,false); |
| } |
| if (m_beginTx) |
| { |
| lock (txlock) |
| { |
| TxInfo txInfo; |
| TransactionId txId = null; |
| txInfo = begin(m_region); |
| txId = (TransactionId)(object)txInfo.getTxId(); |
| activeTxns[txId] = (Object)txInfo; |
| } |
| } |
| |
| if (m_commitTx && m_finish) |
| { |
| finishAllActiveTx(activeTxns, m_commitTx); |
| } |
| } |
| Interlocked.Add(ref m_iters, count - offset); |
| } |
| } |
| |
| public class ResumableTx<TKey,TVal> : FwkTest<TKey,TVal> |
| { |
| // Tx actions (for concurrent resumable transactions with function execution) |
| static protected int BEGIN_TX = 1001; |
| static protected int EXECUTE_TX_OPS = 1002; |
| static protected int EXECUTE_NONTX_OPS = 1003; |
| static protected int COMMIT = 1004; |
| static protected int ROLLBACK = 1005; |
| #region Private constants and statics |
| |
| static protected int SUSPEND = 2; |
| static protected int RESUME = 3; |
| |
| private static Dictionary<string, Dictionary<TKey, TVal>> BeforeTxMap = new Dictionary<string, Dictionary<TKey, TVal>>(); |
| private static Dictionary<string, Dictionary<TKey, TVal>> AfterTxMap = new Dictionary<string, Dictionary<TKey, TVal>>(); |
| protected static Dictionary<TransactionId, object> activeTxns = null; |
| protected static Dictionary<IRegion<TKey,TVal>,Dictionary<TKey, TVal>> OpMap = null; |
| //private List<TransactionId> keyList; |
| //private static int globalTxId = 0; |
| //private static int ExecutionNo = 0; |
| //private static string RRName = null; |
| |
| private object CLASS_LOCK = new object(); |
| |
| #endregion |
| |
| private const string RegionName = "regionName"; |
| private const string ValueSizes = "valueSizes"; |
| private const string OpsSecond = "opsSecond"; |
| private const string EntryCount = "entryCount"; |
| private const string WorkTime = "workTime"; |
| private const string EntryOps = "entryOps"; |
| private const string LargeSetQuery = "largeSetQuery"; |
| private const string UnsupportedPRQuery = "unsupportedPRQuery"; |
| private const string ObjectType = "objectType"; |
| protected const string NumThreads = "numThreads"; |
| protected const string TimedInterval = "timedInterval"; |
| protected const string DistinctKeys = "distinctKeys"; |
| protected const string BEGINTX = "isBeginTX"; |
| protected const string COMMITTX = "isCommitTX"; |
| |
| public class SilenceListener<TKey1, TVal1> : CacheListenerAdapter<TKey1,TVal1> |
| { |
| public override void AfterCreate(EntryEvent<TKey1, TVal1> ev) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkInfo("SilenceListener: AfterCreate key"); |
| FwkTest<TKey, TVal>.CurrentTest.FwkInfo("SilenceListener: AfterCreate key = {0} value = {1}", ev.Key, ev.NewValue); |
| } |
| public override void AfterUpdate(EntryEvent<TKey1, TVal1> ev) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkInfo("SilenceListener: AfterUpdate"); |
| FwkTest<TKey, TVal>.CurrentTest.FwkInfo("SilenceListener: AfterUpdate key = {0} value = {1}", ev.Key, ev.NewValue); |
| } |
| public override void AfterDestroy(EntryEvent<TKey1, TVal1> ev) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkInfo("SilenceListener: AfterDestroy"); |
| } |
| public override void AfterInvalidate(EntryEvent<TKey1, TVal1> ev) |
| { |
| FwkTest<TKey, TVal>.CurrentTest.FwkInfo("SilenceListener: AfterInvalidate"); |
| } |
| } |
| |
| #region Private utility methods |
| |
| private IRegion<TKey, TVal> GetRegion() |
| { |
| return GetRegion(null); |
| } |
| |
| protected IRegion<TKey, TVal> GetRegion(string regionName) |
| { |
| IRegion<TKey, TVal> region; |
| if (regionName == null) |
| { |
| regionName = GetStringValue("regionName"); |
| } |
| if (regionName == null) |
| { |
| region = (IRegion<TKey, TVal>)GetRootRegion(); |
| if (region == null) |
| { |
| IRegion<TKey, TVal>[] rootRegions = CacheHelper<TKey, TVal>.DCache.RootRegions<TKey, TVal>(); |
| if (rootRegions != null && rootRegions.Length > 0) |
| { |
| region = rootRegions[Util.Rand(rootRegions.Length)]; |
| } |
| } |
| } |
| else |
| { |
| region = CacheHelper<TKey, TVal>.GetRegion(regionName); |
| } |
| return region; |
| } |
| #endregion |
| |
| #region Public Methods |
| |
| public static ICacheListener<TKey, TVal> CreateSilenceListener() |
| { |
| return new SilenceListener<TKey, TVal>(); |
| } |
| |
| public virtual void DoCreatePool() |
| { |
| FwkInfo("In DoCreatePool()"); |
| try |
| { |
| CreatePool(); |
| } |
| catch (Exception ex) |
| { |
| FwkException("DoCreatePool() Caught Exception: {0}", ex); |
| } |
| FwkInfo("DoCreatePool() complete."); |
| } |
| |
| |
| public virtual void DoCreateRegion() |
| { |
| FwkInfo("ResumableTx:DoCreateRegion()"); |
| try |
| { |
| IRegion<TKey, TVal> region = CreateRootRegion(); |
| ResetKey("useTransactions"); |
| if (region == null) |
| { |
| FwkException("ResumableTx:DoCreateRegion() could not create region."); |
| } |
| FwkInfo("ResumableTx:DoCreateRegion() Created region '{0}'", region.Name); |
| } |
| catch (Exception ex) |
| { |
| FwkException("ResumableTx:DoCreateRegion() Caught Exception: {0}", ex); |
| } |
| FwkInfo("ResumableTx:DoCreateRegion() complete."); |
| } |
| |
| public void DoCloseCache() |
| { |
| FwkInfo("DoCloseCache() Closing cache and disconnecting from" + |
| " distributed system."); |
| CacheHelper<TKey, TVal>.Close(); |
| } |
| |
| public void DoRegisterAllKeys() |
| { |
| FwkInfo("In DoRegisterAllKeys()"); |
| try |
| { |
| IRegion<TKey, TVal> region = GetRegion(); |
| FwkInfo("DoRegisterAllKeys() region name is {0}", region.Name); |
| bool isDurable = GetBoolValue("isDurableReg"); |
| ResetKey("getInitialValues"); |
| bool isGetInitialValues = GetBoolValue("getInitialValues"); |
| bool checkReceiveVal = GetBoolValue("checkReceiveVal"); |
| bool isReceiveValues = true; |
| if (checkReceiveVal) |
| { |
| ResetKey("receiveValue"); |
| isReceiveValues = GetBoolValue("receiveValue"); |
| } |
| region.GetSubscriptionService().RegisterAllKeys(isDurable, isGetInitialValues, isReceiveValues); |
| } |
| catch (Exception ex) |
| { |
| FwkException("DoRegisterAllKeys() Caught Exception: {0}", ex); |
| } |
| FwkInfo("DoRegisterAllKeys() complete."); |
| } |
| |
| public void DoSerialTxWithOps() |
| { |
| FwkInfo("Inside DoSerialTxWithOps ActiveMap count is {0}", activeTxns.Count); |
| IRegion<TKey, TVal> region = GetRegion(); |
| int entryCount = GetUIntValue(EntryCount); |
| entryCount = (entryCount < 1) ? 10000 : entryCount; |
| OpMap = new Dictionary<IRegion<TKey, TVal>, Dictionary<TKey, TVal>>(); |
| |
| int timedInterval = GetTimeValue("timedInterval") * 1000; |
| int maxTime = 10 * timedInterval; |
| // Loop over key set sizes |
| ResetKey("distinctKeys"); |
| int numKeys = GetUIntValue("distinctKeys"); |
| |
| ResetKey(NumThreads); |
| int numThreads; |
| while ((numThreads = GetUIntValue(NumThreads)) > 0) |
| { |
| EntryTxTask<TKey, TVal> entrytask = new EntryTxTask<TKey, TVal>(region, numKeys / numThreads, activeTxns, OpMap,false,false,false); |
| FwkInfo("Running timed task "); |
| try |
| { |
| RunTask(entrytask, numThreads, -1, timedInterval, maxTime, null); |
| } |
| catch (ClientTimeoutException) |
| { |
| FwkException("In DoSerialTxWithOps() Timed run timed out."); |
| } |
| FwkInfo("Completed timed task "); |
| Thread.Sleep(3000); |
| //entrytask.dumpToBB(); |
| } |
| } |
| |
| public void DoConcTxWithOps() |
| { |
| FwkInfo("Inside ResumableTx:DoConcTxWithOps()"); |
| IRegion<TKey, TVal> region = GetRegion(); |
| activeTxns = new Dictionary<TransactionId, object>(); |
| OpMap = new Dictionary<IRegion<TKey, TVal>, Dictionary<TKey, TVal>>(); |
| int entryCount = GetUIntValue(EntryCount); |
| entryCount = (entryCount < 1) ? 10000 : entryCount; |
| |
| int secondsToRun = GetTimeValue(WorkTime); |
| secondsToRun = (secondsToRun < 1) ? 10 : secondsToRun; |
| |
| DateTime now = DateTime.Now; |
| DateTime end = now.AddSeconds(secondsToRun); |
| |
| int timedInterval = GetTimeValue(TimedInterval) * 1000; |
| int maxTime = 10 * timedInterval; |
| // Loop over key set sizes |
| ResetKey(DistinctKeys); |
| int numKeys = GetUIntValue(DistinctKeys); |
| |
| ResetKey(NumThreads); |
| int numThreads; |
| while ((numThreads = GetUIntValue(NumThreads)) > 0) |
| { |
| EntryTxTask<TKey, TVal> entrytask = new EntryTxTask<TKey, TVal>(region, numKeys / numThreads, activeTxns,OpMap,false,false,false); |
| FwkInfo("Running timed task "); |
| try |
| { |
| RunTask(entrytask, numThreads, -1, timedInterval, maxTime, null); |
| } |
| catch (ClientTimeoutException) |
| { |
| FwkException("In DoConcTxWithOps() Timed run timed out."); |
| } |
| Thread.Sleep(3000); |
| //entrytask.dumpToBB(); |
| } |
| FwkInfo("Completed timed task "); |
| } |
| |
| public void DoBeginTx() |
| { |
| IRegion<TKey, TVal> region = GetRegion(); |
| activeTxns = new Dictionary<TransactionId, object>(); |
| int timedInterval = GetTimeValue(TimedInterval) * 1000; |
| int maxTime = 10 * timedInterval; |
| ResetKey(DistinctKeys); |
| |
| ResetKey(BEGINTX); |
| bool isBeginTx = GetBoolValue(BEGINTX); |
| |
| int numKeys = GetUIntValue(DistinctKeys); |
| ResetKey(NumThreads); |
| int numThreads; |
| while ((numThreads = GetUIntValue(NumThreads)) > 0) |
| { |
| EntryTxTask<TKey, TVal> entrytask = new EntryTxTask<TKey, TVal>(region, numKeys / numThreads, activeTxns,OpMap, false, isBeginTx, false); |
| FwkInfo("Running timed task "); |
| try |
| { |
| RunTask(entrytask, numThreads, -1, timedInterval, maxTime, null); |
| } |
| catch (ClientTimeoutException) |
| { |
| FwkException("In DoBeginTx() Timed run timed out."); |
| } |
| Thread.Sleep(3000); |
| //entrytask.dumpToBB(); |
| } |
| |
| } |
| |
| public void DoFinishAllActiveTx() |
| { |
| FwkInfo("Inside DoFinishActiveTx ActiveMap count is {0}",activeTxns.Count); |
| Dictionary<TKey, TVal> tempmap = new Dictionary<TKey, TVal>(); |
| IRegion<TKey, TVal> region = GetRegion(); |
| int timedInterval = GetTimeValue(TimedInterval) * 1000; |
| int maxTime = 10 * timedInterval; |
| // Loop over key set sizes |
| ResetKey(DistinctKeys); |
| int numKeys = GetUIntValue(DistinctKeys); |
| |
| ResetKey(COMMITTX); |
| bool isCommitTx = GetBoolValue(COMMITTX); |
| //bool Finish=true; |
| ResetKey(NumThreads); |
| int numThreads; |
| while ((numThreads = GetUIntValue(NumThreads)) > 0) |
| { |
| EntryTxTask<TKey, TVal> entrytask = new EntryTxTask<TKey, TVal>(region, numKeys / numThreads, activeTxns,OpMap, true,false,isCommitTx); |
| FwkInfo("Running timed task "); |
| try |
| { |
| RunTask(entrytask, numThreads, -1, timedInterval, maxTime, null); |
| } |
| catch (ClientTimeoutException) |
| { |
| FwkException("In DoFinishAllActiveTx() Timed run timed out."); |
| } |
| Thread.Sleep(3000); |
| //entrytask.dumpToBB(); |
| } |
| |
| } |
| |
| public void DoPopulateRegion() |
| { |
| CacheTransactionManager txManager = CacheHelper<TKey, TVal>.DCache.CacheTransactionManager; |
| //TransactionId txId = CacheHelper<TKey, TVal>.DCache.CacheTransactionManager; |
| IRegion<TKey, TVal> reg = GetRegion(); |
| for (int i = 0; i < 5; i++) |
| { |
| TKey key = (TKey)(object)(i); |
| TVal value = (TVal)(object)"Value_"; |
| txManager.Begin(); |
| reg[key] = value; |
| txManager.Commit(); |
| Util.Log("The Key is={0} and value={1}", key.ToString(), value.ToString()); |
| } |
| } |
| |
| public void DoGet() |
| { |
| IRegion<TKey, TVal> reg = GetRegion(); |
| for (int i = 0; i < 5; i++) |
| { |
| TKey key = (TKey)(object)(i); |
| try |
| { |
| TVal val = reg[key]; |
| Util.Log("The Key is={0} and value={1}", key.ToString(), val.ToString()); |
| } |
| catch (Exception e) { FwkException("SP:Caught {0}",e); } |
| } |
| } |
| |
| public void DoBasicTX() |
| { |
| FwkInfo("Inside ResumableTx:DoBasicTX()"); |
| IRegion<TKey, TVal> region = GetRegion(); |
| string opcode = null; |
| int action = 0; |
| int minNoEx = GetUIntValue("minExecutions"); |
| |
| if (region == null) |
| { |
| FwkSevere("ResumableTx:DoBasicTX(): No region to perform operations on."); |
| } |
| CacheTransactionManager txManager = null; |
| TransactionId txId = null; |
| |
| int numOfEx = 0; |
| try |
| { |
| txManager = CacheHelper<TKey, TVal>.DCache.CacheTransactionManager; |
| if (numOfEx == 0) |
| { |
| FwkInfo("Begin.() tx numOfEx={0} and minNoEx={1}", numOfEx, minNoEx); |
| txManager.Begin(); |
| action = BEGIN_TX; |
| doValidateTxOps(txId, action); |
| } |
| while (!(numOfEx > minNoEx)) |
| { |
| try |
| { |
| FwkInfo("numOfEx={0} and minNoEx={1}", numOfEx, minNoEx); |
| txId = txManager.Suspend(); |
| action = SUSPEND; |
| doValidateTxOps(txId, action); |
| |
| txManager.Resume(txId); |
| int numKeys = 100; |
| Dictionary<TKey, TVal> keyValMap = new Dictionary<TKey, TVal>(); |
| TKey key; |
| TVal value; |
| for (int i = 0; i < numKeys; i++) |
| { |
| string keyName = String.Format("key_{0}", i); |
| key = (TKey)(object)keyName.ToString(); |
| value = (TVal)(object)"Value_"; |
| region[key] = value; |
| keyValMap[key] = value; |
| AfterTxMap["put"] = keyValMap; |
| } |
| action = RESUME; |
| FwkInfo("keyValMap count={0} and region count is {1}",keyValMap.Count,region.Keys.Count); |
| doValidateTxOps(txId, action); |
| } |
| catch (IllegalStateException ex) |
| { |
| FwkException("Got {0}", ex.Message); |
| } |
| catch (CommitConflictException ex) |
| { |
| FwkException("Got {0}", ex.Message); |
| } |
| numOfEx++; |
| } |
| if (numOfEx > minNoEx) |
| { |
| try |
| { |
| txManager.Commit(); |
| } |
| catch (CommitConflictException) |
| { |
| FwkInfo("Got Expected exception as there was a write conflict"); |
| } |
| action = COMMIT; |
| doValidateTxOps(txId, action); |
| } |
| } |
| catch (Exception ex) |
| { |
| FwkException("ResumableTx.DoEntryOpsWithTX() Caught unexpected " + |
| "exception during entry '{0}' operation: {1}.", opcode, ex); |
| } |
| } |
| |
| public void doValidateTxOps(TransactionId txId, int txState) |
| { |
| FwkInfo("Inside Resumabletx:DoValidateTxOps"); |
| FwkInfo("Transactional Id is {0} and transaction state is={1}",txId,txState); |
| //int cnt =0; |
| //TKey key; |
| int txnState = txState; |
| TransactionId txnId = txId; |
| ResetKey("entryCount"); |
| int EntryCount=GetUIntValue("entryCount"); |
| CacheTransactionManager txManager = CacheHelper<TKey, TVal>.DCache.CacheTransactionManager; |
| try |
| { |
| switch (txnState) |
| { |
| case 1001: |
| FwkInfo("...txnState is Begin ..."); |
| checkContainsKey(txnState); |
| break; |
| case 2: |
| FwkInfo("...txnState is Suspend..."); |
| if (!txManager.IsSuspended(txId) && !txManager.Exists(txId)) |
| { |
| FwkException("After Suspend(),the Transaction should have been Suspended and should still exist"); |
| } |
| checkContainsKey(txnState); |
| break; |
| case 3: |
| FwkInfo("...txnState is Resume..."); |
| if (txManager.IsSuspended(txnId)) |
| { |
| FwkException("After Resume(),the Transaction should NOT have been Suspended"); |
| } |
| if (!txManager.Exists(txnId)) |
| { |
| FwkException("After Resume(),the Transaction should still exist"); |
| } |
| checkContainsKey(txnState); |
| break; |
| case 1004: |
| FwkInfo("...txnState is Commit..."); |
| if (txManager.IsSuspended(txnId)) |
| { |
| FwkException("After Commit(),the Transaction should NOT have been Suspended"); |
| } |
| if (txManager.Exists(txnId)) |
| { |
| FwkException("After Commit(),the Transaction should Not exist"); |
| } |
| if (txManager.TryResume(txnId)) |
| { |
| FwkException("After Commit(),the Transaction should Not be resumed"); |
| } |
| checkContainsKey(txnState); |
| FwkInfo("Commit BeforeTxMap.Count={0} ", BeforeTxMap.Count); |
| break; |
| } |
| |
| } |
| catch(Exception ex) |
| { |
| FwkException("doValidateTxOps caught exception {0}",ex.Message); |
| } |
| } |
| |
| public void checkContainsKey(int txnState) |
| { |
| IRegion<TKey, TVal> region = GetRegion(); |
| Dictionary<TKey, TVal> BfTxOpMap = new Dictionary<TKey, TVal>(); |
| Dictionary<TKey, TVal> AfTxOpMap = new Dictionary<TKey, TVal>(); |
| try |
| { |
| if (txnState == 3) |
| { |
| BfTxOpMap = BeforeTxMap["put"]; |
| AfTxOpMap = AfterTxMap["put"]; |
| } |
| else |
| BfTxOpMap = BeforeTxMap["put"]; |
| |
| } |
| catch (Exception e) |
| { |
| FwkException("Throwing this exception {0} BMap cpunt={1} AfMap Count={2}", e.Message, BfTxOpMap.Count, AfTxOpMap.Count); |
| } |
| FwkInfo("BfTxOpmap count is {0} and AfterTxMap is {1}", BfTxOpMap.Count, AfTxOpMap.Count); |
| foreach (KeyValuePair<TKey, TVal> bkp in BfTxOpMap) |
| { |
| FwkInfo("Inside BfTxOpMap"); |
| TKey BfTxKey = bkp.Key; |
| if (!region.ContainsKey(BfTxKey)) |
| FwkException("getKey: expected key {0} is not present in the region key set", bkp.Key.ToString()); |
| } |
| foreach (KeyValuePair<TKey, TVal> akp in AfTxOpMap) |
| { |
| FwkInfo("Inside AfTxOpMap"); |
| TKey AfTxKey = akp.Key; |
| if (!region.ContainsKey(AfTxKey)) |
| FwkException("getKey: expected key {0} is not present in the region key set", akp.Key.ToString()); |
| } |
| } |
| |
| #endregion |
| } |
| } |
| |
