Changeset 1006

Timestamp:

11/28/16 14:02:54 (8 years ago)

Author:

rlacroix

Message:

The workflow is now triggered when using xios_recv_field for fields in read mode received from the servers.

Previously the workflow was triggered upon receiving the data which could cause deadlocks since there are no garanties that clients are receiving data at the same time.

Location:

XIOS/trunk/src

Files:

• filter/filter.cpp (modified) (2 diffs)
• filter/filter.hpp (modified) (1 diff)
• filter/garbage_collector.cpp (modified) (1 diff)
• filter/garbage_collector.hpp (modified) (3 diffs)
• filter/input_pin.cpp (modified) (4 diffs)
• filter/input_pin.hpp (modified) (5 diffs)
• filter/output_pin.cpp (modified) (3 diffs)
• filter/output_pin.hpp (modified) (4 diffs)
• filter/source_filter.cpp (modified) (4 diffs)
• filter/source_filter.hpp (modified) (1 diff)
• filter/spatial_transform_filter.cpp (modified) (1 diff)
• filter/store_filter.cpp (modified) (3 diffs)
• filter/store_filter.hpp (modified) (1 diff)
• node/field.cpp (modified) (3 diffs)

XIOS/trunk/src/filter/filter.cpp

@@ -5 +5 @@
   CFilter::CFilter(CGarbageCollector& gc, size_t inputSlotsCount, IFilterEngine* engine)
     : CInputPin(gc, inputSlotsCount)
-    , COutputPin()
+    , COutputPin(gc)
     , engine(engine)
     , inputSlotCount(inputSlotCount)
@@ -14 +14 @@
     CDataPacketPtr outputPacket = engine->apply(data);
     if (outputPacket)
-      deliverOuput(outputPacket);
+      onOutputReady(outputPacket);
+  }
+
+  void CFilter::setInputTrigger(size_t inputSlot, COutputPin* trigger)
+  {
+    // Was the filter already triggerable? If not, we need to inform
+    // all downstream filters.
+    bool wasTriggerable = canBeTriggered();
+
+    CInputPin::setInputTrigger(inputSlot, trigger);
+
+    if (!wasTriggerable)
+      setOutputTriggers();
+  }
+
+  void CFilter::trigger(Time timestamp)
+  {
+    CInputPin::trigger(timestamp);
+
+    COutputPin::trigger(timestamp);
+  }
+
+  bool CFilter::canBeTriggered() const
+  {
+    return (CInputPin::canBeTriggered() || COutputPin::canBeTriggered());
   }
 } // namespace xios

XIOS/trunk/src/filter/filter.hpp

@@ -26 +26 @@
       CFilter(CGarbageCollector& gc, size_t inputSlotsCount, IFilterEngine* engine);
 
+      /*!
+       * Sets the trigger for a specific input slot.
+       *
+       * \param inputSlot the input slot number
+       * \param trigger the corresponding trigger
+       */
+      void virtual setInputTrigger(size_t inputSlot, COutputPin* trigger);
+
+      /*!
+       * Triggers the filter for the specified timestamp.
+       *
+       * \param timestamp the timestamp for which we are triggering the filter
+       */
+      void virtual trigger(Time timestamp);
+
+      /*!
+       * Tests if the filter can be triggered.
+       *
+       * \return true if the filter can be triggered
+       */
+      bool virtual canBeTriggered() const;
+
     protected:
       IFilterEngine* engine; //!< The filter engine, might be the filter itself

XIOS/trunk/src/filter/garbage_collector.cpp

