From 6d458abe7e82563a5bdac67401d699e7b22129e1 Mon Sep 17 00:00:00 2001 From: Simon Rettberg Date: Tue, 30 Aug 2016 17:57:04 +0200 Subject: Fix some minor handling issues with transfers wrt global image exchange --- .../openslx/filetransfer/util/AbstractTransfer.java | 8 ++++++++ .../filetransfer/util/IncomingTransferBase.java | 19 +++++++++++++++---- .../filetransfer/util/OutgoingTransferBase.java | 6 ++++++ 3 files changed, 29 insertions(+), 4 deletions(-) diff --git a/src/main/java/org/openslx/filetransfer/util/AbstractTransfer.java b/src/main/java/org/openslx/filetransfer/util/AbstractTransfer.java index 75c68e8..636f880 100644 --- a/src/main/java/org/openslx/filetransfer/util/AbstractTransfer.java +++ b/src/main/java/org/openslx/filetransfer/util/AbstractTransfer.java @@ -1,6 +1,7 @@ package org.openslx.filetransfer.util; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import org.openslx.bwlp.thrift.iface.TransferInformation; @@ -43,6 +44,8 @@ public abstract class AbstractTransfer protected final AtomicLong lastActivityTime = new AtomicLong( System.currentTimeMillis() ); private final String transferId; + + protected AtomicInteger connectFails = new AtomicInteger(); public AbstractTransfer( String transferId ) { @@ -77,6 +80,11 @@ public abstract class AbstractTransfer { return getActiveConnectionCount() > 0 || lastActivityTime.get() + HOT_IDLE_TIMEOUT > now; } + + public final int connectFailCount() + { + return connectFails.get(); + } public final String getId() { diff --git a/src/main/java/org/openslx/filetransfer/util/IncomingTransferBase.java b/src/main/java/org/openslx/filetransfer/util/IncomingTransferBase.java index 92c26c9..32fdf6c 100644 --- a/src/main/java/org/openslx/filetransfer/util/IncomingTransferBase.java +++ b/src/main/java/org/openslx/filetransfer/util/IncomingTransferBase.java @@ -342,6 +342,9 @@ public abstract class IncomingTransferBase extends AbstractTransfer implements H cancel(); return null; } + if ( state == TransferState.IDLE ) { + state = TransferState.WORKING; + } return currentChunk.range; } } @@ -365,7 +368,10 @@ public abstract class IncomingTransferBase extends AbstractTransfer implements H public void run() { CbHandler cbh = new CbHandler( connection ); - if ( !connection.download( cbh, cbh ) ) { + if ( connection.download( cbh, cbh ) ) { + connectFails.set( 0 ); + } else { + connectFails.incrementAndGet(); if ( cbh.currentChunk != null ) { // If the download failed and we have a current chunk, put it back into // the queue, so it will be handled again later... @@ -430,7 +436,7 @@ public abstract class IncomingTransferBase extends AbstractTransfer implements H { synchronized ( tmpFileHandle ) { if ( state != TransferState.WORKING ) - throw new IllegalStateException( "Cannot write to file if state != WORKING" ); + throw new IllegalStateException( "Cannot write to file if state != WORKING (is " + state.toString() + ")" ); try { tmpFileHandle.seek( fileOffset ); tmpFileHandle.write( data, 0, dataLength ); @@ -459,9 +465,14 @@ public abstract class IncomingTransferBase extends AbstractTransfer implements H // Fall through case VALID: if ( !chunk.isWrittenToDisk() ) { - writeFileData( chunk.range.startOffset, chunk.range.getLength(), data ); + try { + writeFileData( chunk.range.startOffset, chunk.range.getLength(), data ); + chunks.markCompleted( chunk, true ); + } catch ( Exception e ) { + LOGGER.warn( "Cannot write to file after hash check", e ); + chunks.markFailed( chunk ); + } } - chunks.markCompleted( chunk, true ); chunkStatusChanged( chunk ); if ( chunks.isComplete() ) { finishUploadInternal(); diff --git a/src/main/java/org/openslx/filetransfer/util/OutgoingTransferBase.java b/src/main/java/org/openslx/filetransfer/util/OutgoingTransferBase.java index 12deddb..18296c5 100644 --- a/src/main/java/org/openslx/filetransfer/util/OutgoingTransferBase.java +++ b/src/main/java/org/openslx/filetransfer/util/OutgoingTransferBase.java @@ -78,9 +78,15 @@ public abstract class OutgoingTransferBase extends AbstractTransfer synchronized ( uploads ) { uploads.remove( connection ); } + if ( ret ) { + connectFails.set( 0 ); + } if ( ret && uploads.isEmpty() && potentialFinishTime.get() == 0 ) { potentialFinishTime.set( System.currentTimeMillis() ); } + if ( !ret && uploads.isEmpty() ) { + connectFails.incrementAndGet(); + } lastActivityTime.set( System.currentTimeMillis() ); } } ); -- cgit v1.2.3-55-g7522