-
Notifications
You must be signed in to change notification settings - Fork 8.8k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
HADOOP-18756. S3A prefetch - CachingBlockManager to use AtomicBoolean for closed flag #5718
Changes from 2 commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -38,6 +38,7 @@ | |
import java.util.Set; | ||
import java.util.concurrent.ConcurrentHashMap; | ||
import java.util.concurrent.TimeUnit; | ||
import java.util.concurrent.atomic.AtomicBoolean; | ||
import java.util.concurrent.locks.ReentrantReadWriteLock; | ||
|
||
import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableSet; | ||
|
@@ -68,7 +69,7 @@ public class SingleFilePerBlockCache implements BlockCache { | |
*/ | ||
private int numGets = 0; | ||
|
||
private boolean closed; | ||
private final AtomicBoolean closed; | ||
|
||
private final PrefetchingStatistics prefetchingStatistics; | ||
|
||
|
@@ -174,6 +175,7 @@ private boolean takeLock(LockType lockType, long timeout, TimeUnit unit) { | |
*/ | ||
public SingleFilePerBlockCache(PrefetchingStatistics prefetchingStatistics) { | ||
this.prefetchingStatistics = requireNonNull(prefetchingStatistics); | ||
this.closed = new AtomicBoolean(false); | ||
} | ||
|
||
/** | ||
|
@@ -207,7 +209,7 @@ public int size() { | |
*/ | ||
@Override | ||
public void get(int blockNumber, ByteBuffer buffer) throws IOException { | ||
if (closed) { | ||
if (closed.get()) { | ||
return; | ||
} | ||
|
||
|
@@ -262,7 +264,7 @@ private Entry getEntry(int blockNumber) { | |
@Override | ||
public void put(int blockNumber, ByteBuffer buffer, Configuration conf, | ||
LocalDirAllocator localDirAllocator) throws IOException { | ||
if (closed) { | ||
if (closed.get()) { | ||
return; | ||
} | ||
|
||
|
@@ -333,37 +335,33 @@ protected Path getCacheFilePath(final Configuration conf, | |
|
||
@Override | ||
public void close() throws IOException { | ||
if (closed) { | ||
return; | ||
} | ||
|
||
closed = true; | ||
if (closed.compareAndSet(false, true)) { | ||
LOG.info(getStats()); | ||
int numFilesDeleted = 0; | ||
|
||
LOG.info(getStats()); | ||
int numFilesDeleted = 0; | ||
|
||
for (Entry entry : blocks.values()) { | ||
boolean lockAcquired = entry.takeLock(Entry.LockType.WRITE, PREFETCH_WRITE_LOCK_TIMEOUT, | ||
PREFETCH_WRITE_LOCK_TIMEOUT_UNIT); | ||
if (!lockAcquired) { | ||
LOG.error("Cache file {} deletion would not be attempted as write lock could not" | ||
+ " be acquired within {} {}", entry.path, PREFETCH_WRITE_LOCK_TIMEOUT, | ||
for (Entry entry : blocks.values()) { | ||
boolean lockAcquired = entry.takeLock(Entry.LockType.WRITE, PREFETCH_WRITE_LOCK_TIMEOUT, | ||
PREFETCH_WRITE_LOCK_TIMEOUT_UNIT); | ||
continue; | ||
} | ||
try { | ||
Files.deleteIfExists(entry.path); | ||
prefetchingStatistics.blockRemovedFromFileCache(); | ||
numFilesDeleted++; | ||
} catch (IOException e) { | ||
LOG.debug("Failed to delete cache file {}", entry.path, e); | ||
} finally { | ||
entry.releaseLock(Entry.LockType.WRITE); | ||
if (!lockAcquired) { | ||
LOG.error("Cache file {} deletion would not be attempted as write lock could not" | ||
+ " be acquired within {} {}", entry.path, PREFETCH_WRITE_LOCK_TIMEOUT, | ||
PREFETCH_WRITE_LOCK_TIMEOUT_UNIT); | ||
continue; | ||
} | ||
try { | ||
Files.deleteIfExists(entry.path); | ||
prefetchingStatistics.blockRemovedFromFileCache(); | ||
numFilesDeleted++; | ||
} catch (IOException e) { | ||
LOG.error("Failed to delete cache file {}", entry.path, e); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ok, so this is moving from debug to error. why so? if we do want to print this, it's not an error. at worst a warn. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. sure let me make it at least warn, this is likely going to be rare event and highlighting it would be helpful debugging stale file records in future. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. with this new addendum (log level change: ERROR to WARN), i re-ran the test suite again (without scale profile, since the addendum change is trivial) There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. re-tested the latest state of the PR with |
||
} finally { | ||
entry.releaseLock(Entry.LockType.WRITE); | ||
} | ||
} | ||
} | ||
|
||
if (numFilesDeleted > 0) { | ||
LOG.info("Deleted {} cache files", numFilesDeleted); | ||
if (numFilesDeleted > 0) { | ||
LOG.info("Deleted {} cache files", numFilesDeleted); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. downgrade to debug and remove the if() test; we can log the zero count too |
||
} | ||
} | ||
} | ||
|
||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
could we log this at debug; prefetching is fairly noisy right now
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
agree, it is noisy, let me downgrade it to debug