@@ -3 +3 @@
 namespace xios
 {
-  void CGarbageCollector::registerFilter(CInputPin* inputPin, Time timestamp)
+  void CGarbageCollector::registerObject(InvalidableObject* Object, Time timestamp)
   {
-    registeredFilters[timestamp].insert(inputPin);
+    registeredObjects[timestamp].insert(Object);
   }
 
-  void CGarbageCollector::unregisterFilter(CInputPin* inputPin, Time timestamp)
+  void CGarbageCollector::unregisterObject(InvalidableObject* Object, Time timestamp)
   {
-    std::map<Time, std::set<CInputPin*> >::iterator it = registeredFilters.find(timestamp);
-    if (it != registeredFilters.end())
-      it->second.erase(inputPin);
+    std::map<Time, std::set<InvalidableObject*> >::iterator it = registeredObjects.find(timestamp);
+    if (it != registeredObjects.end())
+      it->second.erase(Object);
   }
 
   void CGarbageCollector::invalidate(Time timestamp)
   {
-    std::map<Time, std::set<CInputPin*> >::iterator it    = registeredFilters.begin(),
-                                                    itEnd = registeredFilters.lower_bound(timestamp);
+    std::map<Time, std::set<InvalidableObject*> >::iterator it    = registeredObjects.begin(),
+                                                            itEnd = registeredObjects.lower_bound(timestamp);
     for (; it != itEnd; ++it)
     {
-      std::set<CInputPin*>::iterator itFilter    = it->second.begin(),
-                                     itFilterEnd = it->second.end();
-      for (; itFilter != itFilterEnd; ++itFilter)
-        (*itFilter)->invalidate(timestamp);
+      std::set<InvalidableObject*>::iterator itObject    = it->second.begin(),
+                                             itObjectEnd = it->second.end();
+      for (; itObject != itObjectEnd; ++itObject)
+        (*itObject)->invalidate(timestamp);
     }
-    registeredFilters.erase(registeredFilters.begin(), itEnd);
+    registeredObjects.erase(registeredObjects.begin(), itEnd);
   }
 } // namespace xios

XIOS/trunk/src/filter/garbage_collector.hpp

@@ -5 +5 @@
 #include <set>
 
-#include "input_pin.hpp"
+#include "date.hpp"
 
 namespace xios
 {
+  /*!
+   * Interface shared by all objects that might need to invalidate packets.
+   */
+  struct InvalidableObject
+  {
+    /*!
+     * Removes all pending packets which are older than the specified timestamp.
+     *
+     * \param timestamp the timestamp used for invalidation
+     */
+    void virtual invalidate(Time timestamp) = 0;
+  }; // struct InvalidableObject
+
   /*!
    * A basic garbage collector which ensures no old packets linger in the filter graph.
@@ -22 +35 @@
 
       /*!
-       * Registers a filter for a specified timestamp.
+       * Registers an object for a specified timestamp.
        *
-       * \param inputPin the input pin of the filter to register
-       * \param timestamp the timestamp for which the filter is registered
+       * \param object the object to register
+       * \param timestamp the timestamp for which the object is registered
        */
-      void registerFilter(CInputPin* inputPin, Time timestamp);
+      void registerObject(InvalidableObject* object, Time timestamp);
 
       /*!
-       * Removes a filter previously registered for a specified timestamp.
+       * Removes a object previously registered for a specified timestamp.
        *
-       * \param inputPin the input pin of the filter to unregister
-       * \param timestamp the timestamp for which the filter is unregistered
+       * \param object the object to unregister
+       * \param timestamp the timestamp for which the object is unregistered
        */
-      void unregisterFilter(CInputPin* inputPin, Time timestamp);
+      void unregisterObject(InvalidableObject* object, Time timestamp);
 
       /*!
-       * Ensures all registered filters invalidate packets older than the specified timestamp.
+       * Ensures all registered objects invalidate packets older than the specified timestamp.
        *
        * \param timestamp the timestamp used for invalidation
@@ -48 +61 @@
       CGarbageCollector& operator=(const CGarbageCollector&);
 
-      std::map<Time, std::set<CInputPin*> > registeredFilters; //!< Currently registered filters
+      std::map<Time, std::set<InvalidableObject*> > registeredObjects; //!< Currently registered objects
   }; // class CGarbageCollector
 } // namespace xios

XIOS/trunk/src/filter/input_pin.cpp

@@ -1 +1 @@
 #include "input_pin.hpp"
+#include "output_pin.hpp"
 #include "garbage_collector.hpp"
 #include "exception.hpp"
@@ -8 +9 @@
     : gc(gc)
     , slotsCount(slotsCount)
+    , triggers(slotsCount)
+    , hasTriggers(false)
   { /* Nothing to do */ }
 
@@ -23 +26 @@
     {
       it = inputs.insert(std::make_pair(packet->timestamp, InputBuffer(slotsCount))).first;
-      gc.registerFilter(this, packet->timestamp);
+      gc.registerObject(this, packet->timestamp);
     }
     it->second.slotsFilled++;
