applications/bulk-o-matic/src/main/java/org/opendaylight/openflowplugin/applications/bulk/o/matic/TableWriter.java

   1 /*
   2  * Copyright (c) 2017 Ericsson 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.openflowplugin.applications.bulk.o.matic;
   9
  10 import com.google.common.util.concurrent.FutureCallback;
  11 import com.google.common.util.concurrent.Futures;
  12 import com.google.common.util.concurrent.MoreExecutors;
  13
  14 import java.util.concurrent.ExecutorService;
  15 import java.util.concurrent.atomic.AtomicInteger;
  16 import java.util.concurrent.atomic.AtomicLong;
  17
  18 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  19 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  20 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  21 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableBuilder;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  24 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  25 import org.slf4j.Logger;
  26 import org.slf4j.LoggerFactory;
  27
  28 public class TableWriter implements FlowCounterMBean {
  29     private static final Logger LOG = LoggerFactory.getLogger(TableWriter.class);
  30
  31     private final AtomicInteger writeOpStatus = new AtomicInteger(FlowCounter.OperationStatus.INIT.status());
  32     private final AtomicLong taskCompletionTime = new AtomicLong(BulkOMaticUtils.DEFAULT_COMPLETION_TIME);
  33     private final AtomicInteger successfulWrites = new AtomicInteger();
  34     private final AtomicInteger failedWrites = new AtomicInteger();
  35     private final DataBroker dataBroker;
  36     private final ExecutorService tablePusher;
  37
  38     public TableWriter(final DataBroker dataBroker, final ExecutorService tablePusher) {
  39         this.dataBroker = dataBroker;
  40         this.tablePusher = tablePusher;
  41     }
  42
  43     public void addTables(final int dpnCount, final short startTableId, final short endTableId) {
  44         LOG.info("Starting to add tables: {} to {} on each of {}", startTableId, endTableId, dpnCount);
  45         TableHandlerTask task = new TableHandlerTask(dpnCount, startTableId, endTableId, true);
  46         tablePusher.execute(task);
  47     }
  48
  49     public void deleteTables(int dpnCount, short startTableId, short endTableId) {
  50         LOG.info("Starting to delete tables: {} to {} on each of {}", startTableId, endTableId, dpnCount);
  51         TableHandlerTask task = new TableHandlerTask(dpnCount, startTableId, endTableId, false);
  52         tablePusher.execute(task);
  53     }
  54
  55     @Override
  56     public int getWriteOpStatus() {
  57         return writeOpStatus.get();
  58     }
  59
  60     @Override
  61     public long getTaskCompletionTime() {
  62         return taskCompletionTime.get();
  63     }
  64
  65     @Override
  66     public long getTableCount() {
  67         return successfulWrites.get();
  68     }
  69
  70     private class TableHandlerTask implements Runnable {
  71
  72         private final short startTableId;
  73         private final short endTableId;
  74         private final int dpnCount;
  75         private final boolean isAdd;
  76
  77         TableHandlerTask(int dpnCount, short startTableId, short endTableId, boolean isAdd) {
  78             this.dpnCount = dpnCount;
  79             this.startTableId = startTableId;
  80             this.endTableId = endTableId;
  81             this.isAdd = isAdd;
  82         }
  83
  84         @Override
  85         public void run() {
  86             writeOpStatus.set(FlowCounter.OperationStatus.IN_PROGRESS.status());
  87             int totalTables = dpnCount * (endTableId - startTableId + 1);
  88
  89             for (int dpn = 1; dpn <= dpnCount; dpn++) {
  90                 String dpId = BulkOMaticUtils.DEVICE_TYPE_PREFIX + String.valueOf(dpn);
  91                 for (short tableId = startTableId; tableId <= endTableId; tableId++) {
  92                     WriteTransaction wtx = dataBroker.newWriteOnlyTransaction();
  93                     Table table = new TableBuilder().setKey(new TableKey(tableId)).setId(tableId).build();
  94                     InstanceIdentifier<Table> tableIId = BulkOMaticUtils.getTableId(tableId, dpId);
  95
  96                     if (isAdd) {
  97                         wtx.put(LogicalDatastoreType.CONFIGURATION, tableIId, table, true);
  98                     } else {
  99                         wtx.delete(LogicalDatastoreType.CONFIGURATION, tableIId);
 100                     }
 101
 102                     Futures.addCallback(wtx.submit(), new FutureCallback<Void>() {
 103                         @Override
 104                         public void onSuccess(Void voidParameter) {
 105                             if (successfulWrites.incrementAndGet() == totalTables) {
 106                                 if (failedWrites.get() > 0) {
 107                                     writeOpStatus.set(FlowCounter.OperationStatus.FAILURE.status());
 108                                 } else {
 109                                     writeOpStatus.set(FlowCounter.OperationStatus.SUCCESS.status());
 110                                 }
 111                             }
 112                         }
 113
 114                         @Override
 115                         public void onFailure(Throwable throwable) {
 116                             LOG.error("Table addition Failed. Error: {}", throwable);
 117                             if (failedWrites.incrementAndGet() == totalTables) {
 118                                 writeOpStatus.set(FlowCounter.OperationStatus.FAILURE.status());
 119                             }
 120                         }
 121                     }, MoreExecutors.directExecutor());
 122                 }
 123             }
 124         }
 125     }
 126 }