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