From a877afccebf3a4ece6bc34eacbbcffd76400802f Mon Sep 17 00:00:00 2001 From: Denny Date: Fri, 3 Mar 2017 10:01:32 -0500 Subject: [PATCH 1/3] Updates buffer processing in recorder manager so that we're not blocking any of the decoding channel producer threads that are trying to push audio packets to the recorder manager. Adds an overflowable packet queue and services the queue once a second followed by an idle recorder check (and disposal). --- src/record/RecorderManager.java | 158 +++++++++++++++++++++----------- 1 file changed, 105 insertions(+), 53 deletions(-) diff --git a/src/record/RecorderManager.java b/src/record/RecorderManager.java index 32957c31..cb30aa08 100644 --- a/src/record/RecorderManager.java +++ b/src/record/RecorderManager.java @@ -25,11 +25,15 @@ import properties.SystemProperties; import record.wave.ComplexBufferWaveRecorder; import record.wave.RealBufferWaveRecorder; import sample.Listener; +import sample.OverflowableTransferQueue; +import sample.real.IOverflowListener; import util.ThreadPool; import java.io.File; +import java.util.ArrayList; import java.util.HashMap; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -42,7 +46,8 @@ public class RecorderManager implements Listener public static final long IDLE_RECORDER_REMOVAL_THRESHOLD = 6000; //6 seconds private Map mRecorders = new HashMap<>(); - private ScheduledFuture mRecorderMonitorFuture; + private OverflowableTransferQueue mAudioPacketQueue = new OverflowableTransferQueue<>(1000, 100); + private ScheduledFuture mBufferProcessorFuture; private boolean mCanStartNewRecorders = true; @@ -55,8 +60,24 @@ public class RecorderManager implements Listener */ public RecorderManager() { - mRecorderMonitorFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(new RecorderMonitor(), 2, - 2, TimeUnit.SECONDS); + mBufferProcessorFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(new BufferProcessor(), 0, + 1, TimeUnit.SECONDS); + + mAudioPacketQueue.setOverflowListener(new IOverflowListener() + { + @Override + public void sourceOverflow(boolean overflow) + { + if(overflow) + { + mLog.warn("overflow - audio packets will be dropped until recording catches up"); + } + else + { + mLog.info("audio recorder packet processing has returned to normal"); + } + } + }); } /** @@ -64,9 +85,9 @@ public class RecorderManager implements Listener */ public void dispose() { - if(mRecorderMonitorFuture != null) + if(mBufferProcessorFuture != null) { - mRecorderMonitorFuture.cancel(true); + mBufferProcessorFuture.cancel(true); } } @@ -79,56 +100,97 @@ public class RecorderManager implements Listener { if(audioPacket.hasMetadata() && audioPacket.getMetadata().isRecordable()) { - String identifier = audioPacket.getMetadata().getUniqueIdentifier(); + mAudioPacketQueue.offer(audioPacket); + } + } - synchronized(mRecorders) + /** + * Process any queued audio buffers and dispatch them to the audio recorders + */ + private void processBuffers() + { + List audioPackets = new ArrayList<>(); + + mAudioPacketQueue.drainTo(audioPackets, 50); + + while(!audioPackets.isEmpty()) + { + for(AudioPacket audioPacket: audioPackets) { - if(mRecorders.containsKey(identifier)) + String identifier = audioPacket.getMetadata().getUniqueIdentifier(); + + synchronized(mRecorders) { - RealBufferWaveRecorder recorder = mRecorders.get(identifier); - - if(audioPacket.getType() == AudioPacket.Type.AUDIO) + if(mRecorders.containsKey(identifier)) { - recorder.receive(audioPacket.getAudioBuffer()); - } - else if(audioPacket.getType() == AudioPacket.Type.END) - { - RealBufferWaveRecorder finished = mRecorders.remove(identifier); - finished.stop(); - } - } - else if(audioPacket.getType() == AudioPacket.Type.AUDIO) - { - if(mCanStartNewRecorders) - { - String filePrefix = getFilePrefix(audioPacket); + RealBufferWaveRecorder recorder = mRecorders.get(identifier); - RealBufferWaveRecorder recorder = null; - - try + if(audioPacket.getType() == AudioPacket.Type.AUDIO) { - recorder = new RealBufferWaveRecorder(AUDIO_SAMPLE_RATE, filePrefix); - - recorder.start(ThreadPool.SCHEDULED); - recorder.receive(audioPacket.getAudioBuffer()); - mRecorders.put(identifier, recorder); } - catch(Exception ioe) + else if(audioPacket.getType() == AudioPacket.Type.END) { - mCanStartNewRecorders = false; + RealBufferWaveRecorder finished = mRecorders.remove(identifier); + finished.stop(); + } + } + else if(audioPacket.getType() == AudioPacket.Type.AUDIO) + { + if(mCanStartNewRecorders) + { + String filePrefix = getFilePrefix(audioPacket); - mLog.error("Error attempting to start new audio wave recorder. All (future) audio recording " + - "is disabled", ioe); + RealBufferWaveRecorder recorder = null; - if(recorder != null) + try { - recorder.stop(); + recorder = new RealBufferWaveRecorder(AUDIO_SAMPLE_RATE, filePrefix); + + recorder.start(ThreadPool.SCHEDULED); + + recorder.receive(audioPacket.getAudioBuffer()); + mRecorders.put(identifier, recorder); + } + catch(Exception ioe) + { + mCanStartNewRecorders = false; + + mLog.error("Error attempting to start new audio wave recorder. All (future) audio recording " + + "is disabled", ioe); + + if(recorder != null) + { + recorder.stop(); + } } } } } } + + audioPackets.clear(); + mAudioPacketQueue.drainTo(audioPackets, 50); + } + } + + /** + * Removes recorders that have not received any new audio buffers in the last 6 seconds. + */ + private void removeIdleRecorders() + { + Iterator> it = mRecorders.entrySet().iterator(); + + while(it.hasNext()) + { + Map.Entry entry = it.next(); + + if(entry.getValue().getLastBufferReceived() + IDLE_RECORDER_REMOVAL_THRESHOLD < System.currentTimeMillis()) + { + mLog.info("Removing idle recorder [" + entry.getKey() + "]"); + it.remove(); + entry.getValue().stop(); + } } } @@ -175,28 +237,18 @@ public class RecorderManager implements Listener } /** - * Monitors currently running recorders and removes/closes any recorders that are idle more than 6 seconds + * Processes queued audio packets and distributes to each of the audio recorders. Removes any idle recorders + * that have not been updated according to an idle threshold period */ - public class RecorderMonitor implements Runnable + public class BufferProcessor implements Runnable { @Override public void run() { synchronized(mRecorders) { - Iterator> it = mRecorders.entrySet().iterator(); - - while(it.hasNext()) - { - Map.Entry entry = it.next(); - - if(entry.getValue().getLastBufferReceived() + IDLE_RECORDER_REMOVAL_THRESHOLD < System.currentTimeMillis()) - { - mLog.info("Removing idle recorder [" + entry.getKey() + "]"); - it.remove(); - entry.getValue().stop(); - } - } + processBuffers(); + removeIdleRecorders(); } } } From ef172a04c6b6284a83276523532081efae5cd480 Mon Sep 17 00:00:00 2001 From: Denny Date: Fri, 3 Mar 2017 10:04:21 -0500 Subject: [PATCH 2/3] Cleanup - removed synchronized blocks since the buffer processor thread is the only thread managing recorders. --- src/record/RecorderManager.java | 82 +++++++++++++++------------------ 1 file changed, 38 insertions(+), 44 deletions(-) diff --git a/src/record/RecorderManager.java b/src/record/RecorderManager.java index cb30aa08..4d9789e0 100644 --- a/src/record/RecorderManager.java +++ b/src/record/RecorderManager.java @@ -60,9 +60,6 @@ public class RecorderManager implements Listener */ public RecorderManager() { - mBufferProcessorFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(new BufferProcessor(), 0, - 1, TimeUnit.SECONDS); - mAudioPacketQueue.setOverflowListener(new IOverflowListener() { @Override @@ -78,6 +75,9 @@ public class RecorderManager implements Listener } } }); + + mBufferProcessorFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(new BufferProcessor(), 0, + 1, TimeUnit.SECONDS); } /** @@ -119,50 +119,47 @@ public class RecorderManager implements Listener { String identifier = audioPacket.getMetadata().getUniqueIdentifier(); - synchronized(mRecorders) + if(mRecorders.containsKey(identifier)) { - if(mRecorders.containsKey(identifier)) - { - RealBufferWaveRecorder recorder = mRecorders.get(identifier); + RealBufferWaveRecorder recorder = mRecorders.get(identifier); - if(audioPacket.getType() == AudioPacket.Type.AUDIO) - { - recorder.receive(audioPacket.getAudioBuffer()); - } - else if(audioPacket.getType() == AudioPacket.Type.END) - { - RealBufferWaveRecorder finished = mRecorders.remove(identifier); - finished.stop(); - } + if(audioPacket.getType() == AudioPacket.Type.AUDIO) + { + recorder.receive(audioPacket.getAudioBuffer()); } - else if(audioPacket.getType() == AudioPacket.Type.AUDIO) + else if(audioPacket.getType() == AudioPacket.Type.END) { - if(mCanStartNewRecorders) + RealBufferWaveRecorder finished = mRecorders.remove(identifier); + finished.stop(); + } + } + else if(audioPacket.getType() == AudioPacket.Type.AUDIO) + { + if(mCanStartNewRecorders) + { + String filePrefix = getFilePrefix(audioPacket); + + RealBufferWaveRecorder recorder = null; + + try { - String filePrefix = getFilePrefix(audioPacket); + recorder = new RealBufferWaveRecorder(AUDIO_SAMPLE_RATE, filePrefix); - RealBufferWaveRecorder recorder = null; + recorder.start(ThreadPool.SCHEDULED); - try + recorder.receive(audioPacket.getAudioBuffer()); + mRecorders.put(identifier, recorder); + } + catch(Exception ioe) + { + mCanStartNewRecorders = false; + + mLog.error("Error attempting to start new audio wave recorder. All (future) audio recording " + + "is disabled", ioe); + + if(recorder != null) { - recorder = new RealBufferWaveRecorder(AUDIO_SAMPLE_RATE, filePrefix); - - recorder.start(ThreadPool.SCHEDULED); - - recorder.receive(audioPacket.getAudioBuffer()); - mRecorders.put(identifier, recorder); - } - catch(Exception ioe) - { - mCanStartNewRecorders = false; - - mLog.error("Error attempting to start new audio wave recorder. All (future) audio recording " + - "is disabled", ioe); - - if(recorder != null) - { - recorder.stop(); - } + recorder.stop(); } } } @@ -245,11 +242,8 @@ public class RecorderManager implements Listener @Override public void run() { - synchronized(mRecorders) - { - processBuffers(); - removeIdleRecorders(); - } + processBuffers(); + removeIdleRecorders(); } } } From 61bfef5798a866c5dcc7ed7adccfa32b9e57c7f1 Mon Sep 17 00:00:00 2001 From: Denny Date: Fri, 3 Mar 2017 10:44:50 -0500 Subject: [PATCH 3/3] Incrementing build to version 0.3.0 beta8 --- build/build.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/build/build.xml b/build/build.xml index 6105114e..d3dd6483 100644 --- a/build/build.xml +++ b/build/build.xml @@ -7,7 +7,7 @@ - +