Revision: 5b5c166b156a
Author: unknown <yulian@Venera.(none)>
Date: Fri Jan 18 07:58:56 2013
Log: Added Transaction manager per ip+port pair.
blocks duplicates for time transction exists ( between request and response
).
http://code.google.com/p/mediaserver/source/detail?r=5b5c166b156a
Added:
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/GlobalTransactionManager.java
Modified:
/client/jsr-309/tck/server.log
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/controller/Controller.java
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/TransactionManager.java
=======================================
--- /dev/null
+++
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/GlobalTransactionManager.java
Fri Jan 18 07:58:56 2013
@@ -0,0 +1,262 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2011, Red Hat, Inc. and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.mobicents.media.server.mgcp.tx;
+
+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.Scheduler;
+import org.mobicents.media.server.scheduler.ConcurrentLinkedList;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import java.util.Enumeration;
+
+/**
+ * Implements pool of transactions.
+ *
+ * @author yulian oifa
+ */
+public class GlobalTransactionManager
+{
+ //one byte block
+ private GlobalTransactionManager[] subManager;
+
+ //ports block
+ private TransactionManager[] managers;
+
+ //MGCP protocol provider
+ protected MgcpProvider provider;
+
+ //Scheduler
+ private Scheduler scheduler;
+
+ //Endpoint naming tree
+ private NamingTree namingService;
+
+ private int poolSize;
+
+ private Lock lock=new Lock();
+
+ public GlobalTransactionManager(Scheduler scheduler)
+ {
+ this.scheduler = scheduler;
+ }
+
+ /**
+ * Associates endpoint naming service.
+ *
+ * @param namingService the endpoint naming service.
+ */
+ public void setNamingService(NamingTree namingService) {
+ this.namingService = namingService;
+ }
+
+ /**
+ * Associates endpoint naming service.
+ *
+ * @param namingService the endpoint naming service.
+ */
+ public void setPoolSize(int poolSize) {
+ this.poolSize = poolSize;
+ }
+
+ /**
+ * Gets the access to the scheduler.
+ *
+ * @return job scheduler.
+ */
+ public Scheduler scheduler() {
+ return scheduler;
+ }
+
+ /**
+ * Provides access to the wall clock.
+ *
+ * @return time measured by wall clock.
+ */
+ public long getTime() {
+ return scheduler.getClock().getTime();
+ }
+
+ /**
+ * Assigns MGCP provider.
+ *
+ * @param provider mgcp provider instance
+ */
+ public void setMgcpProvider(MgcpProvider provider) {
+ this.provider = provider;
+ }
+
+ /**
+ * Find active transaction with specified identifier.
+ *
+ * @param id the transaction identifier.
+ * @return transaction object or null if does not exist.
+ */
+ public Transaction find(InetSocketAddress source,int id) {
+ return find(source.getAddress().getAddress(),source.getPort(),0,id);
+ }
+
+ private Transaction find(byte[] address,int port,int level,int id) {
+ if(level==address.length-1)
+ {
+ checkPort(port-1);
+ return managers[port-1].find(id);
+ }
+
+ checkAddress(address[level]);
+ return subManager[address[level]].find(address, port, level+1, id);
+ }
+
+ public Transaction allocateNew(InetSocketAddress source,int id) {
+ return
allocateNew(source.getAddress().getAddress(),source.getPort(),0,id);
+ }
+
+ private Transaction allocateNew(byte[] address,int port,int level,int
id) {
+ if(level==address.length-1)
+ {
+ checkPort(port-1);
+ Transaction old=managers[port-1].find(id);
+ if(old==null)
+ return managers[port-1].allocateNew(id);
+
+ return null;
+ }
+
+ checkAddress(address[level]);
+ return subManager[address[level]].allocateNew(address, port, level+1,
id);
+ }
+
+ public TransactionManager createTransactionManager() {
+ TransactionManager txManager = new TransactionManager(scheduler,
poolSize);
+ txManager.setNamingService(namingService);
+ txManager.setCallManager(new CallManager());
+ txManager.setMgcpProvider(provider);
+ return txManager;
+ }
+
+ private void checkPort(int portIndex) {
+ if(managers==null)
+ {
+ try
+ {
+ lock.lock();
+ }
+ catch(java.lang.InterruptedException e)
+ {
+
+ }
+
+ if(managers==null)
+ {
+ managers=new TransactionManager[65536];
+ managers[portIndex]=createTransactionManager();
+ }
+
+ lock.unlock();
+ }
+ else if(managers[portIndex]==null)
+ {
+ try
+ {
+ lock.lock();
+ }
+ catch(java.lang.InterruptedException e)
+ {
+
+ }
+
+ if(managers[portIndex]==null)
+ managers[portIndex]=createTransactionManager();
+
+ lock.unlock();
+ }
+ }
+
+ private void checkAddress(byte index) {
+ if(subManager==null)
+ {
+ try
+ {
+ lock.lock();
+ }
+ catch(java.lang.InterruptedException e)
+ {
+
+ }
+
+ if(subManager==null)
+ {
+ subManager=new GlobalTransactionManager[256];
+
+ subManager[index]=new
GlobalTransactionManager(scheduler);
+ subManager[index].setMgcpProvider(provider);
+ subManager[index].setNamingService(namingService);
+ subManager[index].setPoolSize(poolSize);
+ }
+
+ lock.unlock();
+ }
+ else if(subManager[index]==null)
+ {
+ try
+ {
+ lock.lock();
+ }
+ catch(java.lang.InterruptedException e)
+ {
+
+ }
+
+ if(subManager[index]==null)
+ {
+ subManager[index]=new
GlobalTransactionManager(scheduler);
+ subManager[index].setMgcpProvider(provider);
+ subManager[index].setNamingService(namingService);
+ subManager[index].setPoolSize(poolSize);
+ }
+
+ lock.unlock();
+ }
+ }
+
+ private class Lock {
+ protected boolean locked;
+ public Lock() {
+ locked=false;
+ }
+
+ public synchronized void lock() throws InterruptedException {
+ while (locked) wait();
+ locked=true;
+ }
+
+ public synchronized void unlock() {
+ locked=false;
+ notify();
+ }
+ }
+}
=======================================
--- /client/jsr-309/tck/server.log Fri Jan 18 02:21:19 2013
+++ /client/jsr-309/tck/server.log Fri Jan 18 07:58:56 2013
@@ -1,6 +1,6 @@
-2013-01-18 11:27:35,600 DEBUG [TestBase] (main) Loading
file:/opt/mobsource/medianew/client/jsr-309/tck/exclude-list.properties
-2013-01-18 11:27:35,602 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 11:27:35,602 DEBUG [TestBase] (main) Getting a MsControlFactory
instance
-2013-01-18 11:27:35,623 INFO [DriverImpl] (main) Initializing MGCP on
/127.0.0.1:2729
-2013-01-18 11:27:35,637 WARN [JainMgcpStackImpl] (main) Failed to locate
properties file, using default values
-2013-01-18 11:27:35,656 INFO [JainMgcpStackImpl] (main) Jain Mgcp stack
bound to IP /127.0.0.1 and UDP port 2729
+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
=======================================
---
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/controller/Controller.java
Fri Jan 18 02:21:19 2013
+++
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/controller/Controller.java
Fri Jan 18 07:58:56 2013
@@ -23,6 +23,7 @@
package org.mobicents.media.server.mgcp.controller;
import java.net.URL;
+import java.net.InetSocketAddress;
import java.io.FileInputStream;
import java.io.IOException;
import org.apache.log4j.Logger;
@@ -32,7 +33,7 @@
import org.mobicents.media.server.mgcp.MgcpProvider;
import org.mobicents.media.server.mgcp.controller.naming.NamingTree;
import org.mobicents.media.server.mgcp.tx.Transaction;
-import org.mobicents.media.server.mgcp.tx.TransactionManager;
+import org.mobicents.media.server.mgcp.tx.GlobalTransactionManager;
import org.mobicents.media.server.scheduler.Scheduler;
import org.mobicents.media.server.spi.Endpoint;
import org.mobicents.media.server.spi.EndpointInstaller;
@@ -71,10 +72,7 @@
//Endpoint configurator
private Configurator configurator;
- //call manager
- protected CallManager callManager = new CallManager();
-
- protected TransactionManager txManager;
+ protected GlobalTransactionManager txManager;
protected int poolSize=10;
@@ -181,10 +179,10 @@
mgcpProvider = new MgcpProvider(udpInterface, port, scheduler);
}
- public void createTransactionManager() {
- txManager = new TransactionManager(scheduler, poolSize);
- txManager.setNamingService(endpoints);
- txManager.setCallManager(callManager);
+ public void createGlobalTransactionManager() {
+ txManager = new GlobalTransactionManager(scheduler);
+ txManager.setPoolSize(poolSize);
+ txManager.setNamingService(endpoints);
txManager.setMgcpProvider(mgcpProvider);
}
@@ -206,7 +204,7 @@
}
//initialize transaction subsystem
- createTransactionManager();
+ createGlobalTransactionManager();
logger.info("Controller started");
}
@@ -224,9 +222,9 @@
int txID = event.getMessage().getTxID();
Transaction tx;
if(event.getEventID()==MgcpEvent.REQUEST)
- tx = txManager.allocateNew(txID);
+ tx =
txManager.allocateNew((InetSocketAddress)event.getAddress(),txID);
else
- tx = txManager.find(txID);
+ tx = txManager.find((InetSocketAddress)event.getAddress(),txID);
if (tx != null)
tx.process(event);
=======================================
---
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/TransactionManager.java
Fri Jan 18 02:21:19 2013
+++
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/tx/TransactionManager.java
Fri Jan 18 07:58:56 2013
@@ -43,7 +43,7 @@
private static java.util.concurrent.atomic.AtomicInteger ID = new
AtomicInteger(1);
//pool of transaction objects
- protected ConcurrentLinkedList<Transaction> pool=new
ConcurrentLinkedList();
+ private ConcurrentLinkedList<Transaction> pool=new
ConcurrentLinkedList();
//currently active transactions.
private ConcurrentHashMap<Integer,Transaction> active;
@@ -56,6 +56,7 @@
//access to naming service
protected NamingTree namingService;
+ //call manager
protected CallManager callManager;
/**
* Creates new transaction's pool.
@@ -68,13 +69,9 @@
active = new ConcurrentHashMap(size);
- initializePool(size);
- }
-
- public void initializePool(int size){
- for (int i = 0; i < size; i++) {
+ for (int i = 0; i < size; i++) {
pool.offer(new Transaction(this));
- }
+ }
}
/**
@@ -107,8 +104,7 @@
public void setCallManager(CallManager callManager) {
this.callManager = callManager;
}
-
-
+
/**
* Assigns MGCP provider.
*
