content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.celeborn.common.util;
  
  import java.io.*;
  import java.net.InetAddress;
  import java.nio.ByteBuffer;
  import java.nio.channels.ReadableByteChannel;
  import java.nio.charset.StandardCharsets;
  import java.nio.file.Files;
  import java.nio.file.attribute.BasicFileAttributes;
  import java.util.Locale;
  import java.util.Map;
  import java.util.Random;
  import java.util.concurrent.Callable;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.TimeUnit;
  import java.util.function.Function;
  import java.util.regex.Matcher;
  import java.util.regex.Pattern;
  
  import com.google.common.base.Preconditions;
  import com.google.common.collect.ImmutableMap;
  import io.netty.buffer.Unpooled;
  import org.apache.commons.lang3.JavaVersion;
  import org.apache.commons.lang3.SystemUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.celeborn.common.network.util.ByteUnit;
  
  /**
   * General utilities available in the network package. Many of these are sourced from Spark's own
   * Utils, just accessible within this package.
   *
   * <p>Some methods of this class are copied from Apache Spark
   */
  public class JavaUtils {
    private static final Logger logger = LoggerFactory.getLogger(JavaUtils.class);
  
    /** Closes the given object, ignoring IOExceptions. */
    public static void closeQuietly(Closeable closeable) {
      try {
        if (closeable != null) {
          closeable.close();
        }
      } catch (IOException e) {
        logger.error("IOException should not have been thrown.", e);
      }
    }
  
    /** Returns a hash consistent with Spark's Utils.nonNegativeHash(). */
    public static int nonNegativeHash(Object obj) {
      if (obj == null) {
        return 0;
      }
      int hash = obj.hashCode();
      return hash != Integer.MIN_VALUE ? Math.abs(hash) : 0;
    }
  
    /**
     * Convert the given string to a byte buffer. The resulting buffer can be converted back to the
     * same string through {@link #bytesToString(ByteBuffer)}.
     */
    public static ByteBuffer stringToBytes(String s) {
      return Unpooled.wrappedBuffer(s.getBytes(StandardCharsets.UTF_8)).nioBuffer();
    }
  
    /**
     * Convert the given byte buffer to a string. The resulting string can be converted back to the
     * same byte buffer through {@link #stringToBytes(String)}.
     */
    public static String bytesToString(ByteBuffer b) {
      return Unpooled.wrappedBuffer(b).toString(StandardCharsets.UTF_8);
    }
  
    /**
     * Delete a file or directory and its contents recursively. Don't follow directories if they are
     * symlinks.
     *
     * @param file Input file / dir to be deleted
     * @throws IOException if deletion is unsuccessful
     */
    public static void deleteRecursively(File file) throws IOException {
      deleteRecursively(file, null);
    }
  
    /**
     * Delete a file or directory and its contents recursively. Don't follow directories if they are
     * symlinks.
     *
     * @param file Input file / dir to be deleted
     * @param filter A filename filter that make sure only files / dirs with the satisfied filenames
     *     are deleted.
     * @throws IOException if deletion is unsuccessful
     */
    public static void deleteRecursively(File file, FilenameFilter filter) throws IOException {
      if (file == null) {
        return;
      }
  
      // On Unix systems, use operating system command to run faster
      // If that does not work out, fallback to the Java IO way
      if (SystemUtils.IS_OS_UNIX && filter == null) {
        try {
          deleteRecursivelyUsingUnixNative(file);
          return;
        } catch (IOException e) {
          logger.warn(
              "Attempt to delete using native Unix OS command failed for path = {}. "
                  + "Falling back to Java IO way",
              file.getAbsolutePath(),
              e);
        }
      }
  
      deleteRecursivelyUsingJavaIO(file, filter);
    }
  
    private static void deleteRecursivelyUsingJavaIO(File file, FilenameFilter filter)
        throws IOException {
      if (!file.exists()) return;
      BasicFileAttributes fileAttributes =
          Files.readAttributes(file.toPath(), BasicFileAttributes.class);
      if (fileAttributes.isDirectory() && !isSymlink(file)) {
        IOException savedIOException = null;
        for (File child : listFilesSafely(file, filter)) {
          try {
            deleteRecursively(child, filter);
          } catch (IOException e) {
            // In case of multiple exceptions, only last one will be thrown
            savedIOException = e;
          }
        }
        if (savedIOException != null) {
          throw savedIOException;
        }
      }
  
      // Delete file only when it's a normal file or an empty directory.
      if (fileAttributes.isRegularFile()
          || (fileAttributes.isDirectory() && listFilesSafely(file, null).length == 0)) {
        boolean deleted = file.delete();
        // Delete can also fail if the file simply did not exist.
        if (!deleted && file.exists()) {
          throw new IOException("Failed to delete: " + file.getAbsolutePath());
        }
      }
    }
  
    private static void deleteRecursivelyUsingUnixNative(File file) throws IOException {
      ProcessBuilder builder = new ProcessBuilder("rm", "-rf", file.getAbsolutePath());
      Process process = null;
      int exitCode = -1;
  
      try {
        // In order to avoid deadlocks, consume the stdout (and stderr) of the process
        builder.redirectErrorStream(true);
        builder.redirectOutput(new File("/dev/null"));
  
        process = builder.start();
  
        exitCode = process.waitFor();
      } catch (Exception e) {
        throw new IOException("Failed to delete: " + file.getAbsolutePath(), e);
      } finally {
        if (process != null) {
          process.destroy();
        }
      }
  
      if (exitCode != 0 || file.exists()) {
        throw new IOException("Failed to delete: " + file.getAbsolutePath());
      }
    }
  
    private static File[] listFilesSafely(File file, FilenameFilter filter) throws IOException {
      if (file.exists()) {
        File[] files = file.listFiles(filter);
        if (files == null) {
          throw new IOException("Failed to list files for dir: " + file);
        }
        return files;
      } else {
        return new File[0];
      }
    }
  
    private static boolean isSymlink(File file) throws IOException {
      Preconditions.checkNotNull(file);
      File fileInCanonicalDir = null;
      if (file.getParent() == null) {
        fileInCanonicalDir = file;
      } else {
        fileInCanonicalDir = new File(file.getParentFile().getCanonicalFile(), file.getName());
      }
      return !fileInCanonicalDir.getCanonicalFile().equals(fileInCanonicalDir.getAbsoluteFile());
    }
  
    private static final ImmutableMap<String, TimeUnit> timeSuffixes =
        ImmutableMap.<String, TimeUnit>builder()
            .put("us", TimeUnit.MICROSECONDS)
            .put("ms", TimeUnit.MILLISECONDS)
            .put("s", TimeUnit.SECONDS)
            .put("m", TimeUnit.MINUTES)
            .put("min", TimeUnit.MINUTES)
            .put("h", TimeUnit.HOURS)
            .put("d", TimeUnit.DAYS)
            .build();
  
    private static final ImmutableMap<String, ByteUnit> byteSuffixes =
        ImmutableMap.<String, ByteUnit>builder()
            .put("b", ByteUnit.BYTE)
            .put("k", ByteUnit.KiB)
            .put("kb", ByteUnit.KiB)
            .put("kib", ByteUnit.KiB)
            .put("ki", ByteUnit.KiB)
            .put("m", ByteUnit.MiB)
            .put("mb", ByteUnit.MiB)
            .put("mib", ByteUnit.MiB)
            .put("mi", ByteUnit.MiB)
            .put("g", ByteUnit.GiB)
            .put("gb", ByteUnit.GiB)
            .put("gib", ByteUnit.GiB)
            .put("gi", ByteUnit.GiB)
            .put("t", ByteUnit.TiB)
            .put("tb", ByteUnit.TiB)
            .put("tib", ByteUnit.TiB)
            .put("ti", ByteUnit.TiB)
            .put("p", ByteUnit.PiB)
            .put("pb", ByteUnit.PiB)
            .put("pib", ByteUnit.PiB)
            .put("pi", ByteUnit.PiB)
            .build();
  
    private static final Pattern TIME_STRING_PATTERN = Pattern.compile("(-?[0-9]+)([a-z]+)?");
  
    /**
     * Convert a passed time string (e.g. 50s, 100ms, or 250us) to a time count in the given unit. The
     * unit is also considered the default if the given string does not specify a unit.
     */
    public static long timeStringAs(String str, TimeUnit unit) {
      String lower = str.toLowerCase(Locale.ROOT).trim();
  
      try {
        Matcher m = TIME_STRING_PATTERN.matcher(lower);
        if (!m.matches()) {
          throw new NumberFormatException("Failed to parse time string: " + str);
        }
  
        long val = Long.parseLong(m.group(1));
        String suffix = m.group(2);
  
        // Check for invalid suffixes
        if (suffix != null && !timeSuffixes.containsKey(suffix)) {
          throw new NumberFormatException("Invalid suffix: \"" + suffix + "\"");
        }
  
        // If suffix is valid use that, otherwise none was provided and use the default passed
        return unit.convert(val, suffix != null ? timeSuffixes.get(suffix) : unit);
      } catch (NumberFormatException e) {
        String timeError =
            "Time must be specified as seconds (s), "
                + "milliseconds (ms), microseconds (us), minutes (m or min), hour (h), or day (d). "
                + "E.g. 50s, 100ms, or 250us.";
  
        throw new NumberFormatException(timeError + "\n" + e.getMessage());
      }
    }
  
    /**
     * Convert a time parameter such as (50s, 100ms, or 250us) to milliseconds for internal use. If no
     * suffix is provided, the passed number is assumed to be in ms.
     */
    public static long timeStringAsMs(String str) {
      return timeStringAs(str, TimeUnit.MILLISECONDS);
    }
  
    /**
     * Convert a time parameter such as (50s, 100ms, or 250us) to seconds for internal use. If no
     * suffix is provided, the passed number is assumed to be in seconds.
     */
    public static long timeStringAsSec(String str) {
      return timeStringAs(str, TimeUnit.SECONDS);
    }
  
    private static final Pattern BYTE_STRING_PATTERN = Pattern.compile("([0-9]+)([a-z]+)?");
    private static final Pattern BYTE_STRING_FRACTION_PATTERN =
        Pattern.compile("([0-9]+\\.[0-9]+)([a-z]+)?");
  
    /**
     * Convert a passed byte string (e.g. 50b, 100kb, or 250mb) to the given. If no suffix is
     * provided, a direct conversion to the provided unit is attempted.
     */
    public static long byteStringAs(String str, ByteUnit unit) {
      String lower = str.toLowerCase(Locale.ROOT).trim();
  
      try {
        Matcher m = BYTE_STRING_PATTERN.matcher(lower);
        Matcher fractionMatcher = BYTE_STRING_FRACTION_PATTERN.matcher(lower);
  
        if (m.matches()) {
          long val = Long.parseLong(m.group(1));
          String suffix = m.group(2);
  
          // Check for invalid suffixes
          if (suffix != null && !byteSuffixes.containsKey(suffix)) {
            throw new NumberFormatException("Invalid suffix: \"" + suffix + "\"");
          }
  
          // If suffix is valid use that, otherwise none was provided and use the default passed
          return unit.convertFrom(val, suffix != null ? byteSuffixes.get(suffix) : unit);
        } else if (fractionMatcher.matches()) {
          throw new NumberFormatException(
              "Fractional values are not supported. Input was: " + fractionMatcher.group(1));
        } else {
          throw new NumberFormatException("Failed to parse byte string: " + str);
        }
  
      } catch (NumberFormatException e) {
        String byteError =
            "Size must be specified as bytes (b), "
                + "kibibytes (k), mebibytes (m), gibibytes (g), tebibytes (t), or pebibytes(p). "
                + "E.g. 50b, 100k, or 250m.";
  
        throw new NumberFormatException(byteError + "\n" + e.getMessage());
      }
    }
  
    /**
     * Convert a passed byte string (e.g. 50b, 100k, or 250m) to bytes for internal use.
     *
     * <p>If no suffix is provided, the passed number is assumed to be in bytes.
     */
    public static long byteStringAsBytes(String str) {
      return byteStringAs(str, ByteUnit.BYTE);
    }
  
    /**
     * Convert a passed byte string (e.g. 50b, 100k, or 250m) to kibibytes for internal use.
     *
     * <p>If no suffix is provided, the passed number is assumed to be in kibibytes.
     */
    public static long byteStringAsKb(String str) {
      return byteStringAs(str, ByteUnit.KiB);
    }
  
    /**
     * Convert a passed byte string (e.g. 50b, 100k, or 250m) to mebibytes for internal use.
     *
     * <p>If no suffix is provided, the passed number is assumed to be in mebibytes.
     */
    public static long byteStringAsMb(String str) {
      return byteStringAs(str, ByteUnit.MiB);
    }
  
    /**
     * Convert a passed byte string (e.g. 50b, 100k, or 250m) to gibibytes for internal use.
     *
     * <p>If no suffix is provided, the passed number is assumed to be in gibibytes.
     */
    public static long byteStringAsGb(String str) {
      return byteStringAs(str, ByteUnit.GiB);
    }
  
    /**
     * Returns a byte array with the buffer's contents, trying to avoid copying the data if possible.
     */
    public static byte[] bufferToArray(ByteBuffer buffer) {
      if (buffer.hasArray()
          && buffer.arrayOffset() == 0
          && buffer.array().length == buffer.remaining()) {
        return buffer.array();
      } else {
        byte[] bytes = new byte[buffer.remaining()];
        buffer.get(bytes);
        return bytes;
      }
    }
  
    /** Fills a buffer with data read from the channel. */
    public static void readFully(ReadableByteChannel channel, ByteBuffer dst) throws IOException {
      int expected = dst.remaining();
      while (dst.hasRemaining()) {
        if (channel.read(dst) < 0) {
          throw new EOFException(
              String.format("Not enough bytes in channel (expected %d).", expected));
        }
      }
    }
  
    public static String mkString(String[] arr, String sep) {
      StringBuilder sb = new StringBuilder();
      for (int i = 0; i < arr.length; i++) {
        sb.append(arr[i]);
        if (i != arr.length - 1) {
          sb.append(sep);
        }
      }
      return sb.toString();
    }
  
    /**
     * Shuffle the elements of an array into a random order, modifying the original array. Returns the
     * original array.
     */
    public static void randomizeInPlace(int[] arr) {
      Random rand = new Random();
      for (int i = arr.length - 1; i >= 1; i--) {
        int j = rand.nextInt(i + 1);
        int tmp = arr[j];
        arr[j] = arr[i];
        arr[i] = tmp;
      }
    }
  
    public static void shuffleArray(int[] arr, int[] inversed) {
      if (arr == null || inversed == null || arr.length != inversed.length) {
        return;
      } else {
        for (int i = 0; i < arr.length; i++) {
          arr[i] = i;
        }
        JavaUtils.randomizeInPlace(arr);
        for (int i = 0; i < arr.length; i++) {
          inversed[arr[i]] = i;
        }
      }
    }
  
    public static <K, V> ConcurrentHashMap<K, V> newConcurrentHashMap() {
      if (SystemUtils.isJavaVersionAtLeast(JavaVersion.JAVA_9)) {
        return new ConcurrentHashMap<>();
      } else {
        return new ConcurrentHashMapForJDK8<>();
      }
    }
  
    public static <K, V> ConcurrentHashMap<K, V> newConcurrentHashMap(
        Map<? extends K, ? extends V> m) {
      if (SystemUtils.isJavaVersionAtLeast(JavaVersion.JAVA_9)) {
        return new ConcurrentHashMap<>(m);
      } else {
        return new ConcurrentHashMapForJDK8<>(m);
      }
    }
  
    public static <K, V> ConcurrentHashMap<K, V> newConcurrentHashMap(int initialCapacity) {
      if (SystemUtils.isJavaVersionAtLeast(JavaVersion.JAVA_9)) {
        return new ConcurrentHashMap<>(initialCapacity);
      } else {
        return new ConcurrentHashMapForJDK8<>(initialCapacity);
      }
    }
  
    /**
     * For JDK8, there is bug for ConcurrentHashMap#computeIfAbsent, checking the key existence to
     * speed up. See details in CELEBORN-474.
     */
    private static class ConcurrentHashMapForJDK8<K, V> extends ConcurrentHashMap<K, V> {
      public ConcurrentHashMapForJDK8() {
        super();
      }
  
      public ConcurrentHashMapForJDK8(int initialCapacity) {
        super(initialCapacity);
      }
  
      public ConcurrentHashMapForJDK8(Map<? extends K, ? extends V> m) {
        super(m);
      }
  
      @Override
      public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
        V result;
        if (null == (result = get(key))) {
          result = super.computeIfAbsent(key, mappingFunction);
        }
        return result;
      }
    }
  
    public static void timeOutOrMeetCondition(Callable<Boolean> callable) throws Exception {
      int timeout = 10000; // 10s
      while (true) {
        if (callable.call() || timeout < 0) {
          break;
        }
  
        timeout = timeout - 100;
        Thread.sleep(100);
      }
    }
  
    public static String getLocalHost() {
      try {
        return InetAddress.getLocalHost().getHostAddress();
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }
  }
  