close #73: monitoring unfinished arrivals

NEWS.md

@@ -4,6 +4,8 @@
 
 * Simplify Rcpp glue: remove unnecessary `as<>()` calls (ec4e51a).
 * Simplify trajectory's head/tail management (06432a8).
+* Ongoing (unfinished) arrivals are reported with `get_mon_arrivals(ongoing = TRUE)` (#73).
+* Now, `run(until)` runs the simulation exactly until `until`, instead of until the first event scheduled at a time >= `until`.
 
 # Changes in version 3.4.1
 

R/RcppExports.R

@@ -37,12 +37,8 @@ add_resource_manager_ <- function(sim_, name, param, intervals, values, period)
     .Call('simmer_add_resource_manager_', PACKAGE = 'simmer', sim_, name, param, intervals, values, period)
 }
 
-get_mon_arrivals_ <- function(sim_) {
-    .Call('simmer_get_mon_arrivals_', PACKAGE = 'simmer', sim_)
-}
-
-get_mon_arrivals_per_resource_ <- function(sim_) {
-    .Call('simmer_get_mon_arrivals_per_resource_', PACKAGE = 'simmer', sim_)
+get_mon_arrivals_ <- function(sim_, per_resource, ongoing) {
+    .Call('simmer_get_mon_arrivals_', PACKAGE = 'simmer', sim_, per_resource, ongoing)
 }
 
 get_mon_attributes_ <- function(sim_) {

R/simulator.R

@@ -130,11 +130,13 @@ Simmer <- R6Class("simmer",
       self
     },
 
-    get_mon_arrivals = function(per_resource=FALSE) {
+    get_mon_arrivals = function(per_resource=FALSE, ongoing=FALSE) {
+      per_resource <- evaluate_value(per_resource)
+      ongoing <- evaluate_value(ongoing)
       as.data.frame(
-        if (!per_resource) get_mon_arrivals_(private$sim_obj)
-        else get_mon_arrivals_per_resource_(private$sim_obj)
-        , stringsAsFactors=FALSE)
+        get_mon_arrivals_(private$sim_obj, per_resource, ongoing), 
+        stringsAsFactors=FALSE
+      )
     },
 
     get_mon_attributes = function() 
@@ -377,13 +379,15 @@ add_generator <- function(env, name_prefix, trajectory, dist, mon=1,
 #' Simulator getters for obtaining monitored data (if any) about arrivals, attributes and resources.
 #' 
 #' @param envs the simulation environment (or a list of environments).
-#' @param per_resource whether the activity should be reported on a per-resource basis (by default: FALSE).
+#' @param per_resource if \code{TRUE}, statistics will be reported on a per-resource basis.
+#' @param ongoing if \code{TRUE}, ongoing arrivals will be reported. The columns
+#' \code{end_time} and \code{finished} of these arrivals are reported as \code{NA}s.
 #' 
 #' @return Return a data frame.
 #' @name get_mon
 #' @export
-get_mon_arrivals <- function(envs, per_resource=FALSE) 
-  envs_apply(envs, "get_mon_arrivals", per_resource)
+get_mon_arrivals <- function(envs, per_resource=FALSE, ongoing=FALSE)
+  envs_apply(envs, "get_mon_arrivals", per_resource, ongoing)
 
 #' @rdname get_mon
 #' @export

R/wrap.R

@@ -11,8 +11,8 @@ simmer.wrap <- R6Class("simmer.wrap",
       private$peek_val <- env$peek(Inf, TRUE)
       private$res <- env$get_resources()
       private$gen <- env$get_generators()
-      private$arrivals <- env$get_mon_arrivals()
-      private$arrivals_res <- env$get_mon_arrivals(TRUE)
+      private$arrivals <- env$get_mon_arrivals(ongoing=TRUE)
+      private$arrivals_res <- env$get_mon_arrivals(TRUE, ongoing=TRUE)
       private$attributes <- env$get_mon_attributes()
       private$resources_all <- env$get_mon_resources(data=c("counts", "limits"))
       private$resources_counts <- env$get_mon_resources(data="counts")
@@ -62,9 +62,16 @@ simmer.wrap <- R6Class("simmer.wrap",
       else ret # nocov
     },
 
-    get_mon_arrivals = function(per_resource=FALSE) {
-      if (per_resource) private$arrivals_res
-      else private$arrivals 
+    get_mon_arrivals = function(per_resource=FALSE, ongoing=FALSE) {
+      if (per_resource) {
+        if (!ongoing)
+          na.omit(private$arrivals_res)
+        else private$arrivals_res
+      } else {
+        if (!ongoing)
+          na.omit(private$arrivals)
+        else private$arrivals 
+      }
     },
     get_mon_attributes = function() { private$attributes },
     get_mon_resources = function(data=c("counts", "limits")) {

src/RcppExports.cpp

@@ -124,24 +124,15 @@ BEGIN_RCPP
 END_RCPP
 }
 // get_mon_arrivals_
-List get_mon_arrivals_(SEXP sim_);
-RcppExport SEXP simmer_get_mon_arrivals_(SEXP sim_SEXP) {
+List get_mon_arrivals_(SEXP sim_, bool per_resource, bool ongoing);
+RcppExport SEXP simmer_get_mon_arrivals_(SEXP sim_SEXP, SEXP per_resourceSEXP, SEXP ongoingSEXP) {
 BEGIN_RCPP
     Rcpp::RObject __result;
     Rcpp::RNGScope __rngScope;
     Rcpp::traits::input_parameter< SEXP >::type sim_(sim_SEXP);
-    __result = Rcpp::wrap(get_mon_arrivals_(sim_));
-    return __result;
-END_RCPP
-}
-// get_mon_arrivals_per_resource_
-List get_mon_arrivals_per_resource_(SEXP sim_);
-RcppExport SEXP simmer_get_mon_arrivals_per_resource_(SEXP sim_SEXP) {
-BEGIN_RCPP
-    Rcpp::RObject __result;
-    Rcpp::RNGScope __rngScope;
-    Rcpp::traits::input_parameter< SEXP >::type sim_(sim_SEXP);
-    __result = Rcpp::wrap(get_mon_arrivals_per_resource_(sim_));
+    Rcpp::traits::input_parameter< bool >::type per_resource(per_resourceSEXP);
+    Rcpp::traits::input_parameter< bool >::type ongoing(ongoingSEXP);
+    __result = Rcpp::wrap(get_mon_arrivals_(sim_, per_resource, ongoing));
     return __result;
 END_RCPP
 }

src/process.cpp

@@ -26,6 +26,7 @@ void Generator::run() {
       (sim->now() + delay) << std::endl;
 
     // schedule the arrival
+    sim->register_arrival(arrival);
     sim->schedule(delay, arrival, count);
   }
   // schedule the generator
@@ -65,6 +66,13 @@ void Task::run() {
   delete this;
 }
 
+void Arrival::reset() {
+  cancel_timeout();
+  if (!--(*clones))
+    delete clones;
+  sim->unregister_arrival(this);
+}
+
 void Arrival::run() {
   double delay;
 
@@ -160,6 +168,16 @@ int Arrival::set_attribute(std::string key, double value) {
   return 0;
 }
 
+double Arrival::get_start(std::string name) {
+  double start = restime[name].start;
+  if (batch) {
+    double up = batch->get_start(name);
+    if (up >= 0 && (start < 0 || up < start))
+      start = up;
+  }
+  return start;
+}
+
 void Arrival::set_timeout(double timeout, Activity* next) {
   cancel_timeout();
   timer = new Task(sim, "Renege-Timer", boost::bind(&Arrival::renege, this, next));

src/process.h

@@ -168,12 +168,7 @@ class Arrival : public Process {
 
   ~Arrival() { reset(); }
 
-  void reset() {
-    cancel_timeout();
-    if (!--(*clones)) 
-      delete clones;
-  }
-
+  void reset();
   void run();
   void activate();
   void deactivate();
@@ -182,18 +177,20 @@ class Arrival : public Process {
   virtual void terminate(bool finished);
   void renege(Activity* next);
   virtual int set_attribute(std::string key, double value);
+  double get_start(std::string name);
 
   Attr* get_attributes() { return &attributes; }
   double get_remaining() { return lifetime.remaining; }
   void set_start(std::string name, double value) { restime[name].start = value; }
   void set_activity(Activity* ptr) { activity = ptr; }
   void set_activity(double value) { lifetime.activity = value; }
   void set_activity(std::string name, double value) { restime[name].activity = value; }
-  double get_activity() { return lifetime.activity; }
+  double get_start() { return lifetime.start; }
+  double get_activity() { return lifetime.activity - lifetime.remaining; }
   double get_activity(std::string name) { return restime[name].activity; }
+
   void set_selected(int id, Resource* res) { selected[id] = res; }
   Resource* get_selected(int id) { return selected[id]; }
-
   void register_entity(Resource* ptr) { resources.insert(ptr); }
   void register_entity(Batched* ptr) { batch = ptr; }
   void unregister_entity(Resource* ptr) { resources.erase(resources.find(ptr)); }
@@ -215,7 +212,7 @@ class Arrival : public Process {
   SelMap selected;      /**< selected resource */
   Task* timer;          /**< timer that triggers reneging */
   Batched* batch;       /**< batch that contains this arrival */
-  ResMSet resources;     /**< resources that contain this arrival */
+  ResMSet resources;    /**< resources that contain this arrival */
 };
 
 /** 

src/simmer_rcpp.cpp

@@ -74,15 +74,9 @@ bool add_resource_manager_(SEXP sim_, std::string name, std::string param,
 }
 
 //[[Rcpp::export]]
-List get_mon_arrivals_(SEXP sim_) {
+List get_mon_arrivals_(SEXP sim_, bool per_resource, bool ongoing) {
   XPtr<Simulator> sim(sim_);
-  return sim->get_arr_traj_stats();
-}
-
-//[[Rcpp::export]]
-List get_mon_arrivals_per_resource_(SEXP sim_) {
-  XPtr<Simulator> sim(sim_);
-  return sim->get_arr_res_stats();
+  return sim->get_arr_stats(per_resource, ongoing);
 }
 
 //[[Rcpp::export]]

src/simulator.h

@@ -33,6 +33,7 @@ class Simulator {
   typedef MSET<Event> PQueue;
   typedef UMAP<Process*, PQueue::iterator> EvMap;
   typedef UMAP<std::string, Entity*> EntMap;
+  typedef USET<Arrival*> ArrSet;
   typedef UMAP<std::string, Batched*> NamBMap;
   typedef UMAP<Activity*, Batched*> UnnBMap;
 
@@ -80,6 +81,7 @@ class Simulator {
       if (itr.second) delete itr.second;
     foreach_ (UnnBMap::value_type& itr, unnamedb_map)
       if (itr.second) delete itr.second;
+    arrival_set.clear();
     namedb_map.clear();
     unnamedb_map.clear();
     b_count = 0;
@@ -125,11 +127,17 @@ class Simulator {
   /**
    * Process the next event. Only one step, a giant leap for mankind.
    */
-  bool step() {
-    if (event_queue.empty()) return false;
+  bool step(double until = -1) {
+    if (event_queue.empty())
+      return false;
     PQueue::iterator ev = event_queue.begin();
-    event_map.erase(ev->process);
+    if (until >= 0 && until <= ev->time) {
+      if (until > now_)
+        now_ = until;
+      return false;
+    }
     now_ = ev->time;
+    event_map.erase(ev->process);
     ev->process->run();
     event_queue.erase(ev);
     return true;
@@ -141,7 +149,7 @@ class Simulator {
    */
   void run(double until) {
     long int nsteps = 0;
-    while ((now_ < until || until < 0) && step())
+    while (step(until))
       if (++nsteps % 100000 == 0)
         Rcpp::checkUserInterrupt();
   }
@@ -265,6 +273,9 @@ class Simulator {
 
   unsigned int get_batch_count() { return b_count++; }
 
+  void register_arrival(Arrival* arrival) { arrival_set.emplace(arrival); }
+  void unregister_arrival(Arrival* arrival) { arrival_set.erase(arrival); }
+
   /**
    * Record monitoring data.
    */
@@ -301,23 +312,61 @@ class Simulator {
   /**
    * Get monitoring data.
    */
-  Rcpp::List get_arr_traj_stats() {
-    return Rcpp::List::create(
-        Rcpp::Named("name")           = arr_traj_stats.get<std::string>("name"),
-        Rcpp::Named("start_time")     = arr_traj_stats.get<double>("start_time"),
-        Rcpp::Named("end_time")       = arr_traj_stats.get<double>("end_time"),
-        Rcpp::Named("activity_time")  = arr_traj_stats.get<double>("activity_time"),
-        Rcpp::Named("finished")       = arr_traj_stats.get<bool>("finished")
-    );
-  }
-  Rcpp::List get_arr_res_stats() {
-    return Rcpp::List::create(
-      Rcpp::Named("name")             = arr_res_stats.get<std::string>("name"),
-      Rcpp::Named("start_time")       = arr_res_stats.get<double>("start_time"),
-      Rcpp::Named("end_time")         = arr_res_stats.get<double>("end_time"),
-      Rcpp::Named("activity_time")    = arr_res_stats.get<double>("activity_time"),
-      Rcpp::Named("resource")         = arr_res_stats.get<std::string>("resource")
-    );
+  Rcpp::List get_arr_stats(bool per_resource, bool ongoing) {
+    if (!per_resource) {
+      VEC<std::string> name             = arr_traj_stats.get<std::string>("name");
+      VEC<double> start_time            = arr_traj_stats.get<double>("start_time");
+      VEC<double> end_time              = arr_traj_stats.get<double>("end_time");
+      VEC<double> activity_time         = arr_traj_stats.get<double>("activity_time");
+      Rcpp::LogicalVector finished      = Rcpp::wrap(arr_traj_stats.get<bool>("finished"));
+      if (ongoing) {
+        foreach_ (Arrival* arrival, arrival_set) {
+          if (!arrival->is_monitored())
+            continue;
+          name.push_back(arrival->name);
+          start_time.push_back(arrival->get_start());
+          end_time.push_back(R_NaReal);
+          activity_time.push_back(R_NaReal);
+          finished.push_back(R_NaInt);
+        }
+      }
+      return Rcpp::List::create(
+          Rcpp::Named("name")           = name,
+          Rcpp::Named("start_time")     = start_time,
+          Rcpp::Named("end_time")       = end_time,
+          Rcpp::Named("activity_time")  = activity_time,
+          Rcpp::Named("finished")       = finished
+      );
+    } else {
+      VEC<std::string> name             = arr_res_stats.get<std::string>("name");
+      VEC<double> start_time            = arr_res_stats.get<double>("start_time");
+      VEC<double> end_time              = arr_res_stats.get<double>("end_time");
+      VEC<double> activity_time         = arr_res_stats.get<double>("activity_time");
+      VEC<std::string> resource         = arr_res_stats.get<std::string>("resource");
+      if (ongoing) {
+        foreach_ (Arrival* arrival, arrival_set) {
+          if (!arrival->is_monitored())
+            continue;
+          foreach_ (EntMap::value_type& itr, resource_map) {
+            double start = arrival->get_start(itr.second->name);
+            if (start < 0)
+              continue;
+            name.push_back(arrival->name);
+            start_time.push_back(start);
+            end_time.push_back(R_NaReal);
+            activity_time.push_back(R_NaReal);
+            resource.push_back(itr.second->name);
+          }
+        }
+      }
+      return Rcpp::List::create(
+        Rcpp::Named("name")             = name,
+        Rcpp::Named("start_time")       = start_time,
+        Rcpp::Named("end_time")         = end_time,
+        Rcpp::Named("activity_time")    = activity_time,
+        Rcpp::Named("resource")         = resource
+      );
+    }
   }
   Rcpp::List get_attr_stats() {
     return Rcpp::List::create(
@@ -360,6 +409,7 @@ class Simulator {
   EntMap resource_map;      /**< map of resources */
   EntMap process_map;       /**< map of processes */
   EvMap event_map;          /**< map of pending events */
+  ArrSet arrival_set;       /**< set of ongoing arrivals */
   NamBMap namedb_map;       /**< map of named batches */
   UnnBMap unnamedb_map;     /**< map of unnamed batches */
   unsigned int b_count;     /**< unnamed batch counter */