Implement LP-based OMIP blocking analysis

8 files changed, 334 insertions, 1 deletions

diff --git a/native/Makefile b/native/Makefile
index 6f1abe2..54c75d8 100644
--- a/native/Makefile
+++ b/native/Makefile
@@ -118,7 +118,7 @@ ALL = testmain _sched.so _locking.so _sim.so
 
 # Compile LP-based code only if we have a solver.
 ifneq ($(CPLEX_PATH)$(GLPK_PATH),)
-LP_OBJ    = lp_common.o io.o lp_dflp.o lp_dpcp.o lp_mpcp.o lp_fmlp.o
+LP_OBJ    = lp_common.o io.o lp_dflp.o lp_dpcp.o lp_mpcp.o lp_fmlp.o lp_omip.o
 
 ifneq ($(CPLEX_PATH),)
 LP_OBJ   += cplex.o cpx.o
diff --git a/native/include/lp_analysis.h b/native/include/lp_analysis.h
index 998eeb5..862c09e 100644
--- a/native/include/lp_analysis.h
+++ b/native/include/lp_analysis.h
@@ -13,4 +13,9 @@ BlockingBounds* lp_mpcp_bounds(const ResourceSharingInfo& info);
 
 BlockingBounds* lp_part_fmlp_bounds(const ResourceSharingInfo& info);
 
+BlockingBounds* lp_omip_bounds(
+        const ResourceSharingInfo& info,
+        unsigned int num_procs,
+        unsigned int cluster_size);
+
 #endif /* LP_ANALYSYS_H_ */
diff --git a/native/include/lp_common.h b/native/include/lp_common.h
index b459664..8f612ea 100644
--- a/native/include/lp_common.h
+++ b/native/include/lp_common.h
@@ -19,6 +19,9 @@ enum blocking_type
         BLOCKING_PREEMPT
 };
 
+// s-oblivious analysis reuses BLOCKING_DIRECT as a catch-all blocking type.
+#define BLOCKING_SOB BLOCKING_DIRECT
+
 class VarMapper {
 private:
         hashmap<uint64_t, unsigned int> map;
@@ -97,6 +100,12 @@ void set_blocking_objective_part_shm(
         LinearExpression *local_obj = 0,
         LinearExpression *remote_obj = 0);
 
+void set_blocking_objective_sob(
+        VarMapper& vars,
+        const ResourceSharingInfo& info,
+        const TaskInfo& ti,
+        LinearProgram& lp);
+
 void add_mutex_constraints(VarMapper& vars,
                 const ResourceSharingInfo& info,
                 const TaskInfo& ti,  LinearProgram& lp);
diff --git a/native/include/sharedres_types.h b/native/include/sharedres_types.h
index 898434c..fbe8f72 100644
--- a/native/include/sharedres_types.h
+++ b/native/include/sharedres_types.h
@@ -119,6 +119,13 @@ public:
                 return len;
         }
 
+        unsigned int get_num_requests(unsigned int res_id) const
+        {
+                foreach(requests, it)
+                        if (it->get_resource_id() == res_id)
+                                return it->get_num_requests();
+                return 0;
+        }
 };
 
 typedef std::vector<TaskInfo> TaskInfos;
diff --git a/native/src/blocking/linprog/lp_common.cpp b/native/src/blocking/linprog/lp_common.cpp
index 1715d3b..2df1284 100644
--- a/native/src/blocking/linprog/lp_common.cpp
+++ b/native/src/blocking/linprog/lp_common.cpp
@@ -119,6 +119,40 @@ void set_blocking_objective_part_shm(
         }
 }
 