@@ -31 +34 @@
     {
       // Unregister before calling onInputReady in case the filter registers again
-      gc.unregisterFilter(this, packet->timestamp);
+      gc.unregisterObject(this, packet->timestamp);
       onInputReady(it->second.packets);
       inputs.erase(it);
     }
+  }
+
+  void CInputPin::setInputTrigger(size_t inputSlot, COutputPin* trigger)
+  {
+    if (inputSlot >= slotsCount)
+      ERROR("void CInputPin::setInputTrigger(size_t inputSlot, COutputPin* trigger)",
+            "The input slot " << inputSlot << " does not exist.");
+    if (triggers[inputSlot])
+      ERROR("void CInputPin::setInputTrigger(size_t inputSlot, COutputPin* trigger)",
+            "The trigger for input slot " << inputSlot << " has already been set.");
+
+    triggers[inputSlot] = trigger;
+    hasTriggers = true;
+  }
+
+  void CInputPin::trigger(Time timestamp)
+  {
+    if (hasTriggers) // Don't use canBeTriggered here, this function is virtual and can be overriden
+    {
+      std::map<Time, InputBuffer>::iterator it = inputs.find(timestamp);
+      bool nothingReceived = (it == inputs.end());
+
+      for (size_t s = 0; s < slotsCount; s++)
+      {
+        if (triggers[s] && (nothingReceived || !it->second.packets[s]))
+          triggers[s]->trigger(timestamp);
+      }
+    }
+  }
+
+  bool CInputPin::canBeTriggered() const
+  {
+    return hasTriggers;
   }
 

XIOS/trunk/src/filter/input_pin.hpp

@@ -5 +5 @@
 #include <map>
 
+#include "garbage_collector.hpp"
 #include "data_packet.hpp"
 
 namespace xios
 {
-  class CGarbageCollector;
+  class COutputPin;
 
   /*!
    * An input pin handles the data packets received by a filter.
    */
-  class CInputPin
+  class CInputPin : public InvalidableObject
   {
     public:
@@ -27 +28 @@
 
       /*!
+       * Sets the trigger for a specific input slot.
+       *
+       * \param inputSlot the input slot number
+       * \param trigger the corresponding trigger
+       */
+      void virtual setInputTrigger(size_t inputSlot, COutputPin* trigger);
+
+      /*!
        * Receives a data packet from an upstream filter on
        * the specified input slot.
@@ -37 +46 @@
 
       /*!
+       * Triggers the input of any buffered packet for the specified timestamp.
+       *
+       * \param timestamp the timestamp for which we are triggering the input
+       */
+      void virtual trigger(Time timestamp);
+
+      /*!
+       * Tests if the pin can be triggered.
+       *
+       * \return true if the pin can be triggered
+       */
+      bool virtual canBeTriggered() const;
+
+      /*!
        * Removes all pending packets which are older than the specified timestamp.
        *
@@ -44 +67 @@
 
     protected:
-      CGarbageCollector& gc; //!< The garbage collector associated to the input pin
-
       /*!
        * Function triggered when all slots have been filled for a specific timestamp.
@@ -75 +96 @@
       };
 
+      CGarbageCollector& gc; //!< The garbage collector associated to the input pin
+
       size_t slotsCount; //!< The number of slots
 
       //! Input buffer, store the packets until all slots are full for a timestep
       std::map<Time, InputBuffer> inputs;
+
+      //! Store the triggers corresponding to the input slots
+      std::vector<COutputPin*> triggers;
+
+      //! Whether some triggers have been set
+      bool hasTriggers;
   }; // class CInputPin
 } // namespace xios

XIOS/trunk/src/filter/output_pin.cpp

@@ -4 +4 @@
 namespace xios
 {
+  COutputPin::COutputPin(CGarbageCollector& gc, bool manualTrigger /*= false*/)
+    : gc(gc)
+    , manualTrigger(manualTrigger)
+  { /* Nothing to do */ }
+
   void COutputPin::connectOutput(boost::shared_ptr<CInputPin> inputPin, size_t inputSlot)
   {
@@ -11 +16 @@
 
     outputs.push_back(std::make_pair(inputPin, inputSlot));
+
+    if (canBeTriggered())
+      inputPin->setInputTrigger(inputSlot, this);
+  }
+
+  void COutputPin::onOutputReady(CDataPacketPtr packet)
+  {
+    if (!packet)
+      ERROR("void COutputPin::onOutputReady(CDataPacketPtr packet)",
+            "The packet cannot be null.");
+
+    if (manualTrigger) // Don't use canBeTriggered here, this function is virtual and can be overriden
+    {
+      outputPackets[packet->timestamp] = packet;
+      gc.registerObject(this, packet->timestamp);
+    }
+    else
+      deliverOuput(packet);
   }
 
@@ -23 +46 @@
       it->first->setInput(it->second, packet);
   }
