Revision: d5e0b67bcb26
Author: unknown <yulian@Venera.(none)>
Date: Sat Jan 12 12:37:26 2013
Log: Updated scheduler
Changed to use Unsafe.park and not object.wait
http://code.google.com/p/mediaserver/source/detail?r=d5e0b67bcb26
Modified:
/client/jsr-309/tck/deps/mgcp-driver-3.0.0.FINAL-SNAPSHOT.jar
/client/jsr-309/tck/media/decreasingvolume-recorded.wav
/client/jsr-309/tck/server.log
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/handlers/MessageHandler.java
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/handlers/TransactionHandler.java
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/stack/JainMgcpStackImpl.java
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/utils/ConcurrentLinkedList.java
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/utils/PacketRepresentation.java
/component/src/main/java/org/mobicents/media/server/component/audio/AudioComponent.java
/component/src/main/java/org/mobicents/media/server/component/audio/AudioMixer.java
/component/src/main/java/org/mobicents/media/server/component/audio/AudioSplitter.java
/component/src/main/java/org/mobicents/media/server/component/oob/OOBComponent.java
/component/src/main/java/org/mobicents/media/server/component/oob/OOBMixer.java
/component/src/main/java/org/mobicents/media/server/component/oob/OOBSplitter.java
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/controller/Request.java
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/pkg/au/Play.java
/io/rtp/src/main/java/org/mobicents/media/server/impl/rtp/RTPDataChannel.java
/resources/recorder/src/main/java/org/mobicents/media/server/impl/resource/audio/AudioRecorderImpl.java
/scheduler/src/main/java/org/mobicents/media/server/scheduler/ConcurrentLinkedList.java
/scheduler/src/main/java/org/mobicents/media/server/scheduler/IntConcurrentLinkedList.java
/scheduler/src/main/java/org/mobicents/media/server/scheduler/Scheduler.java
=======================================
--- /client/jsr-309/tck/deps/mgcp-driver-3.0.0.FINAL-SNAPSHOT.jar Tue Jan
1 16:15:51 2013
+++ /client/jsr-309/tck/deps/mgcp-driver-3.0.0.FINAL-SNAPSHOT.jar Sat Jan
12 12:37:26 2013
Binary file, no diff available.
=======================================
--- /client/jsr-309/tck/media/decreasingvolume-recorded.wav Sat Jan 5
09:44:27 2013
+++ /client/jsr-309/tck/media/decreasingvolume-recorded.wav Sat Jan 12
12:37:26 2013
Binary file, no diff available.
=======================================
--- /client/jsr-309/tck/server.log Sat Jan 5 09:44:27 2013
+++ /client/jsr-309/tck/server.log Sat Jan 12 12:37:26 2013
@@ -1,6 +1,6 @@
-2013-01-05 18:58:05,329 DEBUG [TestBase] (main) Loading
file:/opt/mobsource/medianew/client/jsr-309/tck/exclude-list.properties
-2013-01-05 18:58:05,335 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-05 18:58:05,335 DEBUG [TestBase] (main) Getting a MsControlFactory
instance
-2013-01-05 18:58:05,357 INFO [DriverImpl] (main) Initializing MGCP on
/127.0.0.1:2729
-2013-01-05 18:58:05,370 WARN [JainMgcpStackImpl] (main) Failed to locate
properties file, using default values
-2013-01-05 18:58:05,380 INFO [JainMgcpStackImpl] (main) Jain Mgcp stack
bound to IP /127.0.0.1 and UDP port 2729
+2013-01-12 21:52:51,813 DEBUG [TestBase] (main) Loading
file:/opt/mobsource/medianew/client/jsr-309/tck/exclude-list.properties
+2013-01-12 21:52:51,815 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-12 21:52:51,815 DEBUG [TestBase] (main) Getting a MsControlFactory
instance
+2013-01-12 21:52:51,837 INFO [DriverImpl] (main) Initializing MGCP on
/127.0.0.1:2729
+2013-01-12 21:52:51,852 WARN [JainMgcpStackImpl] (main) Failed to locate
properties file, using default values
+2013-01-12 21:52:51,862 INFO [JainMgcpStackImpl] (main) Jain Mgcp stack
bound to IP /127.0.0.1 and UDP port 2729
=======================================
---
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/handlers/MessageHandler.java
Sat Jan 5 09:44:27 2013
+++
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/handlers/MessageHandler.java
Sat Jan 12 12:37:26 2013
@@ -299,10 +299,6 @@
}
finally
{
- pr.setEndTime(System.currentTimeMillis());
- if(logger.isDebugEnabled())
- logger.debug("Tx ID:" + remoteTxIdIntegere + ",Received message time:"
+ pr.getReceiveTime() + ",parse start time:" + pr.getParseTime() + ",end
time:" + pr.getEndTime());
-
pr.release();
}
}
=======================================
---
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/handlers/TransactionHandler.java
Tue Jan 1 16:15:51 2013
+++
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/handlers/TransactionHandler.java
Sat Jan 12 12:37:26 2013
@@ -302,6 +302,8 @@
if(originalPacket!=null)
stack.releasePacket(originalPacket);
+
+ originalPacket=null;
}
/**
@@ -444,7 +446,6 @@
if(originalPacket==null)
originalPacket=stack.allocatePacket();
- originalPacket.setReceiveTime(System.currentTimeMillis());
originalPacket.setLength(encode(event,originalPacket.getRawData()));
InetSocketAddress inetSocketAddress = new InetSocketAddress(address,
port);
originalPacket.setRemoteAddress(inetSocketAddress);
@@ -453,7 +454,7 @@
resetTHISTTimerTask(false);
if (logger.isDebugEnabled())
- logger.debug("Send command event to " + address + ", message\n" + new
String(originalPacket.getRawData(),0,originalPacket.getLength()));
+ logger.debug("Send command event to " + address + "remote TX ID:" +
remoteTID + ", message\n" + new
String(originalPacket.getRawData(),0,originalPacket.getLength()));
countOfCommandRetransmitted++;
stack.send(originalPacket);
@@ -484,7 +485,6 @@
if(originalPacket==null)
originalPacket=stack.allocatePacket();
- originalPacket.setReceiveTime(System.currentTimeMillis());
originalPacket.setLength(encode(event,originalPacket.getRawData()));
InetSocketAddress inetSocketAddress = new
InetSocketAddress(remoteAddress, remotePort);
originalPacket.setRemoteAddress(inetSocketAddress);
=======================================
---
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/stack/JainMgcpStackImpl.java
Sat Jan 5 09:44:27 2013
+++
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/stack/JainMgcpStackImpl.java
Sat Jan 12 12:37:26 2013
@@ -315,7 +315,6 @@
public void send(PacketRepresentation pr)
{
- pr.setParseTime(System.currentTimeMillis());
outputQueue.offer(pr);
}
@@ -355,7 +354,6 @@
receiveBuffer.get(pr.getRawData(), 0, length);
pr.setLength(length);
pr.setRemoteAddress(address);
-
pr.setReceiveTime(System.currentTimeMillis());
inputQueue.offer(pr);
}
} while (this.address != null);
@@ -366,10 +364,6 @@
while((pr=outputQueue.poll())!=null)
{
try {
-
pr.setEndTime(System.currentTimeMillis());
- if(logger.isDebugEnabled())
- logger.debug("Accepted message time:" + pr.getReceiveTime()
+ "parse completed time:" + pr.getParseTime() + ",end time:" +
pr.getEndTime());
-
this.sendBuffer.clear();
this.sendBuffer.put(pr.getRawData(),0,pr.getLength());
this.sendBuffer.flip();
@@ -458,7 +452,6 @@
while(active)
try {
PacketRepresentation current=inputQueue.take();
-
current.setParseTime(System.currentTimeMillis());
messageHandler.scheduleMessages(current);
}
catch(Exception e)
=======================================
---
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/utils/ConcurrentLinkedList.java
Tue Jan 1 16:15:51 2013
+++
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/utils/ConcurrentLinkedList.java
Sat Jan 12 12:37:26 2013
@@ -30,6 +30,7 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.locks.LockSupport;
/**
*
* @author oifa yulian
@@ -145,6 +146,13 @@
releaseAccess();
return false;
}
+
+ if(notifierSize>0)
+ {
+ updateNotifier(value);
+ releaseAccess();
+ return true;
+ }
if(cacheSize==0)
//need new node
@@ -153,8 +161,8 @@
{
//obtain node from cache decrease cache size
tempNode=cacheHead.next;
- cacheHead.next=tempNode.next;
- cacheHead.next.previous=cacheHead;
+ tempNode.next.previous=cacheHead;
+ cacheHead.next=tempNode.next;
tempNode.element=value;
cacheSize--;
@@ -167,10 +175,7 @@
tempNode.next=tail;
tail.previous=tempNode;
size++;
-
- if(notifierSize>0)
- updateNotifier(value);
-
+
releaseAccess();
return true;
}
@@ -200,6 +205,13 @@
return false;
aquireAccess();
+
+ if(notifierSize>0)
+ {
+ updateNotifier(value);
+ releaseAccess();
+ return true;
+ }
if(cacheSize==0)
//need new node
@@ -208,8 +220,8 @@
{
//obtain node from cache decrease cache size
tempNode=cacheHead.next;
- cacheHead.next=tempNode.next;
- cacheHead.next.previous=cacheHead;
+ tempNode.next.previous=cacheHead;
+ cacheHead.next=tempNode.next;
tempNode.element=value;
cacheSize--;
@@ -223,9 +235,6 @@
tail.previous=tempNode;
size++;
- if(notifierSize>0)
- updateNotifier(value);
-
releaseAccess();
return true;
}
@@ -354,42 +363,29 @@
notifierTail.previous.next=tempNotifier;
tempNotifier.previous=notifierTail.previous;
- tempNotifier.next=notifierTail;
+ tempNotifier.next=notifierTail;
notifierTail.previous=tempNotifier;
notifierSize++;
+ tempNotifier.element=null;
+ tempNotifier.waitingThread=Thread.currentThread();
- while(size==0)
- {
- //release lock so others will be able to add
nodes and wait
- releaseAccess();
+ //release lock so others will be able to add nodes and
wait
+ releaseAccess();
+
+ while(tempNotifier.element==null)
+ LockSupport.park();
- synchronized(tempNotifier.lock)
- {
- try
- {
- tempNotifier.lock.wait();
- }
- catch(InterruptedException e)
- {
-
- }
- }
-
- aquireAccess();
-
- if(size!=0)
- {
- //take back notifier
-
tempNotifier.previous.next=tempNotifier.next;
-
tempNotifier.next.previous=tempNotifier.previous;
-
-
tempNotifier.next=notifierCacheHead.next;
-
notifierCacheHead.next.previous=tempNotifier;
- notifierCacheHead.next=tempNotifier;
- tempNotifier.previous=notifierCacheHead;
- cacheNotifierSize++;
- }
- }
+ aquireAccess();
+
+ tempNotifier.next=notifierCacheHead.next;
+ notifierCacheHead.next.previous=tempNotifier;
+ notifierCacheHead.next=tempNotifier;
+ tempNotifier.previous=notifierCacheHead;
+ cacheNotifierSize++;
+
+ E result=tempNotifier.element;
+ releaseAccess();
+ return result;
}
//disconnect from main list
@@ -439,57 +435,59 @@
tempNotifier.previous=notifierTail.previous;
tempNotifier.next=notifierTail;
- notifierTail.previous=tempNotifier;
+ notifierTail.previous=tempNotifier;
notifierSize++;
- while(size==0)
+ tempNotifier.element=null;
+ tempNotifier.waitingThread=Thread.currentThread();
+
+ //release lock so others will be able to add nodes and
wait
+ releaseAccess();
+
+ switch(unit)
{
- //release lock so others will be able to add
nodes and wait
- releaseAccess();
+ case SECONDS:
+ TimeUnit.SECONDS.toNanos(timeout);
+ break;
+ case MILLISECONDS:
+ TimeUnit.MILLISECONDS.toNanos(timeout);
+ break;
+ case MICROSECONDS:
+ TimeUnit.MICROSECONDS.toNanos(timeout);
+ break;
+ case NANOSECONDS:
+ LockSupport.parkNanos(timeout);
+ break;
+ }
- synchronized(tempNotifier.lock)
- {
- try {
- switch(unit)
- {
- case MICROSECONDS:
-
tempNotifier.lock.wait(timeout/1000L,(int)((timeout*1000L)%1000000L));
- break;
- case MILLISECONDS:
-
tempNotifier.lock.wait(timeout);
- break;
- case NANOSECONDS:
-
tempNotifier.lock.wait(timeout/1000000L,(int)(timeout%1000000L));
- break;
- case SECONDS:
-
tempNotifier.lock.wait(timeout*1000L);
- break;
- }
- }
- catch(InterruptedException e)
- {
-
- }
- }
+ aquireAccess();
- aquireAccess();
+ if(tempNotifier.element==null)
+ {
+ //take back notifier
+ tempNotifier.previous.next=tempNotifier.next;
+
tempNotifier.next.previous=tempNotifier.previous;
+
+ tempNotifier.next=notifierCacheHead.next;
+ notifierCacheHead.next.previous=tempNotifier;
+ notifierCacheHead.next=tempNotifier;
+ tempNotifier.previous=notifierCacheHead;
+ cacheNotifierSize++;
- if(size==0)
- {
- releaseAccess();
- throw new InterruptedException();
- }
+ releaseAccess();
+ throw new InterruptedException();
}
-
- //take back notifier
- tempNotifier.previous.next=tempNotifier.next;
- tempNotifier.next.previous=tempNotifier.previous;
tempNotifier.next=notifierCacheHead.next;
notifierCacheHead.next.previous=tempNotifier;
notifierCacheHead.next=tempNotifier;
tempNotifier.previous=notifierCacheHead;
cacheNotifierSize++;
+
+ E result=tempNotifier.element;
+ releaseAccess();
+
+ return result;
}
//disconnect from main list
@@ -715,15 +713,15 @@
private void updateNotifier(E newElement)
{
- if(notifierSize==0)
- return;
-
notifierSize--;
- Notifier tempNotifier=notifierHead.next;
- synchronized(tempNotifier.lock)
- {
- tempNotifier.lock.notify();
- }
+ Notifier tempNotifier=notifierHead.next;
+
+ //take back notifier
+ notifierHead.next=tempNotifier.next;
+ notifierHead.next.previous=notifierHead;
+
+ tempNotifier.element=newElement;
+ LockSupport.unpark(tempNotifier.waitingThread);
}
private void aquireAccess()
@@ -744,9 +742,10 @@
}
private class Notifier<E> {
- private Object lock=new Object();
Notifier next;
Notifier previous;
+ E element;
+ Thread waitingThread;
}
private class Node<E> {
=======================================
---
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/utils/PacketRepresentation.java
Tue Jan 1 16:15:51 2013
+++
/client/mgcp/driver/src/main/java/org/mobicents/protocols/mgcp/utils/PacketRepresentation.java
Sat Jan 12 12:37:26 2013
@@ -47,9 +47,6 @@
private byte[] rawData = null;
private InetSocketAddress remoteAddress = null;
private int length = 0;
- private long receiveTime=0L;
- private long parseTime=0L;
- private long endTime=0L;
private PacketRepresentationFactory prFactory = null;
@@ -86,37 +83,9 @@
this.remoteAddress = remoteAddress;
}
- public long getReceiveTime() {
- return this.receiveTime;
- }
-
- public void setReceiveTime(long receiveTime) {
- this.receiveTime = receiveTime;
- }
-
- public long getParseTime() {
- return this.parseTime;
- }
-
- public void setParseTime(long parseTime) {
- this.parseTime = parseTime;
- }
-
- public long getEndTime() {
- return this.endTime;
- }
-
- public void setEndTime(long endTime) {
- this.endTime = endTime;
- }
-
public void release() {
this.prFactory.deallocate(this);
this.remoteAddress=null;
- this.length=0;
- this.receiveTime=0L;
- this.parseTime=0L;
- this.endTime=0L;
+ this.length=0;
}
-
}
=======================================
---
/component/src/main/java/org/mobicents/media/server/component/audio/AudioComponent.java
Tue Jan 1 16:15:51 2013
+++
/component/src/main/java/org/mobicents/media/server/component/audio/AudioComponent.java
Sat Jan 12 12:37:26 2013
@@ -43,7 +43,10 @@
private IntConcurrentLinkedList<AudioInput> inputs = new
IntConcurrentLinkedList();
private IntConcurrentLinkedList<AudioOutput> outputs = new
IntConcurrentLinkedList();
-
+
+ private Iterator<AudioInput> activeInputs=inputs.iterator();
+ private Iterator<AudioOutput> activeOutputs=outputs.iterator();
+
protected Boolean shouldRead=false;
protected Boolean shouldWrite=false;
@@ -97,9 +100,9 @@
public void perform()
{
- first=true;
- Iterator<AudioInput> activeInputs=inputs.iterator();
- while(activeInputs.hasNext())
+ first=true;
+ inputs.resetIterator(activeInputs);
+ while(activeInputs.hasNext())
{
AudioInput input=activeInputs.next();
frame=input.poll();
@@ -157,8 +160,8 @@
frame.setDuration(period);
frame.setFormat(format);
- Iterator<AudioOutput> activeOutputs=outputs.iterator();
- while(activeOutputs.hasNext())
+ outputs.resetIterator(activeOutputs);
+ while(activeOutputs.hasNext())
{
AudioOutput output=activeOutputs.next();
if(!activeOutputs.hasNext())
=======================================
---
/component/src/main/java/org/mobicents/media/server/component/audio/AudioMixer.java
Tue Jan 1 16:15:51 2013
+++
/component/src/main/java/org/mobicents/media/server/component/audio/AudioMixer.java
Sat Jan 12 12:37:26 2013
@@ -48,6 +48,8 @@
//The pool of components
private IntConcurrentLinkedList<AudioComponent> components = new
IntConcurrentLinkedList();
+ Iterator<AudioComponent> activeComponents=components.iterator();
+
private long period = 20000000L;
private int packetSize = (int)(period / 1000000) *
format.getSampleRate()/1000 * format.getSampleSize() / 8;
@@ -124,7 +126,7 @@
public long perform() {
//summarize all
sourcesCount=0;
- Iterator<AudioComponent>
activeComponents=components.iterator();
+ components.resetIterator(activeComponents);
while(activeComponents.hasNext())
{
AudioComponent component=activeComponents.next();
@@ -171,7 +173,7 @@
total[i]=(short)Math.round((double) total[i] *
currGain);
//get data for each component
- activeComponents=components.iterator();
+ components.resetIterator(activeComponents);
while(activeComponents.hasNext())
{
AudioComponent component=activeComponents.next();
=======================================
---
/component/src/main/java/org/mobicents/media/server/component/audio/AudioSplitter.java
Tue Jan 1 16:15:51 2013
+++
/component/src/main/java/org/mobicents/media/server/component/audio/AudioSplitter.java
Sat Jan 12 12:37:26 2013
@@ -49,6 +49,11 @@
private IntConcurrentLinkedList<AudioComponent> insideComponents = new
IntConcurrentLinkedList();
private IntConcurrentLinkedList<AudioComponent> outsideComponents =
new IntConcurrentLinkedList();
+ private Iterator<AudioComponent>
insideRIterator=insideComponents.iterator();
+ private Iterator<AudioComponent>
insideSIterator=insideComponents.iterator();
+
+ private Iterator<AudioComponent>
outsideRIterator=outsideComponents.iterator();
+ private Iterator<AudioComponent>
outsideSIterator=outsideComponents.iterator();
private long period = 20000000L;
private int packetSize = (int)(period / 1000000) *
format.getSampleRate()/1000 * format.getSampleSize() / 8;
@@ -144,10 +149,10 @@
public long perform() {
//summarize all
first=true;
- Iterator<AudioComponent>
activeComponents=insideComponents.iterator();
- while(activeComponents.hasNext())
+ insideComponents.resetIterator(insideRIterator);
+ while(insideRIterator.hasNext())
{
- AudioComponent component=activeComponents.next();
+ AudioComponent component=insideRIterator.next();
component.perform();
current=component.getData();
if(current!=null)
@@ -194,10 +199,10 @@
total[i]=(short)Math.round((double) total[i] *
currGain);
//get data for each component
- activeComponents=outsideComponents.iterator();
- while(activeComponents.hasNext())
+ outsideComponents.resetIterator(outsideSIterator);
+ while(outsideSIterator.hasNext())
{
- AudioComponent component=activeComponents.next();
+ AudioComponent component=outsideSIterator.next();
component.offer(total);
}
@@ -230,10 +235,10 @@
public long perform() {
//summarize all
first=true;
- Iterator<AudioComponent>
activeComponents=outsideComponents.iterator();
- while(activeComponents.hasNext())
+ outsideComponents.resetIterator(outsideRIterator);
+ while(outsideRIterator.hasNext())
{
- AudioComponent component=activeComponents.next();
+ AudioComponent component=outsideRIterator.next();
component.perform();
current=component.getData();
if(current!=null)
@@ -278,10 +283,10 @@
total[i]=(short)Math.round((double) total[i] *
currGain);
//get data for each component
- activeComponents=insideComponents.iterator();
- while(activeComponents.hasNext())
+ insideComponents.resetIterator(insideSIterator);
+ while(insideSIterator.hasNext())
{
- AudioComponent component=activeComponents.next();
+ AudioComponent component=insideSIterator.next();
component.offer(total);
}
=======================================
---
/component/src/main/java/org/mobicents/media/server/component/oob/OOBComponent.java
Fri Jan 4 03:50:24 2013
+++
/component/src/main/java/org/mobicents/media/server/component/oob/OOBComponent.java
Sat Jan 12 12:37:26 2013
@@ -42,6 +42,9 @@
private IntConcurrentLinkedList<OOBInput> inputs = new
IntConcurrentLinkedList();
private IntConcurrentLinkedList<OOBOutput> outputs = new
IntConcurrentLinkedList();
+ Iterator<OOBInput> activeInputs=inputs.iterator();
+ Iterator<OOBOutput> activeOutputs=outputs.iterator();
+
protected Boolean shouldRead=false;
protected Boolean shouldWrite=false;
@@ -87,8 +90,8 @@
public void perform()
{
frame=null;
- Iterator<OOBInput> activeInputs=inputs.iterator();
- while(activeInputs.hasNext())
+ inputs.resetIterator(activeInputs);
+ while(activeInputs.hasNext())
{
OOBInput input=activeInputs.next();
frame=input.poll();
@@ -118,7 +121,7 @@
return;
}
- Iterator<OOBOutput> activeOutputs=outputs.iterator();
+ outputs.resetIterator(activeOutputs);
while(activeOutputs.hasNext())
{
OOBOutput output=activeOutputs.next();
=======================================
---
/component/src/main/java/org/mobicents/media/server/component/oob/OOBMixer.java
Fri Jan 4 03:50:24 2013
+++
/component/src/main/java/org/mobicents/media/server/component/oob/OOBMixer.java
Sat Jan 12 12:37:26 2013
@@ -45,6 +45,8 @@
//The pool of components
private IntConcurrentLinkedList<OOBComponent> components = new
IntConcurrentLinkedList();
+ Iterator<OOBComponent> activeComponents=components.iterator();
+
private MixTask mixer;
private volatile boolean started = false;
@@ -99,8 +101,8 @@
public long perform() {
//summarize all
- Iterator<OOBComponent> activeComponents=components.iterator();
- while(activeComponents.hasNext())
+ components.resetIterator(activeComponents);
+ while(activeComponents.hasNext())
{
OOBComponent component=activeComponents.next();
component.perform();
@@ -120,8 +122,8 @@
}
//get data for each component
- activeComponents=components.iterator();
- while(activeComponents.hasNext())
+ components.resetIterator(activeComponents);
+ while(activeComponents.hasNext())
{
OOBComponent component=activeComponents.next();
if(component.getComponentId()!=sourceComponent)
=======================================
---
/component/src/main/java/org/mobicents/media/server/component/oob/OOBSplitter.java
Fri Jan 4 03:50:24 2013
+++
/component/src/main/java/org/mobicents/media/server/component/oob/OOBSplitter.java
Sat Jan 12 12:37:26 2013
@@ -46,6 +46,12 @@
private IntConcurrentLinkedList<OOBComponent> insideComponents = new
IntConcurrentLinkedList();
private IntConcurrentLinkedList<OOBComponent> outsideComponents = new
IntConcurrentLinkedList();
+ private Iterator<OOBComponent>
insideRIterator=insideComponents.iterator();
+ private Iterator<OOBComponent>
insideSIterator=insideComponents.iterator();
+
+ private Iterator<OOBComponent>
outsideRIterator=outsideComponents.iterator();
+ private Iterator<OOBComponent>
outsideSIterator=outsideComponents.iterator();
+
private InsideMixTask insideMixer;
private OutsideMixTask outsideMixer;
private volatile boolean started = false;
@@ -115,10 +121,10 @@
public long perform() {
//summarize all
current=null;
- Iterator<OOBComponent>
activeComponents=insideComponents.iterator();
- while(activeComponents.hasNext())
+ insideComponents.resetIterator(insideRIterator);
+ while(insideRIterator.hasNext())
{
- OOBComponent component=activeComponents.next();
+ OOBComponent component=insideRIterator.next();
component.perform();
current=component.getData();
if(current!=null)
@@ -133,11 +139,12 @@
}
//get data for each component
- activeComponents=outsideComponents.iterator();
- while(activeComponents.hasNext())
+ outsideComponents.resetIterator(outsideSIterator);
+
+ while(outsideSIterator.hasNext())
{
- OOBComponent component=activeComponents.next();
- if(!activeComponents.hasNext())
+ OOBComponent component=outsideSIterator.next();
+ if(!outsideSIterator.hasNext())
component.offer(current);
else
component.offer(current.clone());
@@ -165,10 +172,10 @@
public long perform() {
//summarize all
current=null;
- Iterator<OOBComponent>
activeComponents=outsideComponents.iterator();
- while(activeComponents.hasNext())
+ outsideComponents.resetIterator(outsideRIterator);
+ while(outsideRIterator.hasNext())
{
- OOBComponent component=activeComponents.next();
+ OOBComponent component=outsideRIterator.next();
component.perform();
current=component.getData();
if(current!=null)
@@ -183,11 +190,11 @@
}
//get data for each component
- activeComponents=insideComponents.iterator();
- while(activeComponents.hasNext())
+ insideComponents.resetIterator(insideSIterator);
+ while(insideSIterator.hasNext())
{
- OOBComponent component=activeComponents.next();
- if(!activeComponents.hasNext())
+ OOBComponent component=insideSIterator.next();
+ if(!insideSIterator.hasNext())
component.offer(current);
else
component.offer(current.clone());
=======================================
---
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/controller/Request.java
Wed Jul 25 12:00:02 2012
+++
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/controller/Request.java
Sat Jan 12 12:37:26 2013
@@ -139,8 +139,10 @@
public void execute() {
if (!executors.isEmpty()) {
currentSignal = executors.poll();
- currentSignal.execute();
- }
+
+ if(currentSignal!=null)
+ currentSignal.execute();
+ }
}
public void onEvent(Text event) {
=======================================
---
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/pkg/au/Play.java
Sat Sep 15 16:38:09 2012
+++
/controls/mgcp/src/main/java/org/mobicents/media/server/mgcp/pkg/au/Play.java
Sat Jan 12 12:37:26 2013
@@ -172,10 +172,11 @@
}
private void terminate() {
- if (player != null) {
- player.removeListener(this);
- player.deactivate();
- player=null;
+ Player currPlayer=player;
+ if (currPlayer != null) {
+ currPlayer.removeListener(this);
+ currPlayer.deactivate();
+ player=null;
}
}
=======================================
---
/io/rtp/src/main/java/org/mobicents/media/server/impl/rtp/RTPDataChannel.java
Fri Jan 4 07:17:31 2013
+++
/io/rtp/src/main/java/org/mobicents/media/server/impl/rtp/RTPDataChannel.java
Sat Jan 12 12:37:26 2013
@@ -335,7 +335,7 @@
dataChannel.disconnect();
}
catch (IOException e) {
- logger.error(e);
+ logger.error(e);
}
connectImmediately=udpManager.connectImmediately((InetSocketAddress)address);
@@ -343,8 +343,9 @@
try {
dataChannel.connect(address);
}
- catch (IOException e) {
- logger.error(e);
+ catch (IOException e) {
+ logger.info("Can not connect to remote address , please check
that you are not using local address - 127.0.0.X to connect to remote");
+ logger.error(e);
}
}
@@ -368,14 +369,14 @@
dataChannel.disconnect();
}
catch(IOException e) {
- logger.error(e);
+ logger.error(e);
}
try {
dataChannel.socket().close();
dataChannel.close();
} catch(IOException e) {
- logger.error(e);
+ logger.error(e);
}
if (controlChannel != null) {
@@ -557,7 +558,7 @@
}
}
catch (Exception e) {
- logger.error(e);
+ logger.error(e);
}
}
@@ -591,11 +592,11 @@
dataChannel.disconnect();
}
catch(IOException ex) {
- logger.error(ex);
+ logger.error(ex);
}
}
catch (IOException e) {
- logger.error(e);
+ logger.error(e);
}
while (currAddress != null) {
@@ -638,11 +639,11 @@
dataChannel.disconnect();
}
catch(IOException ex) {
- logger.error(ex);
+ logger.error(ex);
}
}
catch (Exception e) {
- logger.error(e);
+ logger.error(e);
}
rtpHandler.isReading = false;
@@ -708,11 +709,11 @@
dataChannel.disconnect();
}
catch(IOException ex) {
- logger.error(ex);
+ logger.error(ex);
}
}
catch (Exception e) {
- logger.error(e);
+ logger.error(e);
}
}
@@ -764,11 +765,11 @@
dataChannel.disconnect();
}
catch(IOException ex) {
- logger.error(ex);
+ logger.error(ex);
}
}
catch (Exception e) {
- logger.error(e);
+ logger.error(e);
}
}
}
=======================================
---
/resources/recorder/src/main/java/org/mobicents/media/server/impl/resource/audio/AudioRecorderImpl.java
Fri Jan 4 07:17:31 2013
+++
/resources/recorder/src/main/java/org/mobicents/media/server/impl/resource/audio/AudioRecorderImpl.java
Sat Jan 12 12:37:26 2013
@@ -252,9 +252,7 @@
}
}
else
- this.lastPacketData=scheduler.getClock().getTime();
-
-
+ this.lastPacketData=scheduler.getClock().getTime();
}
public void setRecordDir(String recordDir) {
=======================================
---
/scheduler/src/main/java/org/mobicents/media/server/scheduler/ConcurrentLinkedList.java
Sun Feb 26 13:04:30 2012
+++
/scheduler/src/main/java/org/mobicents/media/server/scheduler/ConcurrentLinkedList.java
Sat Jan 12 12:37:26 2013
@@ -30,6 +30,7 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.locks.LockSupport;
/**
*
* @author oifa yulian
@@ -145,6 +146,13 @@
releaseAccess();
return false;
}
+
+ if(notifierSize>0)
+ {
+ updateNotifier(value);
+ releaseAccess();
+ return true;
+ }
if(cacheSize==0)
//need new node
@@ -153,8 +161,8 @@
{
//obtain node from cache decrease cache size
tempNode=cacheHead.next;
- cacheHead.next=tempNode.next;
- cacheHead.next.previous=cacheHead;
+ tempNode.next.previous=cacheHead;
+ cacheHead.next=tempNode.next;
tempNode.element=value;
cacheSize--;
@@ -167,10 +175,7 @@
tempNode.next=tail;
tail.previous=tempNode;
size++;
-
- if(notifierSize>0)
- updateNotifier(value);
-
+
releaseAccess();
return true;
}
@@ -200,6 +205,13 @@
return false;
aquireAccess();
+
+ if(notifierSize>0)
+ {
+ updateNotifier(value);
+ releaseAccess();
+ return true;
+ }
if(cacheSize==0)
//need new node
@@ -208,8 +220,8 @@
{
//obtain node from cache decrease cache size
tempNode=cacheHead.next;
- cacheHead.next=tempNode.next;
- cacheHead.next.previous=cacheHead;
+ tempNode.next.previous=cacheHead;
+ cacheHead.next=tempNode.next;
tempNode.element=value;
cacheSize--;
@@ -223,9 +235,6 @@
tail.previous=tempNode;
size++;
- if(notifierSize>0)
- updateNotifier(value);
-
releaseAccess();
return true;
}
@@ -354,42 +363,29 @@
notifierTail.previous.next=tempNotifier;
tempNotifier.previous=notifierTail.previous;
- tempNotifier.next=notifierTail;
+ tempNotifier.next=notifierTail;
notifierTail.previous=tempNotifier;
notifierSize++;
+ tempNotifier.element=null;
+ tempNotifier.waitingThread=Thread.currentThread();
- while(size==0)
- {
- //release lock so others will be able to add
nodes and wait
- releaseAccess();
+ //release lock so others will be able to add nodes and
wait
+ releaseAccess();
+
+ while(tempNotifier.element==null)
+ LockSupport.park();
- synchronized(tempNotifier.lock)
- {
- try
- {
- tempNotifier.lock.wait();
- }
- catch(InterruptedException e)
- {
-
- }
- }
-
- aquireAccess();
-
- if(size!=0)
- {
- //take back notifier
-
tempNotifier.previous.next=tempNotifier.next;
-
tempNotifier.next.previous=tempNotifier.previous;
-
-
tempNotifier.next=notifierCacheHead.next;
-
notifierCacheHead.next.previous=tempNotifier;
- notifierCacheHead.next=tempNotifier;
- tempNotifier.previous=notifierCacheHead;
- cacheNotifierSize++;
- }
- }
+ aquireAccess();
+
+ tempNotifier.next=notifierCacheHead.next;
+ notifierCacheHead.next.previous=tempNotifier;
+ notifierCacheHead.next=tempNotifier;
+ tempNotifier.previous=notifierCacheHead;
+ cacheNotifierSize++;
+
+ E result=tempNotifier.element;
+ releaseAccess();
+ return result;
}
//disconnect from main list
@@ -439,57 +435,59 @@
tempNotifier.previous=notifierTail.previous;
tempNotifier.next=notifierTail;
- notifierTail.previous=tempNotifier;
+ notifierTail.previous=tempNotifier;
notifierSize++;
- while(size==0)
+ tempNotifier.element=null;
+ tempNotifier.waitingThread=Thread.currentThread();
+
+ //release lock so others will be able to add nodes and
wait
+ releaseAccess();
+
+ switch(unit)
{
- //release lock so others will be able to add
nodes and wait
- releaseAccess();
+ case SECONDS:
+ TimeUnit.SECONDS.toNanos(timeout);
+ break;
+ case MILLISECONDS:
+ TimeUnit.MILLISECONDS.toNanos(timeout);
+ break;
+ case MICROSECONDS:
+ TimeUnit.MICROSECONDS.toNanos(timeout);
+ break;
+ case NANOSECONDS:
+ LockSupport.parkNanos(timeout);
+ break;
+ }
- synchronized(tempNotifier.lock)
- {
- try {
- switch(unit)
- {
- case MICROSECONDS:
-
tempNotifier.lock.wait(timeout/1000L,(int)((timeout*1000L)%1000000L));
- break;
- case MILLISECONDS:
-
tempNotifier.lock.wait(timeout);
- break;
- case NANOSECONDS:
-
tempNotifier.lock.wait(timeout/1000000L,(int)(timeout%1000000L));
- break;
- case SECONDS:
-
tempNotifier.lock.wait(timeout*1000L);
- break;
- }
- }
- catch(InterruptedException e)
- {
-
- }
- }
+ aquireAccess();
- aquireAccess();
+ if(tempNotifier.element==null)
+ {
+ //take back notifier
+ tempNotifier.previous.next=tempNotifier.next;
+
tempNotifier.next.previous=tempNotifier.previous;
+
+ tempNotifier.next=notifierCacheHead.next;
+ notifierCacheHead.next.previous=tempNotifier;
+ notifierCacheHead.next=tempNotifier;
+ tempNotifier.previous=notifierCacheHead;
+ cacheNotifierSize++;
- if(size==0)
- {
- releaseAccess();
- throw new InterruptedException();
- }
+ releaseAccess();
+ throw new InterruptedException();
}
-
- //take back notifier
- tempNotifier.previous.next=tempNotifier.next;
- tempNotifier.next.previous=tempNotifier.previous;
tempNotifier.next=notifierCacheHead.next;
notifierCacheHead.next.previous=tempNotifier;
notifierCacheHead.next=tempNotifier;
tempNotifier.previous=notifierCacheHead;
cacheNotifierSize++;
+
+ E result=tempNotifier.element;
+ releaseAccess();
+
+ return result;
}
//disconnect from main list
@@ -715,15 +713,15 @@
private void updateNotifier(E newElement)
{
- if(notifierSize==0)
- return;
-
notifierSize--;
- Notifier tempNotifier=notifierHead.next;
- synchronized(tempNotifier.lock)
- {
- tempNotifier.lock.notify();
- }
+ Notifier tempNotifier=notifierHead.next;
+
+ //take back notifier
+ notifierHead.next=tempNotifier.next;
+ notifierHead.next.previous=notifierHead;
+
+ tempNotifier.element=newElement;
+ LockSupport.unpark(tempNotifier.waitingThread);
}
private void aquireAccess()
@@ -744,9 +742,10 @@
}
private class Notifier<E> {
- private Object lock=new Object();
Notifier next;
Notifier previous;
+ E element;
+ Thread waitingThread;
}
private class Node<E> {
=======================================
---
/scheduler/src/main/java/org/mobicents/media/server/scheduler/IntConcurrentLinkedList.java
Fri Feb 17 05:07:24 2012
+++
/scheduler/src/main/java/org/mobicents/media/server/scheduler/IntConcurrentLinkedList.java
Sat Jan 12 12:37:26 2013
@@ -49,8 +49,6 @@
private Integer size=0;
private Integer cacheSize=0;
- private ListIterator iterator=new ListIterator();
-
private Lock lock=new Lock();
public IntConcurrentLinkedList()
@@ -235,9 +233,15 @@
public Iterator<E> iterator()
{
+ ListIterator iterator=new ListIterator();
iterator.currNode=head;
return iterator;
}
+
+ public void resetIterator(Iterator<E> iterator)
+ {
+ ((ListIterator)iterator).currNode=head;
+ }
private void aquireAccess()
{
=======================================
---
/scheduler/src/main/java/org/mobicents/media/server/scheduler/Scheduler.java
Sun Nov 25 14:18:52 2012
+++
/scheduler/src/main/java/org/mobicents/media/server/scheduler/Scheduler.java
Sat Jan 12 12:37:26 2013
@@ -73,7 +73,7 @@
//priority queue
protected OrderedTaskQueue[] taskQueues = new OrderedTaskQueue[7];
- protected OrderedTaskQueue heartBeatQueue;
+ protected OrderedTaskQueue[] heartBeatQueue = new OrderedTaskQueue[5];
//CPU bound threads
private CoreThread coreThread;
@@ -96,7 +96,8 @@
for(int i=0;i<taskQueues.length;i++)
taskQueues[i]=new OrderedTaskQueue();
- heartBeatQueue=new OrderedTaskQueue();
+ for(int i=0;i<heartBeatQueue.length;i++)
+ heartBeatQueue[i]=new OrderedTaskQueue();
coreThread = new CoreThread(String.format("Scheduler"));
criticalThread = new CriticalThread(String.format("Scheduler"));
@@ -150,7 +151,7 @@
*/
public void submitHeatbeat(Task task) {
task.activate(true);
- heartBeatQueue.accept(task);
+ heartBeatQueue[coreThread.runIndex].accept(task);
}
/**
@@ -215,7 +216,8 @@
for(int i=0;i<taskQueues.length;i++)
taskQueues[i].clear();
- heartBeatQueue.clear();
+ for(int i=0;i<heartBeatQueue.length;i++)
+ heartBeatQueue[i].clear();
}
//removed statistics to increase perfomance
@@ -293,16 +295,15 @@
}
runIndex=(runIndex+1)%5;
- if(runIndex==0)
- synchronized(LOCK) {
- if(executeQueue(heartBeatQueue))
- try {
- LOCK.wait();
- }
- catch(InterruptedException e) {
- //lets continue
- }
- }
+ synchronized(LOCK) {
+
if(executeQueue(heartBeatQueue[runIndex]))
+ try {
+ LOCK.wait();
+ }
+ catch(InterruptedException e) {
+ //lets continue
+ }
+ }
synchronized(LOCK) {
if(executeQueue(taskQueues[MANAGEMENT_QUEUE]))
@@ -455,7 +456,7 @@
while(active)
{
waitingTasks.take().run();
- coreThread.notifyCompletion();
+ coreThread.notifyCompletion();
}
}
