Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4382: Fix deadlock in FileWritingMH

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4382

There is a deadlock in the FMWH due to lock ordering.

Writes try to lock the handler monitor while holding the file lock.
The `Flusher` tries to lock the file lock while holding th handler monitor.

- Change the `Flusher`, `stop` and `flushIfNeeded` methods to not call close
  on the file state while holding the handler monitor.

__cherry pick to 4.3.x__ (fix <> for Java 6)

Polishing - PR Comments

* Optimize `flushIfNeeded(FlushPredicate)` to delegate only to the
`flushIfNeeded(MessageFlushPredicate, Message<?>)`

# Conflicts:
#	spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java
This commit is contained in:
Gary Russell
2018-01-19 10:26:42 -05:00
committed by Artem Bilan
parent 59909b64ab
commit 21c4932abe
1 changed files with 91 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java
View File

@@ -365,13 +365,13 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
if (this.destinationDirectoryExpression instanceof LiteralExpression) {
final File directory = new File(this.destinationDirectoryExpression.getValue(
this.evaluationContext, null, String.class));
String directoryPath = this.destinationDirectoryExpression.getValue(this.evaluationContext, String.class);
final File directory = new File(directoryPath);
validateDestinationDirectory(directory, this.autoCreateDirectory);
}
Assert.state(!(this.temporaryFileSuffixSet
&& (FileExistsMode.APPEND.equals(this.fileExistsMode)
&& (FileExistsMode.APPEND.equals(this.fileExistsMode)
|| FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode))),
"'temporaryFileSuffix' can not be set when appending to an existing file");
@@ -391,12 +391,32 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
}
@Override
public synchronized void stop() {
if (this.flushTask != null) {
this.flushTask.cancel(true);
this.flushTask = null;
public void stop() {
synchronized (this) {
if (this.flushTask != null) {
this.flushTask.cancel(true);
this.flushTask = null;
}
}
Flusher flusher = new Flusher();
flusher.run();
boolean needInterrupt = this.fileStates.size() > 0;
int n = 0;
while (n++ < 10 && this.fileStates.size() > 0) {
try {
Thread.sleep(1);
}
catch (InterruptedException e) {
// cancel the interrupt
}
flusher.run();
}
if (this.fileStates.size() > 0) {
this.logger.error("Failed to flush after multiple attempts, while stopping: " + this.fileStates.keySet());
}
if (needInterrupt) {
Thread.currentThread().interrupt();
}
new Flusher().run();
}
@Override
@@ -529,7 +549,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
}
private File handleInputStreamMessage(final InputStream sourceFileInputStream, File originalFile, File tempFile,
final File resultFile) throws IOException {
final File resultFile) throws IOException {
final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode)
|| FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode);
@@ -796,10 +816,10 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
private synchronized FileState getFileState(final File fileToWriteTo, boolean isString)
throws FileNotFoundException {
String absolutePath = fileToWriteTo.getAbsolutePath();
FileState state;
boolean appendNoFlush = FileExistsMode.APPEND_NO_FLUSH.equals(this.fileExistsMode);
if (appendNoFlush) {
String absolutePath = fileToWriteTo.getAbsolutePath();
state = this.fileStates.get(absolutePath);
if (state != null && ((isString && state.stream != null) || (!isString && state.writer != null))) {
state.close();
@@ -872,16 +892,19 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
* @param flushPredicate the {@link FlushPredicate}.
* @since 4.3
*/
public synchronized void flushIfNeeded(FlushPredicate flushPredicate) {
Iterator<Entry<String, FileState>> iterator = FileWritingMessageHandler.this.fileStates.entrySet().iterator();
while (iterator.hasNext()) {
Entry<String, FileState> entry = iterator.next();
FileState state = entry.getValue();
if (flushPredicate.shouldFlush(entry.getKey(), state.lastWrite)) {
iterator.remove();
state.close();
}
}
public void flushIfNeeded(final FlushPredicate flushPredicate) {
flushIfNeeded(new MessageFlushPredicate() {
@Override
public boolean shouldFlush(String fileAbsolutePath, long lastWrite,
Message<?> filterMessage) {
return flushPredicate.shouldFlush(fileAbsolutePath, lastWrite);
}
},
null);
}
/**
@@ -893,16 +916,24 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
* @param filterMessage an optional message passed into the predicate.
* @since 4.3
*/
public synchronized void flushIfNeeded(MessageFlushPredicate flushPredicate, Message<?> filterMessage) {
Iterator<Entry<String, FileState>> iterator = FileWritingMessageHandler.this.fileStates.entrySet().iterator();
while (iterator.hasNext()) {
Entry<String, FileState> entry = iterator.next();
FileState state = entry.getValue();
if (flushPredicate.shouldFlush(entry.getKey(), state.lastWrite, filterMessage)) {
iterator.remove();
state.close();
public void flushIfNeeded(MessageFlushPredicate flushPredicate, Message<?> filterMessage) {
doFlush(findFilesToFlush(flushPredicate, filterMessage));
}
private Map<String, FileState> findFilesToFlush(MessageFlushPredicate flushPredicate, Message<?> filterMessage) {
Map<String, FileState> toRemove = new HashMap<String, FileState>();
synchronized (this) {
Iterator<Entry<String, FileState>> iterator = this.fileStates.entrySet().iterator();
while (iterator.hasNext()) {
Entry<String, FileState> entry = iterator.next();
FileState state = entry.getValue();
if (flushPredicate.shouldFlush(entry.getKey(), state.lastWrite, filterMessage)) {
iterator.remove();
toRemove.put(entry.getKey(), state);
}
}
}
return toRemove;
}
private synchronized void clearState(final File fileToWriteTo, final FileState state) {
@@ -911,6 +942,33 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
}
}
private void doFlush(Map<String, FileState> toRemove) {
Map<String, FileState> toRestore = new HashMap<String, FileState>();
boolean interrupted = false;
for (Entry<String, FileState> entry : toRemove.entrySet()) {
if (!interrupted && entry.getValue().close()) {
if (FileWritingMessageHandler.this.logger.isDebugEnabled()) {
FileWritingMessageHandler.this.logger.debug("Flushed: " + entry.getKey());
}
}
else { // interrupted (stop), re-add
interrupted = true;
toRestore.put(entry.getKey(), entry.getValue());
}
}
if (interrupted) {
if (FileWritingMessageHandler.this.logger.isDebugEnabled()) {
FileWritingMessageHandler.this.logger
.debug("Interrupted during flush; not flushed: " + toRestore.keySet());
}
synchronized (this) {
for (Entry<String, FileState> entry : toRestore.entrySet()) {
this.fileStates.putIfAbsent(entry.getKey(), entry.getValue());
}
}
}
}
private static boolean rename(File source, File target) throws IOException {
return (nioFilesPresent && filesMove(source, target)) || source.renameTo(target);
}
@@ -925,7 +983,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
private final BufferedWriter writer;
private final BufferedOutputStream stream;
private final BufferedOutputStream stream;
private final Lock lock;
@@ -975,6 +1033,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
@Override
public void run() {
Map<String, FileState> toRemove = new HashMap<String, FileState>();
synchronized (FileWritingMessageHandler.this) {
long expired = FileWritingMessageHandler.this.flushTask == null ? Long.MAX_VALUE
: (System.currentTimeMillis() - FileWritingMessageHandler.this.flushInterval);
@@ -984,18 +1043,12 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
FileState state = entry.getValue();
if (state.lastWrite < expired ||
(!FileWritingMessageHandler.this.flushWhenIdle && state.firstWrite < expired)) {
if (state.close()) {
if (FileWritingMessageHandler.this.logger.isDebugEnabled()) {
FileWritingMessageHandler.this.logger.debug("Flushed: " + entry.getKey());
}
iterator.remove();
}
else {
break; // interrupted
}
toRemove.put(entry.getKey(), state);
iterator.remove();
}
}
}
doFlush(toRemove);
}
}