close #177: handle unfinished arrivals (#178)

* initial implementation of #177

* wider verbose output

* add documentation about unfinished arrivals

* fix monitoring bug

* add tests

* update NEWS [ci skip]

Author: Enchufa2

NAMESPACE

@@ -44,6 +44,7 @@ S3method(get_server_count_selected,simmer)
 S3method(get_sources,simmer)
 S3method(get_sources,wrap)
 S3method(get_trajectory,simmer)
+S3method(handle_unfinished,trajectory)
 S3method(join,trajectory)
 S3method(leave,trajectory)
 S3method(length,trajectory)
@@ -116,6 +117,7 @@ export(get_server_count)
 export(get_server_count_selected)
 export(get_sources)
 export(get_trajectory)
+export(handle_unfinished)
 export(join)
 export(leave)
 export(log_)

NEWS.md

@@ -1,9 +1,17 @@
 # simmer 4.1.1
 
+## New features:
+
+- New `handle_unfinished()` activity sets a drop-out trajectory for unfinished
+  arrivals, i.e., those dropped from a resource (due to preemption, resource
+  shrinkage or a rejected `seize`) or those that `leave` a trajectory (#178
+  addressing #177).
+
 ## Minor changes and fixes:
 
 - Fix performance issues in data sources (#176).
 - Update CITATION.
+- Fix monitored activity for preempted arrivals (as part of #178).
 
 # simmer 4.1.0
 

R/RcppExports.R

@@ -165,6 +165,10 @@ Leave__new <- function(prob) {
     .Call(`_simmer_Leave__new`, prob)
 }
 
+HandleUnfinished__new <- function(trj) {
+    .Call(`_simmer_HandleUnfinished__new`, trj)
+}
+
 Leave__new_func <- function(prob) {
     .Call(`_simmer_Leave__new_func`, prob)
 }

R/simmer-methods.R

@@ -181,6 +181,11 @@ peek.simmer <- function(.env, steps=1, verbose=FALSE) .env$peek(steps, verbose)
 #' preempted arrivals go to a dedicated queue, so that \code{queue_size} may be
 #' exceeded. If this option is \code{TRUE}, preempted arrivals go to the standard
 #' queue, and the maximum \code{queue_size} is guaranteed (rejection may occur).
+#' Whenever an arrival is rejected (due to a server drop or a queue drop), it
+#' will set the \code{finished} flag to \code{FALSE} in the output of
+#' \code{\link{get_mon_arrivals}}. Unfinished arrivals can be handled with a
+#' drop-out trajectory that can be set using the \code{\link{handle_unfinished}}
+#' activity.
 #'
 #' @return Returns the simulation environment.
 #' @seealso Convenience functions: \code{\link{schedule}}.

R/trajectory-activities.R

@@ -24,14 +24,20 @@
 #' @param .trj the trajectory object.
 #' @inheritParams select
 #' @param resource the name of the resource.
-#' @param amount the amount to seize/release, accepts either a numeric or a callable
-#' object (a function) which must return a numeric.
+#' @param amount the amount to seize/release, accepts either a numeric or a
+#' callable object (a function) which must return a numeric.
 #' @param continue a boolean (if \code{post.seize} OR \code{reject} is defined)
 #' or a pair of booleans (if \code{post.seize} AND \code{reject} are defined; if
 #' only one value is provided, it will be recycled) to indicate whether these
 #' subtrajectories should continue to the next activity in the main trajectory.
-#' @param post.seize an optional trajectory object which will be followed after a successful seize.
-#' @param reject an optional trajectory object which will be followed if the arrival is rejected.
+#' @param post.seize an optional trajectory object which will be followed after
+#' a successful seize.
+#' @param reject an optional trajectory object which will be followed if the
+#' arrival is rejected. Note that if the arrival is accepted (either in the
+#' queue or in the server) and then it is dropped afterwards due to preemption
+#' or resource shrinkage, then this trajectory won't be executed. Instead, see
+#' \code{\link{handle_unfinished}} for another, more general, method for
+#' handling all kinds of unfinished arrivals.
 #'
 #' @return Returns the trajectory object.
 #' @seealso \code{\link{select}}, \code{\link{set_capacity}}, \code{\link{set_queue_size}},
@@ -353,12 +359,37 @@ rollback.trajectory <- function(.trj, amount, times=Inf, check=NULL) .trj$rollba
 #' @param prob a probability or a function returning a probability.
 #'
 #' @return Returns the trajectory object.
+#'
+#' @details Arrivals that leave the trajectory will set the \code{finished} flag
+#' to \code{FALSE} in the output of \code{\link{get_mon_arrivals}}. Unfinished
+#' arrivals can be handled with a drop-out trajectory that can be set using the
+#' \code{\link{handle_unfinished}} activity.
+#'
+#' @seealso \code{\link{handle_unfinished}}, \code{\link{renege_in}}
 #' @export
 leave <- function(.trj, prob) UseMethod("leave")
 
 #' @export
 leave.trajectory <- function(.trj, prob) .trj$leave(prob)
 
+#' Handle Unfinished Arrivals
+#'
+#' Activity for setting a drop-out trajectory for unfinished arrivals, i.e.,
+#' those dropped from a resource (due to preemption, resource shrinkage or a
+#' rejected \code{\link{seize}}) or those that \code{\link{leave}} a trajectory.
+#'
+#' @inheritParams seize
+#' @param handler trajectory object to handle unfinished arrivals. A \code{NULL}
+#' value will unset the drop-out trajectory.
+#'
+#' @return Returns the trajectory object.
+#' @seealso \code{\link{leave}}, \code{\link{set_capacity}}
+#' @export
+handle_unfinished <- function(.trj, handler) UseMethod("handle_unfinished")
+
+#' @export
+handle_unfinished.trajectory <- function(.trj, handler) .trj$handle_unfinished(handler)
+
 #' Renege on some Condition
 #'
 #' Activities for setting or unsetting a timer or a signal after which the arrival will abandon.
@@ -379,7 +410,7 @@ renege_in.trajectory <- function(.trj, t, out=NULL) .trj$renege_in(t, out)
 #' (a function) which must return a string.
 #'
 #' @rdname renege_in
-#' @seealso \code{\link{send}}
+#' @seealso \code{\link{send}}, \code{\link{leave}}
 #' @export
 renege_if <- function(.trj, signal, out=NULL) UseMethod("renege_if")
 

R/trajectory-class.R

@@ -292,6 +292,12 @@ Trajectory <- R6Class("trajectory",
       )
     },
 
+    handle_unfinished = function(handler) {
+      check_args(handler=c("trajectory", "NULL"))
+      traj <- as.list(c(handler[]))
+      private$add_activity(HandleUnfinished__new(traj))
+    },
+
     renege_in = function(t, out=NULL) {
       check_args(t=c("number", "function"), out=c("trajectory", "NULL"))
       traj <- as.list(c(out[]))

inst/include/simmer/activity/leave.h

@@ -48,6 +48,31 @@ namespace simmer {
     T prob;
   };
 
+  /**
+   * Set a path to handle unfinished arrivals (from 'leave' or resources)
+   */
+  class HandleUnfinished : public Fork {
+  public:
+    CLONEABLE(HandleUnfinished)
+
+    HandleUnfinished(const VEC<REnv>& trj)
+      : Fork("HandleUnfinished", VEC<bool>(trj.size(), false), trj) {}
+
+    void print(unsigned int indent = 0, bool verbose = false, bool brief = false) {
+      Activity::print(indent, verbose, brief);
+      internal::print(brief, false);
+      Fork::print(indent, verbose, brief);
+    }
+
+    double run(Arrival* arrival) {
+      Activity* next = NULL;
+      if (heads.size())
+        next = heads[0];
+      arrival->set_dropout(next);
+      return 0;
+    }
+  };
+
 } // namespace simmer
 
 #endif

inst/include/simmer/process/arrival.h

@@ -65,13 +65,13 @@ namespace simmer {
     Arrival(Simulator* sim, const std::string& name, int mon, Order order,
             Activity* first_activity, int priority = 0)
       : Process(sim, name, mon, priority), order(order), paused(false),
-        clones(new int(0)), activity(first_activity), timer(NULL), batch(NULL)
-        { init(); }
+        clones(new int(0)), activity(first_activity), timer(NULL),
+        dropout(NULL), batch(NULL) { init(); }
 
     Arrival(const Arrival& o)
       : Process(o), order(o.order), paused(o.paused), clones(o.clones),
-        activity(NULL), attributes(o.attributes), timer(NULL), batch(NULL)
-        { init(); }
+        activity(NULL), attributes(o.attributes), timer(NULL),
+        dropout(NULL), batch(NULL) { init(); }
 
     ~Arrival() { reset(); }
 
@@ -186,6 +186,8 @@ namespace simmer {
       batch = NULL;
     }
 
+    void set_dropout(Activity* next) { dropout = next; }
+
     void set_renege(double timeout, Activity* next) {
       cancel_renege();
       timer = new Task(sim, "Renege-Timer",
@@ -222,6 +224,7 @@ namespace simmer {
     SelMap selected;      /**< selected resource */
     Task* timer;          /**< timer that triggers reneging */
     std::string signal;   /**< signal that triggers reneging */
+    Activity* dropout;    /**< drop-out trajectory */
     Batched* batch;       /**< batch that contains this arrival */
     ResMSet resources;    /**< resources that contain this arrival */
 

inst/include/simmer/process/arrival_impl.h

@@ -26,6 +26,12 @@
 namespace simmer {
 
   inline void Arrival::terminate(bool finished) {
+    if (!finished && dropout) {
+      activity = dropout;
+      sim->schedule(0, this, priority);
+      return;
+    }
+
     foreach_ (ResMSet::value_type& itr, resources) {
       Rcpp::warning("'%s': leaving without releasing '%s'", name, itr->name);
       itr->erase(this, true);

inst/include/simmer/resource/preemptive.h

@@ -88,18 +88,22 @@ namespace simmer {
       typename T::iterator first = server.begin();
       if (first == server.end())
         return count;
-      first->arrival->pause();
       if (sim->verbose) print(first->arrival->name, "PREEMPT");
       count += first->amount;
       server_count -= first->amount;
       server_map.erase(first->arrival);
       if (queue_size_strict) {
         if (!room_in_queue(first->amount, first->priority())) {
           if (sim->verbose) print(first->arrival->name, "REJECT");
+          first->arrival->stop();
           first->arrival->unregister_entity(this);
           first->arrival->terminate(false);
-        } else insert_in_queue(first->arrival, first->amount);
+        } else {
+          first->arrival->pause();
+          insert_in_queue(first->arrival, first->amount);
+        }
       } else {
+        first->arrival->pause();
         preempted_map[first->arrival] = preempted.insert(*first);
         queue_count += first->amount;
       }

inst/include/simmer/simulator.h

@@ -141,8 +141,8 @@ namespace simmer {
     {
       Rcpp::Rcout <<
         FMT(10, right) << now_ << " |" <<
-        FMT(12, right) << e_type + ": " << FMT(15, left) << e_name << "|" <<
-        FMT(12, right) << a_type + ": " << FMT(15, left) << a_name << "| " << trail;
+        FMT(12, right) << e_type + ": " << FMT(17, left) << e_name << "|" <<
+        FMT(12, right) << a_type + ": " << FMT(17, left) << a_name << "| " << trail;
       if (flush) Rcpp::Rcout << std::endl;
     }
 

man/add_resource.Rd

@@ -526,6 +526,17 @@ BEGIN_RCPP
     return rcpp_result_gen;
 END_RCPP
 }
+// HandleUnfinished__new
+SEXP HandleUnfinished__new(const std::vector<Environment>& trj);
+RcppExport SEXP _simmer_HandleUnfinished__new(SEXP trjSEXP) {
+BEGIN_RCPP
+    Rcpp::RObject rcpp_result_gen;
+    Rcpp::RNGScope rcpp_rngScope_gen;
+    Rcpp::traits::input_parameter< const std::vector<Environment>& >::type trj(trjSEXP);
+    rcpp_result_gen = Rcpp::wrap(HandleUnfinished__new(trj));
+    return rcpp_result_gen;
+END_RCPP
+}
 // Leave__new_func
 SEXP Leave__new_func(const Function& prob);
 RcppExport SEXP _simmer_Leave__new_func(SEXP probSEXP) {
@@ -1335,6 +1346,7 @@ static const R_CallMethodDef CallEntries[] = {
     {"_simmer_Rollback__new", (DL_FUNC) &_simmer_Rollback__new, 2},
     {"_simmer_Rollback__new_func", (DL_FUNC) &_simmer_Rollback__new_func, 2},
     {"_simmer_Leave__new", (DL_FUNC) &_simmer_Leave__new, 1},
+    {"_simmer_HandleUnfinished__new", (DL_FUNC) &_simmer_HandleUnfinished__new, 1},
     {"_simmer_Leave__new_func", (DL_FUNC) &_simmer_Leave__new_func, 1},
     {"_simmer_Clone__new", (DL_FUNC) &_simmer_Clone__new, 2},
     {"_simmer_Clone__new_func", (DL_FUNC) &_simmer_Clone__new_func, 2},

man/handle_unfinished.Rd

@@ -255,6 +255,11 @@ SEXP Leave__new(double prob) {
   return XPtr<Leave<double> >(new Leave<double>(prob));
 }
 
+//[[Rcpp::export]]
+SEXP HandleUnfinished__new(const std::vector<Environment>& trj) {
+  return XPtr<HandleUnfinished>(new HandleUnfinished(trj));
+}
+
 //[[Rcpp::export]]
 SEXP Leave__new_func(const Function& prob) {
   return XPtr<Leave<Function> >(new Leave<Function>(prob));