keith-turner closed pull request #380: ACCUMULO-4805 Obtain filemanager lock
once when opening files
URL: https://github.com/apache/accumulo/pull/380
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/FileManager.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/FileManager.java
index a707251463..2aaad39ccd 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/FileManager.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/FileManager.java
@@ -172,7 +172,7 @@ public FileManager(AccumuloServerContext context,
VolumeManager fs, int maxOpen,
this.dataCache = dataCache;
this.indexCache = indexCache;
- this.filePermits = new Semaphore(maxOpen, true);
+ this.filePermits = new Semaphore(maxOpen, false);
this.maxOpen = maxOpen;
this.fs = fs;
@@ -234,7 +234,7 @@ private static int
countReaders(Map<String,List<OpenReader>> files) {
return ofl;
}
- private void closeReaders(List<FileSKVIterator> filesToClose) {
+ private void closeReaders(Collection<FileSKVIterator> filesToClose) {
for (FileSKVIterator reader : filesToClose) {
try {
reader.close();
@@ -244,7 +244,7 @@ private void closeReaders(List<FileSKVIterator>
filesToClose) {
}
}
- private List<String> takeOpenFiles(Collection<String> files,
List<FileSKVIterator> reservedFiles, Map<FileSKVIterator,String>
readersReserved) {
+ private List<String> takeOpenFiles(Collection<String> files,
Map<FileSKVIterator,String> readersReserved) {
List<String> filesToOpen = new LinkedList<>(files);
for (Iterator<String> iterator = filesToOpen.iterator();
iterator.hasNext();) {
String file = iterator.next();
@@ -252,7 +252,6 @@ private void closeReaders(List<FileSKVIterator>
filesToClose) {
List<OpenReader> ofl = openFiles.get(file);
if (ofl != null && ofl.size() > 0) {
OpenReader openReader = ofl.remove(ofl.size() - 1);
- reservedFiles.add(openReader.reader);
readersReserved.put(openReader.reader, file);
if (ofl.size() == 0) {
openFiles.remove(file);
@@ -264,23 +263,18 @@ private void closeReaders(List<FileSKVIterator>
filesToClose) {
return filesToOpen;
}
- private synchronized String getReservedReadeFilename(FileSKVIterator reader)
{
- return reservedReaders.get(reader);
- }
-
- private List<FileSKVIterator> reserveReaders(KeyExtent tablet,
Collection<String> files, boolean continueOnFailure) throws IOException {
+ private Map<FileSKVIterator,String> reserveReaders(KeyExtent tablet,
Collection<String> files, boolean continueOnFailure) throws IOException {
if (!tablet.isMeta() && files.size() >= maxOpen) {
throw new IllegalArgumentException("requested files exceeds max open");
}
if (files.size() == 0) {
- return Collections.emptyList();
+ return Collections.emptyMap();
}
List<String> filesToOpen = null;
List<FileSKVIterator> filesToClose = Collections.emptyList();
- List<FileSKVIterator> reservedFiles = new ArrayList<>();
Map<FileSKVIterator,String> readersReserved = new HashMap<>();
if (!tablet.isMeta()) {
@@ -295,7 +289,7 @@ private synchronized String
getReservedReadeFilename(FileSKVIterator reader) {
// a synch block
synchronized (this) {
- filesToOpen = takeOpenFiles(files, reservedFiles, readersReserved);
+ filesToOpen = takeOpenFiles(files, readersReserved);
int numOpen = countReaders(openFiles);
@@ -318,7 +312,6 @@ private synchronized String
getReservedReadeFilename(FileSKVIterator reader) {
// log.debug("Opening "+file + " path " + path);
FileSKVIterator reader =
FileOperations.getInstance().newReaderBuilder().forFile(path.toString(), ns,
ns.getConf())
.withTableConfiguration(context.getServerConfigurationFactory().getTableConfiguration(tablet)).withBlockCache(dataCache,
indexCache).build();
- reservedFiles.add(reader);
readersReserved.put(reader, file);
} catch (Exception e) {
@@ -332,7 +325,7 @@ private synchronized String
getReservedReadeFilename(FileSKVIterator reader) {
log.warn("Failed to open file {} {} continuing...", file,
e.getMessage());
} else {
// close whatever files were opened
- closeReaders(reservedFiles);
+ closeReaders(readersReserved.keySet());
if (!tablet.isMeta()) {
filePermits.release(files.size());
@@ -349,7 +342,7 @@ private synchronized String
getReservedReadeFilename(FileSKVIterator reader) {
reservedReaders.putAll(readersReserved);
}
- return reservedFiles;
+ return readersReserved;
}
private void releaseReaders(KeyExtent tablet, List<FileSKVIterator> readers,
boolean sawIOException) {
@@ -481,14 +474,14 @@ public void setInterruptFlag(AtomicBoolean flag) {
}
}
- private List<FileSKVIterator> openFileRefs(Collection<FileRef> files)
throws TooManyFilesException, IOException {
+ private Map<FileSKVIterator,String> openFileRefs(Collection<FileRef>
files) throws TooManyFilesException, IOException {
List<String> strings = new ArrayList<>(files.size());
for (FileRef ref : files)
strings.add(ref.path().toString());
return openFiles(strings);
}
- private List<FileSKVIterator> openFiles(Collection<String> files) throws
TooManyFilesException, IOException {
+ private Map<FileSKVIterator,String> openFiles(Collection<String> files)
throws TooManyFilesException, IOException {
// one tablet can not open more than maxOpen files, otherwise it could
get stuck
// forever waiting on itself to release files
@@ -497,16 +490,16 @@ public void setInterruptFlag(AtomicBoolean flag) {
+ " files.size()=" + files.size() + " maxOpen=" + maxOpen + "
tablet = " + tablet);
}
- List<FileSKVIterator> newlyReservedReaders = reserveReaders(tablet,
files, continueOnFailure);
+ Map<FileSKVIterator,String> newlyReservedReaders =
reserveReaders(tablet, files, continueOnFailure);
- tabletReservedReaders.addAll(newlyReservedReaders);
+ tabletReservedReaders.addAll(newlyReservedReaders.keySet());
return newlyReservedReaders;
}
public synchronized List<InterruptibleIterator>
openFiles(Map<FileRef,DataFileValue> files, boolean detachable,
SamplerConfigurationImpl samplerConfig)
throws IOException {
- List<FileSKVIterator> newlyReservedReaders =
openFileRefs(files.keySet());
+ Map<FileSKVIterator,String> newlyReservedReaders =
openFileRefs(files.keySet());
ArrayList<InterruptibleIterator> iters = new ArrayList<>();
@@ -518,8 +511,9 @@ public void setInterruptFlag(AtomicBoolean flag) {
}
}
- for (FileSKVIterator reader : newlyReservedReaders) {
- String filename = getReservedReadeFilename(reader);
+ for (Entry<FileSKVIterator,String> entry :
newlyReservedReaders.entrySet()) {
+ FileSKVIterator reader = entry.getKey();
+ String filename = entry.getValue();
InterruptibleIterator iter;
FileSKVIterator source = reader;
@@ -570,10 +564,11 @@ public synchronized void
reattach(SamplerConfigurationImpl samplerConfig) throws
for (FileDataSource fds : dataSources)
files.add(fds.file);
- List<FileSKVIterator> newlyReservedReaders = openFiles(files);
+ Map<FileSKVIterator,String> newlyReservedReaders = openFiles(files);
Map<String,List<FileSKVIterator>> map = new HashMap<>();
- for (FileSKVIterator reader : newlyReservedReaders) {
- String fileName = getReservedReadeFilename(reader);
+ for (Entry<FileSKVIterator,String> entry :
newlyReservedReaders.entrySet()) {
+ FileSKVIterator reader = entry.getKey();
+ String fileName = entry.getValue();
List<FileSKVIterator> list = map.get(fileName);
if (list == null) {
list = new LinkedList<>();
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
