2 * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.openflowplugin.applications.frsync.impl;
10 import com.google.common.util.concurrent.Futures;
11 import com.google.common.util.concurrent.ListenableFuture;
12 import com.google.common.util.concurrent.ListeningExecutorService;
13 import java.util.HashMap;
15 import java.util.concurrent.Semaphore;
16 import org.opendaylight.openflowplugin.applications.frsync.SemaphoreKeeper;
17 import org.opendaylight.openflowplugin.applications.frsync.SyncReactor;
18 import org.opendaylight.openflowplugin.applications.frsync.util.SemaphoreKeeperGuavaImpl;
19 import org.opendaylight.openflowplugin.applications.frsync.util.SyncupEntry;
20 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
21 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
24 * Enriches {@link SyncReactorFutureDecorator} with state compression.
26 public class SyncReactorFutureZipDecorator extends SyncReactorFutureDecorator {
28 private final Map<InstanceIdentifier<FlowCapableNode>, SyncupEntry> compressionQueue = new HashMap<>();
29 private final SemaphoreKeeper<InstanceIdentifier<FlowCapableNode>> semaphoreKeeper =
30 new SemaphoreKeeperGuavaImpl<>(1, true);
32 public SyncReactorFutureZipDecorator(final SyncReactor delegate, final ListeningExecutorService executorService) {
33 super(delegate, executorService);
37 public ListenableFuture<Boolean> syncup(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
38 final SyncupEntry syncupEntry) {
39 Semaphore guard = null;
41 guard = semaphoreKeeper.summonGuardAndAcquire(flowcapableNodePath);
43 return Futures.immediateFuture(Boolean.FALSE);
45 final boolean newTaskNecessary = updateCompressionState(flowcapableNodePath, syncupEntry);
46 if (newTaskNecessary) {
47 super.syncup(flowcapableNodePath, syncupEntry);
49 return Futures.immediateFuture(Boolean.TRUE);
51 semaphoreKeeper.releaseGuard(guard);
56 protected ListenableFuture<Boolean> doSyncupInFuture(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
57 final SyncupEntry syncupEntry) {
58 final SyncupEntry lastCompressionState = removeLastCompressionState(flowcapableNodePath);
60 if (lastCompressionState == null) {
61 return Futures.immediateFuture(Boolean.TRUE);
63 return super.doSyncupInFuture(flowcapableNodePath, lastCompressionState);
68 * If a syncup entry for corresponding the device is present in compression queue and new configuration diff is
69 * coming - update the entry in compression queue (zip). Create new (no entry in queue for device) or replace
70 * entry (config vs. operational is coming) in queue otherwise.
72 private boolean updateCompressionState(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
73 final SyncupEntry syncupEntry) {
74 final SyncupEntry previousEntry = compressionQueue.get(flowcapableNodePath);
76 if (previousEntry != null && syncupEntry.isOptimizedConfigDelta()) {
77 updateOptimizedConfigDelta(flowcapableNodePath, syncupEntry, previousEntry);
79 compressionQueue.put(flowcapableNodePath, syncupEntry);
81 return previousEntry == null;
84 private void updateOptimizedConfigDelta(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
85 final SyncupEntry actual,
86 final SyncupEntry previous) {
87 final SyncupEntry updatedEntry = new SyncupEntry(actual.getAfter(), actual.getDsTypeAfter(),
88 previous.getBefore(), previous.getDsTypeBefore());
89 compressionQueue.put(flowcapableNodePath, updatedEntry);
92 private SyncupEntry removeLastCompressionState(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath) {
93 Semaphore guard = null;
95 guard = semaphoreKeeper.summonGuardAndAcquire(flowcapableNodePath);
96 return guard == null ? null : compressionQueue.remove(flowcapableNodePath);
98 semaphoreKeeper.releaseGuard(guard);