GitHub/sdrtrunk
1
0
Fork 0
mirror of https://github.com/DSheirer/sdrtrunk.git synced 2026-11-11 08:07:24 -05:00
Code Issues Packages Projects Releases 2 Wiki Activity

[wip] reusable buffer support.

Browse Source
This commit is contained in:
Denny
2018-02-09 05:56:20 -05:00
parent 28cb9d9407
commit 28da8b2fa4
7 changed files with 572 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableBufferDisposedListener.java Normal file
View File

@@ -0,0 +1,25 @@
/*******************************************************************************
* sdr-trunk
* Copyright (C) 2014-2018 Dennis Sheirer
*
* This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any
* later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License along with this program.
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
public interface IReusableBufferDisposedListener
{
/**
* Notifies the listener that the timestamped buffer has been disposed and can now be reused.
* @param reusableComplexBuffer that has been disposed
*/
void disposed(ReusableComplexBuffer reusableComplexBuffer);
}

23
src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableComplexBufferListener.java Normal file
View File

@@ -0,0 +1,23 @@
/*******************************************************************************
* sdr-trunk
* Copyright (C) 2014-2018 Dennis Sheirer
*
* This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any
* later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License along with this program.
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
import io.github.dsheirer.sample.Listener;
public interface IReusableComplexBufferListener
{
Listener<ReusableComplexBuffer> getReusableComplexBufferListener();
}

36
src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableComplexBufferProvider.java Normal file
View File

@@ -0,0 +1,36 @@
/*******************************************************************************
* sdr-trunk
* Copyright (C) 2014-2018 Dennis Sheirer
*
* This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any
* later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License along with this program.
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
import io.github.dsheirer.sample.Listener;
public interface IReusableComplexBufferProvider
{
/**
* Adds the listener to receive complex buffer samples
*/
void addBufferListener(Listener<ReusableComplexBuffer> listener);
/**
* Removes the listener from receiving complex buffer samples
*/
void removeBufferListener(Listener<ReusableComplexBuffer> listener);
/**
* Indicates if there are any complex buffer listeners registered
*/
boolean hasBufferListeners();
}

136
src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferAssembler.java Normal file
View File

@@ -0,0 +1,136 @@
/*******************************************************************************
* sdr-trunk
* Copyright (C) 2014-2018 Dennis Sheirer
*
* This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any
* later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License along with this program.
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
import io.github.dsheirer.dsp.filter.channelizer.SampleTimestampManager;
import io.github.dsheirer.sample.Listener;
import org.apache.commons.lang3.Validate;
import java.nio.FloatBuffer;
/**
* Assembles complex float samples into a ComplexBuffer containing an array of floats. Monitors incoming sample count
* to accurately assign a relative timestamp to each assembled buffer.
*/
public class ReusableBufferAssembler
{
private ReusableBufferQueue mReusableBufferQueue = new ReusableBufferQueue();
private SampleTimestampManager mTimestampManager;
private FloatBuffer mBuffer;
private int mBufferSize;
private long mCurrentBufferTimestamp;
private Listener<ReusableComplexBuffer> mListener;
public ReusableBufferAssembler(int bufferSize, double sampleRate)
{
Validate.isTrue(bufferSize % 2 == 0);
mBufferSize = bufferSize;
mBuffer = FloatBuffer.allocate(mBufferSize);
mTimestampManager = new SampleTimestampManager(sampleRate);
}
/**
* Updates the reference timestamp to use for all subsequent samples that are assembled. This timestamp should
* identify the relative timestamp of the first/next sample that is received by this assembler.
* @param timestamp to use relative to the incoming sample arrays.
*/
public void updateTimestamp(long timestamp)
{
mTimestampManager.setReferenceTimestamp(timestamp);
if(mBuffer.position() == 0)
{
mCurrentBufferTimestamp = timestamp;
}
}
public void dispose()
{
mListener = null;
mBuffer = null;
}
/**
* Adds the samples to this assembler. As each buffer is assembled, it will be dispatched to the registered
* listener.
*
* @param samples to assemble
*/
public void receive(float[] samples)
{
if(mBuffer.remaining() >= samples.length)
{
mBuffer.put(samples);
mTimestampManager.increment(samples.length / 2);
if(!mBuffer.hasRemaining())
{
flush();
}
}
else
{
int offset = 0;
while(offset < samples.length)
{
int toCopy = mBuffer.remaining();
if((samples.length - offset) < toCopy)
{
toCopy = samples.length - offset;
}
mBuffer.put(samples, offset, toCopy);
offset += toCopy;
mTimestampManager.increment(toCopy / 2);
if(!mBuffer.hasRemaining())
{
flush();
}
}
}
}
/**
* Flushes the current buffer contents to the registered listener
*/
public void flush()
{
if(mListener != null && mBuffer.position() > 0)
{
mBuffer.rewind();
ReusableComplexBuffer reusableComplexBuffer = mReusableBufferQueue.getBuffer(mBufferSize);
reusableComplexBuffer.reloadFrom(mBuffer, mCurrentBufferTimestamp);
reusableComplexBuffer.setUserCount(1);
mListener.receive(reusableComplexBuffer);
}
mBuffer.rewind();
mCurrentBufferTimestamp = mTimestampManager.getCurrentTimestamp();
}
/**
* Sets the listener to receive the output from this buffer assembler. Note: the listener is responsible for
* managing the user count for any buffers received from this buffer assembler.
*/
public void setListener(Listener<ReusableComplexBuffer> listener)
{
mListener = listener;
}
}

