Source code

001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *   http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing,
013 * software distributed under the License is distributed on an
014 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
015 * KIND, either express or implied.  See the License for the
016 * specific language governing permissions and limitations
017 * under the License.
018 */
019package org.eclipse.aether.named.ipc;
020
021import java.io.Closeable;
022import java.io.DataInputStream;
023import java.io.DataOutputStream;
024import java.io.EOFException;
025import java.io.File;
026import java.io.FileWriter;
027import java.io.IOException;
028import java.io.InterruptedIOException;
029import java.io.PrintWriter;
030import java.io.RandomAccessFile;
031import java.net.SocketAddress;
032import java.net.URL;
033import java.nio.channels.ByteChannel;
034import java.nio.channels.Channels;
035import java.nio.channels.FileLock;
036import java.nio.channels.ServerSocketChannel;
037import java.nio.channels.SocketChannel;
038import java.nio.file.Files;
039import java.nio.file.Path;
040import java.nio.file.Paths;
041import java.util.ArrayList;
042import java.util.Arrays;
043import java.util.Collection;
044import java.util.List;
045import java.util.Locale;
046import java.util.Map;
047import java.util.Objects;
048import java.util.Random;
049import java.util.concurrent.CompletableFuture;
050import java.util.concurrent.ConcurrentHashMap;
051import java.util.concurrent.ExecutionException;
052import java.util.concurrent.ExecutorService;
053import java.util.concurrent.Executors;
054import java.util.concurrent.Future;
055import java.util.concurrent.TimeUnit;
056import java.util.concurrent.TimeoutException;
057import java.util.concurrent.atomic.AtomicInteger;
058
059import static org.eclipse.aether.named.ipc.IpcMessages.REQUEST_ACQUIRE;
060import static org.eclipse.aether.named.ipc.IpcMessages.REQUEST_CLOSE;
061import static org.eclipse.aether.named.ipc.IpcMessages.REQUEST_CONTEXT;
062import static org.eclipse.aether.named.ipc.IpcMessages.REQUEST_STOP;
063import static org.eclipse.aether.named.ipc.IpcMessages.RESPONSE_ACQUIRE;
064import static org.eclipse.aether.named.ipc.IpcMessages.RESPONSE_CLOSE;
065import static org.eclipse.aether.named.ipc.IpcMessages.RESPONSE_CONTEXT;
066import static org.eclipse.aether.named.ipc.IpcMessages.RESPONSE_STOP;
067
068/**
069 * Client side implementation.
070 * The client instance is bound to a given maven repository.
071 *
072 * @since 2.0.1
073 */
074public class IpcClient {
075
076    static final boolean IS_WINDOWS =
077            System.getProperty("os.name").toLowerCase(Locale.ENGLISH).contains("win");
078
079    protected volatile boolean initialized;
080    protected final Path lockPath;
081    protected final Path logPath;
082    protected final Path syncPath;
083    protected final boolean noFork;
084
085    protected volatile SocketChannel socket;
086    protected volatile DataOutputStream output;
087    protected volatile DataInputStream input;
088    protected volatile Thread receiver;
089
090    protected final AtomicInteger requestId = new AtomicInteger();
091    protected final Map<Integer, CompletableFuture<List<String>>> responses = new ConcurrentHashMap<>();
092
093    IpcClient(Path lockPath, Path logPath, Path syncPath) {
094        this.lockPath = lockPath;
095        this.logPath = logPath;
096        this.syncPath = syncPath;
097        this.noFork = Boolean.parseBoolean(
098                System.getProperty(IpcServer.SYSTEM_PROP_NO_FORK, Boolean.toString(IpcServer.DEFAULT_NO_FORK)));
099    }
100
101    void ensureInitialized() throws IOException {
102        if (!initialized) {
103            // caller must block on this method
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                        // ignore
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     * To be used in tests to stop server immediately. Should not be used outside of tests.
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}