Services.java [src/csip/utils] Revision: default  Date:
/*
 * $Id$
 *
 * This file is part of the Cloud Services Integration Platform (CSIP),
 * a Model-as-a-Service framework, API and application suite.
 *
 * 2012-2022, Olaf David and others, OMSLab, Colorado State University.
 *
 * OMSLab licenses this file to you under the MIT license.
 * See the LICENSE file in the project root for more information.
 */
package csip.utils;

import csip.Config;
import csip.ModelDataService;
import java.io.*;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.UriBuilder;
import org.codehaus.jettison.json.JSONArray;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
import org.glassfish.jersey.media.multipart.BodyPart;
import org.glassfish.jersey.media.multipart.BodyPartEntity;
import org.glassfish.jersey.media.multipart.FormDataContentDisposition;

/**
 * service utilities.
 *
 * @author Olaf David
 */
public class Services {

  public static final String LOCAL_IP_ADDR = getLocalIP();
  public static final int ENSEMBLE_THREADS = 10;


  /**
   * Returns the current local IP address or an empty string in error case /
   * when no network connection is up.
   *
   * @return Returns the current local IP address or an empty string in error
   * case.
   * @since 0.1.0
   */
  private static String getLocalIP() {

    String ipOnly = "";
    try {
      Enumeration<NetworkInterface> nifs = NetworkInterface.getNetworkInterfaces();
      if (nifs == null)
        return "";
      while (nifs.hasMoreElements()) {
        NetworkInterface nif = nifs.nextElement();
        if (!nif.isLoopback() && nif.isUp() && !nif.isVirtual()) {
          Enumeration<InetAddress> adrs = nif.getInetAddresses();
          while (adrs.hasMoreElements()) {
            InetAddress adr = adrs.nextElement();
            if (adr != null && !adr.isLoopbackAddress() && (nif.isPointToPoint() || !adr.isLinkLocalAddress())) {
              String adrIP = adr.getHostAddress();
              String adrName = nif.isPointToPoint() ? adrIP : adr.getCanonicalHostName();
              if (!adrName.equals(adrIP))
                return adrIP;
              else
                ipOnly = adrIP;

            }
          }
        }
      }
      if (ipOnly.length() == 0) 
        return null;
      return ipOnly;
    } catch (SocketException ex) {
      return null;
    }
  }


  static {
    Calendar uuidEpoch = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
    uuidEpoch.clear();
    uuidEpoch.set(1582, 9, 15, 0, 0, 0); // 9 = October
    epochMillis = uuidEpoch.getTime().getTime();
  }

  static long epochMillis;
  static SimpleDateFormat f = new SimpleDateFormat("/dd/HH");


  private static synchronized String getPrefix(String uuid) {
    UUID u = UUID.fromString(uuid);
    long time = (u.timestamp() / 10000L) + epochMillis;
    return f.format(new Date(time));
  }


  public static File getResultsDir(String suid) {
    return new File(Config.getString(Config.CSIP_RESULTS_DIR) + getPrefix(suid), suid);
  }


  public static File getWorkDir(String suid) {
    return new File(Config.getString(Config.CSIP_WORK_DIR) + getPrefix(suid), suid);
  }

  public static class FormDataParameter {

    String name;
    InputStream is;
    String filename;
    String value;


    public FormDataParameter(BodyPart bp) {
      FormDataContentDisposition fd = (FormDataContentDisposition) bp.getContentDisposition();
      name = fd.getName();
      if (fd.getFileName() != null) {
        BodyPartEntity bpe = (BodyPartEntity) bp.getEntity();
        is = bpe.getInputStream();
        filename = fd.getFileName();
      } else {
        value = bp.getEntityAs(String.class);
      }
    }


    public InputStream getInputStream() {
      return is;
    }


    public boolean isFile() {
      return filename != null;
    }


    public String getValue() {
      return value;
    }


    public String getFilename() {
      return filename;
    }


    public String getName() {
      return name;
    }
  }


  /**
   * Creates a map of Strings pointing to the input streams fo files
   *
   * @param b
   * @return The form parameter
   */
  public static Map<String, FormDataParameter> getFormParameter(List<BodyPart> b) {
    Map<String, FormDataParameter> m = new HashMap<>();
    for (BodyPart bp : b) {
      FormDataContentDisposition fd = (FormDataContentDisposition) bp.getContentDisposition();
      m.put(fd.getName(), new FormDataParameter(bp));
    }
    return m;
  }

  public interface CallableFactory {

    Callable<?> create(int i);
  }


  static synchronized ExecutorService getES(int nthreads, int bq_len) {
    BlockingQueue<Runnable> bq = new ArrayBlockingQueue<>(nthreads + bq_len);
    RejectedExecutionHandler eh = new ThreadPoolExecutor.CallerRunsPolicy();
    ExecutorService es = new ThreadPoolExecutor(nthreads, nthreads, 0L, TimeUnit.MILLISECONDS, bq, eh);
    return es;
  }