+// This version is for suspension-oblivious shared-memory protocols,
+// where the analysis does not differentiate among the different kinds
+// of blocking (since they are all just added to the execution time anyway).
+void set_blocking_objective_sob(
+        VarMapper& vars,
+        const ResourceSharingInfo& info,
+        const TaskInfo& ti,
+        LinearProgram& lp)
+{
+        LinearExpression *obj;
+
+        obj = lp.get_objective();
+
+        foreach_task_except(info.get_tasks(), ti, tx)
+        {
+                unsigned int t = tx->get_id();
+
+                foreach(tx->get_requests(), request)
+                {
+                        unsigned int q = request->get_resource_id();
+                        double length = request->get_request_length();;
+
+                        foreach_request_instance(*request, ti, v)
+                        {
+                                unsigned int var_id;
+
+                                var_id = vars.lookup(t, q, v, BLOCKING_SOB);
+                                obj->add_term(length, var_id);
+                        }
+                }
+        }
+}
+
+
 // Constraint 1 in [Brandenburg 2013]
 void add_mutex_constraints(
         VarMapper& vars,
diff --git a/native/src/blocking/linprog/lp_omip.cpp b/native/src/blocking/linprog/lp_omip.cpp
new file mode 100644
index 0000000..0e5d864
--- /dev/null
+++ b/native/src/blocking/linprog/lp_omip.cpp
@@ -0,0 +1,268 @@
+#include <iostream>
+#include <set>
+#include <algorithm>
+#include <cmath>
+
+#include "lp_common.h"
+#include "blocking.h"
+#include "stl-hashmap.h"
+
+#include "cpu_time.h"
+
+// Per-cluster, per-resource access counts.
+typedef hashmap<unsigned int, unsigned int> AccessCounts;
+typedef hashmap<unsigned int, AccessCounts> PerClusterACounts;
+
+static AccessCounts count_accesses(
+        const ResourceSharingInfo& info,
+        unsigned int cluster)
+{
+        AccessCounts acount;
+
+        foreach(info.get_tasks(), tx) {
+                if (tx->get_cluster() == cluster) {
+                        foreach(tx->get_requests(), request)
+                        {
+                                unsigned int q = request->get_resource_id();
+                                // count that q is being accessed by tx
+                                acount[q] += 1;
+                        }
+                }
+        }
+        return acount;
+}
+
+static void add_total_constraints(
+        VarMapper& vars,
+        const ResourceSharingInfo& info,
+        const TaskInfo& ti,
+        LinearProgram& lp,
+        unsigned int num_procs)
+{
+        // one constraint for each resource
+        hashmap<unsigned int, LinearExpression *> constraints;
+        LinearExpression *exp;
+
+        foreach_task_except(info.get_tasks(), ti, tx)
+        {
+                unsigned int t = tx->get_id();
+
+                //for all requests accessed by tx
+                foreach(tx->get_requests(), request)
+                {
+                        unsigned int q = request->get_resource_id();
+
+                        if (constraints.find(q) == constraints.end())
+                                constraints[q] = new LinearExpression();
+                        exp = constraints[q];
+                        foreach_request_instance(*request, ti, v)
+                        {
+                                unsigned int var_id;
+                                var_id = vars.lookup(t, q, v, BLOCKING_SOB);
+                                exp->add_var(var_id);
+                        }
+                }
+        }
+
+        // add each per-resource constraint
+        foreach(constraints, it) {
+                unsigned int bound = ti.get_num_requests(it->first);
+                bound *= (2 * num_procs - 1);
+                lp.add_inequality(it->second, bound);
+        }
+}
+
+
+static void add_remote_cluster_constraints(
+        VarMapper& vars,
+        const ResourceSharingInfo& info,
+        const TaskInfo& ti,
+        LinearProgram& lp,
+        AccessCounts& acount,
+        unsigned int num_procs,
+        unsigned int cluster_size)
+{
+        // one constraint for each cluster and each resource
+        hashmap<unsigned int, hashmap<unsigned int, LinearExpression *> > cluster_c;
+        LinearExpression *exp;
+
+        foreach_remote_task(info.get_tasks(), ti, tx)
+        {
+                unsigned int t = tx->get_id();
+                unsigned int c = tx->get_cluster();
+
+                hashmap<unsigned int, LinearExpression *> &constraints = cluster_c[c];
+
+                //for all requests accessed by tx
+                foreach(tx->get_requests(), request)
+                {
+                        unsigned int q = request->get_resource_id();
+
+                        if (constraints.find(q) == constraints.end())
+                                constraints[q] = new LinearExpression();
+                        exp = constraints[q];
+
+                        foreach_request_instance(*request, ti, v)
+                        {
+                                unsigned int var_id;
+                                var_id = vars.lookup(t, q, v, BLOCKING_SOB);
+                                exp->add_var(var_id);
+                        }
+                }
+        }
+
+        // add each per-resource constraint
+        foreach(cluster_c, ct) {
+                assert(ct->first != ti.get_cluster());
+                foreach(ct->second, it) {
+                        unsigned int bound  = ti.get_num_requests(it->first);
+                        unsigned int q = it->first;
+                        if (acount[q] <= 2 * cluster_size) {
+                                // FQ-only case
+                                bound *= acount[q];
+                        } else {
+                                // PQ case
+                                bound *= (cluster_size + num_procs);
+                        }
+                        lp.add_inequality(it->second, bound);
+                }
+        }
+}
+
+static void add_local_cluster_constraints(
+        VarMapper& vars,
+        const ResourceSharingInfo& info,
+        const TaskInfo& ti,
+        LinearProgram& lp,
+        AccessCounts& acount,
+        unsigned int cluster_size)
+{
+        foreach_local_task_except(info.get_tasks(), ti, tx)
+        {
+                unsigned int t = tx->get_id();
+                //for all requests accessed by tx
+                foreach(tx->get_requests(), request)
+                {
+                        unsigned int q = request->get_resource_id();
+                        LinearExpression *exp = new LinearExpression();
+
+                        foreach_request_instance(*request, ti, v)
+                        {
+                                unsigned int var_id;
+                                var_id = vars.lookup(t, q, v, BLOCKING_SOB);
+                                exp->add_var(var_id);
+                        }
+                        unsigned int bound = ti.get_num_requests(q);
+                        if (acount[q] > 2 * cluster_size)
+                                bound *= 2; // PQ case
+                        lp.add_inequality(exp, bound);
+                }
+        }
+}
+
+
+static void add_omip_constraints(
+        VarMapper& vars,
+        const ResourceSharingInfo& info,
+        const TaskInfo& ti,
+        LinearProgram& lp,
+        AccessCounts& acounts,
+        unsigned int num_procs,
+        unsigned int cluster_size)
+{
+        add_total_constraints(vars, info, ti, lp,  num_procs);
+        add_remote_cluster_constraints(vars, info, ti, lp, acounts, num_procs, cluster_size);
+        add_local_cluster_constraints(vars, info, ti, lp, acounts, num_procs);
+}
+
+static void apply_omip_bounds_for_task(
+        unsigned int i,
+        BlockingBounds& bounds,
+        const ResourceSharingInfo& info,
+        PerClusterACounts &pcacounts,
+        unsigned int num_procs,
+        unsigned int cluster_size)
+{
+        LinearProgram lp;
+        VarMapper vars;
+        const TaskInfo& ti = info.get_tasks()[i];
+        unsigned int cluster = ti.get_cluster();
+
+        if (pcacounts.find(cluster) == pcacounts.end())
+                pcacounts[cluster] = count_accesses(info, cluster);
+
+#if DEBUG_LP_OVERHEADS >= 2
+        static DEFINE_CPU_CLOCK(model_gen_cost);
+        static DEFINE_CPU_CLOCK(solver_cost);
+
+        std::cout << "---- " << __FUNCTION__ << " ----" << std::endl;
+
+        model_gen_cost.start();
+#endif
+
+        set_blocking_objective_sob(vars, info, ti, lp);
+        vars.seal();
+
+        add_omip_constraints(vars, info, ti, lp,
+                pcacounts[cluster], num_procs, cluster_size);
+
+#if DEBUG_LP_OVERHEADS >= 2
+        model_gen_cost.stop();
+        std::cout << model_gen_cost << std::endl;
+        solver_cost.start();
+#endif
+
+        Solution *sol = linprog_solve(lp, vars.get_num_vars());
+
+#if DEBUG_LP_OVERHEADS >= 2
+        solver_cost.stop();
+        std::cout << solver_cost << std::endl;
+#endif
+        assert(sol != NULL);
+
+        Interference total;
+
+        total.total_length = lrint(sol->evaluate(*lp.get_objective()));
+        bounds[i] = total;
+
+        delete sol;
+}
+
+
+static BlockingBounds* _lp_omip_bounds(
+        const ResourceSharingInfo& info,
+        unsigned int num_procs,
+        unsigned int cluster_size)
+{
+        PerClusterACounts pcacounts;
+        BlockingBounds* results = new BlockingBounds(info);
+
+        for (unsigned int i = 0; i < info.get_tasks().size(); i++)
+                apply_omip_bounds_for_task(i, *results, info, pcacounts, num_procs, cluster_size);
+
+        return results;
+}
+
+BlockingBounds* lp_omip_bounds(
+        const ResourceSharingInfo& info,
+        unsigned int num_procs,
+        unsigned int cluster_size)
+{
+        assert(num_procs >= cluster_size);
+        assert(num_procs % cluster_size == 0);
+
+#if DEBUG_LP_OVERHEADS >= 1
+        static DEFINE_CPU_CLOCK(cpu_costs);
+
+        cpu_costs.start();
+#endif
+
+        BlockingBounds *results = _lp_omip_bounds(info, num_procs, cluster_size);
+
+#if DEBUG_LP_OVERHEADS >= 1
+        cpu_costs.stop();
+        std::cout << cpu_costs << std::endl;
+#endif
+
+        return results;
+}
diff --git a/schedcat/locking/bounds.py b/schedcat/locking/bounds.py
index 3f31f89..f307b80 100644
--- a/schedcat/locking/bounds.py
+++ b/schedcat/locking/bounds.py
@@ -267,3 +267,9 @@ def apply_lp_part_fmlp_bounds(all_tasks):
         t.locally_blocked = res.get_local_blocking(i)
 
     return res
+
+def apply_omip_bounds(all_tasks, num_cpus, procs_per_cluster):
+    model = get_cpp_model(all_tasks)
+    res = lp_cpp.lp_omip_bounds(model, num_cpus, procs_per_cluster)
+    apply_suspension_oblivious(all_tasks, res)
+    return res
diff --git a/tests/locking.py b/tests/locking.py
index 94dda3b..2a9766f 100644
--- a/tests/locking.py
+++ b/tests/locking.py
@@ -113,6 +113,10 @@ class ApplyBounds(unittest.TestCase):
         lb.apply_global_omlp_bounds(self.ts, 2)
         self.sob_non_zero_blocking()
 
+    def test_omip(self):
+        lb.apply_omip_bounds(self.ts, 2, 1)
+        self.sob_non_zero_blocking()
+
     def test_clustered_omlp(self):
         lb.apply_clustered_omlp_bounds(self.ts, 2)
         self.sob_non_zero_blocking()