39 [WIP] cleans up the reusable buffers and creates both real and

complex variants.
This commit is contained in:
Denny
2018-03-25 09:13:03 -04:00
parent c36f6901af
commit 60ffb18cd9
70 changed files with 1211 additions and 728 deletions

View File

@@ -1,5 +1,6 @@
package io.github.dsheirer.sample;
import io.github.dsheirer.sample.buffer.ReusableBuffer;
import io.github.dsheirer.sample.real.RealBuffer;
import java.nio.ByteBuffer;
@@ -11,61 +12,71 @@ import java.nio.ShortBuffer;
*/
public class ConversionUtils
{
public static float[] convertFromSigned16BitSamples( byte[] bytes )
public static float[] convertFromSigned16BitSamples(byte[] bytes)
{
return convertFromSigned16BitSamples( ByteBuffer.wrap( bytes ) );
return convertFromSigned16BitSamples(ByteBuffer.wrap(bytes));
}
/**
* Converts the byte buffer containing 16-bit samples into a float array
*/
public static float[] convertFromSigned16BitSamples( ByteBuffer buffer )
public static float[] convertFromSigned16BitSamples(ByteBuffer buffer)
{
ShortBuffer byteBuffer = buffer.order( ByteOrder.LITTLE_ENDIAN ).asShortBuffer();
float[] samples = new float[ buffer.limit() / 2 ];
for( int x = 0; x < samples.length; x++ )
{
samples[ x ] = (float)byteBuffer.get() / (float)Short.MAX_VALUE;
}
ShortBuffer byteBuffer = buffer.order(ByteOrder.LITTLE_ENDIAN).asShortBuffer();
return samples;
float[] samples = new float[buffer.limit() / 2];
for(int x = 0; x < samples.length; x++)
{
samples[x] = (float)byteBuffer.get() / (float)Short.MAX_VALUE;
}
return samples;
}
/**
* Converts the float samples into a little-endian 16-bit sample byte buffer.
*
* @param samples - float array of sample data
* @return - little-endian 16-bit sample byte buffer
*/
public static ByteBuffer convertToSigned16BitSamples( float[] samples )
public static ByteBuffer convertToSigned16BitSamples(float[] samples)
{
ByteBuffer converted = ByteBuffer.allocate( samples.length * 2 );
converted.order( ByteOrder.LITTLE_ENDIAN );
ByteBuffer converted = ByteBuffer.allocate(samples.length * 2);
converted.order(ByteOrder.LITTLE_ENDIAN);
for( float sample: samples )
{
converted.putShort( (short)( sample * Short.MAX_VALUE ) );
}
return converted;
for(float sample : samples)
{
converted.putShort((short)(sample * Short.MAX_VALUE));
}
return converted;
}
/**
* Converts the float samples in a complex buffer to a little endian 16-bit
* buffer
*/
public static ByteBuffer convertToSigned16BitSamples( Buffer buffer )
{
return convertToSigned16BitSamples( buffer.getSamples() );
}
/**
* Converts the float samples in a complex buffer to a little endian 16-bit
* buffer
*/
public static ByteBuffer convertToSigned16BitSamples( RealBuffer buffer )
{
return convertToSigned16BitSamples( buffer.getSamples() );
}
/**
* Converts the float samples in a complex buffer to a little endian 16-bit
* buffer
*/
public static ByteBuffer convertToSigned16BitSamples(Buffer buffer)
{
return convertToSigned16BitSamples(buffer.getSamples());
}
/**
* Converts the float samples in a complex buffer to a little endian 16-bit
* buffer
*/
public static ByteBuffer convertToSigned16BitSamples(RealBuffer buffer)
{
return convertToSigned16BitSamples(buffer.getSamples());
}
/**
* Converts the float samples in a complex buffer to a little endian 16-bit
* buffer
*/
public static ByteBuffer convertToSigned16BitSamples(ReusableBuffer buffer)
{
return convertToSigned16BitSamples(buffer.getSamples());
}
}

View File

@@ -15,7 +15,7 @@
******************************************************************************/
package io.github.dsheirer.sample.adapter;
import io.github.dsheirer.sample.complex.reusable.ReusableComplexBuffer;
import io.github.dsheirer.sample.buffer.ReusableComplexBuffer;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;

View File