+
+  void COutputPin::trigger(Time timestamp)
+  {
+    if (manualTrigger) // Don't use canBeTriggered here, this function is virtual and can be overriden
+    {
+      std::map<Time, CDataPacketPtr>::iterator it = outputPackets.find(timestamp);
+      if (it != outputPackets.end())
+      {
+        gc.unregisterObject(this, timestamp);
+        deliverOuput(it->second);
+        outputPackets.erase(it);
+      }
+    }
+  }
+
+  bool COutputPin::canBeTriggered() const
+  {
+    return manualTrigger;
+  }
+
+  void COutputPin::setOutputTriggers()
+  {
+    std::vector<std::pair<boost::shared_ptr<CInputPin>, size_t> >::iterator it, itEnd;
+    for (it = outputs.begin(), itEnd = outputs.end(); it != itEnd; ++it)
+      it->first->setInputTrigger(it->second, this);
+  }
+
+  void COutputPin::invalidate(Time timestamp)
+  {
+    outputPackets.erase(outputPackets.begin(), outputPackets.lower_bound(timestamp));
+  }
 } // namespace xios

XIOS/trunk/src/filter/output_pin.hpp

@@ -2 +2 @@
 #define __XIOS_COutputPin__
 
+#include "garbage_collector.hpp"
 #include "input_pin.hpp"
 
@@ -9 +10 @@
    * An output pin handles the connections with downstream filters.
    */
-  class COutputPin
+  class COutputPin : public InvalidableObject
   {
     public:
+      /*!
+       * Constructs an ouput pin with manual or automatic trigger
+       * and an associated garbage collector.
+       *
+       * \param gc the garbage collector associated with this ouput pin
+       * \param slotsCount the number of slots
+       */
+      COutputPin(CGarbageCollector& gc, bool manualTrigger = false);
+
       /*!
        * Connects to a specific slot of the input pin of a downstream filter.
@@ -21 +31 @@
       void connectOutput(boost::shared_ptr<CInputPin> inputPin, size_t inputSlot);
 
+      /*!
+       * Triggers the output of any buffered packet for the specified timestamp.
+       *
+       * \param timestamp the timestamp for which we are triggering the output
+       */
+      void virtual trigger(Time timestamp);
+
+      /*!
+       * Tests if the pin can be triggered.
+       *
+       * \return true if the pin can be triggered
+       */
+      bool virtual canBeTriggered() const;
+
+      /*!
+       * Removes all pending packets which are older than the specified timestamp.
+       *
+       * \param timestamp the timestamp used for invalidation
+       */
+      void virtual invalidate(Time timestamp);
+
     protected:
+      /*!
+       * Function triggered when a packet is ready to be delivered.
+       *
+       * \param packet the packet ready for output
+       */
+      void onOutputReady(CDataPacketPtr packet);
+
+      /*!
+       * Informs the downstream pins that this output pin should be triggered.
+       */
+      void setOutputTriggers();
+
+    private:
       /*!
        * Delivers an output packet to the downstreams filter.
@@ -29 +73 @@
       void deliverOuput(CDataPacketPtr packet);
 
-    private:
+      CGarbageCollector& gc; //!< The garbage collector associated to the output pin
+
+      //!< Whether the ouput should be triggered manually
+      bool manualTrigger;
+
       //!< The list of connected filters and the corresponding slot numbers
       std::vector<std::pair<boost::shared_ptr<CInputPin>, size_t> > outputs;
+
+      //! Output buffer, store the packets until the output is triggered
+      std::map<Time, CDataPacketPtr> outputPackets;
   }; // class COutputPin
 } // namespace xios

XIOS/trunk/src/filter/source_filter.cpp

@@ -6 +6 @@
 namespace xios
 {
-  CSourceFilter::CSourceFilter(CGrid* grid, const CDuration offset /*= NoneDu*/)
-    : grid(grid)
+  CSourceFilter::CSourceFilter(CGarbageCollector& gc, CGrid* grid,
+                               const CDuration offset /*= NoneDu*/, bool manualTrigger /*= false*/)
+    : COutputPin(gc, manualTrigger)
+    , grid(grid)
     , offset(offset)
   {
@@ -28 +30 @@
     grid->inputField(data, packet->data);
 
-    deliverOuput(packet);
+    onOutputReady(packet);
   }
 
