1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.aether.named.ipc;
20
21 import java.io.Closeable;
22 import java.io.DataInputStream;
23 import java.io.DataOutputStream;
24 import java.io.EOFException;
25 import java.io.File;
26 import java.io.FileWriter;
27 import java.io.IOException;
28 import java.io.InterruptedIOException;
29 import java.io.OutputStream;
30 import java.io.PrintWriter;
31 import java.io.RandomAccessFile;
32 import java.net.SocketAddress;
33 import java.net.URL;
34 import java.nio.channels.ByteChannel;
35 import java.nio.channels.Channels;
36 import java.nio.channels.FileLock;
37 import java.nio.channels.ServerSocketChannel;
38 import java.nio.channels.SocketChannel;
39 import java.nio.charset.StandardCharsets;
40 import java.nio.file.Files;
41 import java.nio.file.Path;
42 import java.nio.file.Paths;
43 import java.security.SecureRandom;
44 import java.util.ArrayList;
45 import java.util.Arrays;
46 import java.util.Collection;
47 import java.util.List;
48 import java.util.Locale;
49 import java.util.Map;
50 import java.util.Objects;
51 import java.util.concurrent.CompletableFuture;
52 import java.util.concurrent.ConcurrentHashMap;
53 import java.util.concurrent.ExecutionException;
54 import java.util.concurrent.ExecutorService;
55 import java.util.concurrent.Executors;
56 import java.util.concurrent.Future;
57 import java.util.concurrent.TimeUnit;
58 import java.util.concurrent.TimeoutException;
59 import java.util.concurrent.atomic.AtomicInteger;
60
61 import static org.eclipse.aether.named.ipc.IpcMessages.REQUEST_ACQUIRE;
62 import static org.eclipse.aether.named.ipc.IpcMessages.REQUEST_CLOSE;
63 import static org.eclipse.aether.named.ipc.IpcMessages.REQUEST_CONTEXT;
64 import static org.eclipse.aether.named.ipc.IpcMessages.REQUEST_STOP;
65 import static org.eclipse.aether.named.ipc.IpcMessages.RESPONSE_ACQUIRE;
66 import static org.eclipse.aether.named.ipc.IpcMessages.RESPONSE_CLOSE;
67 import static org.eclipse.aether.named.ipc.IpcMessages.RESPONSE_CONTEXT;
68 import static org.eclipse.aether.named.ipc.IpcMessages.RESPONSE_STOP;
69
70
71
72
73
74
75
76 public class IpcClient {
77
78 static final boolean IS_WINDOWS =
79 System.getProperty("os.name").toLowerCase(Locale.ENGLISH).contains("win");
80
81 private static final SecureRandom SECURE_RANDOM = new SecureRandom();
82
83 protected volatile boolean initialized;
84 protected final Path lockPath;
85 protected final Path logPath;
86 protected final Path syncPath;
87 protected final boolean noFork;
88
89 protected volatile SocketChannel socket;
90 protected volatile DataOutputStream output;
91 protected volatile DataInputStream input;
92 protected volatile Thread receiver;
93
94 protected final AtomicInteger requestId = new AtomicInteger();
95 protected final Map<Integer, CompletableFuture<List<String>>> responses = new ConcurrentHashMap<>();
96
97
98
99
100
101
102 protected volatile String bootstrapToken;
103
104 IpcClient(Path lockPath, Path logPath, Path syncPath) {
105 this.lockPath = lockPath;
106 this.logPath = logPath;
107 this.syncPath = syncPath;
108 this.noFork = Boolean.parseBoolean(
109 System.getProperty(IpcServer.SYSTEM_PROP_NO_FORK, Boolean.toString(IpcServer.DEFAULT_NO_FORK)));
110 }
111
112 void ensureInitialized() throws IOException {
113 if (!initialized) {
114
115 synchronized (this) {
116 if (!initialized) {
117 socket = createClient();
118 ByteChannel wrapper = new ByteChannelWrapper(socket);
119 input = new DataInputStream(Channels.newInputStream(wrapper));
120 output = new DataOutputStream(Channels.newOutputStream(wrapper));
121 receiver = new Thread(this::receive);
122 receiver.setDaemon(true);
123 receiver.start();
124 initialized = true;
125 }
126 }
127 }
128 }
129
130 SocketChannel createClient() throws IOException {
131 SocketFamily family =
132 SocketFamily.valueOf(System.getProperty(IpcServer.SYSTEM_PROP_FAMILY, IpcServer.DEFAULT_FAMILY));
133
134 Path lockPath = this.lockPath.toAbsolutePath().normalize();
135 Path lockFile =
136 lockPath.resolve(".maven-resolver-ipc-lock-" + family.name().toLowerCase(Locale.ENGLISH));
137 if (!Files.isRegularFile(lockFile)) {
138 if (!Files.isDirectory(lockFile.getParent())) {
139 Files.createDirectories(lockFile.getParent());
140 }
141 }
142
143 try (RandomAccessFile raf = new RandomAccessFile(lockFile.toFile(), "rw")) {
144 try (FileLock lock = raf.getChannel().lock()) {
145 String line = raf.readLine();
146 if (line != null) {
147 try {
148 SocketAddress address = SocketFamily.fromString(line);
149 return SocketChannel.open(address);
150 } catch (IOException e) {
151
152 }
153 }
154
155 ServerSocketChannel ss = family.openServerSocket();
156 String tmpaddr = SocketFamily.toString(ss.getLocalAddress());
157
158 String rand = Long.toHexString(SECURE_RANDOM.nextLong()) + Long.toHexString(SECURE_RANDOM.nextLong());
159 String nativeName =
160 System.getProperty(IpcServer.SYSTEM_PROP_NATIVE_NAME, IpcServer.DEFAULT_NATIVE_NAME);
161 String syncCmd = IS_WINDOWS ? nativeName + ".exe" : nativeName;
162
163 boolean debug = Boolean.parseBoolean(
164 System.getProperty(IpcServer.SYSTEM_PROP_DEBUG, Boolean.toString(IpcServer.DEFAULT_DEBUG)));
165 boolean noNative = Boolean.parseBoolean(System.getProperty(
166 IpcServer.SYSTEM_PROP_NO_NATIVE, Boolean.toString(IpcServer.DEFAULT_NO_NATIVE)));
167 if (!noNative) {
168 noNative = !Files.isExecutable(syncPath.resolve(syncCmd));
169 }
170 Closeable close;
171 Path logFile = logPath.resolve("resolver-ipcsync-" + rand + ".log");
172 List<String> args = new ArrayList<>();
173 if (noNative) {
174 if (noFork) {
175 IpcServer server = IpcServer.runServer(family, tmpaddr, rand);
176 close = server::close;
177 } else {
178 String javaHome = System.getenv("JAVA_HOME");
179 if (javaHome == null) {
180 javaHome = System.getProperty("java.home");
181 }
182 String javaCmd = IS_WINDOWS ? "bin\\java.exe" : "bin/java";
183 String java = Paths.get(javaHome)
184 .resolve(javaCmd)
185 .toAbsolutePath()
186 .toString();
187 args.add(java);
188 String classpath = getJarPath(getClass()) + File.pathSeparator + getJarPath(IpcServer.class);
189 args.add("-cp");
190 args.add(classpath);
191 String timeout = System.getProperty(IpcServer.SYSTEM_PROP_IDLE_TIMEOUT);
192 if (timeout != null) {
193 args.add("-D" + IpcServer.SYSTEM_PROP_IDLE_TIMEOUT + "=" + timeout);
194 }
195 args.add("-D" + IpcServer.SYSTEM_PROP_DEBUG + "=" + debug);
196 args.add(IpcServer.class.getName());
197 args.add(family.name());
198 args.add(tmpaddr);
199
200
201 args.add("-");
202 ProcessBuilder processBuilder = new ProcessBuilder();
203 ProcessBuilder.Redirect discard = ProcessBuilder.Redirect.to(logFile.toFile());
204 Files.createDirectories(logPath);
205 Process process = processBuilder
206 .directory(lockFile.getParent().toFile())
207 .command(args)
208 .redirectOutput(discard)
209 .redirectError(discard)
210 .start();
211 writeBootstrapToken(process, rand);
212 close = process::destroyForcibly;
213 }
214 } else {
215 args.add(syncPath.resolve(syncCmd).toString());
216 String timeout = System.getProperty(IpcServer.SYSTEM_PROP_IDLE_TIMEOUT);
217 if (timeout != null) {
218 args.add("-D" + IpcServer.SYSTEM_PROP_IDLE_TIMEOUT + "=" + timeout);
219 }
220 args.add("-D" + IpcServer.SYSTEM_PROP_DEBUG + "=" + debug);
221 args.add(family.name());
222 args.add(tmpaddr);
223
224 args.add("-");
225 ProcessBuilder processBuilder = new ProcessBuilder();
226 ProcessBuilder.Redirect discard = ProcessBuilder.Redirect.to(logFile.toFile());
227 Files.createDirectories(logPath);
228 Process process = processBuilder
229 .directory(lockFile.getParent().toFile())
230 .command(args)
231 .redirectOutput(discard)
232 .redirectError(discard)
233 .start();
234 writeBootstrapToken(process, rand);
235 close = process::destroyForcibly;
236 }
237
238 ExecutorService es = Executors.newSingleThreadExecutor();
239 Future<String[]> future = es.submit(() -> {
240 SocketChannel s = ss.accept();
241 DataInputStream dis = new DataInputStream(Channels.newInputStream(s));
242 String rand2 = dis.readUTF();
243 String addr2 = dis.readUTF();
244 return new String[] {rand2, addr2};
245 });
246 String[] res;
247 try {
248 res = future.get(5, TimeUnit.SECONDS);
249 } catch (Exception e) {
250 try (PrintWriter writer = new PrintWriter(new FileWriter(logFile.toFile(), true))) {
251 writer.println("Arguments:");
252 args.forEach(writer::println);
253 writer.println();
254 writer.println("Exception:");
255 e.printStackTrace(writer);
256 }
257 close.close();
258 throw e;
259 } finally {
260 es.shutdownNow();
261 ss.close();
262 }
263 if (!Objects.equals(rand, res[0])) {
264 close.close();
265 throw new IllegalStateException("IpcServer did not respond with the correct random");
266 }
267 this.bootstrapToken = rand;
268
269 SocketAddress addr = SocketFamily.fromString(res[1]);
270 SocketChannel socket = SocketChannel.open(addr);
271
272 raf.seek(0);
273 raf.writeBytes(res[1] + "\n");
274 return socket;
275 } catch (Exception e) {
276 throw new RuntimeException("Unable to create and connect to lock server", e);
277 }
278 }
279 }
280
281 private static void writeBootstrapToken(Process process, String token) throws IOException {
282 try (OutputStream os = process.getOutputStream()) {
283 os.write((token + "\n").getBytes(StandardCharsets.UTF_8));
284 }
285 }
286
287 private String getJarPath(Class<?> clazz) {
288 String classpath;
289 String className = clazz.getName().replace('.', '/') + ".class";
290 URL resource = clazz.getResource("/" + className);
291 if (resource == null) {
292 throw new IllegalStateException("Unable to find resource for class " + clazz.getName());
293 }
294 String url = resource.toString();
295 if (url.startsWith("jar:")) {
296 url = url.substring("jar:".length(), url.indexOf("!/"));
297 if (url.startsWith("file:")) {
298 classpath = url.substring("file:".length());
299 } else {
300 throw new IllegalStateException();
301 }
302 } else if (url.startsWith("file:")) {
303 classpath = url.substring("file:".length(), url.indexOf(className));
304 } else {
305 throw new IllegalStateException();
306 }
307 if (IS_WINDOWS) {
308 if (classpath.startsWith("/")) {
309 classpath = classpath.substring(1);
310 }
311 classpath = classpath.replace('/', '\\');
312 }
313
314 return classpath;
315 }
316
317 void receive() {
318 try {
319 while (true) {
320 DataInputStream in = input;
321 if (in == null) {
322 throw new IOException("Connection closed");
323 }
324 int id = in.readInt();
325 int sz = in.readInt();
326 List<String> s = new ArrayList<>(Math.max(0, Math.min(sz, 1024)));
327 for (int i = 0; i < sz; i++) {
328 s.add(in.readUTF());
329 }
330 CompletableFuture<List<String>> f = responses.remove(id);
331 if (f == null) {
332 continue;
333 }
334 if (s.isEmpty()) {
335 f.completeExceptionally(new IOException("Protocol error: empty response"));
336 continue;
337 }
338 f.complete(s);
339 }
340 } catch (EOFException e) {
341 close(new IOException("Server disconnected", e));
342 } catch (Exception e) {
343 close(e);
344 }
345 }
346
347 List<String> send(List<String> request, long time, TimeUnit unit) throws TimeoutException, IOException {
348 ensureInitialized();
349 DataOutputStream out = output;
350 if (out == null) {
351 throw new IOException("Connection closed");
352 }
353 int id = requestId.incrementAndGet();
354 CompletableFuture<List<String>> response = new CompletableFuture<>();
355 responses.put(id, response);
356 synchronized (out) {
357 out.writeInt(id);
358 out.writeInt(request.size());
359 for (String s : request) {
360 out.writeUTF(s);
361 }
362 out.flush();
363 }
364 try {
365 return response.get(time, unit);
366 } catch (InterruptedException e) {
367 responses.remove(id);
368 throw (IOException) new InterruptedIOException("Interrupted").initCause(e);
369 } catch (ExecutionException e) {
370 throw new IOException("Execution error", e);
371 } catch (TimeoutException e) {
372 responses.remove(id);
373 throw e;
374 }
375 }
376
377 void close() {
378 if (noFork) {
379 stopServer();
380 }
381 close(new IOException("Closing"));
382 }
383
384 synchronized void close(Throwable e) {
385 initialized = false;
386 if (socket != null) {
387 try {
388 socket.close();
389 } catch (IOException t) {
390 e.addSuppressed(t);
391 }
392 socket = null;
393 input = null;
394 output = null;
395 }
396 if (receiver != null && Thread.currentThread() != receiver) {
397 receiver.interrupt();
398 try {
399 receiver.join(1000);
400 } catch (InterruptedException t) {
401 e.addSuppressed(t);
402 }
403 }
404 responses.values().forEach(f -> f.completeExceptionally(e));
405 responses.clear();
406 }
407
408 String newContext(boolean shared, long time, TimeUnit unit) throws TimeoutException {
409 RuntimeException error = new RuntimeException("Unable to create new sync context");
410 for (int i = 0; i < 2; i++) {
411 try {
412 List<String> response = send(Arrays.asList(REQUEST_CONTEXT, Boolean.toString(shared)), time, unit);
413 if (response.size() != 2 || !RESPONSE_CONTEXT.equals(response.get(0))) {
414 throw new IOException("Unexpected response: " + response);
415 }
416 return response.get(1);
417 } catch (TimeoutException e) {
418 throw e;
419 } catch (Exception e) {
420 close(e);
421 error.addSuppressed(e);
422 }
423 }
424 throw error;
425 }
426
427 void lock(String contextId, Collection<String> keys, long time, TimeUnit unit) throws TimeoutException {
428 try {
429 List<String> req = new ArrayList<>(keys.size() + 2);
430 req.add(REQUEST_ACQUIRE);
431 req.add(contextId);
432 req.addAll(keys);
433 List<String> response = send(req, time, unit);
434 if (response.size() != 1 || !RESPONSE_ACQUIRE.equals(response.get(0))) {
435 throw new IOException("Unexpected response: " + response);
436 }
437 } catch (TimeoutException e) {
438 throw e;
439 } catch (Exception e) {
440 close(e);
441 throw new RuntimeException("Unable to perform lock (contextId = " + contextId + ")", e);
442 }
443 }
444
445 void unlock(String contextId) {
446 try {
447 List<String> response = send(Arrays.asList(REQUEST_CLOSE, contextId), 10, TimeUnit.SECONDS);
448 if (response.size() != 1 || !RESPONSE_CLOSE.equals(response.get(0))) {
449 throw new IOException("Unexpected response: " + response);
450 }
451 } catch (Exception e) {
452 close(e);
453 throw new RuntimeException("Unable to unlock (contextId = " + contextId + ")", e);
454 }
455 }
456
457
458
459
460 void stopServer() {
461 String token = bootstrapToken;
462 try {
463 List<String> response = send(List.of(REQUEST_STOP, token == null ? "" : token), 30, TimeUnit.SECONDS);
464 if (response.size() != 1 || !RESPONSE_STOP.equals(response.get(0))) {
465 throw new IOException("Unexpected response: " + response);
466 }
467 } catch (Exception e) {
468 close(e);
469 throw new RuntimeException("Unable to stop server", e);
470 }
471 }
472
473 @Override
474 public String toString() {
475 return "IpcClient{"
476 + "lockPath=" + lockPath + ","
477 + "syncServerPath=" + syncPath + ","
478 + "address='" + getAddress() + "'}";
479 }
480
481 private String getAddress() {
482 SocketChannel s = socket;
483 if (s == null) {
484 return "[closed]";
485 }
486 try {
487 return SocketFamily.toString(s.getLocalAddress());
488 } catch (IOException e) {
489 return "[not bound]";
490 }
491 }
492 }