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