Resolves significant memory usage issue - updates Airspy and HackRF with

the recent changes applied to RTL2832 Tuner controller for processing of
libusb transfer buffers.

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<ComplexBuffer> listener )
     {
-    	mComplexBufferBroadcaster.addListener( listener );
+    	synchronized( mComplexBufferBroadcaster )
+    	{
+        	mComplexBufferBroadcaster.addListener( listener );
 
-		if( mBufferProcessor == null || !mBufferProcessor.isRunning() )
+    		if( mBufferProcessor == null || !mBufferProcessor.isRunning() )
-		{
+    		{
-			mBufferProcessor = new BufferProcessor();
+    			mBufferProcessor = new BufferProcessor();
 
-			Thread thread = new Thread( mBufferProcessor );
+    			Thread thread = new Thread( mBufferProcessor );
-			thread.setDaemon( true );
+    			thread.setDaemon( true );
-			thread.setName( "Airspy Buffer Processor" );
+    			thread.setName( "Airspy Buffer Processor" );
 
-			thread.start();
+    			thread.start();
-		}
+    		}
+    	}
     }
 
 	/**
@@ -1212,12 +1238,15 @@ public class AirspyTunerController extends TunerController
 	 */
     public void removeListener( Listener<ComplexBuffer> listener )
     {
-    	mComplexBufferBroadcaster.removeListener( listener );
+    	synchronized( mComplexBufferBroadcaster )
-    	
+    	{
-		if( !mComplexBufferBroadcaster.hasListeners() )
+        	mComplexBufferBroadcaster.removeListener( listener );
-		{
+        	
-			mBufferProcessor.stop();
+    		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<Transfer> mTransfers;
+		private LinkedTransferQueue<Transfer> mAvailableTransfers;
+		private LinkedTransferQueue<Transfer> 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<Transfer> 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 )
+			mTransfersInProgress.remove( transfer );
-			{
-				ByteBuffer buffer = transfer.buffer();
-				
-				byte[] data = new byte[ transfer.actualLength() ];
-				
-				buffer.get( data );
-
-				buffer.rewind();
-
-				mFilledBuffers.add( data );
-				
-				if( !isRunning() )
-				{
-					LibUsb.cancelTransfer( transfer );
-				}
-			}
 			
 			switch( transfer.status() )
 			{
 				case LibUsb.TRANSFER_COMPLETED:
-					/* resubmit the transfer */
+				case LibUsb.TRANSFER_STALL:
-					int result = LibUsb.submitTransfer( transfer );
+					if( transfer.actualLength() > 0 )
-					
-					if( result != LibUsb.SUCCESS )
 					{
-						mLog.error( "couldn't resubmit buffer transfer to tuner" );
+						ByteBuffer buffer = transfer.buffer();
-						LibUsb.freeTransfer( transfer );
+						
-						mTransfers.remove( transfer );
+						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 )
+			mAvailableTransfers.add( transfer );
-			{
+	    }
-				mLog.error( "error handling usb events during cancel [" + 
-						LibUsb.errorName( result ) + "]" );
-			}
-		}
 		
 	    /**
 	     * 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<Transfer>();
+	    	if( mAvailableTransfers == null )
-
-	    	for( int x = 0; x < TRANSFER_BUFFER_POOL_SIZE; x++ )
 	    	{
-	    		Transfer transfer = LibUsb.allocTransfer();
+	    		mAvailableTransfers = new LinkedTransferQueue<>();
+	    		
+		    	for( int x = 0; x < TRANSFER_BUFFER_POOL_SIZE; x++ )
+		    	{
+		    		Transfer transfer = LibUsb.allocTransfer();
 
-	    		if( transfer == null )
+		    		if( transfer == null )
-	    		{
+		    		{
-	    			throw new LibUsbException( "couldn't allocate transfer", 
+		    			throw new LibUsbException( "couldn't allocate transfer", 
-	    						LibUsb.ERROR_NO_MEM );
+		    						LibUsb.ERROR_NO_MEM );
-	    		}
+		    		}
-	    		
+		    		
-	    		final ByteBuffer buffer = 
+		    		final ByteBuffer buffer = 
-	    				ByteBuffer.allocateDirect( mBufferSize );
+		    				ByteBuffer.allocateDirect( mBufferSize );
-	    		
+		    		
-	    		LibUsb.fillBulkTransfer( transfer, 
+		    		LibUsb.fillBulkTransfer( transfer, mDeviceHandle, USB_ENDPOINT, 
-	    								 mDeviceHandle, 
+	    				buffer, BufferProcessor.this, "Buffer", USB_TIMEOUT_MS );
-	    								 USB_ENDPOINT, 
+
-	    								 buffer, 
+		    		mAvailableTransfers.add( transfer );
-	    								 BufferProcessor.this, 
+		    	}
-	    								 "Buffer #" + x,
-	    								 USB_TIMEOUT_MS );
-	    		
-	    		mTransfers.add( transfer );
 	    	}
 	    }
 	}

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<ComplexBuffer> listener )
     {
-    	mComplexBufferBroadcaster.addListener( listener );
+    	synchronized( mComplexBufferBroadcaster )
+    	{
+        	mComplexBufferBroadcaster.addListener( listener );
 
-		if( mBufferProcessor == null || !mBufferProcessor.isRunning() )
+    		if( mBufferProcessor == null || !mBufferProcessor.isRunning() )
-		{
+    		{
-			mBufferProcessor = new BufferProcessor();
+    			mBufferProcessor = new BufferProcessor();
 
-			Thread thread = new Thread( mBufferProcessor );
+    			Thread thread = new Thread( mBufferProcessor );
-			thread.setDaemon( true );
+    			thread.setDaemon( true );
-			thread.setName( "HackRF Sample Processor" );
+    			thread.setName( "HackRF Sample Processor" );
 
-			thread.start();
+    			thread.start();
-		}
+    		}
+    	}
     }
 
 	/**
@@ -906,12 +910,15 @@ public class HackRFTunerController extends TunerController
 	 */
     public void removeListener( Listener<ComplexBuffer> listener )
     {
-    	mComplexBufferBroadcaster.removeListener( listener );
+    	synchronized( mComplexBufferBroadcaster )
-    	
+    	{
-		if( !mComplexBufferBroadcaster.hasListeners() )
+        	mComplexBufferBroadcaster.removeListener( listener );
-		{
+        	
-			mBufferProcessor.stop();
+    		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<Transfer> mTransfers;
+		private LinkedTransferQueue<Transfer> mAvailableTransfers;
+		private LinkedTransferQueue<Transfer> 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<Transfer> 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 )
+			mTransfersInProgress.remove( transfer );
-			{
-				ByteBuffer buffer = transfer.buffer();
-				
-				byte[] data = new byte[ transfer.actualLength() ];
-				
-				buffer.get( data );
-
-				buffer.rewind();
-
-				mFilledBuffers.add( data );
-				
-				if( !isRunning() )
-				{
-					LibUsb.cancelTransfer( transfer );
-				}
-			}
 			
 			switch( transfer.status() )
 			{
 				case LibUsb.TRANSFER_COMPLETED:
-					/* resubmit the transfer */
+				case LibUsb.TRANSFER_STALL:
-					int result = LibUsb.submitTransfer( transfer );
+					if( transfer.actualLength() > 0 )
-					
-					if( result != LibUsb.SUCCESS )
 					{
-						mLog.error( "couldn't resubmit buffer transfer to tuner" );
+						ByteBuffer buffer = transfer.buffer();
-						LibUsb.freeTransfer( transfer );
+						
-						mTransfers.remove( transfer );
+						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 )
+			mAvailableTransfers.add( transfer );
-			{
+	    }
-				mLog.error( "error handling usb events during cancel [" + 
-						LibUsb.errorName( result ) + "]" );
-			}
-		}
 		
 	    /**
 	     * 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<Transfer>();
+	    	if( mAvailableTransfers == null )
-
-	    	for( int x = 0; x < TRANSFER_BUFFER_POOL_SIZE; x++ )
 	    	{
-	    		Transfer transfer = LibUsb.allocTransfer();
+	    		mAvailableTransfers = new LinkedTransferQueue<>();
+	    		
+		    	for( int x = 0; x < TRANSFER_BUFFER_POOL_SIZE; x++ )
+		    	{
+		    		Transfer transfer = LibUsb.allocTransfer();
 
-	    		if( transfer == null )
+		    		if( transfer == null )
-	    		{
+		    		{
-	    			throw new LibUsbException( "couldn't allocate transfer", 
+		    			throw new LibUsbException( "couldn't allocate transfer", 
-	    						LibUsb.ERROR_NO_MEM );
+		    						LibUsb.ERROR_NO_MEM );
-	    		}
+		    		}
-	    		
+		    		
-	    		final ByteBuffer buffer = 
+		    		final ByteBuffer buffer = 
-	    				ByteBuffer.allocateDirect( mBufferSize );
+		    				ByteBuffer.allocateDirect( mBufferSize );
-	    		
+		    		
-	    		LibUsb.fillBulkTransfer( transfer, 
+		    		LibUsb.fillBulkTransfer( transfer, mDeviceHandle,USB_ENDPOINT, 
-	    								 mDeviceHandle, 
+	    				buffer, BufferProcessor.this, "Buffer", USB_TIMEOUT_MS );
-	    								 USB_ENDPOINT, 
+
-	    								 buffer, 
+		    		mAvailableTransfers.add( transfer );
-	    								 BufferProcessor.this, 
+		    	}
-	    								 "Buffer #" + x,
-	    								 TIMEOUT_US );
-	    		
-	    		mTransfers.add( transfer );
 	    	}
 	    }
 	}

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<Transfer> mAvailableTransfers;
 		private LinkedTransferQueue<Transfer> mTransfersInProgress = new LinkedTransferQueue<>();
-        private CopyOnWriteArrayList<Transfer> 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 );