programming/impl/src/main/java/org/opendaylight/bgpcep/programming/impl/DefaultInstructionScheduler.java

   1 /*
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.bgpcep.programming.impl;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.annotations.VisibleForTesting;
  13 import com.google.common.collect.ImmutableClassToInstanceMap;
  14 import com.google.common.util.concurrent.FluentFuture;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.Futures;
  17 import com.google.common.util.concurrent.ListenableFuture;
  18 import com.google.common.util.concurrent.MoreExecutors;
  19 import com.google.common.util.concurrent.SettableFuture;
  20 import io.netty.util.Timeout;
  21 import io.netty.util.Timer;
  22 import java.math.BigInteger;
  23 import java.util.ArrayList;
  24 import java.util.HashMap;
  25 import java.util.HashSet;
  26 import java.util.Iterator;
  27 import java.util.List;
  28 import java.util.Map;
  29 import java.util.Set;
  30 import java.util.concurrent.Executor;
  31 import java.util.concurrent.TimeUnit;
  32 import org.checkerframework.checker.lock.qual.GuardedBy;
  33 import org.opendaylight.bgpcep.programming.NanotimeUtil;
  34 import org.opendaylight.bgpcep.programming.spi.ExecutionResult;
  35 import org.opendaylight.bgpcep.programming.spi.Instruction;
  36 import org.opendaylight.bgpcep.programming.spi.InstructionScheduler;
  37 import org.opendaylight.bgpcep.programming.spi.SchedulerException;
  38 import org.opendaylight.bgpcep.programming.spi.SuccessfulRpcResult;
  39 import org.opendaylight.mdsal.binding.api.DataBroker;
  40 import org.opendaylight.mdsal.binding.api.NotificationPublishService;
  41 import org.opendaylight.mdsal.binding.api.RpcProviderService;
  42 import org.opendaylight.mdsal.binding.api.WriteTransaction;
  43 import org.opendaylight.mdsal.common.api.CommitInfo;
  44 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  45 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonService;
  46 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  47 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
  48 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.CancelInstruction;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.CancelInstructionInput;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.CancelInstructionOutput;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.CancelInstructionOutputBuilder;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.CleanInstructions;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.CleanInstructionsInput;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.CleanInstructionsOutput;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.CleanInstructionsOutputBuilder;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.DeadOnArrival;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.DuplicateInstructionId;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.InstructionId;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.InstructionStatus;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.InstructionStatusChangedBuilder;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.InstructionsQueue;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.InstructionsQueueBuilder;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.InstructionsQueueKey;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.Nanotime;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.SubmitInstructionInput;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.UnknownInstruction;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.UnknownPreconditionId;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.instruction.queue.InstructionBuilder;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.instruction.queue.InstructionKey;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.instruction.status.changed.Details;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.submit.instruction.output.result.failure._case.FailureBuilder;
  73 import org.opendaylight.yangtools.concepts.Registration;
  74 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  75 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  76 import org.opendaylight.yangtools.yang.binding.Rpc;
  77 import org.opendaylight.yangtools.yang.common.RpcResult;
  78 import org.slf4j.Logger;
  79 import org.slf4j.LoggerFactory;
  80
  81 final class DefaultInstructionScheduler implements ClusterSingletonService, InstructionScheduler {
  82     private static final Logger LOG = LoggerFactory.getLogger(DefaultInstructionScheduler.class);
  83
  84     private final Map<InstructionId, InstructionImpl> insns = new HashMap<>();
  85     private final InstanceIdentifier<InstructionsQueue> qid;
  86     private final NotificationPublishService notifs;
  87     private final Executor executor;
  88     private final DataBroker dataProvider;
  89     private final Timer timer;
  90     private final String instructionId;
  91     private final ServiceGroupIdentifier sgi;
  92     private final ClusterSingletonServiceRegistration csspReg;
  93     private final RpcProviderService rpcProviderRegistry;
  94     @GuardedBy("this")
  95     private Registration reg;
  96
  97     private final class InstructionPusher implements QueueInstruction {
  98         private final InstructionBuilder builder = new InstructionBuilder();
  99
 100         InstructionPusher(final InstructionId id, final Nanotime deadline) {
 101             builder.setDeadline(deadline);
 102             builder.setId(id);
 103             builder.withKey(new InstructionKey(id));
 104             builder.setStatus(InstructionStatus.Queued);
 105         }
 106
 107         @Override
 108         public void instructionUpdated(final InstructionStatus status, final Details details) {
 109             if (!status.equals(builder.getStatus())) {
 110                 builder.setStatus(status);
 111
 112                 final WriteTransaction wt = dataProvider.newWriteOnlyTransaction();
 113                 wt.put(LogicalDatastoreType.OPERATIONAL,
 114                         qid.child(
 115                                 org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming
 116                                         .rev150720.instruction.queue.Instruction.class,
 117                                 new InstructionKey(builder.getId())), builder.build());
 118                 wt.commit().addCallback(new FutureCallback<CommitInfo>() {
 119                     @Override
 120                     public void onSuccess(final CommitInfo result) {
 121                         LOG.debug("Instruction Queue {} updated", qid);
 122                     }
 123
 124                     @Override
 125                     public void onFailure(final Throwable trw) {
 126                         LOG.error("Failed to update Instruction Queue {}", qid, trw);
 127                     }
 128                 }, MoreExecutors.directExecutor());
 129             }
 130
 131             try {
 132                 notifs.putNotification(new InstructionStatusChangedBuilder()
 133                         .setId(builder.getId()).setStatus(status).setDetails(details).build());
 134             } catch (final InterruptedException e) {
 135                 LOG.debug("Failed to publish notification", e);
 136             }
 137         }
 138
 139         @Override
 140         public void instructionRemoved() {
 141             final WriteTransaction wt = dataProvider.newWriteOnlyTransaction();
 142             wt.delete(LogicalDatastoreType.OPERATIONAL, qid.child(
 143                     org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev150720.instruction
 144                             .queue.Instruction.class,
 145                     new InstructionKey(builder.getId())));
 146             wt.commit().addCallback(new FutureCallback<CommitInfo>() {
 147                 @Override
 148                 public void onSuccess(final CommitInfo result) {
 149                     LOG.debug("Instruction Queue {} removed", qid);
 150                 }
 151
 152                 @Override
 153                 public void onFailure(final Throwable trw) {
 154                     LOG.error("Failed to remove Instruction Queue {}", qid, trw);
 155                 }
 156             }, MoreExecutors.directExecutor());
 157         }
 158     }
 159
 160     DefaultInstructionScheduler(final DataBroker dataProvider, final NotificationPublishService notifs,
 161             final Executor executor, final RpcProviderService rpcProviderRegistry,
 162             final ClusterSingletonServiceProvider cssp, final Timer timer, final String instructionId) {
 163         this.dataProvider = requireNonNull(dataProvider);
 164         this.instructionId = requireNonNull(instructionId);
 165         this.notifs = requireNonNull(notifs);
 166         this.executor = requireNonNull(executor);
 167         this.rpcProviderRegistry = requireNonNull(rpcProviderRegistry);
 168         this.timer = requireNonNull(timer);
 169         qid = KeyedInstanceIdentifier.builder(InstructionsQueue.class,
 170                 new InstructionsQueueKey(this.instructionId)).build();
 171         sgi = ServiceGroupIdentifier.create(this.instructionId + "-service-group");
 172         LOG.info("Creating Programming Service {}.", sgi.getName());
 173         csspReg = cssp.registerClusterSingletonService(this);
 174     }
 175
 176     @Override
 177     public synchronized void instantiateServiceInstance() {
 178         LOG.info("Instruction Queue service {} instantiated", sgi.getName());
 179         reg = rpcProviderRegistry.registerRpcImplementations(ImmutableClassToInstanceMap.<Rpc<?, ?>>builder()
 180             .put(CancelInstruction.class, this::cancelInstruction)
 181             .put(CleanInstructions.class, this::cleanInstructions)
 182             .build());
 183
 184         final WriteTransaction wt = dataProvider.newWriteOnlyTransaction();
 185         wt.put(LogicalDatastoreType.OPERATIONAL, qid, new InstructionsQueueBuilder()
 186                 .withKey(new InstructionsQueueKey(instructionId)).setInstruction(Map.of()).build());
 187         wt.commit().addCallback(new FutureCallback<CommitInfo>() {
 188             @Override
 189             public void onSuccess(final CommitInfo result) {
 190                 LOG.debug("Instruction Queue {} added", qid);
 191             }
 192
 193             @Override
 194             public void onFailure(final Throwable trw) {
 195                 LOG.error("Failed to add Instruction Queue {}", qid, trw);
 196             }
 197         }, MoreExecutors.directExecutor());
 198     }
 199
 200     @Override
 201     public ServiceGroupIdentifier getIdentifier() {
 202         return sgi;
 203     }
 204
 205     @VisibleForTesting
 206     ListenableFuture<RpcResult<CancelInstructionOutput>> cancelInstruction(final CancelInstructionInput input) {
 207         return Futures.submit(() -> realCancelInstruction(input), executor);
 208     }
 209
 210     @VisibleForTesting
 211     ListenableFuture<RpcResult<CleanInstructionsOutput>> cleanInstructions(final CleanInstructionsInput input) {
 212         return Futures.submit(() -> realCleanInstructions(input), executor);
 213     }
 214
 215     private synchronized RpcResult<CancelInstructionOutput> realCancelInstruction(final CancelInstructionInput input) {
 216         final InstructionImpl instruction = insns.get(input.getId());
 217         if (instruction == null) {
 218             LOG.debug("Instruction {} not present in the graph", input.getId());
 219
 220             final CancelInstructionOutput out = new CancelInstructionOutputBuilder()
 221                     .setFailure(UnknownInstruction.VALUE).build();
 222             return SuccessfulRpcResult.create(out);
 223         }
 224
 225         return SuccessfulRpcResult.create(new CancelInstructionOutputBuilder()
 226                 .setFailure(instruction.tryCancel(null)).build());
 227     }
 228
 229     private synchronized RpcResult<CleanInstructionsOutput> realCleanInstructions(final CleanInstructionsInput input) {
 230         final Set<InstructionId> failed = new HashSet<>();
 231
 232         for (final InstructionId id : input.getId()) {
 233             // Find the instruction
 234             final InstructionImpl instruction = insns.get(id);
 235             if (instruction == null) {
 236                 LOG.debug("Instruction {} not present in the graph", input.getId());
 237                 failed.add(id);
 238                 continue;
 239             }
 240
 241             // Check its status
 242             switch (instruction.getStatus()) {
 243                 case Cancelled:
 244                 case Failed:
 245                 case Successful:
 246                     break;
 247                 case Executing:
 248                 case Queued:
 249                 case Scheduled:
 250                 case Unknown:
 251                     LOG.debug("Instruction {} cannot be cleaned because of it's in state {}",
 252                             id, instruction.getStatus());
 253                     failed.add(id);
 254                     continue;
 255                 default:
 256                     break;
 257             }
 258
 259             // The instruction is in a terminal state, we need to just unlink
 260             // it from its dependencies and dependents
 261             instruction.clean();
 262
 263             insns.remove(id);
 264             LOG.debug("Instruction {} cleaned successfully", id);
 265         }
 266
 267         final CleanInstructionsOutputBuilder ob = new CleanInstructionsOutputBuilder();
 268         ob.setUnflushed(failed);
 269
 270         return SuccessfulRpcResult.create(ob.build());
 271     }
 272
 273     private List<InstructionImpl> checkDependencies(final SubmitInstructionInput input) throws SchedulerException {
 274         final List<InstructionImpl> dependencies = collectDependencies(input);
 275         // Check if all dependencies are non-failed
 276         final Set<InstructionId> unmet = checkIfUnfailed(dependencies);
 277         /*
 278          *  Some dependencies have failed, declare the request dead-on-arrival
 279          *  and fail the operation.
 280          */
 281         if (!unmet.isEmpty()) {
 282             throw new SchedulerException("Instruction's dependencies are already unsuccessful", new FailureBuilder()
 283                     .setType(DeadOnArrival.VALUE).setFailedPreconditions(unmet).build());
 284         }
 285         return dependencies;
 286     }
 287
 288     private List<InstructionImpl> collectDependencies(final SubmitInstructionInput input) throws SchedulerException {
 289         final List<InstructionImpl> dependencies = new ArrayList<>();
 290         for (final InstructionId pid : input.getPreconditions()) {
 291             final InstructionImpl instruction = insns.get(pid);
 292             if (instruction == null) {
 293                 LOG.info("Instruction {} depends on {}, which is not a known instruction", input.getId(), pid);
 294                 throw new SchedulerException("Unknown dependency ID specified",
 295                         new FailureBuilder().setType(UnknownPreconditionId.VALUE).build());
 296             }
 297             dependencies.add(instruction);
 298         }
 299         return dependencies;
 300     }
 301
 302     private static Set<InstructionId> checkIfUnfailed(final List<InstructionImpl> dependencies) {
 303         final Set<InstructionId> unmet = new HashSet<>();
 304         for (final InstructionImpl d : dependencies) {
 305             switch (d.getStatus()) {
 306                 case Cancelled:
 307                 case Failed:
 308                 case Unknown:
 309                     unmet.add(d.getId());
 310                     break;
 311                 case Executing:
 312                 case Queued:
 313                 case Scheduled:
 314                 case Successful:
 315                     break;
 316                 default:
 317                     break;
 318             }
 319         }
 320         return unmet;
 321     }
 322
 323     @Override
 324     public synchronized ListenableFuture<Instruction> scheduleInstruction(final SubmitInstructionInput input) throws
 325             SchedulerException {
 326         final InstructionId id = input.getId();
 327         if (insns.get(id) != null) {
 328             LOG.info("Instruction ID {} already present", id);
 329             throw new SchedulerException("Instruction ID currently in use",
 330                     new FailureBuilder().setType(DuplicateInstructionId.VALUE).build());
 331         }
 332
 333         // First things first: check the deadline
 334         final Nanotime now = NanotimeUtil.currentTime();
 335         final BigInteger left = input.getDeadline().getValue().toJava().subtract(now.getValue().toJava());
 336
 337         if (left.compareTo(BigInteger.ZERO) <= 0) {
 338             LOG.debug("Instruction {} deadline has already passed by {}ns", id, left);
 339             throw new SchedulerException("Instruction arrived after specified deadline",
 340                     new FailureBuilder().setType(DeadOnArrival.VALUE).build());
 341         }
 342
 343         // Resolve dependencies
 344         final List<InstructionImpl> dependencies = checkDependencies(input);
 345
 346         /*
 347          * All pre-flight checks done are at this point, the following
 348          * steps can only fail in catastrophic scenarios (OOM and the
 349          * like).
 350          */
 351
 352         // Schedule a timeout for the instruction
 353         final Timeout t = timer.newTimeout(timeout -> timeoutInstruction(input.getId()), left.longValue(),
 354                 TimeUnit.NANOSECONDS);
 355
 356         // Put it into the instruction list
 357         final SettableFuture<Instruction> ret = SettableFuture.create();
 358         final InstructionImpl instruction = new InstructionImpl(new InstructionPusher(id, input.getDeadline()), ret, id,
 359                 dependencies, t);
 360         insns.put(id, instruction);
 361
 362         // Attach it into its dependencies
 363         for (final InstructionImpl d : dependencies) {
 364             d.addDependant(instruction);
 365         }
 366
 367         /*
 368          * All done. The next part is checking whether the instruction can
 369          * run, which we can figure out after sending out the acknowledgement.
 370          * This task should be ingress-weighed, so we reinsert it into the
 371          * same execution service.
 372          */
 373         executor.execute(() -> tryScheduleInstruction(instruction));
 374
 375         return ret;
 376     }
 377
 378     @Override
 379     public String getInstructionID() {
 380         return instructionId;
 381     }
 382
 383     private synchronized void timeoutInstruction(final InstructionId id) {
 384         final InstructionImpl instruction = insns.get(id);
 385         if (instruction == null) {
 386             LOG.warn("Instruction {} timed out, but not found in the queue", id);
 387             return;
 388         }
 389
 390         instruction.timeout();
 391     }
 392
 393     private synchronized void tryScheduleDependants(final InstructionImpl instruction) {
 394         // Walk all dependants and try to schedule them
 395         final Iterator<InstructionImpl> it = instruction.getDependants();
 396         while (it.hasNext()) {
 397             tryScheduleInstruction(it.next());
 398         }
 399     }
 400
 401     private synchronized void tryScheduleInstruction(final InstructionImpl instruction) {
 402         final ListenableFuture<ExecutionResult<Details>> f = instruction.ready();
 403         if (f != null) {
 404             Futures.addCallback(f, new FutureCallback<ExecutionResult<Details>>() {
 405                 @Override
 406                 public void onSuccess(final ExecutionResult<Details> result) {
 407                     tryScheduleDependants(instruction);
 408                 }
 409
 410                 @Override
 411                 public void onFailure(final Throwable trw) {
 412                     LOG.error("Instruction {} failed to execute", instruction.getId(), trw);
 413                 }
 414             }, MoreExecutors.directExecutor());
 415         }
 416
 417     }
 418
 419     @Override
 420     public synchronized FluentFuture<? extends CommitInfo> closeServiceInstance() {
 421         LOG.info("Closing Instruction Queue service {}", sgi.getName());
 422
 423         if (reg != null) {
 424             reg.close();
 425             reg = null;
 426         }
 427         for (final InstructionImpl instruction : insns.values()) {
 428             instruction.tryCancel(null);
 429         }
 430         // Workaround for BUG-2283
 431         final WriteTransaction wt = dataProvider.newWriteOnlyTransaction();
 432         wt.delete(LogicalDatastoreType.OPERATIONAL, qid);
 433
 434         final FluentFuture<? extends CommitInfo> future = wt.commit();
 435         future.addCallback(new FutureCallback<CommitInfo>() {
 436             @Override
 437             public void onSuccess(final CommitInfo result) {
 438                 LOG.debug("Instruction Queue {} removed", qid);
 439             }
 440
 441             @Override
 442             public void onFailure(final Throwable trw) {
 443                 LOG.error("Failed to shutdown Instruction Queue {}", qid, trw);
 444             }
 445         }, MoreExecutors.directExecutor());
 446
 447         return future;
 448     }
 449
 450     @Override
 451     public synchronized void close() {
 452         csspReg.close();
 453     }
 454 }