openstack/net-virt-providers/src/main/java/org/opendaylight/ovsdb/openstack/netvirt/providers/openflow13/PipelineOrchestratorImpl.java

   1 /*
   2  * Copyright (C) 2014 Red Hat, Inc.
   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  * Authors : Dave Tucker, Madhu Venugopal
   9  */
  10
  11 package org.opendaylight.ovsdb.openstack.netvirt.providers.openflow13;
  12
  13 import com.google.common.collect.Lists;
  14 import com.google.common.collect.Maps;
  15 import java.util.List;
  16 import java.util.Map;
  17 import java.util.concurrent.BlockingQueue;
  18 import java.util.concurrent.ExecutorService;
  19 import java.util.concurrent.Executors;
  20 import java.util.concurrent.LinkedBlockingQueue;
  21 import org.opendaylight.ovsdb.openstack.netvirt.MdsalUtils;
  22 import org.opendaylight.ovsdb.openstack.netvirt.api.Action;
  23 import org.opendaylight.ovsdb.openstack.netvirt.api.NodeCacheListener;
  24 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  25 import org.osgi.framework.ServiceReference;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28
  29 public class PipelineOrchestratorImpl implements NodeCacheListener, PipelineOrchestrator {
  30
  31     private static final Logger logger = LoggerFactory.getLogger(PipelineOrchestratorImpl.class);
  32     private List<Service> staticPipeline = Lists.newArrayList(
  33             Service.CLASSIFIER,
  34             Service.ARP_RESPONDER,
  35             Service.INBOUND_NAT,
  36             Service.EGRESS_ACL,
  37             Service.LOAD_BALANCER,
  38             Service.ROUTING,
  39             Service.L3_FORWARDING,
  40             Service.L2_REWRITE,
  41             Service.INGRESS_ACL,
  42             Service.OUTBOUND_NAT,
  43             Service.L2_FORWARDING
  44     );
  45     Map<Service, AbstractServiceInstance> serviceRegistry = Maps.newConcurrentMap();
  46     private volatile BlockingQueue<Node> queue;
  47     private ExecutorService eventHandler;
  48     public PipelineOrchestratorImpl() {
  49     }
  50
  51     public void registerService(final ServiceReference ref, AbstractServiceInstance serviceInstance){
  52         Service service = (Service)ref.getProperty(AbstractServiceInstance.SERVICE_PROPERTY);
  53         logger.info("registerService {} - {}", serviceInstance, service);
  54         serviceRegistry.put(service, serviceInstance);
  55     }
  56
  57     public void unregisterService(final ServiceReference ref) {
  58         serviceRegistry.remove(ref.getProperty(AbstractServiceInstance.SERVICE_PROPERTY));
  59     }
  60     @Override
  61     public Service getNextServiceInPipeline(Service service) {
  62         int index = staticPipeline.indexOf(service);
  63         if (index >= staticPipeline.size() - 1) return null;
  64         return staticPipeline.get(index + 1);
  65     }
  66
  67     @Override
  68     public AbstractServiceInstance getServiceInstance(Service service) {
  69         if (service == null) return null;
  70         return serviceRegistry.get(service);
  71     }
  72
  73     public void init() {
  74         eventHandler = Executors.newSingleThreadExecutor();
  75         this.queue = new LinkedBlockingQueue<Node>();
  76         logger.info(">>>>> init PipelineOrchestratorImpl");
  77     }
  78
  79     public void start() {
  80         eventHandler.submit(new Runnable()  {
  81             @Override
  82             public void run() {
  83                 try {
  84                     while (true) {
  85                         Node node = queue.take();
  86                         /*
  87                          * Since we are hooking on OpendaylightInventoryListener and as observed in
  88                          * Bug 1997 multiple Threads trying to write to a same table at the same time
  89                          * causes programming issues. Hence delaying the programming by a second to
  90                          * avoid the clash. This hack/workaround should be removed once Bug 1997 is resolved.
  91                          */
  92                         logger.info(">>>>> dequeue: {}", node);
  93                         Thread.sleep(1000);
  94                         for (Service service : staticPipeline) {
  95                             AbstractServiceInstance serviceInstance = getServiceInstance(service);
  96                             //logger.info("pipeline: {} - {}", service, serviceInstance);
  97                             if (serviceInstance != null) {
  98                                 if (MdsalUtils.getBridge(node) != null) {
  99                                     serviceInstance.programDefaultPipelineRule(node);
 100                                 }
 101                             }
 102                         }
 103                     }
 104                 } catch (Exception e) {
 105                     logger.warn("Processing interrupted, terminating ", e);
 106                 }
 107
 108                 while (!queue.isEmpty()) {
 109                     queue.poll();
 110                 }
 111                 queue = null;
 112             }
 113         });
 114     }
 115
 116     public void stop() {
 117         queue.clear();
 118         eventHandler.shutdownNow();
 119     }
 120
 121     @Override
 122     public void enqueue(Node node) {
 123         logger.info(">>>>> enqueue: {}", node);
 124         try {
 125             queue.put(node);
 126         } catch (InterruptedException e) {
 127             logger.warn("Failed to enqueue operation {}", node, e);
 128         }
 129     }
 130
 131     @Override
 132     public void notifyNode(Node node, Action action) {
 133         if (action == Action.ADD) {
 134             enqueue(node);
 135         } else {
 136             logger.info("update ignored: {}", node);
 137         }
 138     }
 139 }