001/* 002 * Copyright (C) 2012 The Guava Authors 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except 005 * in compliance with the License. You may obtain a copy of the License at 006 * 007 * http://www.apache.org/licenses/LICENSE-2.0 008 * 009 * Unless required by applicable law or agreed to in writing, software distributed under the License 010 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express 011 * or implied. See the License for the specific language governing permissions and limitations under 012 * the License. 013 */ 014 015package com.google.common.util.concurrent; 016 017import static com.google.common.base.Preconditions.checkArgument; 018import static com.google.common.base.Preconditions.checkNotNull; 019import static com.google.common.base.Preconditions.checkState; 020import static com.google.common.base.Predicates.equalTo; 021import static com.google.common.base.Predicates.in; 022import static com.google.common.base.Predicates.instanceOf; 023import static com.google.common.base.Predicates.not; 024import static com.google.common.util.concurrent.Internal.toNanosSaturated; 025import static com.google.common.util.concurrent.MoreExecutors.directExecutor; 026import static com.google.common.util.concurrent.Service.State.FAILED; 027import static com.google.common.util.concurrent.Service.State.NEW; 028import static com.google.common.util.concurrent.Service.State.RUNNING; 029import static com.google.common.util.concurrent.Service.State.STARTING; 030import static com.google.common.util.concurrent.Service.State.STOPPING; 031import static com.google.common.util.concurrent.Service.State.TERMINATED; 032import static java.util.concurrent.TimeUnit.MILLISECONDS; 033 034import com.google.common.annotations.GwtIncompatible; 035import com.google.common.annotations.J2ktIncompatible; 036import com.google.common.base.Function; 037import com.google.common.base.MoreObjects; 038import com.google.common.base.Stopwatch; 039import com.google.common.collect.Collections2; 040import com.google.common.collect.ImmutableCollection; 041import com.google.common.collect.ImmutableList; 042import com.google.common.collect.ImmutableMap; 043import com.google.common.collect.ImmutableSet; 044import com.google.common.collect.ImmutableSetMultimap; 045import com.google.common.collect.Lists; 046import com.google.common.collect.Maps; 047import com.google.common.collect.MultimapBuilder; 048import com.google.common.collect.Multimaps; 049import com.google.common.collect.Multiset; 050import com.google.common.collect.Ordering; 051import com.google.common.collect.SetMultimap; 052import com.google.common.util.concurrent.Service.State; 053import com.google.errorprone.annotations.CanIgnoreReturnValue; 054import com.google.errorprone.annotations.concurrent.GuardedBy; 055import com.google.j2objc.annotations.J2ObjCIncompatible; 056import com.google.j2objc.annotations.WeakOuter; 057import java.lang.ref.WeakReference; 058import java.time.Duration; 059import java.util.Collections; 060import java.util.EnumSet; 061import java.util.IdentityHashMap; 062import java.util.List; 063import java.util.Map.Entry; 064import java.util.concurrent.Executor; 065import java.util.concurrent.TimeUnit; 066import java.util.concurrent.TimeoutException; 067import java.util.logging.Level; 068 069/** 070 * A manager for monitoring and controlling a set of {@linkplain Service services}. This class 071 * provides methods for {@linkplain #startAsync() starting}, {@linkplain #stopAsync() stopping} and 072 * {@linkplain #servicesByState inspecting} a collection of {@linkplain Service services}. 073 * Additionally, users can monitor state transitions with the {@linkplain Listener listener} 074 * mechanism. 075 * 076 * <p>While it is recommended that service lifecycles be managed via this class, state transitions 077 * initiated via other mechanisms do not impact the correctness of its methods. For example, if the 078 * services are started by some mechanism besides {@link #startAsync}, the listeners will be invoked 079 * when appropriate and {@link #awaitHealthy} will still work as expected. 080 * 081 * <p>Here is a simple example of how to use a {@code ServiceManager} to start a server. 082 * 083 * <pre>{@code 084 * class Server { 085 * public static void main(String[] args) { 086 * Set<Service> services = ...; 087 * ServiceManager manager = new ServiceManager(services); 088 * manager.addListener(new Listener() { 089 * public void stopped() {} 090 * public void healthy() { 091 * // Services have been initialized and are healthy, start accepting requests... 092 * } 093 * public void failure(Service service) { 094 * // Something failed, at this point we could log it, notify a load balancer, or take 095 * // some other action. For now we will just exit. 096 * System.exit(1); 097 * } 098 * }, 099 * MoreExecutors.directExecutor()); 100 * 101 * Runtime.getRuntime().addShutdownHook(new Thread() { 102 * public void run() { 103 * // Give the services 5 seconds to stop to ensure that we are responsive to shutdown 104 * // requests. 105 * try { 106 * manager.stopAsync().awaitStopped(5, TimeUnit.SECONDS); 107 * } catch (TimeoutException timeout) { 108 * // stopping timed out 109 * } 110 * } 111 * }); 112 * manager.startAsync(); // start all the services asynchronously 113 * } 114 * } 115 * }</pre> 116 * 117 * <p>This class uses the ServiceManager's methods to start all of its services, to respond to 118 * service failure and to ensure that when the JVM is shutting down all the services are stopped. 119 * 120 * @author Luke Sandberg 121 * @since 14.0 122 */ 123@J2ktIncompatible 124@GwtIncompatible 125@ElementTypesAreNonnullByDefault 126public final class ServiceManager implements ServiceManagerBridge { 127 private static final LazyLogger logger = new LazyLogger(ServiceManager.class); 128 private static final ListenerCallQueue.Event<Listener> HEALTHY_EVENT = 129 new ListenerCallQueue.Event<Listener>() { 130 @Override 131 public void call(Listener listener) { 132 listener.healthy(); 133 } 134 135 @Override 136 public String toString() { 137 return "healthy()"; 138 } 139 }; 140 private static final ListenerCallQueue.Event<Listener> STOPPED_EVENT = 141 new ListenerCallQueue.Event<Listener>() { 142 @Override 143 public void call(Listener listener) { 144 listener.stopped(); 145 } 146 147 @Override 148 public String toString() { 149 return "stopped()"; 150 } 151 }; 152 153 /** 154 * A listener for the aggregate state changes of the services that are under management. Users 155 * that need to listen to more fine-grained events (such as when each particular {@linkplain 156 * Service service} starts, or terminates), should attach {@linkplain Service.Listener service 157 * listeners} to each individual service. 158 * 159 * @author Luke Sandberg 160 * @since 15.0 (present as an interface in 14.0) 161 */ 162 public abstract static class Listener { 163 /** Constructor for use by subclasses. */ 164 public Listener() {} 165 166 /** 167 * Called when the service initially becomes healthy. 168 * 169 * <p>This will be called at most once after all the services have entered the {@linkplain 170 * State#RUNNING running} state. If any services fail during start up or {@linkplain 171 * State#FAILED fail}/{@linkplain State#TERMINATED terminate} before all other services have 172 * started {@linkplain State#RUNNING running} then this method will not be called. 173 */ 174 public void healthy() {} 175 176 /** 177 * Called when the all of the component services have reached a terminal state, either 178 * {@linkplain State#TERMINATED terminated} or {@linkplain State#FAILED failed}. 179 */ 180 public void stopped() {} 181 182 /** 183 * Called when a component service has {@linkplain State#FAILED failed}. 184 * 185 * @param service The service that failed. 186 */ 187 public void failure(Service service) {} 188 } 189 190 /** 191 * An encapsulation of all of the state that is accessed by the {@linkplain ServiceListener 192 * service listeners}. This is extracted into its own object so that {@link ServiceListener} could 193 * be made {@code static} and its instances can be safely constructed and added in the {@link 194 * ServiceManager} constructor without having to close over the partially constructed {@link 195 * ServiceManager} instance (i.e. avoid leaking a pointer to {@code this}). 196 */ 197 private final ServiceManagerState state; 198 199 private final ImmutableList<Service> services; 200 201 /** 202 * Constructs a new instance for managing the given services. 203 * 204 * @param services The services to manage 205 * @throws IllegalArgumentException if not all services are {@linkplain State#NEW new} or if there 206 * are any duplicate services. 207 */ 208 public ServiceManager(Iterable<? extends Service> services) { 209 ImmutableList<Service> copy = ImmutableList.copyOf(services); 210 if (copy.isEmpty()) { 211 // Having no services causes the manager to behave strangely. Notably, listeners are never 212 // fired. To avoid this we substitute a placeholder service. 213 logger 214 .get() 215 .log( 216 Level.WARNING, 217 "ServiceManager configured with no services. Is your application configured" 218 + " properly?", 219 new EmptyServiceManagerWarning()); 220 copy = ImmutableList.<Service>of(new NoOpService()); 221 } 222 this.state = new ServiceManagerState(copy); 223 this.services = copy; 224 WeakReference<ServiceManagerState> stateReference = new WeakReference<>(state); 225 for (Service service : copy) { 226 service.addListener(new ServiceListener(service, stateReference), directExecutor()); 227 // We check the state after adding the listener as a way to ensure that our listener was added 228 // to a NEW service. 229 checkArgument(service.state() == NEW, "Can only manage NEW services, %s", service); 230 } 231 // We have installed all of our listeners and after this point any state transition should be 232 // correct. 233 this.state.markReady(); 234 } 235 236 /** 237 * Registers a {@link Listener} to be {@linkplain Executor#execute executed} on the given 238 * executor. The listener will not have previous state changes replayed, so it is suggested that 239 * listeners are added before any of the managed services are {@linkplain Service#startAsync 240 * started}. 241 * 242 * <p>{@code addListener} guarantees execution ordering across calls to a given listener but not 243 * across calls to multiple listeners. Specifically, a given listener will have its callbacks 244 * invoked in the same order as the underlying service enters those states. Additionally, at most 245 * one of the listener's callbacks will execute at once. However, multiple listeners' callbacks 246 * may execute concurrently, and listeners may execute in an order different from the one in which 247 * they were registered. 248 * 249 * <p>RuntimeExceptions thrown by a listener will be caught and logged. Any exception thrown 250 * during {@code Executor.execute} (e.g., a {@code RejectedExecutionException}) will be caught and 251 * logged. 252 * 253 * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See 254 * the discussion in the {@link ListenableFuture#addListener ListenableFuture.addListener} 255 * documentation. 256 * 257 * @param listener the listener to run when the manager changes state 258 * @param executor the executor in which the listeners callback methods will be run. 259 */ 260 public void addListener(Listener listener, Executor executor) { 261 state.addListener(listener, executor); 262 } 263 264 /** 265 * Initiates service {@linkplain Service#startAsync startup} on all the services being managed. It 266 * is only valid to call this method if all of the services are {@linkplain State#NEW new}. 267 * 268 * @return this 269 * @throws IllegalStateException if any of the Services are not {@link State#NEW new} when the 270 * method is called. 271 */ 272 @CanIgnoreReturnValue 273 public ServiceManager startAsync() { 274 for (Service service : services) { 275 checkState(service.state() == NEW, "Not all services are NEW, cannot start %s", this); 276 } 277 for (Service service : services) { 278 try { 279 state.tryStartTiming(service); 280 service.startAsync(); 281 } catch (IllegalStateException e) { 282 // This can happen if the service has already been started or stopped (e.g. by another 283 // service or listener). Our contract says it is safe to call this method if 284 // all services were NEW when it was called, and this has already been verified above, so we 285 // don't propagate the exception. 286 logger.get().log(Level.WARNING, "Unable to start Service " + service, e); 287 } 288 } 289 return this; 290 } 291 292 /** 293 * Waits for the {@link ServiceManager} to become {@linkplain #isHealthy() healthy}. The manager 294 * will become healthy after all the component services have reached the {@linkplain State#RUNNING 295 * running} state. 296 * 297 * @throws IllegalStateException if the service manager reaches a state from which it cannot 298 * become {@linkplain #isHealthy() healthy}. 299 */ 300 public void awaitHealthy() { 301 state.awaitHealthy(); 302 } 303 304 /** 305 * Waits for the {@link ServiceManager} to become {@linkplain #isHealthy() healthy} for no more 306 * than the given time. The manager will become healthy after all the component services have 307 * reached the {@linkplain State#RUNNING running} state. 308 * 309 * @param timeout the maximum time to wait 310 * @throws TimeoutException if not all of the services have finished starting within the deadline 311 * @throws IllegalStateException if the service manager reaches a state from which it cannot 312 * become {@linkplain #isHealthy() healthy}. 313 * @since 28.0 (but only since 33.4.0 in the Android flavor) 314 */ 315 public void awaitHealthy(Duration timeout) throws TimeoutException { 316 awaitHealthy(toNanosSaturated(timeout), TimeUnit.NANOSECONDS); 317 } 318 319 /** 320 * Waits for the {@link ServiceManager} to become {@linkplain #isHealthy() healthy} for no more 321 * than the given time. The manager will become healthy after all the component services have 322 * reached the {@linkplain State#RUNNING running} state. 323 * 324 * @param timeout the maximum time to wait 325 * @param unit the time unit of the timeout argument 326 * @throws TimeoutException if not all of the services have finished starting within the deadline 327 * @throws IllegalStateException if the service manager reaches a state from which it cannot 328 * become {@linkplain #isHealthy() healthy}. 329 */ 330 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 331 public void awaitHealthy(long timeout, TimeUnit unit) throws TimeoutException { 332 state.awaitHealthy(timeout, unit); 333 } 334 335 /** 336 * Initiates service {@linkplain Service#stopAsync shutdown} if necessary on all the services 337 * being managed. 338 * 339 * @return this 340 */ 341 @CanIgnoreReturnValue 342 public ServiceManager stopAsync() { 343 for (Service service : services) { 344 service.stopAsync(); 345 } 346 return this; 347 } 348 349 /** 350 * Waits for the all the services to reach a terminal state. After this method returns all 351 * services will either be {@linkplain Service.State#TERMINATED terminated} or {@linkplain 352 * Service.State#FAILED failed}. 353 */ 354 public void awaitStopped() { 355 state.awaitStopped(); 356 } 357 358 /** 359 * Waits for the all the services to reach a terminal state for no more than the given time. After 360 * this method returns all services will either be {@linkplain Service.State#TERMINATED 361 * terminated} or {@linkplain Service.State#FAILED failed}. 362 * 363 * @param timeout the maximum time to wait 364 * @throws TimeoutException if not all of the services have stopped within the deadline 365 * @since 28.0 (but only since 33.4.0 in the Android flavor) 366 */ 367 public void awaitStopped(Duration timeout) throws TimeoutException { 368 awaitStopped(toNanosSaturated(timeout), TimeUnit.NANOSECONDS); 369 } 370 371 /** 372 * Waits for the all the services to reach a terminal state for no more than the given time. After 373 * this method returns all services will either be {@linkplain Service.State#TERMINATED 374 * terminated} or {@linkplain Service.State#FAILED failed}. 375 * 376 * @param timeout the maximum time to wait 377 * @param unit the time unit of the timeout argument 378 * @throws TimeoutException if not all of the services have stopped within the deadline 379 */ 380 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 381 public void awaitStopped(long timeout, TimeUnit unit) throws TimeoutException { 382 state.awaitStopped(timeout, unit); 383 } 384 385 /** 386 * Returns true if all services are currently in the {@linkplain State#RUNNING running} state. 387 * 388 * <p>Users who want more detailed information should use the {@link #servicesByState} method to 389 * get detailed information about which services are not running. 390 */ 391 public boolean isHealthy() { 392 for (Service service : services) { 393 if (!service.isRunning()) { 394 return false; 395 } 396 } 397 return true; 398 } 399 400 /** 401 * Provides a snapshot of the current state of all the services under management. 402 * 403 * <p>N.B. This snapshot is guaranteed to be consistent, i.e. the set of states returned will 404 * correspond to a point in time view of the services. 405 * 406 * @since 29.0 (present with return type {@code ImmutableMultimap} since 14.0) 407 */ 408 @Override 409 public ImmutableSetMultimap<State, Service> servicesByState() { 410 return state.servicesByState(); 411 } 412 413 /** 414 * Returns the service load times. This value will only return startup times for services that 415 * have finished starting. 416 * 417 * @return Map of services and their corresponding startup time in millis, the map entries will be 418 * ordered by startup time. 419 */ 420 public ImmutableMap<Service, Long> startupTimes() { 421 return state.startupTimes(); 422 } 423 424 /** 425 * Returns the service load times. This value will only return startup times for services that 426 * have finished starting. 427 * 428 * @return Map of services and their corresponding startup time, the map entries will be ordered 429 * by startup time. 430 * @since 31.0 (but only since 33.4.0 in the Android flavor) 431 */ 432 @J2ObjCIncompatible 433 public ImmutableMap<Service, Duration> startupDurations() { 434 return ImmutableMap.copyOf( 435 Maps.<Service, Long, Duration>transformValues(startupTimes(), Duration::ofMillis)); 436 } 437 438 @Override 439 public String toString() { 440 return MoreObjects.toStringHelper(ServiceManager.class) 441 .add("services", Collections2.filter(services, not(instanceOf(NoOpService.class)))) 442 .toString(); 443 } 444 445 /** 446 * An encapsulation of all the mutable state of the {@link ServiceManager} that needs to be 447 * accessed by instances of {@link ServiceListener}. 448 */ 449 private static final class ServiceManagerState { 450 final Monitor monitor = new Monitor(); 451 452 @GuardedBy("monitor") 453 final SetMultimap<State, Service> servicesByState = 454 MultimapBuilder.enumKeys(State.class).linkedHashSetValues().build(); 455 456 @GuardedBy("monitor") 457 final Multiset<State> states = servicesByState.keys(); 458 459 @GuardedBy("monitor") 460 final IdentityHashMap<Service, Stopwatch> startupTimers = new IdentityHashMap<>(); 461 462 /** 463 * These two booleans are used to mark the state as ready to start. 464 * 465 * <p>{@link #ready}: is set by {@link #markReady} to indicate that all listeners have been 466 * correctly installed 467 * 468 * <p>{@link #transitioned}: is set by {@link #transitionService} to indicate that some 469 * transition has been performed. 470 * 471 * <p>Together, they allow us to enforce that all services have their listeners installed prior 472 * to any service performing a transition, then we can fail in the ServiceManager constructor 473 * rather than in a Service.Listener callback. 474 */ 475 @GuardedBy("monitor") 476 boolean ready; 477 478 @GuardedBy("monitor") 479 boolean transitioned; 480 481 final int numberOfServices; 482 483 /** 484 * Controls how long to wait for all the services to either become healthy or reach a state from 485 * which it is guaranteed that it can never become healthy. 486 */ 487 final Monitor.Guard awaitHealthGuard = new AwaitHealthGuard(); 488 489 @WeakOuter 490 final class AwaitHealthGuard extends Monitor.Guard { 491 AwaitHealthGuard() { 492 super(ServiceManagerState.this.monitor); 493 } 494 495 @Override 496 @GuardedBy("ServiceManagerState.this.monitor") 497 public boolean isSatisfied() { 498 // All services have started or some service has terminated/failed. 499 return states.count(RUNNING) == numberOfServices 500 || states.contains(STOPPING) 501 || states.contains(TERMINATED) 502 || states.contains(FAILED); 503 } 504 } 505 506 /** Controls how long to wait for all services to reach a terminal state. */ 507 final Monitor.Guard stoppedGuard = new StoppedGuard(); 508 509 @WeakOuter 510 final class StoppedGuard extends Monitor.Guard { 511 StoppedGuard() { 512 super(ServiceManagerState.this.monitor); 513 } 514 515 @Override 516 @GuardedBy("ServiceManagerState.this.monitor") 517 public boolean isSatisfied() { 518 return states.count(TERMINATED) + states.count(FAILED) == numberOfServices; 519 } 520 } 521 522 /** The listeners to notify during a state transition. */ 523 final ListenerCallQueue<Listener> listeners = new ListenerCallQueue<>(); 524 525 /** 526 * It is implicitly assumed that all the services are NEW and that they will all remain NEW 527 * until all the Listeners are installed and {@link #markReady()} is called. It is our caller's 528 * responsibility to only call {@link #markReady()} if all services were new at the time this 529 * method was called and when all the listeners were installed. 530 */ 531 ServiceManagerState(ImmutableCollection<Service> services) { 532 this.numberOfServices = services.size(); 533 servicesByState.putAll(NEW, services); 534 } 535 536 /** 537 * Attempts to start the timer immediately prior to the service being started via {@link 538 * Service#startAsync()}. 539 */ 540 void tryStartTiming(Service service) { 541 monitor.enter(); 542 try { 543 Stopwatch stopwatch = startupTimers.get(service); 544 if (stopwatch == null) { 545 startupTimers.put(service, Stopwatch.createStarted()); 546 } 547 } finally { 548 monitor.leave(); 549 } 550 } 551 552 /** 553 * Marks the {@link State} as ready to receive transitions. Returns true if no transitions have 554 * been observed yet. 555 */ 556 void markReady() { 557 monitor.enter(); 558 try { 559 if (!transitioned) { 560 // nothing has transitioned since construction, good. 561 ready = true; 562 } else { 563 // This should be an extremely rare race condition. 564 List<Service> servicesInBadStates = Lists.newArrayList(); 565 for (Service service : servicesByState().values()) { 566 if (service.state() != NEW) { 567 servicesInBadStates.add(service); 568 } 569 } 570 throw new IllegalArgumentException( 571 "Services started transitioning asynchronously before " 572 + "the ServiceManager was constructed: " 573 + servicesInBadStates); 574 } 575 } finally { 576 monitor.leave(); 577 } 578 } 579 580 void addListener(Listener listener, Executor executor) { 581 listeners.addListener(listener, executor); 582 } 583 584 void awaitHealthy() { 585 monitor.enterWhenUninterruptibly(awaitHealthGuard); 586 try { 587 checkHealthy(); 588 } finally { 589 monitor.leave(); 590 } 591 } 592 593 void awaitHealthy(long timeout, TimeUnit unit) throws TimeoutException { 594 monitor.enter(); 595 try { 596 if (!monitor.waitForUninterruptibly(awaitHealthGuard, timeout, unit)) { 597 throw new TimeoutException( 598 "Timeout waiting for the services to become healthy. The " 599 + "following services have not started: " 600 + Multimaps.filterKeys(servicesByState, in(ImmutableSet.of(NEW, STARTING)))); 601 } 602 checkHealthy(); 603 } finally { 604 monitor.leave(); 605 } 606 } 607 608 void awaitStopped() { 609 monitor.enterWhenUninterruptibly(stoppedGuard); 610 monitor.leave(); 611 } 612 613 void awaitStopped(long timeout, TimeUnit unit) throws TimeoutException { 614 monitor.enter(); 615 try { 616 if (!monitor.waitForUninterruptibly(stoppedGuard, timeout, unit)) { 617 throw new TimeoutException( 618 "Timeout waiting for the services to stop. The following " 619 + "services have not stopped: " 620 + Multimaps.filterKeys(servicesByState, not(in(EnumSet.of(TERMINATED, FAILED))))); 621 } 622 } finally { 623 monitor.leave(); 624 } 625 } 626 627 ImmutableSetMultimap<State, Service> servicesByState() { 628 ImmutableSetMultimap.Builder<State, Service> builder = ImmutableSetMultimap.builder(); 629 monitor.enter(); 630 try { 631 for (Entry<State, Service> entry : servicesByState.entries()) { 632 if (!(entry.getValue() instanceof NoOpService)) { 633 builder.put(entry); 634 } 635 } 636 } finally { 637 monitor.leave(); 638 } 639 return builder.build(); 640 } 641 642 ImmutableMap<Service, Long> startupTimes() { 643 List<Entry<Service, Long>> loadTimes; 644 monitor.enter(); 645 try { 646 loadTimes = Lists.newArrayListWithCapacity(startupTimers.size()); 647 // N.B. There will only be an entry in the map if the service has started 648 for (Entry<Service, Stopwatch> entry : startupTimers.entrySet()) { 649 Service service = entry.getKey(); 650 Stopwatch stopwatch = entry.getValue(); 651 if (!stopwatch.isRunning() && !(service instanceof NoOpService)) { 652 loadTimes.add(Maps.immutableEntry(service, stopwatch.elapsed(MILLISECONDS))); 653 } 654 } 655 } finally { 656 monitor.leave(); 657 } 658 Collections.sort( 659 loadTimes, 660 Ordering.natural() 661 .onResultOf( 662 new Function<Entry<Service, Long>, Long>() { 663 @Override 664 public Long apply(Entry<Service, Long> input) { 665 return input.getValue(); 666 } 667 })); 668 return ImmutableMap.copyOf(loadTimes); 669 } 670 671 /** 672 * Updates the state with the given service transition. 673 * 674 * <p>This method performs the main logic of ServiceManager in the following steps. 675 * 676 * <ol> 677 * <li>Update the {@link #servicesByState()} 678 * <li>Update the {@link #startupTimers} 679 * <li>Based on the new state queue listeners to run 680 * <li>Run the listeners (outside of the lock) 681 * </ol> 682 */ 683 void transitionService(final Service service, State from, State to) { 684 checkNotNull(service); 685 checkArgument(from != to); 686 monitor.enter(); 687 try { 688 transitioned = true; 689 if (!ready) { 690 return; 691 } 692 // Update state. 693 checkState( 694 servicesByState.remove(from, service), 695 "Service %s not at the expected location in the state map %s", 696 service, 697 from); 698 checkState( 699 servicesByState.put(to, service), 700 "Service %s in the state map unexpectedly at %s", 701 service, 702 to); 703 // Update the timer 704 Stopwatch stopwatch = startupTimers.get(service); 705 if (stopwatch == null) { 706 // This means the service was started by some means other than ServiceManager.startAsync 707 stopwatch = Stopwatch.createStarted(); 708 startupTimers.put(service, stopwatch); 709 } 710 if (to.compareTo(RUNNING) >= 0 && stopwatch.isRunning()) { 711 // N.B. if we miss the STARTING event then we may never record a startup time. 712 stopwatch.stop(); 713 if (!(service instanceof NoOpService)) { 714 logger.get().log(Level.FINE, "Started {0} in {1}.", new Object[] {service, stopwatch}); 715 } 716 } 717 // Queue our listeners 718 719 // Did a service fail? 720 if (to == FAILED) { 721 enqueueFailedEvent(service); 722 } 723 724 if (states.count(RUNNING) == numberOfServices) { 725 // This means that the manager is currently healthy. N.B. If other threads call isHealthy 726 // they are not guaranteed to get 'true', because any service could fail right now. 727 enqueueHealthyEvent(); 728 } else if (states.count(TERMINATED) + states.count(FAILED) == numberOfServices) { 729 enqueueStoppedEvent(); 730 } 731 } finally { 732 monitor.leave(); 733 // Run our executors outside of the lock 734 dispatchListenerEvents(); 735 } 736 } 737 738 void enqueueStoppedEvent() { 739 listeners.enqueue(STOPPED_EVENT); 740 } 741 742 void enqueueHealthyEvent() { 743 listeners.enqueue(HEALTHY_EVENT); 744 } 745 746 void enqueueFailedEvent(final Service service) { 747 listeners.enqueue( 748 new ListenerCallQueue.Event<Listener>() { 749 @Override 750 public void call(Listener listener) { 751 listener.failure(service); 752 } 753 754 @Override 755 public String toString() { 756 return "failed({service=" + service + "})"; 757 } 758 }); 759 } 760 761 /** Attempts to execute all the listeners in {@link #listeners}. */ 762 void dispatchListenerEvents() { 763 checkState( 764 !monitor.isOccupiedByCurrentThread(), 765 "It is incorrect to execute listeners with the monitor held."); 766 listeners.dispatch(); 767 } 768 769 @GuardedBy("monitor") 770 void checkHealthy() { 771 if (states.count(RUNNING) != numberOfServices) { 772 IllegalStateException exception = 773 new IllegalStateException( 774 "Expected to be healthy after starting. The following services are not running: " 775 + Multimaps.filterKeys(servicesByState, not(equalTo(RUNNING)))); 776 for (Service service : servicesByState.get(State.FAILED)) { 777 exception.addSuppressed(new FailedService(service)); 778 } 779 throw exception; 780 } 781 } 782 } 783 784 /** 785 * A {@link Service} that wraps another service and times how long it takes for it to start and 786 * also calls the {@link ServiceManagerState#transitionService(Service, State, State)}, to record 787 * the state transitions. 788 */ 789 private static final class ServiceListener extends Service.Listener { 790 final Service service; 791 // We store the state in a weak reference to ensure that if something went wrong while 792 // constructing the ServiceManager we don't pointlessly keep updating the state. 793 final WeakReference<ServiceManagerState> state; 794 795 ServiceListener(Service service, WeakReference<ServiceManagerState> state) { 796 this.service = service; 797 this.state = state; 798 } 799 800 @Override 801 public void starting() { 802 ServiceManagerState state = this.state.get(); 803 if (state != null) { 804 state.transitionService(service, NEW, STARTING); 805 if (!(service instanceof NoOpService)) { 806 logger.get().log(Level.FINE, "Starting {0}.", service); 807 } 808 } 809 } 810 811 @Override 812 public void running() { 813 ServiceManagerState state = this.state.get(); 814 if (state != null) { 815 state.transitionService(service, STARTING, RUNNING); 816 } 817 } 818 819 @Override 820 public void stopping(State from) { 821 ServiceManagerState state = this.state.get(); 822 if (state != null) { 823 state.transitionService(service, from, STOPPING); 824 } 825 } 826 827 @Override 828 public void terminated(State from) { 829 ServiceManagerState state = this.state.get(); 830 if (state != null) { 831 if (!(service instanceof NoOpService)) { 832 logger 833 .get() 834 .log( 835 Level.FINE, 836 "Service {0} has terminated. Previous state was: {1}", 837 new Object[] {service, from}); 838 } 839 state.transitionService(service, from, TERMINATED); 840 } 841 } 842 843 @Override 844 public void failed(State from, Throwable failure) { 845 ServiceManagerState state = this.state.get(); 846 if (state != null) { 847 // Log before the transition, so that if the process exits in response to server failure, 848 // there is a higher likelihood that the cause will be in the logs. 849 boolean log = !(service instanceof NoOpService); 850 /* 851 * We have already exposed startup exceptions to the user in the form of suppressed 852 * exceptions. We don't need to log those exceptions again. 853 */ 854 log &= from != State.STARTING; 855 if (log) { 856 logger 857 .get() 858 .log( 859 Level.SEVERE, 860 "Service " + service + " has failed in the " + from + " state.", 861 failure); 862 } 863 state.transitionService(service, from, FAILED); 864 } 865 } 866 } 867 868 /** 869 * A {@link Service} instance that does nothing. This is only useful as a placeholder to ensure 870 * that the {@link ServiceManager} functions properly even when it is managing no services. 871 * 872 * <p>The use of this class is considered an implementation detail of ServiceManager and as such 873 * it is excluded from {@link #servicesByState}, {@link #startupTimes}, {@link #toString} and all 874 * logging statements. 875 */ 876 private static final class NoOpService extends AbstractService { 877 @Override 878 protected void doStart() { 879 notifyStarted(); 880 } 881 882 @Override 883 protected void doStop() { 884 notifyStopped(); 885 } 886 } 887 888 /** This is never thrown but only used for logging. */ 889 private static final class EmptyServiceManagerWarning extends Throwable {} 890 891 private static final class FailedService extends Throwable { 892 FailedService(Service service) { 893 super( 894 service.toString(), 895 service.failureCause(), 896 false /* don't enable suppression */, 897 false /* don't calculate a stack trace. */); 898 } 899 } 900}