@@ -15,7 +15,7 @@
******************************************************************************/
package io.github.dsheirer.sample.adapter;
import io.github.dsheirer.sample.complex.reusable.ReusableComplexBuffer;
import io.github.dsheirer.sample.buffer.ReusableComplexBuffer;
public interface IComplexSampleAdapter
{

View File

@@ -13,7 +13,7 @@
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
package io.github.dsheirer.sample.buffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -21,21 +21,22 @@ import org.slf4j.LoggerFactory;
import java.util.Queue;
import java.util.concurrent.LinkedTransferQueue;
public class ReusableBufferQueue implements IReusableBufferDisposedListener
public abstract class AbstractReusableBufferQueue<T extends ReusableBuffer> implements IReusableBufferDisposedListener<T>
{
private final static Logger mLog = LoggerFactory.getLogger(ReusableBufferQueue.class);
private Queue<ReusableComplexBuffer> mReusableBufferQueue = new LinkedTransferQueue<>();
private final static Logger mLog = LoggerFactory.getLogger(AbstractReusableBufferQueue.class);
private Queue<T> mReusableBufferQueue = new LinkedTransferQueue<>();
private int mBufferCount = 0;
private String mDebugName;
private int mBufferUserCounter;
public ReusableBufferQueue(String debugName)
public AbstractReusableBufferQueue(String debugName)
{
mDebugName = debugName;
}
public ReusableBufferQueue()
public AbstractReusableBufferQueue()
{
}
public void dispose()
@@ -47,12 +48,17 @@ public class ReusableBufferQueue implements IReusableBufferDisposedListener
* 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
* @param reusableBuffer that has been disposed
*/
@Override
public void disposed(ReusableComplexBuffer reusableComplexBuffer)
public void disposed(T reusableBuffer)
{
mReusableBufferQueue.offer(reusableComplexBuffer);
mReusableBufferQueue.offer(reusableBuffer);
// if(mDebugName != null && mDebugName.contentEquals("NativeBufferConverter"))
// {
// mLog.debug("Reclaimed: " + reusableBuffer.name());
// }
}
/**
@@ -62,18 +68,37 @@ public class ReusableBufferQueue implements IReusableBufferDisposedListener
* @param size of the samples that will be loaded into the buffer
* @return a reusable buffer
*/
public ReusableComplexBuffer getBuffer(int size)
public T getBuffer(int size)
{
ReusableComplexBuffer buffer = mReusableBufferQueue.poll();
T 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]);
buffer = createBuffer(size);
mBufferCount++;
mLog.debug("Buffer Created - Count:" + mBufferCount + (mDebugName != null ? " [" + mDebugName + "]" : ""));
buffer.setDebugName("Owner:" + mDebugName + " Number:" + mBufferUserCounter++ + " NEW BUFFER");
// if(mDebugName != null && mDebugName.contentEquals("NativeBufferConverter"))
// {
// mLog.debug("Buffer Created - Count:" + mBufferCount + (mDebugName != null ? " [" + mDebugName + "]" : ""));
// }
}
else
{
String previous = buffer.name();
buffer.setDebugName("Owner:" + mDebugName + " Number:" + mBufferUserCounter++);
// if(mDebugName != null && mDebugName.contentEquals("NativeBufferConverter"))
// {
// mLog.debug("Buffer Reused: " + previous + " New Name:" + buffer.name());
// }
}
return buffer;
}
/**
* Create a new buffer of the specified size
*/
protected abstract T createBuffer(int size);
}

View File

@@ -13,13 +13,13 @@
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
package io.github.dsheirer.sample.buffer;
public interface IReusableBufferDisposedListener
public interface IReusableBufferDisposedListener<T extends ReusableBuffer>
{
/**
* Notifies the listener that the timestamped buffer has been disposed and can now be reused.
* @param reusableComplexBuffer that has been disposed
* @param t reusableBuffer that has been disposed
*/
void disposed(ReusableComplexBuffer reusableComplexBuffer);
void disposed(T t);
}

View File

