tighten engine blocking terms for few-task case

1 files changed, 54 insertions, 19 deletions

diff --git a/rtss14/gpusync.py b/rtss14/gpusync.py
index 3f6242a..1a67e16 100644
--- a/rtss14/gpusync.py
+++ b/rtss14/gpusync.py
@@ -7,7 +7,7 @@ import sets
 import itertools
 import numpy as np
 
-from math import ceil
+from math import ceil, floor
 from copy import deepcopy
 from collections import defaultdict
 
@@ -95,7 +95,9 @@ class FifoExecEngineLock(FifoEngineLock):
         sz = sum([len(o) for o in others])
         # collapse into a single list
         iters = itertools.chain.from_iterable(others)
-        take = (self.ntokens-1)*nrequests
+
+        terms = min(self.ntokens-1, len(self.users)-1)
+        take = terms*nrequests
         b = np.sum(nmax(np.fromiter(iters, np.float32, count=sz), take))
         return b
  
@@ -112,7 +114,8 @@ class FifoCopyEngineLockNoMigrate(FifoEngineLock):
             for u in self.users if t is not u and 0 in u.resmodel[self].requests]
         sz = sum([len(o) for o in others])
         iters = itertools.chain.from_iterable(others)
-        take = (self.ntokens-1)*nrequests
+        terms = min(self.ntokens-1, len(self.users)-1)
+        take = terms*nrequests
         b = np.sum(nmax(np.fromiter(iters, np.float32, count=sz), take))
         return b
 
@@ -153,11 +156,16 @@ class FifoCopyEngineLockP2P(FifoEngineLock):
 
         assert self.alltokens >= 2
 
+        normal_terms = min(self.ntokens - 1, len(self.users) - 1)
+        trans_terms = min(self.alltokens - 1, len(self.users) - 1)
+
         # hard: take (rho*g-2 normal) and 1 migration, per nhard request
-        # easy: take rho-1 normal, per neasy request
-        nnormal = (self.alltokens - 2) * nhard + (self.ntokens - 1) * neasy
+        nnormal = (trans_terms - 1) * nhard
         nmigs = nhard
 
+        # easy: take rho-1 normal, per neasy request
+        nnormal += normal_terms * neasy
+
         if len(normal) < nnorm:
             # not enough normal requests. take underflow from migrations
             nmigs += nnorm - len(normal)
@@ -214,7 +222,8 @@ class FifoCopyEngineLockP2P(FifoEngineLock):
             assert len(migs) == 0
             assert nhard == 0
 
-            ntake = (self.ntokens - 1) * n
+            terms = min(self.ntokens - 1, len(self.users) - 1)
+            ntake = terms * n
             sz = sum([len(l) for l in _local])
             local = np.fromiter(itertools.chain.from_iterable(_local), np.float32, count=sz)
             b = np.sum(nmax(local, ntake))
@@ -238,9 +247,21 @@ class FifoCopyEngineLockP2P(FifoEngineLock):
             normal_from_cache = True
             normal = self.normal_cache[t]
 
-        nnorm = (self.alltokens - 2)*nhard + (self.ntokens - 1) * neasy
+
+        assert self.alltokens >= 2
+
+        # FIFO queueing. blocked by at most one request from each unique other task
+        # take len(tasks)-1 terms instead of rho-1 / rho*g - 1 tasks if that is smaller.
+        normal_terms = min(self.ntokens - 1, len(self.users) - 1)
+        trans_terms = min(self.alltokens - 1, len(self.users) - 1)
+
+        # hard: take (rho*g-2 normal) and 1 migration, per nhard request
+        nnorm = (trans_terms - 1) * nhard
         nmigs = nhard
 
+        # easy: take rho-1 normal, per neasy request
+        nnorm += normal_terms * neasy
+
         if len(normal) < nnorm:
             # not enough normal requests. take underflow from migrations
             nmigs += nnorm - len(normal)
@@ -300,17 +321,20 @@ class FifoCopyEngineLockAlternateP2P(FifoEngineLock):
         # every migration request is blocked by:
         #  (rho*g-1) normal requests and (rho*g-1) migration requests
 
+        normal_terms = min(self.ntokens - 1, len(self.users) - 1)
+        trans_terms = min(self.alltokens - 1, len(self.users) - 1)
+
         # compute blocking for normal requests, adjusting for overflow