67
src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferBroadcaster.java Normal file
View File

@@ -0,0 +1,67 @@
/*******************************************************************************
* sdr-trunk
* Copyright (C) 2014-2018 Dennis Sheirer
*
* This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any
* later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License along with this program.
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
import io.github.dsheirer.sample.Broadcaster;
import io.github.dsheirer.sample.Listener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ReusableBufferBroadcaster extends Broadcaster<ReusableComplexBuffer>
{
private final static Logger mLog = LoggerFactory.getLogger(ReusableBufferBroadcaster.class);
/**
* Increments the user count for the reusable complex buffer and then broadcasts the buffer to all registered
* listeners.
*
* The total user count is established and applied to the buffer prior to dispatching. If we were to simply
* increment the user count prior to sending to each consumer, there is a possibility that the consumer could
* immediately decrement the user count and prematurely signal that the buffer is ready for disposal before we
* send the buffer to all consumers.
*/
@Override
public void broadcast(ReusableComplexBuffer reusableComplexBuffer)
{
//There is a minuscule possibility that the listener list will change from the time that we get the size until
//the for/each iterator is created below, therefore we have to accurately track the user count. If a listener
//is added in the time between getting the list size and creating the iterator, that listener will NOT receive
//a copy of the buffer, but will instead get the next buffer. Likewise, if a listener is removed, we have to
//decrement the user count on the buffer after iterating the listener list.
int listenerCount = mListeners.size();
reusableComplexBuffer.incrementUserCount(listenerCount);
for(Listener<ReusableComplexBuffer> listener : mListeners)
{
if(listenerCount > 0)
{
listener.receive(reusableComplexBuffer);
}
listenerCount--;
}
while(listenerCount > 0)
{
reusableComplexBuffer.decrementUserCount();
listenerCount--;
}
//Decrement user counter for this broadcaster
reusableComplexBuffer.decrementUserCount();
}
}

74
src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferQueue.java Normal file
View File

@@ -0,0 +1,74 @@
/*******************************************************************************
* sdr-trunk
* Copyright (C) 2014-2018 Dennis Sheirer
*
* This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any
* later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License along with this program.
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Queue;
import java.util.concurrent.LinkedTransferQueue;
public class ReusableBufferQueue implements IReusableBufferDisposedListener
{
private final static Logger mLog = LoggerFactory.getLogger(ReusableBufferQueue.class);
private Queue<ReusableComplexBuffer> mReusableBufferQueue = new LinkedTransferQueue<>();
private int mBufferCount = 0;
private String mDebugName;
public ReusableBufferQueue(String debugName)
{
mDebugName = debugName;
}
public ReusableBufferQueue()
{
}
/**
* Implements buffer disposed listener interface. Disposed (ie user count = 0) buffers will automatically callback
* to this method to indicate when they are disposed.
*
* @param reusableComplexBuffer that has been disposed
*/
@Override
public void disposed(ReusableComplexBuffer reusableComplexBuffer)
{
mReusableBufferQueue.offer(reusableComplexBuffer);
}
/**
* Returns a reusable buffer from an internal recycling queue, or creates a new buffer if there are currently no
* buffers available for reuse.
*
* @param size of the samples that will be loaded into the buffer
* @return a reusable buffer
*/
public ReusableComplexBuffer getBuffer(int size)
{
ReusableComplexBuffer buffer = mReusableBufferQueue.poll();
//Create a new buffer if we currently don't have any buffers to reuse
if(buffer == null)
{
buffer = new ReusableComplexBuffer(this, new float[size]);
mBufferCount++;
mLog.debug("Buffer Created - Count:" + mBufferCount + (mDebugName != null ? " [" + mDebugName + "]" : ""));
}
return buffer;
}
}

211
src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableComplexBuffer.java Normal file
View File