@@ -64 +66 @@
     }
 
-    deliverOuput(packet);
+    onOutputReady(packet);
   }
 
@@ -73 +75 @@
     packet->timestamp = date;
     packet->status = CDataPacket::END_OF_STREAM;
-    deliverOuput(packet);
+    onOutputReady(packet);
   }
 } // namespace xios

XIOS/trunk/src/filter/source_filter.hpp

@@ -19 +19 @@
        * Constructs a source filter accepting data attached to the specified grid.
        *
+       * \param gc the garbage collector associated with this filter
        * \param grid the grid to which the data is attached
        * \param offset the offset applied to the timestamp of all packets
+       * \param manualTrigger whether the output should be triggered manually
        */
-      CSourceFilter(CGrid* grid, const CDuration offset = NoneDu);
+      CSourceFilter(CGarbageCollector& gc, CGrid* grid,
+                    const CDuration offset = NoneDu, bool manualTrigger = false);
 
       /*!

XIOS/trunk/src/filter/spatial_transform_filter.cpp

@@ -53 +53 @@
     CDataPacketPtr outputPacket = spaceFilter->applyFilter(data, outputDefaultValue);
     if (outputPacket)
-      deliverOuput(outputPacket);
+      onOutputReady(outputPacket);
   }
 

XIOS/trunk/src/filter/store_filter.cpp

@@ -8 +8 @@
   CStoreFilter::CStoreFilter(CGarbageCollector& gc, CContext* context, CGrid* grid)
     : CInputPin(gc, 1)
+    , gc(gc)
     , context(context)
     , grid(grid)
@@ -27 +28 @@
     do
     {
+      if (canBeTriggered())
+        trigger(timestamp);
+
       timer.resume();
 
@@ -74 +78 @@
     // The packet is always destroyed by the garbage collector
     // so we register but never unregister
-    gc.registerFilter(this, data[0]->timestamp);
+    gc.registerObject(this, data[0]->timestamp);
   }
 

XIOS/trunk/src/filter/store_filter.hpp

@@ -65 +65 @@
 
     private:
+      CGarbageCollector& gc; //!< The garbage collector associated to the filter
       CContext* context; //!< The context to which the data belongs
       CGrid* grid; //!< The grid attached to the data the filter can accept

XIOS/trunk/src/node/field.cpp

@@ -800 +800 @@
        // Check if the data is to be read from a file
        else if (file && !file->mode.isEmpty() && file->mode == CFile::mode_attr::read)
-         instantDataFilter = serverSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(grid,
-                                                                                                     freq_offset.isEmpty() ? NoneDu : freq_offset));
+         instantDataFilter = serverSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(gc, grid,
+                                                                                                     freq_offset.isEmpty() ? NoneDu : freq_offset,
+                                                                                                     true));
        else // The data might be passed from the model
-         instantDataFilter = clientSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(grid));
+         instantDataFilter = clientSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(gc, grid));
      }
 
@@ -874 +875 @@
        {
          if (!serverSourceFilter)
-           serverSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(grid,
-                                                                                   freq_offset.isEmpty() ? NoneDu : freq_offset));
+           serverSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(gc, grid,
+                                                                                   freq_offset.isEmpty() ? NoneDu : freq_offset,
+                                                                                   true));
 
          selfReferenceFilter = serverSourceFilter;
@@ -888 +890 @@
        {
          if (!clientSourceFilter)
-           clientSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(grid));
+           clientSourceFilter = boost::shared_ptr<CSourceFilter>(new CSourceFilter(gc, grid));
 
          selfReferenceFilter = clientSourceFilter;