gerrit.opnfv Code Review - onosfw.git/blob

   1 /*
   2  * Copyright 2014-2015 Open Networking Laboratory
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 package org.onosproject.store.packet.impl;
  17
  18 import com.google.common.collect.ImmutableSet;
  19 import org.apache.felix.scr.annotations.Activate;
  20 import org.apache.felix.scr.annotations.Component;
  21 import org.apache.felix.scr.annotations.Deactivate;
  22 import org.apache.felix.scr.annotations.Reference;
  23 import org.apache.felix.scr.annotations.ReferenceCardinality;
  24 import org.apache.felix.scr.annotations.Service;
  25 import org.onlab.util.KryoNamespace;
  26 import org.onosproject.cluster.ClusterService;
  27 import org.onosproject.cluster.NodeId;
  28 import org.onosproject.mastership.MastershipService;
  29 import org.onosproject.net.flow.TrafficSelector;
  30 import org.onosproject.net.packet.OutboundPacket;
  31 import org.onosproject.net.packet.PacketEvent;
  32 import org.onosproject.net.packet.PacketEvent.Type;
  33 import org.onosproject.net.packet.PacketRequest;
  34 import org.onosproject.net.packet.PacketStore;
  35 import org.onosproject.net.packet.PacketStoreDelegate;
  36 import org.onosproject.store.AbstractStore;
  37 import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
  38 import org.onosproject.store.cluster.messaging.MessageSubject;
  39 import org.onosproject.store.serializers.KryoNamespaces;
  40 import org.onosproject.store.serializers.KryoSerializer;
  41 import org.onosproject.store.service.ConsistentMap;
  42 import org.onosproject.store.service.Serializer;
  43 import org.onosproject.store.service.StorageService;
  44 import org.onosproject.store.service.Versioned;
  45 import org.slf4j.Logger;
  46
  47 import java.util.HashSet;
  48 import java.util.Set;
  49 import java.util.concurrent.ExecutorService;
  50 import java.util.concurrent.Executors;
  51
  52 import static org.onlab.util.Tools.groupedThreads;
  53 import static org.slf4j.LoggerFactory.getLogger;
  54
  55 /**
  56  * Distributed packet store implementation allowing packets to be sent to
  57  * remote instances.
  58  */
  59 @Component(immediate = true)
  60 @Service
  61 public class DistributedPacketStore
  62         extends AbstractStore<PacketEvent, PacketStoreDelegate>
  63         implements PacketStore {
  64
  65     private final Logger log = getLogger(getClass());
  66
  67     // TODO: make this configurable.
  68     private static final int MESSAGE_HANDLER_THREAD_POOL_SIZE = 4;
  69
  70     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  71     protected MastershipService mastershipService;
  72
  73     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  74     protected ClusterService clusterService;
  75
  76     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  77     protected ClusterCommunicationService communicationService;
  78
  79     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  80     protected StorageService storageService;
  81
  82     private PacketRequestTracker tracker;
  83
  84     private static final MessageSubject PACKET_OUT_SUBJECT =
  85             new MessageSubject("packet-out");
  86
  87     private static final KryoSerializer SERIALIZER = new KryoSerializer() {
  88         @Override
  89         protected void setupKryoPool() {
  90             serializerPool = KryoNamespace.newBuilder()
  91                     .register(KryoNamespaces.API)
  92                     .nextId(KryoNamespaces.BEGIN_USER_CUSTOM_ID)
  93                     .build();
  94         }
  95     };
  96
  97     private ExecutorService messageHandlingExecutor;
  98
  99     @Activate
 100     public void activate() {
 101         messageHandlingExecutor = Executors.newFixedThreadPool(
 102                 MESSAGE_HANDLER_THREAD_POOL_SIZE,
 103                 groupedThreads("onos/store/packet", "message-handlers"));
 104
 105         communicationService.<OutboundPacket>addSubscriber(PACKET_OUT_SUBJECT,
 106                 SERIALIZER::decode,
 107                 packet -> notifyDelegate(new PacketEvent(Type.EMIT, packet)),
 108                 messageHandlingExecutor);
 109
 110         tracker = new PacketRequestTracker();
 111
 112         log.info("Started");
 113     }
 114
 115     @Deactivate
 116     public void deactivate() {
 117         communicationService.removeSubscriber(PACKET_OUT_SUBJECT);
 118         messageHandlingExecutor.shutdown();
 119         log.info("Stopped");
 120     }
 121
 122     @Override
 123     public void emit(OutboundPacket packet) {
 124         NodeId myId = clusterService.getLocalNode().id();
 125         NodeId master = mastershipService.getMasterFor(packet.sendThrough());
 126
 127         if (master == null) {
 128             return;
 129         }
 130
 131         if (myId.equals(master)) {
 132             notifyDelegate(new PacketEvent(Type.EMIT, packet));
 133             return;
 134         }
 135
 136         communicationService.unicast(packet, PACKET_OUT_SUBJECT, SERIALIZER::encode, master)
 137                             .whenComplete((r, error) -> {
 138                                 if (error != null) {
 139                                     log.warn("Failed to send packet-out to {}", master, error);
 140                                 }
 141                             });
 142     }
 143
 144     @Override
 145     public boolean requestPackets(PacketRequest request) {
 146         return tracker.add(request);
 147     }
 148
 149     @Override
 150     public boolean cancelPackets(PacketRequest request) {
 151         return tracker.remove(request);
 152     }
 153
 154     @Override
 155     public Set<PacketRequest> existingRequests() {
 156         return tracker.requests();
 157     }
 158
 159     private class PacketRequestTracker {
 160
 161         private ConsistentMap<TrafficSelector, Set<PacketRequest>> requests;
 162
 163         public PacketRequestTracker() {
 164             requests = storageService.<TrafficSelector, Set<PacketRequest>>consistentMapBuilder()
 165                     .withName("onos-packet-requests")
 166                     .withPartitionsDisabled()
 167                     .withSerializer(Serializer.using(KryoNamespaces.API))
 168                     .build();
 169         }
 170
 171         public boolean add(PacketRequest request) {
 172             Versioned<Set<PacketRequest>> old = requests.get(request.selector());
 173             if (old != null && old.value().contains(request)) {
 174                 return false;
 175             }
 176             // FIXME: add retry logic using a random delay
 177             Set<PacketRequest> newSet = new HashSet<>();
 178             newSet.add(request);
 179             if (old == null) {
 180                 return requests.putIfAbsent(request.selector(), newSet) == null;
 181             }
 182             newSet.addAll(old.value());
 183             return requests.replace(request.selector(), old.version(), newSet);
 184         }
 185
 186         public boolean remove(PacketRequest request) {
 187             Versioned<Set<PacketRequest>> old = requests.get(request.selector());
 188             if (old == null || !old.value().contains(request)) {
 189                 return false;
 190             }
 191             // FIXME: add retry logic using a random delay
 192             Set<PacketRequest> newSet = new HashSet<>(old.value());
 193             newSet.remove(request);
 194             if (newSet.isEmpty()) {
 195                 return requests.remove(request.selector(), old.version());
 196             }
 197             return requests.replace(request.selector(), old.version(), newSet);
 198         }
 199
 200         public Set<PacketRequest> requests() {
 201             ImmutableSet.Builder<PacketRequest> builder = ImmutableSet.builder();
 202             requests.values().forEach(v -> builder.addAll(v.value()));
 203             return builder.build();
 204         }
 205
 206     }
 207 }