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 java.util.HashMap;
14 import java.util.concurrent.Executor;
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 {
27 private final Map<InstanceIdentifier<FlowCapableNode>, SyncupEntry> compressionQueue = new HashMap<>();
28 private final SemaphoreKeeper<InstanceIdentifier<FlowCapableNode>> semaphoreKeeper =
29 new SemaphoreKeeperGuavaImpl<>(1, true);
31 public SyncReactorFutureZipDecorator(final SyncReactor delegate, final Executor executor) {
32 super(delegate, executor);
36 public ListenableFuture<Boolean> syncup(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
37 final SyncupEntry syncupEntry) {
38 Semaphore guard = null;
40 guard = semaphoreKeeper.summonGuardAndAcquire(flowcapableNodePath);
42 return Futures.immediateFuture(Boolean.FALSE);
44 final boolean newTaskNecessary = updateCompressionState(flowcapableNodePath, syncupEntry);
45 if (newTaskNecessary) {
46 super.syncup(flowcapableNodePath, syncupEntry);
48 return Futures.immediateFuture(Boolean.TRUE);
50 semaphoreKeeper.releaseGuard(guard);
55 protected ListenableFuture<Boolean> doSyncupInFuture(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
56 final SyncupEntry syncupEntry) {
57 final var lastCompressionState = removeLastCompressionState(flowcapableNodePath);
58 if (lastCompressionState == null) {
59 return Futures.immediateFuture(Boolean.TRUE);
61 return super.doSyncupInFuture(flowcapableNodePath, lastCompressionState);
66 * If a syncup entry for corresponding the device is present in compression queue and new configuration diff is
67 * coming - update the entry in compression queue (zip). Create new (no entry in queue for device) or replace
68 * entry (config vs. operational is coming) in queue otherwise.
70 private boolean updateCompressionState(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
71 final SyncupEntry syncupEntry) {
72 final var previousEntry = compressionQueue.get(flowcapableNodePath);
73 if (previousEntry != null && syncupEntry.isOptimizedConfigDelta()) {
74 updateOptimizedConfigDelta(flowcapableNodePath, syncupEntry, previousEntry);
76 compressionQueue.put(flowcapableNodePath, syncupEntry);
78 return previousEntry == null;
81 private void updateOptimizedConfigDelta(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath,
82 final SyncupEntry actual, final SyncupEntry previous) {
83 compressionQueue.put(flowcapableNodePath, new SyncupEntry(actual.getAfter(), actual.getDsTypeAfter(),
84 previous.getBefore(), previous.getDsTypeBefore()));
87 private SyncupEntry removeLastCompressionState(final InstanceIdentifier<FlowCapableNode> flowcapableNodePath) {
88 Semaphore guard = null;
90 guard = semaphoreKeeper.summonGuardAndAcquire(flowcapableNodePath);
91 return guard == null ? null : compressionQueue.remove(flowcapableNodePath);
93 semaphoreKeeper.releaseGuard(guard);