-        nnorm = min((self.ntokens-1)*nrequests, len(normal))
-        nmig   = min(min((self.ntokens-1)*nrequests, len(migs)), nnorm)
+        nnorm = min(normal_terms*nrequests, len(normal))
+        nmig   = min(min(normal_terms*nrequests, len(migs)), nnorm)
 
         # number of interfering requests remaining in each category
         nremain = max(0, len(normal) - nnorm)
         nmigremain = max(0, len(migs) - nmig)
 
         # compute blocking for migration requests
-        nmig_extra = min((self.alltokens-1)*nmigrequests, nmigremain)
-        nnorm_extra = min(min((self.alltokens-1)*nmigrequests, nmig_extra), nremain)
+        nmig_extra = min(trans_terms*nmigrequests, nmigremain)
+        nnorm_extra = min(min(trans_terms*nmigrequests, nmig_extra), nremain)
 
         # accumulate total blocking
         nnorm += nnorm_extra
@@ -354,17 +378,20 @@ class FifoCopyEngineLockAlternateP2P(FifoEngineLock):
         local = np.fromiter(itertools.chain.from_iterable(_local), np.float32, count=sum([len(n) for n in _local]))
         remote = np.fromiter(itertools.chain.from_iterable(_remote), np.float32, count=sum([len(n) for n in _remote]))
 
+        normal_terms = min(self.ntokens - 1, len(self.users) - 1)
+        trans_terms = min(self.alltokens - 1, len(self.users) - 1)
+
         # compute blocking for local requests, adjusting for overflow
-        nlocal  = min((self.ntokens-1)*nrequests, len(local))
-        nmig    = min(min((self.ntokens-1)*nrequests, len(migs)), nlocal)
+        nlocal  = min(normal_terms*nrequests, len(local))
+        nmig    = min(min(normal_terms*nrequests, len(migs)), nlocal)
 
         # number of interfering requests remaining in each category
         nremain = max(0, len(local) + len(remote) - nlocal)
         nmigremain = max(0, len(migs) - nmig)
 
         # compute blocking for migration requests
-        nmig_extra = min((self.alltokens-1)*nmigrequests, nmigremain)
-        nnorm_extra = min(min((self.alltokens-1)*nmigrequests, nmig_extra), nremain)
+        nmig_extra = min(trans_terms*nmigrequests, nmigremain)
+        nnorm_extra = min(min(trans_terms*nmigrequests, nmig_extra), nremain)
 
         nmig += nmig_extra
 
@@ -431,11 +458,18 @@ class KfmlpTokenLock(TokenLock):
 class R2dglpTokenLock(KfmlpTokenLock):
     def __init__(self, tasks, ncpu, ngpu, ntokens):
         super(R2dglpTokenLock,self).__init__(tasks, ngpu, ntokens)
-        fifo_len = int(ceil(float(ncpu)/self.k))
-        self.nterms = 2*fifo_len - 1
-        self.fifo_capacity = fifo_len * self.k
+        if self.k > 0:
+            fifo_len = int(ceil(float(ncpu)/self.k))
+            self.nterms = 2*fifo_len - 1
+            self.fifo_capacity = fifo_len * self.k
+        else:
+            self.nterms = None
+            self.fifo_capacity = None
 
     def blocking(self, t):
+        assert self.nterms is not None
+        assert self.fifo_capacity is not None
+
         if len(self.users) <= self.fifo_capacity:
             # Few enough users that R2DGLP operates like k-FMLP
             return super(R2dglpTokenLock,self).blocking(t)
@@ -457,7 +491,7 @@ class CkomlpTokenLock(TokenLock):
         super(CkomlpTokenLock,self).__init__(tasks, ngpu, ntokens)
         self.k = ngpu*ntokens
         self.ncpu = ncpu
-        self.nterms = int(ceil(float(ncpu)/self.k)) - 1
+        self.nterms = int(ceil(float(ncpu)/self.k)) - 1 if self.k > 0 else None
         self.other_locks = sets.Set()
         # array of unique cluster ids
         self.clusters = set([u.partition for u in self.users])
@@ -466,6 +500,7 @@ class CkomlpTokenLock(TokenLock):
 
     def request_blocking(self, t):
         assert t in self.users
+        assert self.nterms is not None
 
         # return previously computed blocking term if it exists
         if t in self.cache: