Implement finer synchronization granularity for portaudio streams

Use dedicated syncronization object in InputStream and MasterStream to
implement a synchronization policy with finer granularity. This
speeds up processing if SC uses several InputStream, for example several
calls that are not on hold.

src/net/java/sip/communicator/impl/neomedia/portaudio/streams/InputPortAudioStream.java

@@ -66,6 +66,7 @@ public InputPortAudioStream(MasterPortAudioStream parentStream)
         this.parentStream = parentStream;
     }
 
+    private Object readSync = new Object();
     /**
      * Reads audio data from this <tt>InputPortAudioStream</tt> into a specific
      * <tt>Buffer</tt> blocking until audio data is indeed available.
@@ -82,41 +83,53 @@ public void read(Buffer buffer)
             buffer.setLength(0);
             return;
         }
-
-        synchronized (parentStream)
+        
+        synchronized (readSync)
         {
-            if (bufferData == null)
-            {
-                parentStream.read(buffer);
-            }
-            else
+            while (true)
             {
-                /*
-                 * If buffer conatins a data area then check if the type and 
-                 * length fits. If yes then use it, otherwise allocate a new
-                 * area and set it in buffer.
-                 * 
-                 * In case we can re-use the data area: copy the data, don't
-                 * just set the bufferData into buffer because bufferData 
-                 * points to a data area in another buffer instance.
-                 */
-                Object data = buffer.getData();
-                byte[] tmpArray;
-                if (data instanceof byte[] && ((byte[])data).length >= bufferLength) {
-                    tmpArray = (byte[])data;
+                if (bufferData == null) {
+                    // parent read returns false if read is already active, wait
+                    // until notified by setBuffer below.
+                    if (!parentStream.read(buffer)) {
+                        try {
+                            readSync.wait();
+                        } catch (InterruptedException e) {
+                            continue;
+                        }
+                        continue;  // bufferData was set by setBuffer
+                    }
+                    // parent read returned true, break loop below
                 }
                 else
                 {
-                    tmpArray = new byte[bufferLength];
-                    buffer.setData(tmpArray);
+                    /*
+                     * If buffer conatins a data area then check if the type and 
+                     * length fits. If yes then use it, otherwise allocate a new
+                     * area and set it in buffer.
+                     * 
+                     * In case we can re-use the data area: copy the data, don't
+                     * just set the bufferData into buffer because bufferData 
+                     * points to a data area in another buffer instance.
+                     */
+                    Object data = buffer.getData();
+                    byte[] tmpArray;
+                    if (data instanceof byte[] && ((byte[])data).length >= bufferLength) {
+                        tmpArray = (byte[])data;
+                    }
+                    else
+                    {
+                        tmpArray = new byte[bufferLength];
+                        buffer.setData(tmpArray);
+                    }
+                    System.arraycopy(bufferData, 0, tmpArray, 0, bufferLength);
+                    buffer.setFlags(Buffer.FLAG_SYSTEM_TIME);
+                    buffer.setLength(bufferLength);
+                    buffer.setOffset(0);
+                    buffer.setTimeStamp(bufferTimeStamp);
                 }
-                System.arraycopy(bufferData, 0, tmpArray, 0, bufferLength);
-                buffer.setFlags(Buffer.FLAG_SYSTEM_TIME);
-                buffer.setLength(bufferLength);
-                buffer.setOffset(0);
-                buffer.setTimeStamp(bufferTimeStamp);
+                break;
             }
-
             /*
              * The bufferData of this InputPortAudioStream has been consumed so
              * make sure a new piece of audio data will be read the next time.
@@ -142,11 +155,12 @@ public void read(Buffer buffer)
      */
     void setBuffer(byte[] bufferData, int bufferLength, long bufferTimeStamp)
     {
-        synchronized (parentStream)
+        synchronized (readSync)
         {
             this.bufferData = bufferData;
             this.bufferLength = bufferLength;
             this.bufferTimeStamp = bufferTimeStamp;
+            readSync.notifyAll();
         }
     }
 

src/net/java/sip/communicator/impl/neomedia/portaudio/streams/MasterPortAudioStream.java

@@ -79,6 +79,10 @@ public class MasterPortAudioStream
      */
     private long stream = 0;
 
+    /**
+     * Whether a read is active.
+     */
+    private boolean readActive = false;
     /**
      * Creates new stream.
      * @param deviceIndex the device to use.
@@ -144,6 +148,8 @@ private void initStream()
             null);
     }
 
+    private Object readSync = new Object();
+    
     /**
      * Reads audio data from this <tt>MasterPortAudioStream</tt> into a specific
      * <tt>Buffer</tt> blocking until audio data is indeed available.
@@ -151,14 +157,21 @@ private void initStream()
      * @param buffer the <tt>Buffer</tt> into which the audio data read from
      * this <tt>MasterPortAudioStream</tt> is to be returned
      * @throws PortAudioException if an error occurs while reading
-     */
-    public synchronized void read(Buffer buffer)
+     */    
+    public boolean read(Buffer buffer)
         throws PortAudioException
     {
-        if (!started)
-        {
-            buffer.setLength(0);
-            return;
+        synchronized (readSync) {
+            if (readActive)
+            {
+                return false;
+            }
+            readActive = true;
+            if (!started)
+            {
+                buffer.setLength(0);
+                return true;
+            }
         }
 
         /*
@@ -197,6 +210,11 @@ public synchronized void read(Buffer buffer)
                 .get(slaveIndex)
                     .setBuffer(bufferData, bytesPerBuffer, bufferTimeStamp);
         }
+        synchronized(readSync) {
+            readActive = false;
+            readSync.notify();
+        }
+        return true;
     }
 
     /**
@@ -281,11 +299,21 @@ synchronized void stop(InputPortAudioStream slave)
 
         if(slaves.isEmpty())
         {
-            // stop
-            PortAudio.Pa_CloseStream(stream);
-            stream = 0;
-            started = false;
-            PortAudioManager.getInstance().stoppedInputPortAudioStream(this);
+            synchronized (readSync) {
+                while (readActive) 
+                {
+                    try {
+                        readSync.wait();
+                    } catch (InterruptedException e) {
+                        continue;
+                    }
+                }
+                // stop
+                PortAudio.Pa_CloseStream(stream);
+                stream = 0;
+                started = false;
+                PortAudioManager.getInstance().stoppedInputPortAudioStream(this);
+            }
         }
     }
 }