Issue #59 - reorganizes management of the libusb transfers queue to

better handle processing and shutdowns.

-Reuses transfers between tuner start/stop invocations

-Moves libusb transfer status direct byte buffer allocation out of
buffer processing thread and reuses it, saving a lot of memory
allocations over time.
This commit is contained in:
denny
2016-02-26 14:35:34 -05:00
parent 22d3871d0b
commit 5f2d8c8fe6

View File

@@ -27,6 +27,7 @@ import java.nio.charset.Charset;
import java.text.DecimalFormat; import java.text.DecimalFormat;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedTransferQueue; import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledFuture;
@@ -1378,9 +1379,13 @@ public abstract class RTL2832TunerController extends TunerController
public class BufferProcessor implements Runnable, TransferCallback public class BufferProcessor implements Runnable, TransferCallback
{ {
private ScheduledFuture<?> mSampleDispatcherTask; private ScheduledFuture<?> mSampleDispatcherTask;
private LinkedTransferQueue<Transfer> mAvailableTransfers;
private LinkedTransferQueue<Transfer> mTransfersInProgress = new LinkedTransferQueue<>();
private CopyOnWriteArrayList<Transfer> mTransfers; private CopyOnWriteArrayList<Transfer> mTransfers;
private AtomicBoolean mRunning = new AtomicBoolean(); private AtomicBoolean mRunning = new AtomicBoolean();
private ThreadPoolManager mThreadPoolManager; private ThreadPoolManager mThreadPoolManager;
private ByteBuffer mLibUsbHandlerStatus;
private boolean mCancel = false;
public BufferProcessor( ThreadPoolManager manager ) public BufferProcessor( ThreadPoolManager manager )
{ {
@@ -1407,18 +1412,6 @@ public abstract class RTL2832TunerController extends TunerController
prepareTransfers(); prepareTransfers();
for( Transfer transfer: mTransfers )
{
int result = LibUsb.submitTransfer( transfer );
if( result != LibUsb.SUCCESS )
{
mLog.error( "error submitting transfer [" +
LibUsb.errorName( result ) + "]" );
break;
}
}
try try
{ {
mSampleDispatcherTask = mThreadPoolManager.scheduleFixedRate( mSampleDispatcherTask = mThreadPoolManager.scheduleFixedRate(
@@ -1429,20 +1422,62 @@ public abstract class RTL2832TunerController extends TunerController
{ {
mLog.error( "NPE! = tpm null: " + ( mThreadPoolManager == null ), npe ); mLog.error( "NPE! = tpm null: " + ( mThreadPoolManager == null ), npe );
} }
while( mRunning.get() ) 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( int result = LibUsb.handleEventsTimeoutCompleted(
null, TIMEOUT_US, completed.asIntBuffer() ); null, TIMEOUT_US, mLibUsbHandlerStatus.asIntBuffer() );
if( result != LibUsb.SUCCESS ) if( result != LibUsb.SUCCESS )
{ {
mLog.error( "error handling events for libusb" ); 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, TIMEOUT_US, mLibUsbHandlerStatus.asIntBuffer() );
if( result != LibUsb.SUCCESS )
{
mLog.error( "error handling events for libusb" );
}
mLibUsbHandlerStatus.rewind();
}
} }
} }
@@ -1451,10 +1486,10 @@ public abstract class RTL2832TunerController extends TunerController
*/ */
public void stop() public void stop()
{ {
mCancel = true;
if( mRunning.compareAndSet( true, false ) ) if( mRunning.compareAndSet( true, false ) )
{ {
cancel();
if( mSampleDispatcherTask != null ) if( mSampleDispatcherTask != null )
{ {
mThreadPoolManager.cancel( mSampleDispatcherTask ); mThreadPoolManager.cancel( mSampleDispatcherTask );
@@ -1475,6 +1510,8 @@ public abstract class RTL2832TunerController extends TunerController
@Override @Override
public void processTransfer( Transfer transfer ) public void processTransfer( Transfer transfer )
{ {
mTransfersInProgress.remove( transfer );
switch( transfer.status() ) switch( transfer.status() )
{ {
case LibUsb.TRANSFER_COMPLETED: case LibUsb.TRANSFER_COMPLETED:
@@ -1564,30 +1601,31 @@ public abstract class RTL2832TunerController extends TunerController
*/ */
private void prepareTransfers() throws LibUsbException 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<>();
if( transfer == null )
{
throw new LibUsbException( "couldn't allocate transfer",
LibUsb.ERROR_NO_MEM );
}
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 );
}
mAvailableTransfers.add( transfer );
}
}
for( Transfer transfer: mAvailableTransfers )
{
final ByteBuffer buffer = final ByteBuffer buffer =
ByteBuffer.allocateDirect( mBufferSize ); ByteBuffer.allocateDirect( mBufferSize );
LibUsb.fillBulkTransfer( transfer, LibUsb.fillBulkTransfer( transfer, mDeviceHandle, BULK_ENDPOINT_IN,
mDeviceHandle, buffer, BufferProcessor.this, "Buffer", TIMEOUT_US );
BULK_ENDPOINT_IN,
buffer,
BufferProcessor.this,
"Buffer #" + x,
TIMEOUT_US );
mTransfers.add( transfer );
} }
} }
} }