partial ILP for BSP scheduling resurrected

Author: papp-pal-andras

include/osp/bsp/model/BspScheduleCS.hpp

@@ -475,24 +475,122 @@ class BspScheduleCS : public BspSchedule<Graph_t> {
 
     virtual void shrinkSchedule() override {
 
-        std::vector<bool> comm_phase_empty(this->number_of_supersteps, true);
+        std::vector<unsigned> comm_phase_latest_dependency(this->number_of_supersteps, 0);
+        std::vector<std::vector<unsigned> > first_at = getFirstPresence();
+
         for (auto const &[key, val] : commSchedule)
-            comm_phase_empty[val] = false;
+            if(this->assignedProcessor(std::get<0>(key)) != std::get<1>(key))
+                comm_phase_latest_dependency[val] = std::max(comm_phase_latest_dependency[val], first_at[std::get<0>(key)][std::get<1>(key)]);
+        
+
+        for (const auto &node : BspSchedule<Graph_t>::instance->getComputationalDag().vertices())
+            for (const auto &child : BspSchedule<Graph_t>::instance->getComputationalDag().children(node))
+                if(this->assignedProcessor(node) != this->assignedProcessor(child))
+                    comm_phase_latest_dependency[this->assignedSuperstep(child)] = std::max(comm_phase_latest_dependency[this->assignedSuperstep(child)], first_at[node][this->assignedProcessor(child)]);
+
+        std::vector<bool> comm_phase_deleted(this->number_of_supersteps, false);
+        for(unsigned step = this->number_of_supersteps-1; step < this->number_of_supersteps; --step)
+        {
+            unsigned limit = 0;
+            while(step > limit)
+            {
+                limit = std::max(limit, comm_phase_latest_dependency[step]);
+                if(step > limit)
+                {
+                    comm_phase_deleted[step] = true;
+                    --step;
+                }
+            }
+        }
 
         std::vector<unsigned> new_step_index(this->number_of_supersteps);
-        unsigned current_index = 0;
+        unsigned current_index = std::numeric_limits<unsigned>::max();
         for(unsigned step = 0; step < this->number_of_supersteps; ++step)
         {
-            new_step_index[step] = current_index;
-            if(!comm_phase_empty[step])
+            if(!comm_phase_deleted[step])
                 current_index++;
+
+            new_step_index[step] = current_index;
         }
         for (const auto& node : this->instance->vertices())
             this->node_to_superstep_assignment[node] = new_step_index[this->node_to_superstep_assignment[node]];
         for (auto &[key, val] : commSchedule)
             val = new_step_index[val];
 
-        this->setNumberOfSupersteps(current_index);
+        this->setNumberOfSupersteps(current_index+1);
+    }
+
+    std::vector<std::vector<unsigned> > getFirstPresence() const {
+
+        std::vector<std::vector<unsigned> > first_at(BspSchedule<Graph_t>::instance->numberOfVertices(),
+            std::vector<unsigned>(BspSchedule<Graph_t>::instance->numberOfProcessors(), std::numeric_limits<unsigned>::max()));
+
+        for (const auto &node : BspSchedule<Graph_t>::instance->getComputationalDag().vertices())
+            first_at[node][this->assignedProcessor(node)] = this->assignedSuperstep(node);
+
+        for (auto const &[key, val] : commSchedule)
+            first_at[std::get<0>(key)][std::get<2>(key)] =
+                std::min(first_at[std::get<0>(key)][std::get<2>(key)], val + 1); // TODO: replace by staleness after merge
+
+        return first_at;
+    }
+
+    // remove unneeded comm. schedule entries - these can happen in ILPs, partial ILPs, etc.
+    void cleanCommSchedule(){
+
+        // data that is already present before it arrives
+        std::vector<std::vector<std::multiset<unsigned> > > arrives_at(BspSchedule<Graph_t>::instance->numberOfVertices(),
+            std::vector<std::multiset<unsigned> >(BspSchedule<Graph_t>::instance->numberOfProcessors()));
+        for (const auto &node : BspSchedule<Graph_t>::instance->getComputationalDag().vertices())
+            arrives_at[node][this->assignedProcessor(node)].insert(this->assignedSuperstep(node));
+
+        for (auto const &[key, val] : commSchedule)
+            arrives_at[std::get<0>(key)][std::get<2>(key)].insert(val);
+
+        std::vector<KeyTriple> toErase;
+        for (auto const &[key, val] : commSchedule)
+        {
+            auto itr = arrives_at[std::get<0>(key)][std::get<2>(key)].begin();
+            if(*itr < val)
+                toErase.push_back(key);
+            else if(*itr == val && ++itr != arrives_at[std::get<0>(key)][std::get<2>(key)].end() && *itr == val)
+            {
+                toErase.push_back(key);
+                arrives_at[std::get<0>(key)][std::get<2>(key)].erase(itr);
+            }
+        }
+
+        for(const KeyTriple& key : toErase)
+            commSchedule.erase(key);
+
+        // data that is not used after being sent
+        std::vector<std::vector<std::multiset<unsigned> > > used_at(BspSchedule<Graph_t>::instance->numberOfVertices(),
+            std::vector<std::multiset<unsigned> >(BspSchedule<Graph_t>::instance->numberOfProcessors()));
+        for (const auto &node : BspSchedule<Graph_t>::instance->getComputationalDag().vertices())
+            for (const auto &child : BspSchedule<Graph_t>::instance->getComputationalDag().children(node))
+                used_at[node][this->assignedProcessor(child)].insert(this->assignedSuperstep(child));
+
+        for (auto const &[key, val] : commSchedule)
+            used_at[std::get<0>(key)][std::get<1>(key)].insert(val);
+        
+        // (need to visit cs entries in reverse superstep order here)
+        std::vector<std::vector<KeyTriple> > entries(this->number_of_supersteps);
+        for (auto const &[key, val] : commSchedule)
+            entries[val].push_back(key);
+
+        toErase.clear();
+        for(unsigned step = this->number_of_supersteps-1; step < this->number_of_supersteps; --step)
+            for(const KeyTriple& key : entries[step])
+                if(used_at[std::get<0>(key)][std::get<2>(key)].empty() ||
+                    *used_at[std::get<0>(key)][std::get<2>(key)].rbegin() <= step)
+                {
+                    toErase.push_back(key);
+                    auto itr =  used_at[std::get<0>(key)][std::get<1>(key)].find(step);
+                    used_at[std::get<0>(key)][std::get<1>(key)].erase(itr);
+                }
+        
+        for(const KeyTriple& key : toErase)
+            commSchedule.erase(key);
     }
 };