Source code

001/*
002 * This file is part of the Kompics component model runtime.
003 *
004 * Copyright (C) 2009 Swedish Institute of Computer Science (SICS) Copyright (C)
005 * 2009 Royal Institute of Technology (KTH)
006 *
007 * Kompics is free software; you can redistribute it and/or modify it under the
008 * terms of the GNU General Public License as published by the Free Software
009 * Foundation; either version 2 of the License, or (at your option) any later
010 * version.
011 *
012 * This program is distributed in the hope that it will be useful, but WITHOUT
013 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
014 * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
015 * details.
016 *
017 * You should have received a copy of the GNU General Public License along with
018 * this program; if not, write to the Free Software Foundation, Inc., 59 Temple
019 * Place - Suite 330, Boston, MA 02111-1307, USA.
020 */
021package se.sics.kompics;
022
023import java.lang.reflect.Constructor;
024import java.lang.reflect.InvocationTargetException;
025import java.util.concurrent.atomic.AtomicInteger;
026import org.slf4j.Logger;
027import org.slf4j.LoggerFactory;
028import se.sics.kompics.Component.State;
029import se.sics.kompics.Fault.ResolveAction;
030import se.sics.kompics.config.Config;
031import se.sics.kompics.config.TypesafeConfig;
032import se.sics.kompics.scheduler.ForkJoinScheduler;
033import se.sics.kompics.scheduler.WorkStealingScheduler;
034
035/**
036 * The <code>Kompics</code> class.
037 *
038 * @author Cosmin Arad {@literal <[email protected]>}
039 * @author Jim Dowling {@literal <[email protected]>}
040 * @author Lars Kroll {@literal <[email protected]>}
041 * @version $Id$
042 */
043public final class Kompics {
044
045    public static final long SHUTDOWN_TIMEOUT = 5000;
046    public static final Logger logger = LoggerFactory.getLogger("Kompics");
047    public static final AtomicInteger maxNumOfExecutedEvents = new AtomicInteger(1);
048    private static boolean on = false;
049    private static Scheduler scheduler;
050    private static ComponentCore mainCore;
051    private static final Kompics obj = new Kompics();
052    private static final FaultHandler defaultFaultHandler = new FaultHandler() {
053
054        @Override
055        public Fault.ResolveAction handle(Fault f) {
056            return ResolveAction.ESCALATE;
057        }
058    };
059    private static FaultHandler faultHandler = defaultFaultHandler;
060    private static Config config;
061
062    /**
063     * Set the Kompics runtime scheduler.
064     * 
065     * @param sched
066     *            the scheduler to set
067     * 
068     * @throws java.lang.RuntimeException
069     *             if the Kompics runtime is already running
070     */
071    public static void setScheduler(Scheduler sched) {
072        synchronized (obj) {
073            if (on) {
074                throw new RuntimeException("Kompics already created");
075            }
076            scheduler = sched;
077        }
078    }
079
080    /**
081     * Get the current Kompics runtime scheduler.
082     * 
083     * @return the current scheduler
084     */
085    public static Scheduler getScheduler() {
086        synchronized (obj) {
087            return scheduler;
088        }
089    }
090
091    /**
092     * Set the root fault handler to its default.
093     * 
094     * @param fh
095     *            the new root fault handler
096     * 
097     * @throws java.lang.RuntimeException
098     *             if the Kompics runtime is already running
099     */
100    public static void setFaultHandler(FaultHandler fh) {
101        synchronized (obj) {
102            if (on) {
103                throw new RuntimeException("Kompics already created");
104            }
105            faultHandler = fh;
106        }
107    }
108
109    /**
110     * Reset the root fault handler to its default.
111     * 
112     * @throws java.lang.RuntimeException
113     *             if the Kompics runtime is already running
114     */
115    public static void resetFaultHandler() {
116        synchronized (obj) {
117            if (on) {
118                throw new RuntimeException("Kompics already created");
119            }
120            faultHandler = defaultFaultHandler;
121        }
122    }
123
124    /**
125     * Get the root fault handler.
126     * 
127     * @return the root fault handler
128     */
129    public static FaultHandler getFaultHandler() {
130        synchronized (obj) {
131            return faultHandler;
132        }
133    }
134
135    /**
136     * Set the root config to the default value.
137     * 
138     * @param conf
139     *            the new config to set
140     * 
141     * @throws java.lang.RuntimeException
142     *             if the Kompics runtime is already running
143     */
144    public static void setConfig(Config conf) {
145        synchronized (obj) {
146            if (on) {
147                throw new RuntimeException("Kompics already created");
148            }
149            config = conf;
150        }
151    }
152
153    /**
154     * Reset the root config to the default value.
155     * 
156     * @throws java.lang.RuntimeException
157     *             if the Kompics runtime is already running
158     */
159    public static void resetConfig() {
160        synchronized (obj) {
161            if (on) {
162                throw new RuntimeException("Kompics already created");
163            }
164            config = TypesafeConfig.load();
165        }
166    }
167
168    /**
169     * Get the top-level config instance.
170     * 
171     * Will load a default config, if none was set before.
172     * 
173     * @return the root config
174     */
175    public static Config getConfig() {
176        synchronized (obj) {
177            if (config == null) {
178                config = TypesafeConfig.load();
179            }
180            return config;
181        }
182    }
183
184    /**
185     * Checks if the Kompics runtime is currently running.
186     * 
187     * @return true, if its running
188     */
189    public static boolean isOn() {
190        synchronized (obj) {
191            return on;
192        }
193    }
194
195    /**
196     * Creates the top-level component and starts it together with the Kompics runtime.
197     * 
198     * The value {@code Init.NONE} is passed to the root component.
199     *
200     * @param <C>
201     *            the type of the root component
202     * @param main
203     *            the class instance of the root component
204     */
205    @SuppressWarnings("unchecked")
206    public static <C extends ComponentDefinition> void createAndStart(Class<C> main) {
207        // createAndStart(main, Runtime.getRuntime().availableProcessors());
208        createAndStart(main, (Init<C>) Init.NONE, 1);
209    }
210
211    /**
212     * Creates the top-level component and starts it together with the Kompics runtime.
213     * 
214     * Passes the provided init event to the root component.
215     *
216     * @param <C>
217     *            the type of the root component
218     * @param main
219     *            the class instance of the root component
220     * @param initEvent
221     *            the init event to pass to the root component
222     */
223    public static <C extends ComponentDefinition> void createAndStart(Class<C> main, Init<C> initEvent) {
224        createAndStart(main, initEvent, 1);
225    }
226
227    /**
228     * Creates the top-level component and starts it together with the Kompics runtime.
229     * 
230     * The value {@code Init.NONE} is passed to the root component.
231     *
232     * @param <C>
233     *            the type of the root component
234     * @param main
235     *            the class instance of the root component
236     * @param workers
237     *            the number of workers to use in the runtime's threadpool
238     */
239    @SuppressWarnings("unchecked")
240    public static <C extends ComponentDefinition> void createAndStart(Class<C> main, int workers) {
241        createAndStart(main, (Init<C>) Init.NONE, workers, 1);
242    }
243
244    /**
245     * Creates the top-level component and starts it together with the Kompics runtime.
246     * 
247     * The value {@code Init.NONE} is passed to the root component.
248     *
249     * @param <C>
250     *            the type of the root component
251     * @param main
252     *            the class instance of the root component
253     * @param initEvent
254     *            the init event to pass to the root component
255     * @param workers
256     *            the number of workers to use in the runtime's threadpool
257     */
258    public static <C extends ComponentDefinition> void createAndStart(Class<C> main, Init<C> initEvent, int workers) {
259        createAndStart(main, initEvent, workers, 1);
260    }
261
262    /**
263     * Creates the top-level component and starts it together with the Kompics runtime.
264     * 
265     * The value {@code Init.NONE} is passed to the root component.
266     *
267     * @param <C>
268     *            the type of the root component
269     * @param main
270     *            the class instance of the root component
271     * @param workers
272     *            the number of workers to use in the runtime's threadpool
273     * @param maxEventExecuteNumber
274     *            the maximum number of events to execute before forcing a rescheduling of a component
275     */
276    @SuppressWarnings("unchecked")
277    public static <C extends ComponentDefinition> void createAndStart(Class<C> main, int workers,
278            int maxEventExecuteNumber) {
279        createAndStart(main, (Init<C>) Init.NONE, workers, maxEventExecuteNumber);
280    }
281
282    /**
283     * Creates the top-level component and starts it together with the Kompics runtime.
284     * 
285     * The value {@code Init.NONE} is passed to the root component.
286     *
287     * @param <C>
288     *            the type of the root component
289     * @param main
290     *            the class instance of the root component
291     * @param initEvent
292     *            the init event to pass to the root component
293     * @param workers
294     *            the number of workers to use in the runtime's threadpool
295     * @param maxEventExecuteNumber
296     *            the maximum number of events to execute before forcing a rescheduling of a component
297     */
298    public static <C extends ComponentDefinition> void createAndStart(Class<C> main, Init<C> initEvent, int workers,
299            int maxEventExecuteNumber) {
300        synchronized (obj) {
301            if (on) {
302                throw new RuntimeException("Kompics already created");
303            }
304            on = true;
305
306            if (scheduler == null) {
307                // scheduler = new WorkStealingScheduler(workers);
308                scheduler = new ForkJoinScheduler(workers);
309                // scheduler = new ThreadPoolScheduler(workers);
310            }
311
312            Kompics.maxNumOfExecutedEvents.lazySet(maxEventExecuteNumber);
313
314            try {
315                ComponentDefinition mainComponent;
316                if (initEvent == Init.NONE) { // NONE instance
317                    mainComponent = main.newInstance();
318                } else {
319                    Constructor<C> constr = main.getConstructor(initEvent.getClass());
320                    mainComponent = constr.newInstance(initEvent);
321                }
322                mainCore = mainComponent.getComponentCore();
323                mainCore.setScheduler(scheduler);
324
325                // mainCore.workCount.incrementAndGet();
326                // start Main
327                ((PortCore<ControlPort>) mainCore.getControl()).doTrigger(Start.event, 0, mainCore);
328            } catch (InstantiationException | IllegalAccessException | IllegalArgumentException
329                    | InvocationTargetException | NoSuchMethodException | SecurityException e) {
330                throw new RuntimeException("Cannot create main component " + main.getCanonicalName(), e);
331            }
332
333            scheduler.proceed();
334        }
335    }
336
337    private Kompics() {
338    }
339
340    /**
341     * Asynchronously shut down the Kompics runtime.
342     * <p>
343     * Can be used safely from within components.
344     */
345    public static void asyncShutdown() {
346        Runnable r = new Runnable() {
347
348            @Override
349            public void run() {
350                Kompics.shutdown();
351                ;
352            }
353        };
354        Thread t = new Thread(r);
355        t.start();
356    }
357
358    /**
359     * Synchronously shut down the Kompics runtime.
360     * <p>
361     * Can <b>not</b> be used safely from within Kompics components. Use {@link #asyncShutdown()} instead.
362     */
363    public static void shutdown() {
364        synchronized (obj) {
365            if (mainCore != null) {
366                if (mainCore.state == State.ACTIVE) {
367                    mainCore.control().doTrigger(Kill.event, mainCore.wid, mainCore);
368                }
369                synchronized (mainCore) { // Potential deadlock spot
370                    long start = System.currentTimeMillis();
371                    while (mainCore.state != Component.State.PASSIVE && mainCore.state != Component.State.DESTROYED) {
372                        try {
373                            mainCore.wait(SHUTDOWN_TIMEOUT);
374                        } catch (InterruptedException ex) {
375                            logger.warn("Failed orderly Kompics shutdown", ex);
376                        }
377                        if ((System.currentTimeMillis() - start) > SHUTDOWN_TIMEOUT) {
378                            logger.warn("Failed to shutdown Kompics in time. Forcing shutdown.");
379                            break;
380                        }
381                    }
382                    mainCore.cleanPorts();
383                }
384                mainCore.cleanPorts();
385            }
386            if (scheduler != null) {
387                scheduler.shutdown();
388            }
389            on = false;
390            scheduler = null;
391            obj.notifyAll();
392        }
393    }
394
395    /**
396     * Force an immediate shutdown of the Kompics runtime.
397     * <p>
398     * Can <b>not</b> be used safely from within Kompics components.
399     */
400    public static void forceShutdown() {
401        synchronized (obj) {
402            if (scheduler != null) {
403                scheduler.shutdown();
404            }
405            on = false;
406            scheduler = null;
407
408            obj.notifyAll();
409        }
410    }
411
412    /**
413     * Wait until the Kompics runtime is shut down.
414     * <p>
415     * Can <b>not</b> be used safely from within Kompics components.
416     * 
417     * @throws InterruptedException
418     *             if the waiting is interrupted
419     */
420    public static void waitForTermination() throws InterruptedException {
421        synchronized (obj) {
422            while (on) {
423                obj.wait();
424            }
425        }
426    }
427
428    static void handleFault(final Fault f) {
429        final FaultHandler fh = faultHandler;
430        Thread t = new Thread(new Runnable() {
431
432            @Override
433            public void run() {
434                ResolveAction ra = fh.handle(f);
435                switch (ra) {
436                case RESOLVED:
437                    Kompics.logger.info("Fault {} was resolved by user.", f);
438                    break;
439                case IGNORE:
440                    Kompics.logger.info("Fault {} was declared to be ignored by user. Resuming component...", f);
441                    f.source.markSubtreeAs(Component.State.PASSIVE);
442                    f.source.control().doTrigger(Start.event, 0, mainCore);
443                    break;
444                case DESTROY:
445                    Kompics.logger.info("User declared that Fault {} should quit Kompics...", f);
446                    Kompics.forceShutdown(); {
447                    try {
448                        Kompics.waitForTermination();
449                    } catch (InterruptedException ex) {
450                        Kompics.logger.error("Interrupted while waiting for Kompics termination...");
451                        System.exit(1);
452                    }
453                }
454                    Kompics.logger.info("finished quitting Kompics.");
455                    break;
456                default:
457                    Kompics.logger.info("User declared that Fault {} should quit JVM...", f);
458                    System.exit(1);
459                }
460            }
461
462        });
463        t.start();
464    }
465
466    /**
467     * Log statistics if the @{link WorkStealingScheduler} is used.
468     * 
469     * @deprecated since 1.2.0, don't use the the @{link WorkStealingScheduler}!
470     */
471    @Deprecated
472    public static void logStats() {
473        if (scheduler instanceof WorkStealingScheduler) {
474            ((WorkStealingScheduler) scheduler).logStats();
475        }
476    }
477}