1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.filedistribution;
import com.google.common.util.concurrent.SettableFuture;
import com.yahoo.config.FileReference;
import com.yahoo.log.LogLevel;
import com.yahoo.vespa.config.ConnectionPool;
import com.yahoo.vespa.defaults.Defaults;
import java.io.File;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.logging.Logger;
/**
* Handles downloads of files (file references only for now)
*
* @author hmusum
*/
public class FileDownloader {
private final static Logger log = Logger.getLogger(FileDownloader.class.getName());
private final File downloadDirectory;
private final Duration timeout;
private final FileReferenceDownloader fileReferenceDownloader;
public FileDownloader(ConnectionPool connectionPool) {
this(connectionPool,
new File(Defaults.getDefaults().underVespaHome("var/db/vespa/filedistribution")),
new File(Defaults.getDefaults().underVespaHome("tmp")),
Duration.ofMinutes(15));
}
FileDownloader(ConnectionPool connectionPool, File downloadDirectory, File tmpDirectory, Duration timeout) {
this.downloadDirectory = downloadDirectory;
this.timeout = timeout;
this.fileReferenceDownloader = new FileReferenceDownloader(downloadDirectory, tmpDirectory, connectionPool, timeout);
}
public Optional<File> getFile(FileReference fileReference) {
try {
return getFutureFile(fileReference).get(timeout.toMillis(), TimeUnit.MILLISECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
return Optional.empty();
}
}
private Future<Optional<File>> getFutureFile(FileReference fileReference) {
Objects.requireNonNull(fileReference, "file reference cannot be null");
File directory = new File(downloadDirectory, fileReference.value());
log.log(LogLevel.DEBUG, "Checking if there is a file in '" + directory.getAbsolutePath() + "' ");
Optional<File> file = getFileFromFileSystem(fileReference, directory);
if (file.isPresent()) {
SettableFuture<Optional<File>> future = SettableFuture.create();
future.set(file);
return future;
} else {
log.log(LogLevel.INFO, "File reference '" + fileReference.value() + "' not found in " +
directory.getAbsolutePath() + ", starting download");
return queueForAsyncDownload(fileReference, timeout);
}
}
// Start downloading, but there is no Future used get file being downloaded
public void queueForAsyncDownload(List<FileReference> fileReferences) {
fileReferences.forEach(fileReference -> {
if (fileReferenceDownloader.isDownloading(fileReference)) {
log.log(LogLevel.DEBUG, "Already downloading '" + fileReference.value() + "'");
} else {
queueForAsyncDownload(fileReference);
}
});
}
void receiveFile(FileReferenceData fileReferenceData) {
fileReferenceDownloader.receiveFile(fileReferenceData);
}
double downloadStatus(FileReference fileReference) {
return fileReferenceDownloader.downloadStatus(fileReference.value());
}
public Map<FileReference, Double> downloadStatus() {
return fileReferenceDownloader.downloadStatus();
}
File downloadDirectory() {
return downloadDirectory;
}
private Optional<File> getFileFromFileSystem(FileReference fileReference, File directory) {
File[] files = directory.listFiles();
if (directory.exists() && directory.isDirectory() && files != null && files.length > 0) {
File file = files[0];
if (!file.exists()) {
throw new RuntimeException("File with reference '" + fileReference.value() + "' does not exist");
} else if (!file.canRead()) {
throw new RuntimeException("File with reference '" + fileReference.value() + "'exists, but unable to read it");
} else {
fileReferenceDownloader.setDownloadStatus(fileReference, 1.0);
return Optional.of(file);
}
}
return Optional.empty();
}
private synchronized Future<Optional<File>> queueForAsyncDownload(FileReference fileReference, Duration timeout) {
Future<Optional<File>> inProgress = fileReferenceDownloader.addDownloadListener(fileReference, () -> getFile(fileReference));
if (inProgress != null) {
log.log(LogLevel.DEBUG, "Already downloading '" + fileReference.value() + "'");
return inProgress;
}
Future<Optional<File>> future = queueForAsyncDownload(fileReference);
log.log(LogLevel.INFO, "Queued '" + fileReference.value() + "' for download with timeout " + timeout);
return future;
}
private Future<Optional<File>> queueForAsyncDownload(FileReference fileReference) {
FileReferenceDownload fileReferenceDownload = new FileReferenceDownload(fileReference, SettableFuture.create());
fileReferenceDownloader.addToDownloadQueue(fileReferenceDownload);
return fileReferenceDownload.future();
}
public FileReferenceDownloader fileReferenceDownloader() {
return fileReferenceDownloader;
}
}
|