@@ -0,0 +1,211 @@
/*******************************************************************************
* sdr-trunk
* Copyright (C) 2014-2018 Dennis Sheirer
*
* This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any
* later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License along with this program.
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
import io.github.dsheirer.sample.complex.ComplexBuffer;
import org.apache.commons.lang3.Validate;
import java.nio.FloatBuffer;
import java.util.concurrent.atomic.AtomicInteger;
public class ReusableComplexBuffer extends ComplexBuffer
{
private IReusableBufferDisposedListener mDisposalListener;
private long mTimestamp;
private AtomicInteger mUserCount = new AtomicInteger();
/**
* Creates a reusable, timestamped complex buffer using the specified time in milliseconds.
*
* @param disposalListener to be notified when all consumers/users are finished using the buffer
* @param samples of data
* @param timestamp in millis for the buffer
*/
public ReusableComplexBuffer(IReusableBufferDisposedListener disposalListener, float[] samples, long timestamp)
{
super(samples);
Validate.notNull(disposalListener, "Timestamped Buffer Listener cannot be null");
mDisposalListener = disposalListener;
mTimestamp = timestamp;
}
/**
* Constructs a timestamped complex buffer using the current system time in milliseconds.
*
* @param disposalListener to be notified when all consumers are finished using the buffer
* @param samples of data
*/
public ReusableComplexBuffer(IReusableBufferDisposedListener disposalListener, float[] samples)
{
this(disposalListener, samples, System.currentTimeMillis());
}
/**
* Creates a copy of the samples from this buffer
*/
public float[] getSamplesCopy()
{
float[] samples = getSamples();
float[] copy = new float[samples.length];
System.arraycopy(samples, 0, copy, 0, samples.length);
return copy;
}
/**
* Number of complex samples contained in this buffer
*/
public int getSampleLength()
{
return getSamples().length / 2;
}
/**
* Reference timestamp for this complex buffer. Timestamp is relative to the first sample in the buffer.
* @return timestamp as milliseconds since epoch
*/
public long getTimestamp()
{
return mTimestamp;
}
/**
* Reloads this buffer by copying the sample array into this buffer. Internal buffer length will be automatically
* resized to fit the new sample array length.
*
* This is convenient if the source of the sample array originates via a native buffer.
* @param samples to copy into this buffer
* @param timestamp for the samples in millis
*/
public void reloadFrom(float[] samples, long timestamp)
{
if(mUserCount.get() > 0)
{
throw new IllegalStateException("New data cannot be loaded into this reusable buffer while the user count " +
"is above zero - user count:" + mUserCount.get());
}
mUserCount.set(0);
if(mSamples.length != samples.length)
{
mSamples = new float[samples.length];
}
System.arraycopy(samples, 0, mSamples, 0, samples.length);
mTimestamp = timestamp;
}
/**
* Reloads this buffer by filling this buffer from the float buffer (ie copying the buffer contents). Internal
* buffer length will be automatically resized to accommodate the float buffer's capacity/size.
*
* @param floatBuffer to copy into this buffer
* @param timestamp for the samples in millis
*/
public void reloadFrom(FloatBuffer floatBuffer, long timestamp)
{
if(mUserCount.get() > 0)
{
throw new IllegalStateException("New data cannot be loaded into this reusable buffer while the user count " +
"is above zero - user count:" + mUserCount.get());
}
mUserCount.set(0);
if(mSamples.length != floatBuffer.capacity())
{
mSamples = new float[floatBuffer.capacity()];
}
floatBuffer.rewind();
floatBuffer.get(mSamples);
mTimestamp = timestamp;
}
/**
* Sets the number of users that will receive this buffer. Each user is expected to invoke the
* decrementUserCount() method to indicate that they are finished using this buffer so that this buffer can be
* reused once the user count is zero.
*
* @param userCount that will receive this reusable buffer
*/
public void setUserCount(int userCount)
{
mUserCount.set(userCount);
}
/**
* Decrements the user count for this buffer to indicate that the user/receiver of this buffer is finished using
* the buffer.
*
* User count is established prior to sending this buffer to each listener and each listener invokes this method to
* indicate that they have finished processing the buffer so that when the user count reaches zero, this buffer
* can be reused.
*
* This method is thread-safe.
*/
public ReusableComplexBuffer decrementUserCount()
{
int currentCount = mUserCount.decrementAndGet();
if(currentCount == 0)
{
mDisposalListener.disposed(this);
}
else if(currentCount < 0)
{
throw new IllegalStateException("User count is below zero. This indicates that this buffer's decrement" +
" user count was invoked by more than the expected user count");
}
return this;
}
/**
* Increments the user count to indicate that this buffer will be sent to another user.
*
* Users/receivers of this buffer that wish to forward the buffer to additional users should increment the expected
* user count prior to invoking the decrementUserCount() to indicate that the original receiver of the buffer is
* finished to avoid the user count reaching zero and the buffer being flagged as ready for reuse.
*
* This method is thread-safe.
*/
public ReusableComplexBuffer incrementUserCount()
{
mUserCount.incrementAndGet();
return this;
}
/**
* Increments the user count by the specified count to indicate that this buffer will be sent to additional users.
*
* Users/receivers of this buffer that wish to forward the buffer to additional users should increment the expected
* user count prior to invoking the decrementUserCount() to indicate that the original receiver of the buffer is
* finished to avoid the user count reaching zero and the buffer being flagged as ready for reuse.
*
* This method is thread-safe.
*
* @param additionalUserCount to add to the current user count
*/
public ReusableComplexBuffer incrementUserCount(int additionalUserCount)
{
mUserCount.addAndGet(additionalUserCount);
return this;
}
}