<commit_before>/*
 * Capsule
 * Copyright (c) 2014, Parallel Universe Software Co. All rights reserved.
 * 
 * This program and the accompanying materials are licensed under the terms 
 * of the Eclipse Public License v1.0, available at
 * http://www.eclipse.org/legal/epl-v10.html
 */
package co.paralleluniverse.capsule.container;

import co.paralleluniverse.capsule.CapsuleLauncher;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import javax.management.InstanceAlreadyExistsException;
import javax.management.MBeanNotificationInfo;
import javax.management.MBeanRegistrationException;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.NotCompliantMBeanException;
import javax.management.Notification;
import javax.management.NotificationBroadcasterSupport;
import javax.management.ObjectName;
import javax.management.StandardEmitterMBean;

/**
 *
 * @author pron
 */
public class CapsuleContainer implements CapsuleContainerMXBean {
    private final AtomicLong notificationSequence = new AtomicLong();
    private final ConcurrentMap<String, ProcessInfo> processes = new ConcurrentHashMap<String, ProcessInfo>();
    private final AtomicInteger counter = new AtomicInteger();
    private final Path cacheDir;
    private final StandardEmitterMBean mbean;
    private volatile Map<String, Path> javaHomes;

    @SuppressWarnings("OverridableMethodCallInConstructor")
    public CapsuleContainer(Path cacheDir) {
        this.cacheDir = cacheDir;
        this.mbean = registerMBean("co.paralleluniverse:type=CapsuleContainer", getMBeanInterface());
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {

            @Override
            public void run() {
                killAll();
            }
        }));
    }

    protected Class<?> getMBeanInterface() {
        return CapsuleContainerMXBean.class;
    }

    protected NotificationBroadcasterSupport createEmitter() {
        final MBeanNotificationInfo launch = new MBeanNotificationInfo(
                new String[]{CapsuleProcessLaunched.CAPSULE_PROCESS_LAUNCHED},
                CapsuleProcessLaunched.class.getName(),
                "Notification about a capsule process having launched.");
        final MBeanNotificationInfo death = new MBeanNotificationInfo(
                new String[]{CapsuleProcessKilled.CAPSULE_PROCESS_KILLED},
                CapsuleProcessKilled.class.getName(),
                "Notification about a capsule process having launched.");
        return new NotificationBroadcasterSupport(launch, death);
    }

    private StandardEmitterMBean registerMBean(String name, Class<?> mbeanInterface) {
        try {
            final StandardEmitterMBean _mbean = new StandardEmitterMBean(this, (Class) mbeanInterface, createEmitter());
            final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
            final ObjectName mxbeanName = ObjectName.getInstance(name);
            mbs.registerMBean(_mbean, mxbeanName);
            return _mbean;
        } catch (InstanceAlreadyExistsException ex) {
            throw new RuntimeException(ex);
        } catch (MBeanRegistrationException ex) {
            throw new RuntimeException(ex);
        } catch (NotCompliantMBeanException ex) {
            throw new AssertionError(ex);
        } catch (MalformedObjectNameException ex) {
            throw new AssertionError(ex);
        }
    }

    public String launchCapsule(Path capsulePath, List<String> jvmArgs, List<String> args) throws IOException {
        final Object capsule = CapsuleLauncher.newCapsule(capsulePath, cacheDir, javaHomes);
        if (javaHomes == null) // benign race
            javaHomes = CapsuleLauncher.getJavaHomes(capsule);
        return launchCapsule(capsule, jvmArgs, args);
    }

    private String launchCapsule(Object capsule, List<String> jvmArgs, List<String> args) throws IOException {
        if (jvmArgs == null)
            jvmArgs = Collections.emptyList();
        if (args == null)
            args = Collections.emptyList();

        try {
            final String capsuleId = CapsuleLauncher.getAppId(capsule);
            ProcessBuilder pb = CapsuleLauncher.prepareForLaunch(capsule, CapsuleLauncher.enableJMX(jvmArgs), args);
            pb = configureCapsuleProcess(pb);

            final Process p = pb.start();
            final String id = createProcessId(capsuleId, p);

            final ProcessInfo pi = mountProcess(p, id, capsuleId, jvmArgs, args);
            processes.put(id, pi);

            final Notification n = new CapsuleProcessLaunched(mbean, notificationSequence.incrementAndGet(), System.currentTimeMillis(), id, "args: " + args + " jvmArgs: " + jvmArgs);
            mbean.sendNotification(n);
            onProcessLaunch(id, pi);
            monitorProcess(id, p);

            return id;
        } catch (Exception e) {
            if (e instanceof RuntimeException)
                throw (RuntimeException) e;
            throw new RuntimeException(e);
        }
    }

    protected ProcessInfo mountProcess(Process p, String id, String capsuleId, List<String> jvmArgs, List<String> args) throws IOException, InstanceAlreadyExistsException {
        return new ProcessInfo(p, capsuleId, jvmArgs, args);
    }

    /**
     * May be overriden to pipe app IO streams.
     *
     * @param pb The capsule's {@link ProcessBuilder}.
     * @return {@code pb}
     */
    protected ProcessBuilder configureCapsuleProcess(ProcessBuilder pb) throws IOException {
        return pb;
    }

    protected void processDied(String id, Process p, int exitValue) {
        final Notification n = new CapsuleProcessKilled(mbean, notificationSequence.incrementAndGet(), System.currentTimeMillis(), id, exitValue);
        mbean.sendNotification(n);
        onProcessDeath(id, getProcessInfo(id), exitValue);
    }

    protected void onProcessLaunch(String id, ProcessInfo pi) {
    }

    protected void onProcessDeath(String id, ProcessInfo pi, int exitValue) {
    }

    private void monitorProcess(final String id, final Process p) {
        new Thread("process-monitor-" + id) {
            @Override
            public void run() {
                try {
                    int exit = p.waitFor();
                    processDied(id, p, exit);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }.start();
    }

    protected String createProcessId(String appId, Process p) {
        return appId + "-" + counter.incrementAndGet();
    }
<commit_msg>notifications<commit_after>/*
 * Capsule
 * Copyright (c) 2014, Parallel Universe Software Co. All rights reserved.
 * 
 * This program and the accompanying materials are licensed under the terms 
 * of the Eclipse Public License v1.0, available at
 * http://www.eclipse.org/legal/epl-v10.html
 */
package co.paralleluniverse.capsule.container;

import co.paralleluniverse.capsule.CapsuleLauncher;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import javax.management.InstanceAlreadyExistsException;
import javax.management.MBeanNotificationInfo;
import javax.management.MBeanRegistrationException;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.NotCompliantMBeanException;
import javax.management.Notification;
import javax.management.NotificationBroadcasterSupport;
import javax.management.ObjectName;
import javax.management.StandardEmitterMBean;

/**
 *
 * @author pron
 */
public class CapsuleContainer implements CapsuleContainerMXBean {
    private final AtomicLong notificationSequence = new AtomicLong();
    private final ConcurrentMap<String, ProcessInfo> processes = new ConcurrentHashMap<String, ProcessInfo>();
    private final AtomicInteger counter = new AtomicInteger();
    private final Path cacheDir;
    private final StandardEmitterMBean mbean;
    private volatile Map<String, Path> javaHomes;

    @SuppressWarnings("OverridableMethodCallInConstructor")
    public CapsuleContainer(Path cacheDir) {
        this.cacheDir = cacheDir;
        this.mbean = registerMBean("co.paralleluniverse:type=CapsuleContainer", getMBeanInterface());
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {

            @Override
            public void run() {
                killAll();
            }
        }));
    }

    protected Class<?> getMBeanInterface() {
        return CapsuleContainerMXBean.class;
    }

    protected NotificationBroadcasterSupport createEmitter() {
        final MBeanNotificationInfo launch = new MBeanNotificationInfo(
                new String[]{CapsuleProcessLaunched.CAPSULE_PROCESS_LAUNCHED},
                CapsuleProcessLaunched.class.getName(),
                "Notification about a capsule process having launched.");
        final MBeanNotificationInfo death = new MBeanNotificationInfo(
                new String[]{CapsuleProcessKilled.CAPSULE_PROCESS_KILLED},
                CapsuleProcessKilled.class.getName(),
                "Notification about a capsule process having launched.");
        return new NotificationBroadcasterSupport(launch, death);
    }

    private StandardEmitterMBean registerMBean(String name, Class<?> mbeanInterface) {
        try {
            final StandardEmitterMBean _mbean = new StandardEmitterMBean(this, (Class) mbeanInterface, createEmitter());
            final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
            final ObjectName mxbeanName = ObjectName.getInstance(name);
            mbs.registerMBean(_mbean, mxbeanName);
            return _mbean;
        } catch (InstanceAlreadyExistsException ex) {
            throw new RuntimeException(ex);
        } catch (MBeanRegistrationException ex) {
            throw new RuntimeException(ex);
        } catch (NotCompliantMBeanException ex) {
            throw new AssertionError(ex);
        } catch (MalformedObjectNameException ex) {
            throw new AssertionError(ex);
        }
    }

    public String launchCapsule(Path capsulePath, List<String> jvmArgs, List<String> args) throws IOException {
        final Object capsule = CapsuleLauncher.newCapsule(capsulePath, cacheDir, javaHomes);
        if (javaHomes == null) // benign race
            javaHomes = CapsuleLauncher.getJavaHomes(capsule);
        return launchCapsule(capsule, jvmArgs, args);
    }

    private String launchCapsule(Object capsule, List<String> jvmArgs, List<String> args) throws IOException {
        if (jvmArgs == null)
            jvmArgs = Collections.emptyList();
        if (args == null)
            args = Collections.emptyList();

        try {
            final String capsuleId = CapsuleLauncher.getAppId(capsule);
            ProcessBuilder pb = CapsuleLauncher.prepareForLaunch(capsule, CapsuleLauncher.enableJMX(jvmArgs), args);
            pb = configureCapsuleProcess(pb);

            final Process p = pb.start();
            final String id = createProcessId(capsuleId, p);

            final ProcessInfo pi = mountProcess(p, id, capsuleId, jvmArgs, args);
            processes.put(id, pi);

            mbean.sendNotification(processLaunchedNotification(id, jvmArgs, args));
            onProcessLaunch(id, pi);
            monitorProcess(id, p);

            return id;
        } catch (Exception e) {
            if (e instanceof RuntimeException)
                throw (RuntimeException) e;
            throw new RuntimeException(e);
        }
    }

    protected ProcessInfo mountProcess(Process p, String id, String capsuleId, List<String> jvmArgs, List<String> args) throws IOException, InstanceAlreadyExistsException {
        return new ProcessInfo(p, capsuleId, jvmArgs, args);
    }

    /**
     * May be overriden to pipe app IO streams.
     *
     * @param pb The capsule's {@link ProcessBuilder}.
     * @return {@code pb}
     */
    protected ProcessBuilder configureCapsuleProcess(ProcessBuilder pb) throws IOException {
        return pb;
    }

    protected void processDied(String id, Process p, int exitValue) {
        mbean.sendNotification(processDiedNotification(id, exitValue));
        onProcessDeath(id, getProcessInfo(id), exitValue);
    }

    protected void onProcessLaunch(String id, ProcessInfo pi) {
    }

    protected void onProcessDeath(String id, ProcessInfo pi, int exitValue) {
    }

    private Notification processLaunchedNotification(String id, List<String> jvmArgs, List<String> args) {
        return new CapsuleProcessLaunched(mbean, notificationSequence.incrementAndGet(), System.currentTimeMillis(), id, "args: " + args + " jvmArgs: " + jvmArgs);
    }

    private Notification processDiedNotification(String id, int exitValue) {
        return new CapsuleProcessKilled(mbean, notificationSequence.incrementAndGet(), System.currentTimeMillis(), id, exitValue);
    }

    private void monitorProcess(final String id, final Process p) {
        new Thread("process-monitor-" + id) {
            @Override
            public void run() {
                try {
                    int exit = p.waitFor();
                    processDied(id, p, exit);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }.start();
    }

    protected String createProcessId(String appId, Process p) {
        return appId + "-" + counter.incrementAndGet();
    }

{'fraction_non_alphanumeric': 0.08676244943506765, 'fraction_numerical': 0.0011159157483609987, 'mean_word_length': 3.0689557321225878, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3407819', 'n_tokens_mistral': 3815, 'n_tokens_neox': 3718, 'n_words': 939}