Services.java [src/csip/utils] Revision: 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);
}
}
}