  public static void runParallel(int count, CallableFactory factory) {
    runParallel(count, Config.getInt("csip.service.peers", 4), factory);
  }


  public static void runParallel(int count, int threads, CallableFactory factory) {
    runParallel(count, threads, Config.getInt("csip.internal.call.attempts", 4),
        Config.getInt("csip.internal.bq", 4), factory);
  }


  public static void runParallel(int count, int threads, final int attempts, int bq, CallableFactory factory) {

    // have the number of threads being bound by count
    int threads_ = Math.min(count, threads);
    final ExecutorService exec = getES(threads_, bq);
    final CountDownLatch latch = new CountDownLatch(count);
    for (int i = 0; i < count; i++) {
      final Callable<?> c = factory.create(i);
      exec.submit(new Runnable() {
        @Override
        public void run() {
          int a = attempts;
          Exception Ex = null;
          while (a > 0) {
            try {
              c.call();
              break;
            } catch (Exception E) {
              System.err.println("Failed #" + a);
              Ex = E;
              a--;
            }
          }
          if (Ex != null) {
            System.err.println("Failed all attempts, last exception:");
            Ex.printStackTrace(System.err);
            exec.shutdownNow();
          }
          latch.countDown();
        }
      });
    }
    try {
      latch.await();
    } catch (InterruptedException ex) {
    }
    exec.shutdownNow();
  }


  /**
   * run all models at once.
   *
   * @param models
   * @return the list of futures
   * @throws ExecutionException
   */
  public static List<Future<JSONObject>> runEnsemble(List<Callable<JSONObject>> models) throws ExecutionException {

    final ExecutorService executor = Executors.newFixedThreadPool(Config.getInt("codebase.threadpool", ENSEMBLE_THREADS));
    final CountDownLatch barrier = new CountDownLatch(models.size());
    final List<Future<JSONObject>> results = new ArrayList<>();
    // Model callables
    for (final Callable<JSONObject> ca : models) {
      results.add(executor.submit(new Callable<JSONObject>() {
        @Override
        public JSONObject call() {
          JSONObject res = null;
          try {
            res = ca.call();
          } catch (Exception E) {
            executor.shutdownNow();
          }
          barrier.countDown();
          return res;
        }
      }));
    }

    try {
      barrier.await();
    } catch (InterruptedException E) {
    }

    executor.shutdown();
    return results;
  }


  /**
   * Slice a original request into single runs.
   *
   * @param req
   * @param path
   * @return the mapped list of ensembles.
   */
  public static List<Callable<JSONObject>> mapEnsemble(JSONObject req, String path) throws JSONException {

    String codebase = Config.getString("codebase.url", "http://csip.engr.colostate.edu:8081/rest");

    JSONObject metainfo = req.getJSONObject(ModelDataService.KEY_METAINFO);
    if (!req.has(ModelDataService.KEY_METAINFO) || !metainfo.has(ModelDataService.KEY_PARAMETERSETS)) {
      return null;
    }
    List<Callable<JSONObject>> runs = new ArrayList<>();
    if (metainfo.has(ModelDataService.KEY_PARAMETERSETS)) {
      JSONArray psets = req.getJSONArray(ModelDataService.KEY_PARAMETER);
      for (int i = 0; i < metainfo.getInt(ModelDataService.KEY_PARAMETERSETS); i++) {
        JSONArray pset = psets.getJSONArray(i);
        JSONObject single_req = JSONUtils.newRequest(pset, new JSONObject());
        RestCallable mv = new RestCallable(single_req, codebase + path);
        runs.add(mv);
      }
    }
    return runs;
  }


  static boolean isFailed(JSONObject res) throws JSONException {
    return res.getJSONObject(ModelDataService.KEY_METAINFO).getString(ModelDataService.KEY_STATUS).equals("Failed");
  }


  public static JSONObject reduceEnsemble(List<Future<JSONObject>> ens, JSONObject orig_req) throws Exception {
    JSONArray results = new JSONArray();
    for (Future<JSONObject> future : ens) {
      JSONObject res = future.get();
      if (isFailed(res)) {
        orig_req.getJSONObject(ModelDataService.KEY_METAINFO).put(ModelDataService.KEY_STATUS, "Failed");
      }
      results.put(res.get(ModelDataService.KEY_RESULT));
    }
    return JSONUtils.newResponse(orig_req.getJSONArray(ModelDataService.KEY_PARAMETER), results, orig_req.getJSONObject(ModelDataService.KEY_METAINFO));
  }

  static public class RestCallable implements Callable<JSONObject> {

    JSONObject req;
    String url;


    public RestCallable(JSONObject req, String url) {
      this.req = req;
      this.url = url;
    }


    @Override
    public JSONObject call() throws Exception {
      Client client = ClientBuilder.newClient();
      WebTarget service = client.target(UriBuilder.fromUri(url).build());
      return service.request(MediaType.APPLICATION_JSON).post(Entity.json(req), JSONObject.class);
    }
  }

}