Revision: b16146640918
Author: unknown <yulian@Venera.(none)>
Date: Fri Jan 18 13:27:08 2013
Log: Added 500ms cache to block duplicated
Possibly should be extended up to 16 seconds to
disallowed resend by client , but may cause high memory.
Resolved byte to int translation
http://code.google.com/p/mediaserver/source/detail?r=b16146640918
Modified:
/client/jsr-309/tck/server.log
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/GlobalTransactionManager.java
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/Transaction.java
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/TransactionManager.java
/controls/mgcp/src/test/java/org/mobicents/media/server/mgcp/tx/TransactionManagerTest.java
=======================================
--- /client/jsr-309/tck/server.log Fri Jan 18 07:58:56 2013
+++ /client/jsr-309/tck/server.log Fri Jan 18 13:27:08 2013
@@ -1,6 +1,6 @@
-2013-01-18 17:19:57,313 DEBUG [TestBase] (main) Loading
file:/opt/mobsource/medianew/client/jsr-309/tck/exclude-list.properties
-2013-01-18 17:19:57,315 INFO [Runner] (main) Start
mandatory.functional.mediagroup.PlayerTest.test_2_1_1_1_Play
javadoc::Joinable::1 javadoc::MediaGroup::1 javadoc::MediaGroup::6
javadoc::Player::1 javadoc::PlayerEvent::15
-2013-01-18 17:19:57,315 DEBUG [TestBase] (main) Getting a MsControlFactory
instance
-2013-01-18 17:19:57,337 INFO [DriverImpl] (main) Initializing MGCP on
/127.0.0.1:2729
-2013-01-18 17:19:57,351 WARN [JainMgcpStackImpl] (main) Failed to locate
properties file, using default values
-2013-01-18 17:19:57,362 INFO [JainMgcpStackImpl] (main) Jain Mgcp stack
bound to IP /127.0.0.1 and UDP port 2729
+2013-01-18 22:57:15,696 DEBUG [TestBase] (main) Loading
file:/opt/mobsource/medianew/client/jsr-309/tck/exclude-list.properties
+2013-01-18 22:57:15,697 INFO [Runner] (main) Start
mandatory.functional.mediagroup.PlayerTest.test_2_1_1_1_Play
javadoc::Joinable::1 javadoc::MediaGroup::1 javadoc::MediaGroup::6
javadoc::Player::1 javadoc::PlayerEvent::15
+2013-01-18 22:57:15,697 DEBUG [TestBase] (main) Getting a MsControlFactory
instance
+2013-01-18 22:57:15,719 INFO [DriverImpl] (main) Initializing MGCP on
/127.0.0.1:2729
+2013-01-18 22:57:15,732 WARN [JainMgcpStackImpl] (main) Failed to locate
properties file, using default values
+2013-01-18 22:57:15,742 INFO [JainMgcpStackImpl] (main) Jain Mgcp stack
bound to IP /127.0.0.1 and UDP port 2729
=======================================
---
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/GlobalTransactionManager.java
Fri Jan 18 07:58:56 2013
+++
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/GlobalTransactionManager.java
Fri Jan 18 13:27:08 2013
@@ -155,6 +155,7 @@
txManager.setNamingService(namingService);
txManager.setCallManager(new CallManager());
txManager.setMgcpProvider(provider);
+ txManager.start();
return txManager;
}
@@ -197,6 +198,8 @@
}
private void checkAddress(byte index) {
+ int intIndex = ((int)index) & 0xFF;
+
if(subManager==null)
{
try
@@ -212,15 +215,15 @@
{
subManager=new GlobalTransactionManager[256];
- subManager[index]=new
GlobalTransactionManager(scheduler);
- subManager[index].setMgcpProvider(provider);
- subManager[index].setNamingService(namingService);
- subManager[index].setPoolSize(poolSize);
+ subManager[intIndex]=new
GlobalTransactionManager(scheduler);
+ subManager[intIndex].setMgcpProvider(provider);
+ subManager[intIndex].setNamingService(namingService);
+ subManager[intIndex].setPoolSize(poolSize);
}
lock.unlock();
}
- else if(subManager[index]==null)
+ else if(subManager[intIndex]==null)
{
try
{
@@ -231,12 +234,12 @@
}
- if(subManager[index]==null)
+ if(subManager[intIndex]==null)
{
- subManager[index]=new
GlobalTransactionManager(scheduler);
- subManager[index].setMgcpProvider(provider);
- subManager[index].setNamingService(namingService);
- subManager[index].setPoolSize(poolSize);
+ subManager[intIndex]=new
GlobalTransactionManager(scheduler);
+ subManager[intIndex].setMgcpProvider(provider);
+ subManager[intIndex].setNamingService(namingService);
+ subManager[intIndex].setPoolSize(poolSize);
}
lock.unlock();
=======================================
---
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/Transaction.java
Tue Oct 23 05:48:15 2012
+++
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/Transaction.java
Fri Jan 18 13:27:08 2013
@@ -43,6 +43,8 @@
protected int id;
protected int uniqueId;
+ protected boolean completed=false;
+
//Transaction manager instance
private TransactionManager txManager;
@@ -154,9 +156,14 @@
logger.info("tx=" + id + " was executed normaly");
if (action != null && action.getEvent() != null) {
action.getEvent().recycle();
+ action=null;
}
- txManager.terminate(this);
+ if(!completed)
+ {
+ completed=true;
+ txManager.terminate(this);
+ }
}
public void onFailure(Exception e) {
@@ -176,7 +183,13 @@
logger.info("tx=" + id + " Rolled back");
if (action.getEvent() != null) {
action.getEvent().recycle();
+ action=null;
+ }
+
+ if(!completed)
+ {
+ completed=true;
+ txManager.terminate(this);
}
- txManager.terminate(this);
}
}
=======================================
---
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/TransactionManager.java
Fri Jan 18 07:58:56 2013
+++
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/TransactionManager.java
Fri Jan 18 13:27:08 2013
@@ -25,6 +25,7 @@
import org.mobicents.media.server.mgcp.MgcpProvider;
import org.mobicents.media.server.mgcp.controller.CallManager;
import org.mobicents.media.server.mgcp.controller.naming.NamingTree;
+import org.mobicents.media.server.scheduler.Task;
import org.mobicents.media.server.scheduler.Scheduler;
import org.mobicents.media.server.scheduler.ConcurrentLinkedList;
@@ -44,6 +45,12 @@
//pool of transaction objects
private ConcurrentLinkedList<Transaction> pool=new
ConcurrentLinkedList();
+
+ //cache size in 100ms units
+ private static final int cacheSize=5;
+ private ConcurrentLinkedList<Transaction>[] cache=new
ConcurrentLinkedList[cacheSize];
+ private int cleanIndex=0;
+
//currently active transactions.
private ConcurrentHashMap<Integer,Transaction> active;
@@ -58,6 +65,10 @@
//call manager
protected CallManager callManager;
+
+ //cache heartbeat
+ private Heartbeat cacheHeartbeat;
+
/**
* Creates new transaction's pool.
*
@@ -71,7 +82,17 @@
for (int i = 0; i < size; i++) {
pool.offer(new Transaction(this));
- }
+ }
+
+ for(int i=0;i<cache.length;i++)
+ cache[i]=new ConcurrentLinkedList<Transaction>();
+
+ cacheHeartbeat = new Heartbeat();
+ }
+
+ public void start()
+ {
+ scheduler.submitHeatbeat(cacheHeartbeat);
}
/**
@@ -158,10 +179,7 @@
* @param t the transaction to be terminated
*/
protected void terminate(Transaction t) {
- active.remove(t.uniqueId);
- t.id = 0;
- t.uniqueId=0;
- pool.offer(t);
+ cache[cleanIndex].offer(t);
}
/**
@@ -182,4 +200,35 @@
protected int remainder() {
return pool.size();
}
+
+ private class Heartbeat extends Task {
+ int queueToClean;
+
+ public Heartbeat() {
+ super();
+ }
+
+ @Override
+ public long perform() {
+ queueToClean=(cleanIndex+1)%cacheSize;
+ Transaction current=cache[queueToClean].poll();
+ while(current!=null)
+ {
+ active.remove(current.uniqueId);
+ current.id = 0;
+ current.uniqueId=0;
+ current.completed=false;
+ pool.offer(current);
+ current=cache[queueToClean].poll();
+ }
+
+ cleanIndex=(cleanIndex+1)%cacheSize;
+ scheduler.submitHeatbeat(this);
+ return 0;
+ }
+
+ public int getQueueNumber() {
+ return scheduler.HEARTBEAT_QUEUE;
+ }
+ }
}
=======================================
---
/controls/mgcp/src/test/java/org/mobicents/media/server/mgcp/tx/TransactionManagerTest.java
Sat Jan 5 09:44:27 2013
+++
/controls/mgcp/src/test/java/org/mobicents/media/server/mgcp/tx/TransactionManagerTest.java
Fri Jan 18 13:27:08 2013
@@ -65,7 +65,9 @@
scheduler.setClock(clock);
scheduler.start();
- txManager = new TransactionManager(scheduler, 3);
+ txManager = new TransactionManager(scheduler, 3);
+ txManager.start();
+
action = new Action();
TaskChain actionHandler = new TaskChain(2,scheduler);
@@ -105,6 +107,16 @@
assertEquals(2, txManager.remainder());
txManager.terminate(tx);
+
+ try
+ {
+ Thread.sleep(700);
+ }
+ catch(InterruptedException e)
+ {
+
+ }
+
assertEquals(3, txManager.remainder());
}
@@ -113,7 +125,7 @@
{
Transaction tx = txManager.allocateNew(id);
tx.process(action);
- Thread.sleep(1500);
+ Thread.sleep(700);
assertEquals(3, txManager.remainder());
}
catch(Exception e)
