IntenseFinance/nosqlbench
3
0
Fork 0
mirror of https://github.com/nosqlbench/nosqlbench.git synced 2024-12-26 16:51:06 -06:00
Code Issues Packages Projects Releases Wiki Activity

safely remove status timer on retry

Browse Source
This commit is contained in:
Jonathan Shook 2024-06-26 10:29:32 -05:00
parent e935600a01
commit 8ce9cba559
1 changed files with 49 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
nb-apis/nb-api/src/main/java/io/nosqlbench/nb/api/nbio/ResolverForNBIOCache.java
View File

@ -58,11 +58,13 @@ public class ResolverForNBIOCache implements ContentResolver {
/** /**
* This method is used to resolve the path of a given URI. * This method is used to resolve the path of a given URI.
* It first checks if the URI has a scheme (http or https) and if it does, it tries to resolve the path from the cache. * It first checks if the URI has a scheme (http or https) and if it does, it tries to resolve the path from the
* cache.
* If the file is not in the cache, it tries to download it from the remote URL. * If the file is not in the cache, it tries to download it from the remote URL.
* If the URI does not have a scheme, it returns null. * If the URI does not have a scheme, it returns null.
* *
* @param uri the URI to resolve the path for * @param uri
* the URI to resolve the path for
* @return the resolved Path object, or null if the URI does not have a scheme or the path could not be resolved * @return the resolved Path object, or null if the URI does not have a scheme or the path could not be resolved
*/ */
private Path resolvePath(URI uri) { private Path resolvePath(URI uri) {
@ -72,8 +74,7 @@ public class ResolverForNBIOCache implements ContentResolver {
Path cachedFilePath = Path.of(cacheDir + uri.getPath()); Path cachedFilePath = Path.of(cacheDir + uri.getPath());
if (Files.isReadable(cachedFilePath)) { if (Files.isReadable(cachedFilePath)) {
return pathFromLocalCache(cachedFilePath, uri); return pathFromLocalCache(cachedFilePath, uri);
} } else {
else {
return pathFromRemoteUrl(uri); return pathFromRemoteUrl(uri);
} }
} }
@ -97,7 +98,7 @@ public class ResolverForNBIOCache implements ContentResolver {
public void run() { public void run() {
double progress = (double) totalBytesRead / fileSize * 100; double progress = (double) totalBytesRead / fileSize * 100;
logger.info(() -> "Progress: " + String.format("%.2f", progress) + "% completed"); logger.info(() -> "Progress: " + String.format("%.2f", progress) + "% completed");
if (totalBytesRead==fileSize) { if (totalBytesRead == fileSize) {
cancel(); cancel();
} }
} }
@ -109,36 +110,41 @@ public class ResolverForNBIOCache implements ContentResolver {
while (retries < maxRetries) { while (retries < maxRetries) {
try { try {
if (this.remoteFileExists(uri)) { if (this.remoteFileExists(uri)) {
logger.info(() -> "Downloading remote file " + uri + " to cache at " + cachedFilePath);
ReadableByteChannel channel = Channels.newChannel(uri.toURL().openStream());
FileOutputStream outputStream = new FileOutputStream(cachedFilePath.toFile());
long fileSize = uri.toURL().openConnection().getContentLengthLong();
long totalBytesRead = 0;
FileChannel fileChannel = outputStream.getChannel();
ByteBuffer buffer = ByteBuffer.allocate(32768);
Timer timer = new Timer(); Timer timer = new Timer();
ProgressPrinter printer = new ProgressPrinter(fileSize, 0); try {
timer.scheduleAtFixedRate(printer, 2000, 2000); logger.info(() -> "Downloading remote file " + uri + " to cache at " + cachedFilePath);
while (channel.read(buffer) != -1) { ReadableByteChannel channel = Channels.newChannel(uri.toURL().openStream());
buffer.flip(); FileOutputStream outputStream = new FileOutputStream(cachedFilePath.toFile());
totalBytesRead += fileChannel.write(buffer); long fileSize = uri.toURL().openConnection().getContentLengthLong();
printer.update(totalBytesRead); long totalBytesRead = 0;
buffer.clear(); FileChannel fileChannel = outputStream.getChannel();
} ByteBuffer buffer = ByteBuffer.allocate(32768);
outputStream.close();
channel.close(); ProgressPrinter printer = new ProgressPrinter(fileSize, 0);
logger.info(() -> "Downloaded remote file to cache at " + cachedFilePath); timer.scheduleAtFixedRate(printer, 2000, 2000);
if(checksum == null || verifyChecksum(cachedFilePath, checksum)) { while (channel.read(buffer) != -1) {
success = true; buffer.flip();
break; totalBytesRead += fileChannel.write(buffer);
printer.update(totalBytesRead);
buffer.clear();
}
outputStream.close();
channel.close();
logger.info(() -> "Downloaded remote file to cache at " + cachedFilePath);
if (checksum == null || verifyChecksum(cachedFilePath, checksum)) {
success = true;
break;
}
} finally {
timer.cancel();
} }
} else { } else {
logger.error(() -> "Error downloading remote file to cache at " + cachedFilePath + ", retrying..."); logger.error(() -> "Error downloading remote file to cache at " + cachedFilePath + ", retrying...");
retries++; retries++;
} }
} catch (IOException e) { } catch (IOException e) {
logger.error(() -> "Error downloading remote file to cache at " + cachedFilePath + ", retrying..."); logger.error(() -> "Error downloading remote file to cache at " + cachedFilePath + ":" + e + ", " +
"retrying...");
retries++; retries++;
} }
} }
@ -174,19 +180,21 @@ public class ResolverForNBIOCache implements ContentResolver {
* Then it tries to download the file and if successful, it generates a SHA256 checksum for the downloaded file. * Then it tries to download the file and if successful, it generates a SHA256 checksum for the downloaded file.
* It then compares the generated checksum with the remote checksum. * It then compares the generated checksum with the remote checksum.
* If the checksums match, it returns the path to the cached file. * If the checksums match, it returns the path to the cached file.
* If the checksums don't match or if there was an error during the download, it cleans up the cache and throws a RuntimeException. * If the checksums don't match or if there was an error during the download, it cleans up the cache and throws a
* RuntimeException.
* *
* @param uri the URI of the remote file to download * @param uri
* the URI of the remote file to download
* @return the Path to the downloaded file in the local cache * @return the Path to the downloaded file in the local cache
* @throws RuntimeException if there was an error during the download or if the checksums don't match * @throws RuntimeException
* if there was an error during the download or if the checksums don't match
*/ */
private Path pathFromRemoteUrl(URI uri) { private Path pathFromRemoteUrl(URI uri) {
Path cachedFilePath = Path.of(cacheDir + uri.getPath()); Path cachedFilePath = Path.of(cacheDir + uri.getPath());
createCacheDir(cachedFilePath); createCacheDir(cachedFilePath);
if (!verifyChecksum) { if (!verifyChecksum) {
return execute(NBIOResolverConditions.UPDATE_NO_VERIFY, cachedFilePath, uri); return execute(NBIOResolverConditions.UPDATE_NO_VERIFY, cachedFilePath, uri);
} } else {
else {
return execute(NBIOResolverConditions.UPDATE_AND_VERIFY, cachedFilePath, uri); return execute(NBIOResolverConditions.UPDATE_AND_VERIFY, cachedFilePath, uri);
} }
} }
@ -214,7 +222,7 @@ public class ResolverForNBIOCache implements ContentResolver {
private Path execute(NBIOResolverConditions condition, Path cachedFilePath, URI uri) { private Path execute(NBIOResolverConditions condition, Path cachedFilePath, URI uri) {
String remoteChecksumFileStr = uri.getPath() + ".sha256"; String remoteChecksumFileStr = uri.getPath() + ".sha256";
URLContent checksum = resolveURI(URI.create(uri.toString().replace(uri.getPath(), remoteChecksumFileStr))); URLContent checksum = resolveURI(URI.create(uri.toString().replace(uri.getPath(), remoteChecksumFileStr)));
switch(condition) { switch (condition) {
case UPDATE_AND_VERIFY: case UPDATE_AND_VERIFY:
if (checksum == null) { if (checksum == null) {
logger.warn(() -> "Remote checksum file " + remoteChecksumFileStr + " does not exist. Proceeding without verification"); logger.warn(() -> "Remote checksum file " + remoteChecksumFileStr + " does not exist. Proceeding without verification");
@ -244,8 +252,7 @@ public class ResolverForNBIOCache implements ContentResolver {
if (localChecksum.equals(remoteChecksum)) { if (localChecksum.equals(remoteChecksum)) {
logger.info(() -> "Checksums match, returning cached file " + cachedFilePath); logger.info(() -> "Checksums match, returning cached file " + cachedFilePath);
return cachedFilePath; return cachedFilePath;
} } else {
else {
logger.warn(() -> "Checksums do not match, rehydrating cache " + cachedFilePath); logger.warn(() -> "Checksums do not match, rehydrating cache " + cachedFilePath);
return pathFromRemoteUrl(uri); return pathFromRemoteUrl(uri);
} }
@ -269,10 +276,13 @@ public class ResolverForNBIOCache implements ContentResolver {
* If the checksums don't match, it deletes the cached file and downloads it from the remote URL. * If the checksums don't match, it deletes the cached file and downloads it from the remote URL.
* If the remote file or checksum does not exist, it returns the cached file. * If the remote file or checksum does not exist, it returns the cached file.
* *
* @param cachedFilePath the Path to the cached file * @param cachedFilePath
* @param uri the URI of the remote file * the Path to the cached file
* @param uri
* the URI of the remote file
* @return the Path to the cached file * @return the Path to the cached file
* @throws RuntimeException if there was an error during the checksum comparison or if the checksums don't match * @throws RuntimeException
* if there was an error during the checksum comparison or if the checksums don't match
*/ */
private Path pathFromLocalCache(Path cachedFilePath, URI uri) { private Path pathFromLocalCache(Path cachedFilePath, URI uri) {