R/heartbeat_loops.R

In rush: Rapid Asynchronous and Distributed Computing

#' @title Heartbeat Loop
#'
#' @description
#' The heartbeat loop updates the heartbeat key if the worker is still alive.
#' If the kill key is set, the worker is killed.
#' This function is called in a \CRANpkg{callr} session.
#'
#' @param pid (`integer(1)`)\cr
#' Process ID of the worker.
#' @param config ([redux::redis_config])\cr
#' Redis configuration options.
#' @param heartbeat_key (`character(1)`)\cr
#' Heartbeat key.
#'
#' @template param_network_id
#' @template param_worker_id
#' @template param_heartbeat_period
#' @template param_heartbeat_expire
#'
#' @return `NULL`
#' @keywords internal
#' @export
heartbeat = function(network_id, config, worker_id, heartbeat_key, heartbeat_period, heartbeat_expire, pid) {
  r = redux::hiredis(config)
  kill_key = sprintf("%s:%s:kill", network_id, worker_id)

  repeat {
    r$command(c("EXPIRE", heartbeat_key, heartbeat_expire))
    kill = r$command(c("BLPOP", kill_key, heartbeat_period))[[2]]
    if (!is.null(kill)) {
      r$command(c("DEL", heartbeat_key, kill_key))
      r$command(c(
        "SMOVE",
        sprintf("%s:running_worker_ids", network_id),
        sprintf("%s:terminated_worker_ids", network_id),
        worker_id
      ))
      tools::pskill(pid, tools::SIGKILL)
      break
    }
  }

  NULL
}