mdsalutil/mdsalutil-api/src/main/java/org/opendaylight/genius/utils/batching/ResourceBatchingManager.java

   1 /*
   2  * Copyright (c) 2015 - 2018 Ericsson India Global Services Pvt Ltd. 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.genius.utils.batching;
   9
  10 import com.google.common.base.Preconditions;
  11 import com.google.common.util.concurrent.FluentFuture;
  12 import com.google.common.util.concurrent.FutureCallback;
  13 import com.google.common.util.concurrent.Futures;
  14 import com.google.common.util.concurrent.ListenableFuture;
  15 import com.google.common.util.concurrent.MoreExecutors;
  16 import com.google.common.util.concurrent.SettableFuture;
  17 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  18 import java.util.ArrayList;
  19 import java.util.HashMap;
  20 import java.util.List;
  21 import java.util.Map;
  22 import java.util.Optional;
  23 import java.util.Set;
  24 import java.util.concurrent.BlockingQueue;
  25 import java.util.concurrent.ConcurrentHashMap;
  26 import java.util.concurrent.ExecutionException;
  27 import java.util.concurrent.LinkedBlockingQueue;
  28 import java.util.concurrent.ScheduledExecutorService;
  29 import java.util.concurrent.TimeUnit;
  30 import org.apache.commons.lang3.tuple.ImmutablePair;
  31 import org.apache.commons.lang3.tuple.Pair;
  32 import org.eclipse.jdt.annotation.NonNull;
  33 import org.opendaylight.infrautils.utils.concurrent.Executors;
  34 import org.opendaylight.mdsal.binding.api.DataBroker;
  35 import org.opendaylight.mdsal.binding.api.ReadTransaction;
  36 import org.opendaylight.mdsal.binding.api.ReadWriteTransaction;
  37 import org.opendaylight.mdsal.binding.api.WriteTransaction;
  38 import org.opendaylight.mdsal.common.api.CommitInfo;
  39 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  40 import org.opendaylight.mdsal.common.api.ReadFailedException;
  41 import org.opendaylight.yangtools.util.concurrent.FluentFutures;
  42 import org.opendaylight.yangtools.yang.binding.DataObject;
  43 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  44 import org.slf4j.Logger;
  45 import org.slf4j.LoggerFactory;
  46
  47 /**
  48  * This class lets other modules submit their CRUD methods to it. This class
  49  * will then supply a single transaction to such CRUD methods of the
  50  * subscribers, on which such subscribers write data to that transaction.
  51  * Finally the framework attempts to reliably write this single transaction
  52  * which represents a batch of an ordered list of entities owned by that subscriber,
  53  * to be written/updated/removed from a specific datastore as registered by the subscriber.
  54  */
  55 public class ResourceBatchingManager implements AutoCloseable {
  56
  57     private static final Logger LOG = LoggerFactory.getLogger(ResourceBatchingManager.class);
  58
  59     private static final int INITIAL_DELAY = 3000;
  60     private static final TimeUnit TIME_UNIT = TimeUnit.MILLISECONDS;
  61
  62     private static final int PERIODICITY_IN_MS = 500;
  63     private static final int BATCH_SIZE = 1000;
  64
  65     public enum ShardResource {
  66         CONFIG_TOPOLOGY(LogicalDatastoreType.CONFIGURATION),
  67         OPERATIONAL_TOPOLOGY(LogicalDatastoreType.OPERATIONAL),
  68         CONFIG_INVENTORY(LogicalDatastoreType.CONFIGURATION),
  69         OPERATIONAL_INVENTORY(LogicalDatastoreType.OPERATIONAL);
  70
  71         BlockingQueue<ActionableResource> queue = new LinkedBlockingQueue<>();
  72         LogicalDatastoreType datastoreType;
  73
  74         ShardResource(LogicalDatastoreType datastoreType) {
  75             this.datastoreType = datastoreType;
  76         }
  77
  78         public LogicalDatastoreType getDatastoreType() {
  79             return datastoreType;
  80         }
  81
  82         BlockingQueue<ActionableResource> getQueue() {
  83             return queue;
  84         }
  85     }
  86
  87     private final ConcurrentHashMap<String, Pair<BlockingQueue<ActionableResource>, ResourceHandler>>
  88             resourceHandlerMapper = new ConcurrentHashMap<>();
  89
  90     private final ConcurrentHashMap<String, ScheduledExecutorService>
  91             resourceBatchingThreadMapper = new ConcurrentHashMap<>();
  92
  93     private final Map<String, Set<InstanceIdentifier<?>>> pendingModificationByResourceType = new ConcurrentHashMap<>();
  94
  95     private static ResourceBatchingManager instance;
  96
  97     static {
  98         instance = new ResourceBatchingManager();
  99     }
 100
 101     public static ResourceBatchingManager getInstance() {
 102         return instance;
 103     }
 104
 105     @Override
 106     public void close() {
 107         LOG.trace("ResourceBatchingManager Closed, closing all batched resources");
 108         resourceBatchingThreadMapper.values().forEach(ScheduledExecutorService::shutdown);
 109     }
 110
 111     public void registerBatchableResource(
 112             String resourceType, final BlockingQueue<ActionableResource> resQueue, final ResourceHandler resHandler) {
 113         Preconditions.checkNotNull(resQueue, "ResourceQueue to use for batching cannot not be null.");
 114         Preconditions.checkNotNull(resHandler, "ResourceHandler cannot not be null.");
 115
 116         resourceHandlerMapper.put(resourceType, new ImmutablePair<>(resQueue, resHandler));
 117         ScheduledExecutorService resDelegatorService =
 118                 Executors.newListeningScheduledThreadPool(1, "ResourceBatchingManager", LOG);
 119         resourceBatchingThreadMapper.put(resourceType, resDelegatorService);
 120         LOG.info("Registered resourceType {} with batchSize {} and batchInterval {}", resourceType,
 121                 resHandler.getBatchSize(), resHandler.getBatchInterval());
 122         resDelegatorService.scheduleWithFixedDelay(
 123                 new Batcher(resourceType), resHandler.getBatchInterval(), resHandler.getBatchInterval(), TIME_UNIT);
 124         pendingModificationByResourceType.putIfAbsent(resourceType, ConcurrentHashMap.newKeySet());
 125     }
 126
 127     public void registerDefaultBatchHandlers(DataBroker broker) {
 128         LOG.trace("Registering default batch handlers");
 129         Integer batchSize = Integer.getInteger("resource.manager.batch.size", BATCH_SIZE);
 130         Integer batchInterval = Integer.getInteger("resource.manager.batch.periodicity.ms", PERIODICITY_IN_MS);
 131
 132         for (ShardResource shardResource : ShardResource.values()) {
 133             if (resourceHandlerMapper.containsKey(shardResource.name())) {
 134                 continue;
 135             }
 136             DefaultBatchHandler batchHandler = new DefaultBatchHandler(broker, shardResource.datastoreType, batchSize,
 137                     batchInterval);
 138             registerBatchableResource(shardResource.name(), shardResource.getQueue(), batchHandler);
 139         }
 140     }
 141
 142     private void beforeModification(String resoureType, InstanceIdentifier<?> iid) {
 143         pendingModificationByResourceType.get(resoureType).add(iid);
 144     }
 145
 146     @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
 147         justification = "https://github.com/spotbugs/spotbugs/issues/811")
 148     private void afterModification(String resoureType, InstanceIdentifier<?> iid) {
 149         pendingModificationByResourceType.get(resoureType).remove(iid);
 150     }
 151
 152     /**
 153      * Reads the identifier of the given resource type.
 154      * Not to be used by the applications  which uses their own resource queue
 155      *
 156      * @param resourceType resource type that was registered with batch manager
 157      * @param identifier   identifier to be read
 158      * @return a CheckFuture containing the result of the read
 159      */
 160     public <T extends DataObject> FluentFuture<Optional<T>> read(
 161             String resourceType, InstanceIdentifier<T> identifier) {
 162         BlockingQueue<ActionableResource> queue = getQueue(resourceType);
 163         if (queue != null) {
 164             if (pendingModificationByResourceType.get(resourceType).contains(identifier)) {
 165                 SettableFuture<Optional<T>> readFuture = SettableFuture.create();
 166                 queue.add(new ActionableReadResource<>(identifier, readFuture));
 167                 return FluentFuture.from(Futures.makeChecked(readFuture, ReadFailedException.MAPPER));
 168             } else {
 169                 ResourceHandler resourceHandler = resourceHandlerMapper.get(resourceType).getRight();
 170                 try (ReadTransaction tx = resourceHandler.getResourceBroker().newReadOnlyTransaction()) {
 171                     return tx.read(resourceHandler.getDatastoreType(), identifier);
 172                 }
 173             }
 174         }
 175
 176         return FluentFutures.immediateFailedFluentFuture(new ReadFailedException(
 177                 "No batch handler was registered for resource " + resourceType));
 178     }
 179
 180     public ListenableFuture<Void> merge(ShardResource shardResource, InstanceIdentifier<?> identifier,
 181                                         DataObject updatedData) {
 182         BlockingQueue<ActionableResource> queue = shardResource.getQueue();
 183         if (queue != null) {
 184             beforeModification(shardResource.name(), identifier);
 185             ActionableResource actResource = new ActionableResourceImpl(
 186                     identifier, ActionableResource.UPDATE, updatedData, null/*oldData*/);
 187             queue.add(actResource);
 188             return actResource.getResultFuture();
 189         }
 190         return Futures
 191                 .immediateFailedFuture(new IllegalStateException("Queue missing for provided shardResource "
 192                         + shardResource.name()));
 193     }
 194
 195     public void merge(String resourceType, InstanceIdentifier<?> identifier, DataObject updatedData) {
 196         BlockingQueue<ActionableResource> queue = getQueue(resourceType);
 197         if (queue != null) {
 198             beforeModification(resourceType, identifier);
 199             ActionableResource actResource = new ActionableResourceImpl(
 200                     identifier, ActionableResource.UPDATE, updatedData, null/*oldData*/);
 201             queue.add(actResource);
 202         }
 203     }
 204
 205     public ListenableFuture<Void> delete(ShardResource shardResource, InstanceIdentifier<?> identifier) {
 206         BlockingQueue<ActionableResource> queue = shardResource.getQueue();
 207         if (queue != null) {
 208             beforeModification(shardResource.name(), identifier);
 209             ActionableResource actResource = new ActionableResourceImpl(
 210                     identifier, ActionableResource.DELETE, null, null/*oldData*/);
 211             queue.add(actResource);
 212             return actResource.getResultFuture();
 213         }
 214         return Futures
 215                 .immediateFailedFuture(new IllegalStateException("Queue missing for provided shardResource "
 216                         + shardResource.name()));
 217     }
 218
 219     public void delete(String resourceType, InstanceIdentifier<?> identifier) {
 220         BlockingQueue<ActionableResource> queue = getQueue(resourceType);
 221         if (queue != null) {
 222             beforeModification(resourceType, identifier);
 223             ActionableResource actResource = new ActionableResourceImpl(
 224                     identifier, ActionableResource.DELETE, null, null/*oldData*/);
 225             queue.add(actResource);
 226         }
 227     }
 228
 229     public ListenableFuture<Void> put(ShardResource shardResource, InstanceIdentifier<?> identifier,
 230                                       DataObject updatedData) {
 231         BlockingQueue<ActionableResource> queue = shardResource.getQueue();
 232         if (queue != null) {
 233             beforeModification(shardResource.name(), identifier);
 234             ActionableResource actResource = new ActionableResourceImpl(
 235                     identifier, ActionableResource.CREATE, updatedData, null/*oldData*/);
 236             queue.add(actResource);
 237             return actResource.getResultFuture();
 238         }
 239         return Futures
 240                 .immediateFailedFuture(new IllegalStateException("Queue missing for provided shardResource "
 241                         + shardResource.name()));
 242     }
 243
 244     public void put(String resourceType, InstanceIdentifier<?> identifier, DataObject updatedData) {
 245         BlockingQueue<ActionableResource> queue = getQueue(resourceType);
 246         if (queue != null) {
 247             beforeModification(resourceType, identifier);
 248             ActionableResource actResource = new ActionableResourceImpl(
 249                     identifier, ActionableResource.CREATE, updatedData, null/*oldData*/);
 250             queue.add(actResource);
 251         }
 252     }
 253
 254     private BlockingQueue<ActionableResource> getQueue(String resourceType) {
 255         if (resourceHandlerMapper.containsKey(resourceType)) {
 256             return resourceHandlerMapper.get(resourceType).getLeft();
 257         }
 258         return null;
 259     }
 260
 261     public void deregisterBatchableResource(String resourceType) {
 262         ScheduledExecutorService scheduledThreadPoolExecutor = resourceBatchingThreadMapper.get(resourceType);
 263         if (scheduledThreadPoolExecutor != null) {
 264             scheduledThreadPoolExecutor.shutdown();
 265         }
 266         resourceHandlerMapper.remove(resourceType);
 267         resourceBatchingThreadMapper.remove(resourceType);
 268     }
 269
 270     private class Batcher implements Runnable {
 271         private final String resourceType;
 272
 273         Batcher(String resourceType) {
 274             this.resourceType = resourceType;
 275         }
 276
 277         @Override
 278         public void run() {
 279             List<ActionableResource> resList = new ArrayList<>();
 280
 281             try {
 282                 Pair<BlockingQueue<ActionableResource>, ResourceHandler> resMapper =
 283                         resourceHandlerMapper.get(resourceType);
 284                 if (resMapper == null) {
 285                     LOG.error("Unable to find resourceMapper for batching the ResourceType {}", resourceType);
 286                     return;
 287                 }
 288                 BlockingQueue<ActionableResource> resQueue = resMapper.getLeft();
 289                 ResourceHandler resHandler = resMapper.getRight();
 290                 resList.add(resQueue.take());
 291                 resQueue.drainTo(resList);
 292
 293                 long start = System.currentTimeMillis();
 294                 int batchSize = resHandler.getBatchSize();
 295
 296                 int batches = resList.size() / batchSize;
 297                 if (resList.size() > batchSize) {
 298                     LOG.info("Batched up resources of size {} into batches {} for resourcetype {}",
 299                             resList.size(), batches, resourceType);
 300                     for (int i = 0, j = 0; i < batches; j = j + batchSize,i++) {
 301                         new MdsalDsTask<>(resourceType, resList.subList(j, j + batchSize)).process();
 302                     }
 303                     // process remaining routes
 304                     LOG.trace("Picked up 1 size {} ", resList.subList(batches * batchSize, resList.size()).size());
 305                     new MdsalDsTask<>(resourceType, resList.subList(batches * batchSize, resList.size())).process();
 306                 } else {
 307                     // process less than OR == batchsize routes
 308                     LOG.trace("Picked up 2 size {}", resList.size());
 309                     new MdsalDsTask<>(resourceType, resList).process();
 310                 }
 311
 312                 long timetaken = System.currentTimeMillis() - start;
 313                 LOG.debug("Total taken ##time = {}ms for resourceList of size {} for resourceType {}",
 314                         timetaken, resList.size(), resourceType);
 315
 316             } catch (InterruptedException e) {
 317                 LOG.error("InterruptedException during run()", e);
 318             }
 319
 320         }
 321     }
 322
 323     private class MdsalDsTask<T extends DataObject> {
 324         String resourceType;
 325         List<ActionableResource> actResourceList;
 326
 327         MdsalDsTask(String resourceType, List<ActionableResource> actResourceList) {
 328             this.resourceType = resourceType;
 329             this.actResourceList = actResourceList;
 330         }
 331
 332         @SuppressWarnings("unchecked")
 333         public void process() {
 334             LOG.trace("Picked up 3 size {} of resourceType {}", actResourceList.size(), resourceType);
 335             Pair<BlockingQueue<ActionableResource>, ResourceHandler> resMapper =
 336                     resourceHandlerMapper.get(resourceType);
 337             if (resMapper == null) {
 338                 LOG.error("Unable to find resourceMapper for batching the ResourceType {}", resourceType);
 339                 return;
 340             }
 341             ResourceHandler resHandler = resMapper.getRight();
 342             DataBroker broker = resHandler.getResourceBroker();
 343             LogicalDatastoreType dsType = resHandler.getDatastoreType();
 344             ReadWriteTransaction tx = broker.newReadWriteTransaction();
 345             List<SubTransaction> transactionObjects = new ArrayList<>();
 346             Map<SubTransaction, SettableFuture<Void>> txMap = new HashMap<>();
 347             for (ActionableResource actResource : actResourceList) {
 348                 int startSize = transactionObjects.size();
 349                 switch (actResource.getAction()) {
 350                     case ActionableResource.CREATE:
 351                         resHandler.create(tx, dsType, actResource.getInstanceIdentifier(), actResource.getInstance(),
 352                                 transactionObjects);
 353                         break;
 354                     case ActionableResource.UPDATE:
 355                         Object updated = actResource.getInstance();
 356                         Object original = actResource.getOldInstance();
 357                         resHandler.update(tx, dsType, actResource.getInstanceIdentifier(), original,
 358                                 updated,transactionObjects);
 359                         break;
 360                     case ActionableResource.UPDATECONTAINER:
 361                         Object updatedContainer = actResource.getInstance();
 362                         Object originalContainer = actResource.getOldInstance();
 363                         resHandler.updateContainer(tx, dsType, actResource.getInstanceIdentifier(),
 364                                 originalContainer, updatedContainer,transactionObjects);
 365                         break;
 366                     case ActionableResource.DELETE:
 367                         resHandler.delete(tx, dsType, actResource.getInstanceIdentifier(), actResource.getInstance(),
 368                                 transactionObjects);
 369                         break;
 370                     case ActionableResource.READ:
 371                         ActionableReadResource<DataObject> readAction = (ActionableReadResource<DataObject>)actResource;
 372                         ListenableFuture<Optional<DataObject>> future =
 373                                 tx.read(dsType, readAction.getInstanceIdentifier());
 374                         Futures.addCallback(future, new FutureCallback<Optional<DataObject>>() {
 375                             @Override
 376                             public void onSuccess(Optional<DataObject> result) {
 377                                 readAction.getReadFuture().set(result);
 378                             }
 379
 380                             @Override
 381                             public void onFailure(Throwable failure) {
 382                                 readAction.getReadFuture().setException(failure);
 383                             }
 384                         }, MoreExecutors.directExecutor());
 385                         break;
 386                     default:
 387                         LOG.error("Unable to determine Action for ResourceType {} with ResourceKey {}",
 388                                 resourceType, actResource);
 389                 }
 390                 int endSize = transactionObjects.size();
 391                 if (endSize > startSize) {
 392                     txMap.put(transactionObjects.get(endSize - 1),
 393                             (SettableFuture<Void>) actResource.getResultFuture());
 394                 }
 395             }
 396
 397
 398             long start = System.currentTimeMillis();
 399             FluentFuture<? extends @NonNull CommitInfo> futures = tx.commit();
 400
 401             try {
 402                 futures.get();
 403                 actResourceList.forEach(actionableResource -> {
 404                     ((SettableFuture<Void>) actionableResource.getResultFuture()).set(null);
 405                     postCommit(actionableResource.getAction(), actionableResource.getInstanceIdentifier());
 406                 });
 407                 long time = System.currentTimeMillis() - start;
 408                 LOG.trace("##### Time taken for {} = {}ms", actResourceList.size(), time);
 409
 410             } catch (InterruptedException | ExecutionException e) {
 411                 LOG.error("Exception occurred while batch writing to datastore", e);
 412                 LOG.info("Trying to submit transaction operations one at a time for resType {}", resourceType);
 413                 for (SubTransaction object : transactionObjects) {
 414                     WriteTransaction writeTransaction = broker.newWriteOnlyTransaction();
 415                     switch (object.getAction()) {
 416                         case SubTransaction.CREATE:
 417                             writeTransaction.put(dsType, object.getInstanceIdentifier(),
 418                                     (DataObject) object.getInstance(), true);
 419                             break;
 420                         case SubTransaction.DELETE:
 421                             writeTransaction.delete(dsType, object.getInstanceIdentifier());
 422                             break;
 423                         case SubTransaction.UPDATE:
 424                             writeTransaction.merge(dsType, object.getInstanceIdentifier(),
 425                                     (DataObject) object.getInstance(), true);
 426                             break;
 427                         default:
 428                             LOG.error("Unable to determine Action for transaction object with id {}",
 429                                     object.getInstanceIdentifier());
 430                     }
 431                     FluentFuture<? extends @NonNull CommitInfo> futureOperation = writeTransaction.commit();
 432                     try {
 433                         futureOperation.get();
 434                         if (txMap.containsKey(object)) {
 435                             txMap.get(object).set(null);
 436                         } else {
 437                             LOG.error("Subtx object {} has no Actionable-resource associated with it !! ",
 438                                     object.getInstanceIdentifier());
 439                         }
 440                     } catch (InterruptedException | ExecutionException exception) {
 441                         if (txMap.containsKey(object)) {
 442                             txMap.get(object).setException(exception);
 443                         }
 444                         LOG.error("Error {} to datastore (path, data) : ({}, {})", object.getAction(),
 445                                 object.getInstanceIdentifier(), object.getInstance(), exception);
 446                     } finally {
 447                         postCommit(object.getAction(), object.getInstanceIdentifier());
 448                     }
 449                 }
 450             }
 451         }
 452
 453         private void postCommit(int action, InstanceIdentifier iid) {
 454             switch (action) {
 455                 case ActionableResource.CREATE:
 456                 case ActionableResource.UPDATE:
 457                 case ActionableResource.DELETE:
 458                     afterModification(resourceType, iid);
 459                     break;
 460                 default:
 461                     break;
 462             }
 463         }
 464     }
 465
 466     private static class ActionableReadResource<T extends DataObject> extends ActionableResourceImpl {
 467         private final SettableFuture<Optional<T>> readFuture;
 468
 469         ActionableReadResource(InstanceIdentifier<T> identifier, SettableFuture<Optional<T>> readFuture) {
 470             super(identifier, ActionableResource.READ, null, null);
 471             this.readFuture = readFuture;
 472         }
 473
 474         SettableFuture<Optional<T>> getReadFuture() {
 475             return readFuture;
 476         }
 477     }
 478 }