Changeset 639

Timestamp:

07/17/15 13:58:12 (9 years ago)

Author:

rlacroix

Message:

Add a basic garbage collector to ensure no packets linger in the filter graph.

Location:

XIOS/trunk/src

Files:

• filter/file_writer_filter.cpp (modified) (1 diff)
• filter/file_writer_filter.hpp (modified) (1 diff)
• filter/filter.cpp (modified) (1 diff)
• filter/filter.hpp (modified) (1 diff)
• filter/garbage_collector.cpp (added)
• filter/garbage_collector.hpp (added)
• filter/input_pin.cpp (modified) (3 diffs)
• filter/input_pin.hpp (modified) (3 diffs)
• filter/store_filter.cpp (modified) (2 diffs)
• filter/store_filter.hpp (modified) (2 diffs)
• interface/c/iccalendar.cpp (modified) (1 diff)
• node/context.cpp (modified) (1 diff)
• node/context.hpp (modified) (2 diffs)

XIOS/trunk/src/filter/file_writer_filter.cpp

@@ -5 +5 @@
 namespace xios
 {
-  CFileWriterFilter::CFileWriterFilter(CField* field)
-    : CInputPin(1)
+  CFileWriterFilter::CFileWriterFilter(CGarbageCollector& gc, CField* field)
+    : CInputPin(gc, 1)
     , field(field)
   {

XIOS/trunk/src/filter/file_writer_filter.hpp

@@ -15 +15 @@
     public:
       /*!
-       * Constructs the filter (with one input slot) associated to the specified field.
+       * Constructs the filter (with one input slot) associated to the specified field
+       * and a garbage collector.
        *
+       * \param gc the associated garbage collector
        * \param field the associated field
        */
-      CFileWriterFilter(CField* field);
+      CFileWriterFilter(CGarbageCollector& gc, CField* field);
 
     protected:

XIOS/trunk/src/filter/filter.cpp

@@ -3 +3 @@
 namespace xios
 {
-  CFilter::CFilter(size_t inputSlotsCount, IFilterEngine* engine)
-    : CInputPin(inputSlotsCount)
+  CFilter::CFilter(CGarbageCollector& gc, size_t inputSlotsCount, IFilterEngine* engine)
+    : CInputPin(gc, inputSlotsCount)
     , COutputPin()
     , engine(engine)

XIOS/trunk/src/filter/filter.hpp

@@ -20 +20 @@
        * and the specified engine.
        *
+       * \param gc the associated garbage collector
        * \param inputSlotsCount the number of input slots
        * \param engine the filter engine
        */
-      CFilter(size_t inputSlotsCount, IFilterEngine* engine);
+      CFilter(CGarbageCollector& gc, size_t inputSlotsCount, IFilterEngine* engine);
 
     protected:

XIOS/trunk/src/filter/input_pin.cpp

@@ -1 +1 @@
 #include "input_pin.hpp"
+#include "garbage_collector.hpp"
 #include "exception.hpp"
 
 namespace xios
 {
-  CInputPin::CInputPin(size_t slotsCount)
-    : slotsCount(slotsCount)
+  CInputPin::CInputPin(CGarbageCollector& gc, size_t slotsCount)
+    : gc(gc)
+    , slotsCount(slotsCount)
   { /* Nothing to do */ }
 
@@ -19 +21 @@
     std::map<Time, InputBuffer>::iterator it = inputs.find(packet->timestamp);
     if (it == inputs.end())
+    {
       it = inputs.insert(std::make_pair(packet->timestamp, InputBuffer(slotsCount))).first;
+      gc.registerFilter(this, packet->timestamp);
+    }
     it->second.slotsFilled++;
     it->second.packets[inputSlot] = packet;
@@ -25 +30 @@
     if (it->second.slotsFilled == slotsCount)
     {
+      // Unregister before calling onInputReady in case the filter registers again
+      gc.unregisterFilter(this, packet->timestamp);
       onInputReady(it->second.packets);
       inputs.erase(it);
     }
   }
+
+  void CInputPin::invalidate(Time timestamp)
+  {
+    inputs.erase(inputs.begin(), inputs.lower_bound(timestamp));
+  }
 } // namespace xios

XIOS/trunk/src/filter/input_pin.hpp

@@ -9 +9 @@
 namespace xios
 {
+  class CGarbageCollector;
+
   /*!
    * An input pin handles the data packets received by a filter.
@@ -16 +18 @@
     public:
       /*!
-       * Constructs an input pin with the specified number of slots.
+       * Constructs an input pin with the specified number of slots
+       * and an associated garbage collector.
        *
+       * \param gc the garbage collector associated with this input pin
        * \param slotsCount the number of slots
        */
-      CInputPin(size_t slotsCount);
+      CInputPin(CGarbageCollector& gc, size_t slotsCount);
 
       /*!
@@ -32 +36 @@
       void setInput(size_t inputSlot, CDataPacketPtr packet);
 
+      /*!
+       * Removes all pending packets which are older than the specified timestamp.
+       *
+       * \param timestamp the timestamp used for invalidation
+       */
+      void virtual invalidate(Time timestamp);
+
     protected:
+      CGarbageCollector& gc; //!< The garbage collector associated to the input pin
+
       /*!
        * Function triggered when all slots have been filled for a specific timestamp.

XIOS/trunk/src/filter/store_filter.cpp

@@ -6 +6 @@
 namespace xios
 {
-  CStoreFilter::CStoreFilter(CContext* context, CGrid* grid)
-    : CInputPin(1)
+  CStoreFilter::CStoreFilter(CGarbageCollector& gc, CContext* context, CGrid* grid)
+    : CInputPin(gc, 1)
     , context(context)
     , grid(grid)
@@ -63 +63 @@
   {
     packets.insert(std::make_pair(data[0]->timestamp, data[0]));
+    // The packet is always destroyed by the garbage collector
+    // so we register but never unregister
+    gc.registerFilter(this, data[0]->timestamp);
+  }
+
+  void CStoreFilter::invalidate(Time timestamp)
+  {
+    CInputPin::invalidate(timestamp);
+    packets.erase(packets.begin(), packets.lower_bound(timestamp));
   }
 } // namespace xios

XIOS/trunk/src/filter/store_filter.hpp

@@ -16 +16 @@
     public:
       /*!
-       * Constructs the filter with one input slot for the specified grid
-       * and context.
+       * Constructs the filter with one input slot and an associated
+       * garbage collector for the specified grid and context.
        *
+       * \param gc the garbage collector associated with this input pin
        * \param context the context to which the data belongs
        * \param grid the grid to which the data is attached
        */
-      CStoreFilter(CContext* context, CGrid* grid);
+      CStoreFilter(CGarbageCollector& gc, CContext* context, CGrid* grid);
 
       /*!
@@ -48 +49 @@
       CDataPacket::StatusCode getData(Time timestamp, CArray<double, N>& data);
 
+      /*!
+       * Removes all pending packets which are older than the specified timestamp.
+       *
+       * \param timestamp the timestamp used for invalidation
+       */
+      void virtual invalidate(Time timestamp);
+
     protected:
       /*!

XIOS/trunk/src/interface/c/iccalendar.cpp

@@ -16 +16 @@
     context->updateCalendar(step);
     context->sendUpdateCalendar(step);
-    context->checkPrefetchingOfEnabledReadModeFiles();
     CTimer::get("XIOS").suspend();
   }

XIOS/trunk/src/node/context.cpp

@@ -931 +931 @@
    void CContext::updateCalendar(int step)
    {
-      info(50)<<"updateCalendar : before : "<<calendar->getCurrentDate()<<endl;
+      info(50) << "updateCalendar : before : " << calendar->getCurrentDate() << endl;
       calendar->update(step);
-      info(50)<<"updateCalendar : after : "<<calendar->getCurrentDate()<<endl;
+      info(50) << "updateCalendar : after : " << calendar->getCurrentDate() << endl;
+
+      if (hasClient)
+      {
+        checkPrefetchingOfEnabledReadModeFiles();
+        garbageCollector.invalidate(calendar->getCurrentDate());
+      }
    }
 

XIOS/trunk/src/node/context.hpp

@@ -11 +11 @@
 //#include "context_server.hpp"
 #include "data_output.hpp"
+#include "garbage_collector.hpp"
 
 #include "mpi.hpp"
@@ -218 +219 @@
          std::map<int, StdSize> dataSize_;
          StdString idServer_;
+         CGarbageCollector garbageCollector;