joerghoh commented on code in PR #2607:
URL: https://github.com/apache/jackrabbit-oak/pull/2607#discussion_r2494276543
##########
oak-run/src/main/java/org/apache/jackrabbit/oak/run/Downloader.java:
##########
@@ -170,29 +183,86 @@ public ItemResponse call() throws Exception {
Path destinationPath = Paths.get(item.destination);
Files.createDirectories(destinationPath.getParent());
+ long segmentSize = MAX_LENGTH_SINGLE_THREADED;
long size = 0;
- try (InputStream inputStream = sourceUrl.getInputStream();
- FileOutputStream outputStream = new
FileOutputStream(destinationPath.toFile())) {
- byte[] buffer = new byte[bufferSize];
- int bytesRead;
- while ((bytesRead = inputStream.read(buffer)) != -1) {
- if (md != null) {
- md.update(buffer, 0, bytesRead);
+ if (item.length >= segmentSize) {
+ size = item.length;
+ LOG.info("Downloading large file {}: {} bytes",
destinationPath.toString(), item.length);
+ String fileName = destinationPath.getFileName().toString();
+ long numSegments = (item.length + segmentSize - 1) /
segmentSize;
+ ArrayList<Path> segmentFiles = new ArrayList<>();
+ ArrayList<Future<Boolean>> downloadTasks = new ArrayList<>();
+ for (int i = 0; i < numSegments; i++) {
+ long startByte = i * segmentSize;
+ long endByte = Math.min(startByte + segmentSize - 1,
item.length - 1);
+ Path segmentFile =
destinationPath.getParent().resolve(fileName + "_" + i + ".tmp");
+ segmentFiles.add(segmentFile);
+ downloadTasks.add(executorService.submit(
+ new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ Exception lastException = null;
+ for (int i = 0; i < maxRetries; i++) {
+ try {
+ return
tryDownloadRange(item.source, connectTimeoutMs, readTimeoutMs,
+ segmentFile, startByte,
endByte);
+ } catch (Exception e) {
+ LOG.warn("Range download try # {}
failed", i, e);
+ lastException = e;
+ // retry
+ }
+ }
+ throw lastException;
+ }
+ }
+ ));
+ }
+ // wait for threads
+ boolean allSuccess = true;
+ for (int i = 0; i < downloadTasks.size(); i++) {
+ try {
+ boolean success = downloadTasks.get(i).get();
+ if (!success) {
+ allSuccess = false;
+ break;
+ }
+ } catch (Exception e) {
+ allSuccess = false;
+ break;
+ }
+ }
+ // merge
+ if (allSuccess) {
+ try (OutputStream fileOut =
Files.newOutputStream(destinationPath)) {
+ OutputStream out = md == null ? fileOut : new
DigestOutputStream(fileOut, md);
+ for (Path segmentFile : segmentFiles) {
+ if (Files.exists(segmentFile)) {
+ Files.copy(segmentFile, out);
+ Files.delete(segmentFile);
+ }
+ }
+ LOG.info("Download {} size {}, {} parts",
destinationPath.toString(), size, downloadTasks.size());
Review Comment:
given the async nature of these steps, we end up with the 2 log statements
for a large file being written quite apart from each other (this line + line
190 above). Wouldn't it make sense to just log once per file (still the
question if INFO or DEBUG).
##########
oak-run/src/main/java/org/apache/jackrabbit/oak/run/Downloader.java:
##########
@@ -170,29 +183,86 @@ public ItemResponse call() throws Exception {
Path destinationPath = Paths.get(item.destination);
Files.createDirectories(destinationPath.getParent());
+ long segmentSize = MAX_LENGTH_SINGLE_THREADED;
long size = 0;
- try (InputStream inputStream = sourceUrl.getInputStream();
- FileOutputStream outputStream = new
FileOutputStream(destinationPath.toFile())) {
- byte[] buffer = new byte[bufferSize];
- int bytesRead;
- while ((bytesRead = inputStream.read(buffer)) != -1) {
- if (md != null) {
- md.update(buffer, 0, bytesRead);
+ if (item.length >= segmentSize) {
+ size = item.length;
+ LOG.info("Downloading large file {}: {} bytes",
destinationPath.toString(), item.length);
+ String fileName = destinationPath.getFileName().toString();
+ long numSegments = (item.length + segmentSize - 1) /
segmentSize;
+ ArrayList<Path> segmentFiles = new ArrayList<>();
+ ArrayList<Future<Boolean>> downloadTasks = new ArrayList<>();
+ for (int i = 0; i < numSegments; i++) {
+ long startByte = i * segmentSize;
+ long endByte = Math.min(startByte + segmentSize - 1,
item.length - 1);
+ Path segmentFile =
destinationPath.getParent().resolve(fileName + "_" + i + ".tmp");
+ segmentFiles.add(segmentFile);
+ downloadTasks.add(executorService.submit(
+ new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ Exception lastException = null;
+ for (int i = 0; i < maxRetries; i++) {
+ try {
+ return
tryDownloadRange(item.source, connectTimeoutMs, readTimeoutMs,
+ segmentFile, startByte,
endByte);
+ } catch (Exception e) {
+ LOG.warn("Range download try # {}
failed", i, e);
+ lastException = e;
+ // retry
+ }
+ }
+ throw lastException;
+ }
+ }
+ ));
+ }
+ // wait for threads
+ boolean allSuccess = true;
+ for (int i = 0; i < downloadTasks.size(); i++) {
+ try {
+ boolean success = downloadTasks.get(i).get();
+ if (!success) {
+ allSuccess = false;
+ break;
+ }
+ } catch (Exception e) {
+ allSuccess = false;
Review Comment:
With this we loose the actual exception. I would suggest to convert this
``boolean`` into an ``Exception`` type, and populate it here, so you can later
log it.
##########
oak-run/src/main/java/org/apache/jackrabbit/oak/run/Downloader.java:
##########
@@ -170,29 +183,86 @@ public ItemResponse call() throws Exception {
Path destinationPath = Paths.get(item.destination);
Files.createDirectories(destinationPath.getParent());
+ long segmentSize = MAX_LENGTH_SINGLE_THREADED;
long size = 0;
- try (InputStream inputStream = sourceUrl.getInputStream();
- FileOutputStream outputStream = new
FileOutputStream(destinationPath.toFile())) {
- byte[] buffer = new byte[bufferSize];
- int bytesRead;
- while ((bytesRead = inputStream.read(buffer)) != -1) {
- if (md != null) {
- md.update(buffer, 0, bytesRead);
+ if (item.length >= segmentSize) {
+ size = item.length;
Review Comment:
the code in this branch is quite lengthy... would it make sense to extract
it into a dedicated method?
##########
oak-run/src/main/java/org/apache/jackrabbit/oak/run/Downloader.java:
##########
@@ -170,29 +183,86 @@ public ItemResponse call() throws Exception {
Path destinationPath = Paths.get(item.destination);
Files.createDirectories(destinationPath.getParent());
+ long segmentSize = MAX_LENGTH_SINGLE_THREADED;
long size = 0;
- try (InputStream inputStream = sourceUrl.getInputStream();
- FileOutputStream outputStream = new
FileOutputStream(destinationPath.toFile())) {
- byte[] buffer = new byte[bufferSize];
- int bytesRead;
- while ((bytesRead = inputStream.read(buffer)) != -1) {
- if (md != null) {
- md.update(buffer, 0, bytesRead);
+ if (item.length >= segmentSize) {
+ size = item.length;
+ LOG.info("Downloading large file {}: {} bytes",
destinationPath.toString(), item.length);
Review Comment:
how frequently does that happen? If too frequently I would suggest to move
this to DEBUG otherwise it will pollute the log.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