@@ -13,21 +13,21 @@
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
package io.github.dsheirer.sample.buffer;
import io.github.dsheirer.sample.Listener;
public interface IReusableComplexBufferProvider
public interface IReusableBufferProvider<T extends ReusableBuffer>
{
/**
* Adds the listener to receive complex buffer samples
*/
void addBufferListener(Listener<ReusableComplexBuffer> listener);
void addBufferListener(Listener<T> listener);
/**
* Removes the listener from receiving complex buffer samples
*/
void removeBufferListener(Listener<ReusableComplexBuffer> listener);
void removeBufferListener(Listener<T> listener);
/**
* Indicates if there are any complex buffer listeners registered

View File

@@ -13,11 +13,11 @@
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
package io.github.dsheirer.sample.buffer;
import io.github.dsheirer.sample.Listener;
public interface IReusableComplexBufferListener
public interface IReusableComplexBufferListener<T extends ReusableBuffer>
{
Listener<ReusableComplexBuffer> getReusableComplexBufferListener();
Listener<T> getReusableComplexBufferListener();
}

View File

@@ -13,11 +13,11 @@
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
package io.github.dsheirer.sample.buffer;
import io.github.dsheirer.sample.OverflowableTransferQueue;
public class OverflowableReusableBufferTransferQueue extends OverflowableTransferQueue<ReusableComplexBuffer>
public class OverflowableReusableBufferTransferQueue<T extends ReusableBuffer> extends OverflowableTransferQueue<T>
{
/**
* Concurrent transfer queue that couples a higher-throughput linked transfer queue with an atomic integer for
@@ -38,12 +38,12 @@ public class OverflowableReusableBufferTransferQueue extends OverflowableTransfe
* Overrides the overflow method to decrement the user count on any buffers that are being discarded when the queue
* is in an overflow state.
*
* @param reusableComplexBuffer that will be discarded
* @param t reusableBuffer that will be discarded
*/
@Override
protected void overflow(ReusableComplexBuffer reusableComplexBuffer)
protected void overflow(T t)
{
reusableComplexBuffer.decrementUserCount();
t.decrementUserCount();
}
/**
@@ -54,7 +54,7 @@ public class OverflowableReusableBufferTransferQueue extends OverflowableTransfe
{
synchronized(mQueue)
{
ReusableComplexBuffer buffer = mQueue.poll();
T buffer = mQueue.poll();
while(buffer != null)
{

View File

@@ -13,19 +13,20 @@
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
package io.github.dsheirer.sample.buffer;
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
public class ReusableBuffer
{
private IReusableBufferDisposedListener mDisposalListener;
private float[] mSamples;
private IReusableBufferDisposedListener mBufferDisposedListener;
private long mTimestamp;
private AtomicInteger mUserCount = new AtomicInteger();
private String mDebugName;
/**
* Creates a reusable, timestamped complex buffer using the specified time in milliseconds.
@@ -34,32 +35,40 @@ public class ReusableComplexBuffer extends ComplexBuffer
* buffer should not modify the buffer contents. Each component should also increment the user count before
* sending this buffer to another component and should decrement the user count when finished using this buffer.
*
* @param disposalListener to be notified when all consumers/users are finished using the buffer
* @param bufferDisposedListener 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)
ReusableBuffer(IReusableBufferDisposedListener bufferDisposedListener, float[] samples, long timestamp)
{
super(samples);
mSamples = samples;
Validate.notNull(disposalListener, "Timestamped Buffer Listener cannot be null");
mDisposalListener = disposalListener;
Validate.notNull(bufferDisposedListener, "Reusable Buffer Listener cannot be null");
mBufferDisposedListener = bufferDisposedListener;
mTimestamp = timestamp;
}
/**
* Constructs a timestamped complex buffer using the current system time in milliseconds.
*
* NOTE: reusability of this buffer requires strict user count tracking. Each component that receives this
* buffer should not modify the buffer contents. Each component should also increment the user count before
* sending this buffer to another component and should decrement the user count when finished using this buffer.
*
* @param disposalListener to be notified when all consumers are finished using the buffer
* @param samples of data
*/
public ReusableComplexBuffer(IReusableBufferDisposedListener disposalListener, float[] samples)
public void setDebugName(String debugName)
{
this(disposalListener, samples, System.currentTimeMillis());
mDebugName = debugName;
}
public String name()
{
return (mDebugName != null ? mDebugName : "(null)") + " User Count:" + getUserCount();
}
public String toString()
{
return name();
}
/**
* Samples for this buffer
*/
public float[] getSamples()
{
return mSamples;
}
/**
@@ -74,11 +83,26 @@ public class ReusableComplexBuffer extends ComplexBuffer
}
/**
* Number of complex samples contained in this buffer
* Number of samples contained in this buffer
*/
public int getSampleLength()
public int getSampleCount()
{
return getSamples().length / 2;
return getSamples().length;
}
/**
* Constructs a timestamped complex buffer using the current system time in milliseconds.
*
* NOTE: reusability of this buffer requires strict user count tracking. Each component that receives this
* buffer should not modify the buffer contents. Each component should also increment the user count before
* sending this buffer to another component and should decrement the user count when finished using this buffer.
*
* @param bufferDisposedListener to be notified when all consumers are finished using the buffer
* @param samples of data
*/
public ReusableBuffer(IReusableBufferDisposedListener bufferDisposedListener, float[] samples)
{
this(bufferDisposedListener, samples, System.currentTimeMillis());
}
/**
@@ -173,21 +197,19 @@ public class ReusableComplexBuffer extends ComplexBuffer
*
* This method is thread-safe.
*/
public ReusableComplexBuffer decrementUserCount()
public void decrementUserCount()
{
int currentCount = mUserCount.decrementAndGet();
if(currentCount == 0)
{
mDisposalListener.disposed(this);
mBufferDisposedListener.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;
}
/**
@@ -199,10 +221,9 @@ public class ReusableComplexBuffer extends ComplexBuffer
*
* This method is thread-safe.
*/
public ReusableComplexBuffer incrementUserCount()
public void incrementUserCount()
{
mUserCount.incrementAndGet();
return this;
}
/**
@@ -216,9 +237,16 @@ public class ReusableComplexBuffer extends ComplexBuffer
*
* @param additionalUserCount to add to the current user count
*/
public ReusableComplexBuffer incrementUserCount(int additionalUserCount)
public void incrementUserCount(int additionalUserCount)
{
mUserCount.addAndGet(additionalUserCount);
return this;
}
/**
* Number of users currently registered for this buffer
*/
public int getUserCount()
{
return mUserCount.get();
}
}

View File

@@ -13,7 +13,7 @@
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
package io.github.dsheirer.sample.buffer;
import io.github.dsheirer.dsp.filter.channelizer.SampleTimestampManager;
import io.github.dsheirer.sample.Listener;
@@ -27,7 +27,7 @@ import java.nio.FloatBuffer;
*/
public class ReusableBufferAssembler
{
private ReusableBufferQueue mReusableBufferQueue = new ReusableBufferQueue();
private ReusableComplexBufferQueue mReusableComplexBufferQueue = new ReusableComplexBufferQueue("ReusableBufferAssembler");
private SampleTimestampManager mTimestampManager;
private FloatBuffer mBuffer;
private int mBufferSize;
@@ -115,7 +115,7 @@ public class ReusableBufferAssembler
if(mListener != null && mBuffer.position() > 0)
{
mBuffer.rewind();
ReusableComplexBuffer reusableComplexBuffer = mReusableBufferQueue.getBuffer(mBufferSize);
ReusableComplexBuffer reusableComplexBuffer = mReusableComplexBufferQueue.getBuffer(mBufferSize);
reusableComplexBuffer.reloadFrom(mBuffer, mCurrentBufferTimestamp);
reusableComplexBuffer.setUserCount(1);
mListener.receive(reusableComplexBuffer);

View File

@@ -13,7 +13,7 @@
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.sample.complex.reusable;
package io.github.dsheirer.sample.buffer;
import io.github.dsheirer.sample.Broadcaster;
import io.github.dsheirer.sample.Listener;
@@ -36,29 +36,10 @@ public class ReusableBufferBroadcaster extends Broadcaster<ReusableComplexBuffer
@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--;
reusableComplexBuffer.incrementUserCount();
listener.receive(reusableComplexBuffer);
}
//Decrement user counter for this broadcaster

View File

@@ -0,0 +1,39 @@
/*******************************************************************************
* 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.buffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ReusableBufferQueue extends AbstractReusableBufferQueue<ReusableBuffer>
{
private final static Logger mLog = LoggerFactory.getLogger(ReusableBufferQueue.class);
public ReusableBufferQueue(String debugName)
{
super(debugName);
}
public ReusableBufferQueue()
{
}
@Override
protected ReusableBuffer createBuffer(int size)
{
return new ReusableBuffer(this, new float[size]);
}
}

View File

@@ -0,0 +1,59 @@
/*******************************************************************************
* 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.buffer;
public class ReusableComplexBuffer extends ReusableBuffer
{
/**
* Creates a reusable, timestamped complex buffer using the specified time in milliseconds.
*
* NOTE: reusability of this buffer requires strict user count tracking. Each component that receives this
* buffer should not modify the buffer contents. Each component should also increment the user count before
* sending this buffer to another component and should decrement the user count when finished using this buffer.
*
* @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
*/
ReusableComplexBuffer(IReusableBufferDisposedListener disposalListener, float[] samples, long timestamp)
{
super(disposalListener, samples, timestamp);
}
/**
* Constructs a timestamped complex buffer using the current system time in milliseconds.
*
* NOTE: reusability of this buffer requires strict user count tracking. Each component that receives this
* buffer should not modify the buffer contents. Each component should also increment the user count before
* sending this buffer to another component and should decrement the user count when finished using this buffer.
*
* @param disposalListener to be notified when all consumers are finished using the buffer
* @param samples of data
*/
protected ReusableComplexBuffer(IReusableBufferDisposedListener disposalListener, float[] samples)
{
this(disposalListener, samples, System.currentTimeMillis());
}
/**
* Number of complex samples contained in this buffer
*/
@Override
public int getSampleCount()
{
return getSamples().length / 2;
}
}

View File

@@ -0,0 +1,39 @@
/*******************************************************************************
* 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.buffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ReusableComplexBufferQueue extends AbstractReusableBufferQueue<ReusableComplexBuffer>
{
private final static Logger mLog = LoggerFactory.getLogger(ReusableComplexBufferQueue.class);
public ReusableComplexBufferQueue(String debugName)
{
super(debugName);
}
public ReusableComplexBufferQueue()
{
}
@Override
protected ReusableComplexBuffer createBuffer(int size)
{
return new ReusableComplexBuffer(this, new float[size]);
}
}