From cf2c0ca07d5c73dcd61e8d66344e8fa223a48484 Mon Sep 17 00:00:00 2001 From: denny Date: Sat, 27 Feb 2016 06:19:35 -0500 Subject: [PATCH] Resolves significant memory usage issue - updates Airspy and HackRF with the recent changes applied to RTL2832 Tuner controller for processing of libusb transfer buffers. --- .../tuner/airspy/AirspyTunerController.java | 240 ++++++++++-------- .../tuner/hackrf/HackRFTunerController.java | 220 ++++++++-------- .../tuner/rtl/RTL2832TunerController.java | 6 +- 3 files changed, 252 insertions(+), 214 deletions(-) diff --git a/src/source/tuner/airspy/AirspyTunerController.java b/src/source/tuner/airspy/AirspyTunerController.java index 755465d0..947d1f40 100644 --- a/src/source/tuner/airspy/AirspyTunerController.java +++ b/src/source/tuner/airspy/AirspyTunerController.java @@ -262,7 +262,30 @@ public class AirspyTunerController extends TunerController } } - @Override + public static String getTransferStatus( int status ) + { + switch( status ) + { + case 0: + return "TRANSFER COMPLETED (0)"; + case 1: + return "TRANSFER ERROR (1)"; + case 2: + return "TRANSFER TIMED OUT (2)"; + case 3: + return "TRANSFER CANCELLED (3)"; + case 4: + return "TRANSFER STALL (4)"; + case 5: + return "TRANSFER NO DEVICE (5)"; + case 6: + return "TRANSFER OVERFLOW (6)"; + default: + return "UNKNOWN TRANSFER STATUS (" + status + ")"; + } + } + + @Override public void apply( TunerConfiguration config ) throws SourceException { if( config instanceof AirspyTunerConfiguration ) @@ -1192,18 +1215,21 @@ public class AirspyTunerController extends TunerController */ public void addListener( Listener listener ) { - mComplexBufferBroadcaster.addListener( listener ); + synchronized( mComplexBufferBroadcaster ) + { + mComplexBufferBroadcaster.addListener( listener ); - if( mBufferProcessor == null || !mBufferProcessor.isRunning() ) - { - mBufferProcessor = new BufferProcessor(); + if( mBufferProcessor == null || !mBufferProcessor.isRunning() ) + { + mBufferProcessor = new BufferProcessor(); - Thread thread = new Thread( mBufferProcessor ); - thread.setDaemon( true ); - thread.setName( "Airspy Buffer Processor" ); + Thread thread = new Thread( mBufferProcessor ); + thread.setDaemon( true ); + thread.setName( "Airspy Buffer Processor" ); - thread.start(); - } + thread.start(); + } + } } /** @@ -1212,12 +1238,15 @@ public class AirspyTunerController extends TunerController */ public void removeListener( Listener listener ) { - mComplexBufferBroadcaster.removeListener( listener ); - - if( !mComplexBufferBroadcaster.hasListeners() ) - { - mBufferProcessor.stop(); - } + synchronized( mComplexBufferBroadcaster ) + { + mComplexBufferBroadcaster.removeListener( listener ); + + if( !mComplexBufferBroadcaster.hasListeners() ) + { + mBufferProcessor.stop(); + } + } } /** @@ -1227,8 +1256,11 @@ public class AirspyTunerController extends TunerController public class BufferProcessor implements Runnable, TransferCallback { private ScheduledFuture mSampleDispatcherTask; - private CopyOnWriteArrayList mTransfers; + private LinkedTransferQueue mAvailableTransfers; + private LinkedTransferQueue mTransfersInProgress = new LinkedTransferQueue<>(); private AtomicBoolean mRunning = new AtomicBoolean(); + private ByteBuffer mLibUsbHandlerStatus; + private boolean mCancel = false; @Override public void run() @@ -1237,33 +1269,64 @@ public class AirspyTunerController extends TunerController { prepareTransfers(); - for( Transfer transfer: mTransfers ) - { - int result = LibUsb.submitTransfer( transfer ); - - if( result != LibUsb.SUCCESS ) - { - mLog.error( "error submitting transfer [" + - LibUsb.errorName( result ) + "]" ); - break; - } - } - mSampleDispatcherTask = mThreadPoolManager.scheduleFixedRate( ThreadType.SOURCE_SAMPLE_PROCESSING, new BufferDispatcher(), 20, TimeUnit.MILLISECONDS ); + mLibUsbHandlerStatus = ByteBuffer.allocateDirect( 4 ); + + List transfers = new ArrayList<>(); + + mCancel = false; + while( mRunning.get() ) { - ByteBuffer completed = ByteBuffer.allocateDirect( 4 ); - + mAvailableTransfers.drainTo( transfers ); + + for( Transfer transfer: transfers ) + { + int result = LibUsb.submitTransfer( transfer ); + + if( result == LibUsb.SUCCESS ) + { + mTransfersInProgress.add( transfer ); + } + else + { + mLog.error( "error submitting transfer [" + + LibUsb.errorName( result ) + "]" ); + } + } + int result = LibUsb.handleEventsTimeoutCompleted( - null, USB_TIMEOUT_MS, completed.asIntBuffer() ); + null, USB_TIMEOUT_MS, mLibUsbHandlerStatus.asIntBuffer() ); if( result != LibUsb.SUCCESS ) { mLog.error( "error handling events for libusb" ); } + + transfers.clear(); + + mLibUsbHandlerStatus.rewind(); + } + + if( mCancel ) + { + for( Transfer transfer: mTransfersInProgress ) + { + LibUsb.cancelTransfer( transfer ); + } + + int result = LibUsb.handleEventsTimeoutCompleted( null, + USB_TIMEOUT_MS, mLibUsbHandlerStatus.asIntBuffer() ); + + if( result != LibUsb.SUCCESS ) + { + mLog.error( "error handling events for libusb" ); + } + + mLibUsbHandlerStatus.rewind(); } } } @@ -1273,10 +1336,10 @@ public class AirspyTunerController extends TunerController */ public void stop() { + mCancel = true; + if( mRunning.compareAndSet( true, false ) ) { - cancel(); - if( mSampleDispatcherTask != null ) { mSampleDispatcherTask.cancel( true ); @@ -1296,66 +1359,39 @@ public class AirspyTunerController extends TunerController @Override public void processTransfer( Transfer transfer ) { - if( transfer.status() == LibUsb.TRANSFER_COMPLETED ) - { - ByteBuffer buffer = transfer.buffer(); - - byte[] data = new byte[ transfer.actualLength() ]; - - buffer.get( data ); - - buffer.rewind(); - - mFilledBuffers.add( data ); - - if( !isRunning() ) - { - LibUsb.cancelTransfer( transfer ); - } - } + mTransfersInProgress.remove( transfer ); switch( transfer.status() ) { case LibUsb.TRANSFER_COMPLETED: - /* resubmit the transfer */ - int result = LibUsb.submitTransfer( transfer ); - - if( result != LibUsb.SUCCESS ) + case LibUsb.TRANSFER_STALL: + if( transfer.actualLength() > 0 ) { - mLog.error( "couldn't resubmit buffer transfer to tuner" ); - LibUsb.freeTransfer( transfer ); - mTransfers.remove( transfer ); + ByteBuffer buffer = transfer.buffer(); + + byte[] data = new byte[ transfer.actualLength() ]; + + buffer.get( data ); + + buffer.rewind(); + + if( isRunning() ) + { + mFilledBuffers.add( data ); + } } break; case LibUsb.TRANSFER_CANCELLED: - /* free the transfer and remove it */ - LibUsb.freeTransfer( transfer ); - mTransfers.remove( transfer ); break; default: /* unexpected error */ - mLog.error( "transfer error" ); + mLog.error( "transfer error [" + + getTransferStatus( transfer.status() ) + + "] transferred actual: " + transfer.actualLength() ); } - } - - private void cancel() - { - for( Transfer transfer: mTransfers ) - { - LibUsb.cancelTransfer( transfer ); - } - - ByteBuffer completed = ByteBuffer.allocateDirect( 4 ); - - int result = LibUsb.handleEventsTimeoutCompleted( - null, USB_TIMEOUT_MS, completed.asIntBuffer() ); - if( result != LibUsb.SUCCESS ) - { - mLog.error( "error handling usb events during cancel [" + - LibUsb.errorName( result ) + "]" ); - } - } + mAvailableTransfers.add( transfer ); + } /** * Prepares (allocates) a set of transfer buffers for use in @@ -1363,30 +1399,28 @@ public class AirspyTunerController extends TunerController */ private void prepareTransfers() throws LibUsbException { - mTransfers = new CopyOnWriteArrayList(); - - for( int x = 0; x < TRANSFER_BUFFER_POOL_SIZE; x++ ) + if( mAvailableTransfers == null ) { - Transfer transfer = LibUsb.allocTransfer(); + mAvailableTransfers = new LinkedTransferQueue<>(); + + for( int x = 0; x < TRANSFER_BUFFER_POOL_SIZE; x++ ) + { + Transfer transfer = LibUsb.allocTransfer(); - if( transfer == null ) - { - throw new LibUsbException( "couldn't allocate transfer", - LibUsb.ERROR_NO_MEM ); - } - - final ByteBuffer buffer = - ByteBuffer.allocateDirect( mBufferSize ); - - LibUsb.fillBulkTransfer( transfer, - mDeviceHandle, - USB_ENDPOINT, - buffer, - BufferProcessor.this, - "Buffer #" + x, - USB_TIMEOUT_MS ); - - mTransfers.add( transfer ); + if( transfer == null ) + { + throw new LibUsbException( "couldn't allocate transfer", + LibUsb.ERROR_NO_MEM ); + } + + final ByteBuffer buffer = + ByteBuffer.allocateDirect( mBufferSize ); + + LibUsb.fillBulkTransfer( transfer, mDeviceHandle, USB_ENDPOINT, + buffer, BufferProcessor.this, "Buffer", USB_TIMEOUT_MS ); + + mAvailableTransfers.add( transfer ); + } } } } diff --git a/src/source/tuner/hackrf/HackRFTunerController.java b/src/source/tuner/hackrf/HackRFTunerController.java index bdc3bce2..6c1074ba 100644 --- a/src/source/tuner/hackrf/HackRFTunerController.java +++ b/src/source/tuner/hackrf/HackRFTunerController.java @@ -56,6 +56,7 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.util.ArrayList; import java.util.Arrays; +import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.LinkedTransferQueue; import java.util.concurrent.ScheduledFuture; @@ -91,7 +92,7 @@ public class HackRFTunerController extends TunerController private final static Logger mLog = LoggerFactory.getLogger( HackRFTunerController.class ); - public final static long TIMEOUT_US = 1000000l; //uSeconds + public final static long USB_TIMEOUT_MS = 1000000l; //uSeconds public static final byte USB_ENDPOINT = (byte)0x81; public static final byte USB_INTERFACE = (byte)0x0; @@ -394,7 +395,7 @@ public class HackRFTunerController extends TunerController (short)value, (short)index, buffer, - TIMEOUT_US ); + USB_TIMEOUT_MS ); if( transferred < 0 ) { @@ -467,7 +468,7 @@ public class HackRFTunerController extends TunerController (short)value, (short)index, buffer, - TIMEOUT_US ); + USB_TIMEOUT_MS ); if( transferred < 0 ) { @@ -886,18 +887,21 @@ public class HackRFTunerController extends TunerController */ public void addListener( Listener listener ) { - mComplexBufferBroadcaster.addListener( listener ); + synchronized( mComplexBufferBroadcaster ) + { + mComplexBufferBroadcaster.addListener( listener ); - if( mBufferProcessor == null || !mBufferProcessor.isRunning() ) - { - mBufferProcessor = new BufferProcessor(); + if( mBufferProcessor == null || !mBufferProcessor.isRunning() ) + { + mBufferProcessor = new BufferProcessor(); - Thread thread = new Thread( mBufferProcessor ); - thread.setDaemon( true ); - thread.setName( "HackRF Sample Processor" ); + Thread thread = new Thread( mBufferProcessor ); + thread.setDaemon( true ); + thread.setName( "HackRF Sample Processor" ); - thread.start(); - } + thread.start(); + } + } } /** @@ -906,12 +910,15 @@ public class HackRFTunerController extends TunerController */ public void removeListener( Listener listener ) { - mComplexBufferBroadcaster.removeListener( listener ); - - if( !mComplexBufferBroadcaster.hasListeners() ) - { - mBufferProcessor.stop(); - } + synchronized( mComplexBufferBroadcaster ) + { + mComplexBufferBroadcaster.removeListener( listener ); + + if( !mComplexBufferBroadcaster.hasListeners() ) + { + mBufferProcessor.stop(); + } + } } /** @@ -921,45 +928,77 @@ public class HackRFTunerController extends TunerController public class BufferProcessor implements Runnable, TransferCallback { private ScheduledFuture mSampleDispatcherTask; - private CopyOnWriteArrayList mTransfers; + private LinkedTransferQueue mAvailableTransfers; + private LinkedTransferQueue mTransfersInProgress = new LinkedTransferQueue<>(); private AtomicBoolean mRunning = new AtomicBoolean(); + private ByteBuffer mLibUsbHandlerStatus; + private boolean mCancel = false; @Override public void run() { if( mRunning.compareAndSet( false, true ) ) { - mLog.info( "starting HackRF sample fetch thread" ); - prepareTransfers(); - for( Transfer transfer: mTransfers ) - { - int result = LibUsb.submitTransfer( transfer ); - - if( result != LibUsb.SUCCESS ) - { - mLog.error( "error submitting transfer [" + - LibUsb.errorName( result ) + "]" ); - break; - } - } - mSampleDispatcherTask = mThreadPoolManager.scheduleFixedRate( ThreadType.SOURCE_SAMPLE_PROCESSING, new BufferDispatcher(), 20, TimeUnit.MILLISECONDS ); + mLibUsbHandlerStatus = ByteBuffer.allocateDirect( 4 ); + + List transfers = new ArrayList<>(); + + mCancel = false; + while( mRunning.get() ) { - ByteBuffer completed = ByteBuffer.allocateDirect( 4 ); + mAvailableTransfers.drainTo( transfers ); + + for( Transfer transfer: transfers ) + { + int result = LibUsb.submitTransfer( transfer ); + + if( result == LibUsb.SUCCESS ) + { + mTransfersInProgress.add( transfer ); + } + else + { + mLog.error( "error submitting transfer [" + + LibUsb.errorName( result ) + "]" ); + } + } int result = LibUsb.handleEventsTimeoutCompleted( - null, TIMEOUT_US, completed.asIntBuffer() ); + null, USB_TIMEOUT_MS, mLibUsbHandlerStatus.asIntBuffer() ); if( result != LibUsb.SUCCESS ) { mLog.error( "error handling events for libusb" ); } + + transfers.clear(); + + mLibUsbHandlerStatus.rewind(); + } + + if( mCancel ) + { + for( Transfer transfer: mTransfersInProgress ) + { + LibUsb.cancelTransfer( transfer ); + } + + int result = LibUsb.handleEventsTimeoutCompleted( + null, USB_TIMEOUT_MS, mLibUsbHandlerStatus.asIntBuffer() ); + + if( result != LibUsb.SUCCESS ) + { + mLog.error( "error cancelling transfers in progress" ); + } + + mLibUsbHandlerStatus.rewind(); } } } @@ -969,12 +1008,10 @@ public class HackRFTunerController extends TunerController */ public void stop() { + mCancel = true; + if( mRunning.compareAndSet( true, false ) ) { - mLog.info( "stopping HackRF sample fetch thread" ); - - cancel(); - if( mSampleDispatcherTask != null ) { mSampleDispatcherTask.cancel( true ); @@ -994,41 +1031,29 @@ public class HackRFTunerController extends TunerController @Override public void processTransfer( Transfer transfer ) { - if( transfer.status() == LibUsb.TRANSFER_COMPLETED ) - { - ByteBuffer buffer = transfer.buffer(); - - byte[] data = new byte[ transfer.actualLength() ]; - - buffer.get( data ); - - buffer.rewind(); - - mFilledBuffers.add( data ); - - if( !isRunning() ) - { - LibUsb.cancelTransfer( transfer ); - } - } + mTransfersInProgress.remove( transfer ); switch( transfer.status() ) { case LibUsb.TRANSFER_COMPLETED: - /* resubmit the transfer */ - int result = LibUsb.submitTransfer( transfer ); - - if( result != LibUsb.SUCCESS ) + case LibUsb.TRANSFER_STALL: + if( transfer.actualLength() > 0 ) { - mLog.error( "couldn't resubmit buffer transfer to tuner" ); - LibUsb.freeTransfer( transfer ); - mTransfers.remove( transfer ); + ByteBuffer buffer = transfer.buffer(); + + byte[] data = new byte[ transfer.actualLength() ]; + + buffer.get( data ); + + buffer.rewind(); + + if( isRunning() ) + { + mFilledBuffers.add( data ); + } } break; case LibUsb.TRANSFER_CANCELLED: - /* free the transfer and remove it */ - LibUsb.freeTransfer( transfer ); - mTransfers.remove( transfer ); break; default: /* unexpected error */ @@ -1036,26 +1061,9 @@ public class HackRFTunerController extends TunerController getTransferStatus( transfer.status() ) + "] transferred actual: " + transfer.actualLength() ); } - } - - private void cancel() - { - for( Transfer transfer: mTransfers ) - { - LibUsb.cancelTransfer( transfer ); - } - - ByteBuffer completed = ByteBuffer.allocateDirect( 4 ); - - int result = LibUsb.handleEventsTimeoutCompleted( - null, TIMEOUT_US, completed.asIntBuffer() ); - if( result != LibUsb.SUCCESS ) - { - mLog.error( "error handling usb events during cancel [" + - LibUsb.errorName( result ) + "]" ); - } - } + mAvailableTransfers.add( transfer ); + } /** * Prepares (allocates) a set of transfer buffers for use in @@ -1063,30 +1071,28 @@ public class HackRFTunerController extends TunerController */ private void prepareTransfers() throws LibUsbException { - mTransfers = new CopyOnWriteArrayList(); - - for( int x = 0; x < TRANSFER_BUFFER_POOL_SIZE; x++ ) + if( mAvailableTransfers == null ) { - Transfer transfer = LibUsb.allocTransfer(); + mAvailableTransfers = new LinkedTransferQueue<>(); + + for( int x = 0; x < TRANSFER_BUFFER_POOL_SIZE; x++ ) + { + Transfer transfer = LibUsb.allocTransfer(); - if( transfer == null ) - { - throw new LibUsbException( "couldn't allocate transfer", - LibUsb.ERROR_NO_MEM ); - } - - final ByteBuffer buffer = - ByteBuffer.allocateDirect( mBufferSize ); - - LibUsb.fillBulkTransfer( transfer, - mDeviceHandle, - USB_ENDPOINT, - buffer, - BufferProcessor.this, - "Buffer #" + x, - TIMEOUT_US ); - - mTransfers.add( transfer ); + if( transfer == null ) + { + throw new LibUsbException( "couldn't allocate transfer", + LibUsb.ERROR_NO_MEM ); + } + + final ByteBuffer buffer = + ByteBuffer.allocateDirect( mBufferSize ); + + LibUsb.fillBulkTransfer( transfer, mDeviceHandle,USB_ENDPOINT, + buffer, BufferProcessor.this, "Buffer", USB_TIMEOUT_MS ); + + mAvailableTransfers.add( transfer ); + } } } } diff --git a/src/source/tuner/rtl/RTL2832TunerController.java b/src/source/tuner/rtl/RTL2832TunerController.java index df6c1750..fe37f8ad 100644 --- a/src/source/tuner/rtl/RTL2832TunerController.java +++ b/src/source/tuner/rtl/RTL2832TunerController.java @@ -28,7 +28,6 @@ import java.text.DecimalFormat; import java.util.ArrayList; import java.util.Arrays; import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.LinkedTransferQueue; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -72,7 +71,7 @@ public abstract class RTL2832TunerController extends TunerController public final static int TWO_TO_22_POWER = 4194304; public final static byte USB_INTERFACE = (byte)0x0; - public final static byte BULK_ENDPOINT_IN = (byte)0x81; + public final static byte USB_BULK_ENDPOINT = (byte)0x81; public final static byte CONTROL_ENDPOINT_IN = (byte)( LibUsb.ENDPOINT_IN | LibUsb.REQUEST_TYPE_VENDOR ); public final static byte CONTROL_ENDPOINT_OUT = @@ -1387,7 +1386,6 @@ public abstract class RTL2832TunerController extends TunerController private ScheduledFuture mSampleDispatcherTask; private LinkedTransferQueue mAvailableTransfers; private LinkedTransferQueue mTransfersInProgress = new LinkedTransferQueue<>(); - private CopyOnWriteArrayList mTransfers; private AtomicBoolean mRunning = new AtomicBoolean(); private ThreadPoolManager mThreadPoolManager; private ByteBuffer mLibUsbHandlerStatus; @@ -1577,7 +1575,7 @@ public abstract class RTL2832TunerController extends TunerController final ByteBuffer buffer = ByteBuffer.allocateDirect( mBufferSize ); - LibUsb.fillBulkTransfer( transfer, mDeviceHandle, BULK_ENDPOINT_IN, + LibUsb.fillBulkTransfer( transfer, mDeviceHandle, USB_BULK_ENDPOINT, buffer, BufferProcessor.this, "Buffer", TIMEOUT_US ); mAvailableTransfers.add( transfer );