R/map-abstract.R

In dipsaus: A Dipping Sauce for Data Analysis and Visualizations

#' Abstract Map to store key-value pairs
AbstractMap <- R6::R6Class(
  classname = "AbstractMap",
  portable = TRUE,
  cloneable = TRUE,
  private = list(
    .id = character(0),

    # Lock file that each queue should have
    # If lock file is locked, then we should wait till the next transaction period
    .lockfile = character(0),
    lock = NULL,

    # Run expr making sure that locker is locked to be exclusive (for write-only)
    exclusive = function(expr, ...) {
      stopifnot2(private$valid, msg = "Map is not valid")
      custom_locker <- is.function(self$get_locker) && is.function(self$free_locker)
      if (self$has_locker) {

        if (custom_locker) {
          self$get_locker(...)
        } else {
          private$default_get_locker(...)
        }


        on.exit({
          if (custom_locker) {
            self$free_locker()
          } else {
            private$default_free_locker()
          }
        }, add = FALSE, after = FALSE)


        force(expr)

        if (custom_locker) {
          self$free_locker()
        } else {
          private$default_free_locker()
        }
        on.exit({}, add = FALSE, after = FALSE)

      } else {
        force(expr)
      }

    },

    default_get_locker = function(timeout = Inf) {

      timeout <- as.numeric(timeout)
      if (is.na(timeout)) {
        timeout <- Inf
      }

      dipsaus_lock(name = self$lockfile, timeout = timeout)

    },
    default_free_locker = function() {
      dipsaus_unlock(self$lockfile)
    },

    map = NULL,
    valid = TRUE

  ),
  public = list(

    # By default, queue uses file locker, if you have customized locker, please
    # implement these two methods as functions:
    #   get_locker obtain and lock access (exclusive)
    #   free_locker free the lock
    # private$exclusive will take care the rest
    get_locker = NULL,
    free_locker = NULL,
    has_locker = TRUE,
    missing_default = NULL,



    `@remove` = function(keys) {
      not_implemented()
      private$map$remove(keys)
    },
    remove = function(keys) {
      private$exclusive({
        self$`@remove`( keys )
      })
    },
    reset = function(...) {
      keys <- self$keys(include_signatures = FALSE)
      self$remove( keys )
    },



    keys = function(include_signatures = FALSE) {
      not_implemented()

      keys <- private$map$keys()
      if ( include_signatures ) {
        # Returns two columns: key digest

        keys <- t(sapply(keys, function(k) {
          c(k, private$map$get(k)$signature)
        }))
      }

      keys
    },

    size = function() {
      length(self$keys( include_signatures = FALSE ))
    },


    digest = function(signature) {
      digest::digest(signature)
    },



    has = function(keys, signature, sig_encoded = FALSE) {
      stopifnot2(is.character(keys) || is.null(keys), msg = "`keys` must be a character vector or NULL")
      all_keys <- self$keys(include_signatures = TRUE)

      if (!length(all_keys)) { return(rep(FALSE, length(keys))) }

      has_sig <- !missing(signature)

      if ( !sig_encoded && has_sig ) {
        signature <- self$digest(signature)
      }

      vapply(keys, function(k) {
        sel <- all_keys[, 1] == k
        has_key <- any(sel)
        if ( has_sig && has_key ) { has_key <- all_keys[sel, 2] == signature }
        has_key
      }, FUN.VALUE = FALSE)
    },




    `@set` = function(key, value, signature) {
      not_implemented()
      private$map$set(key = key, value = list(
        signature = signature,
        value = value
      ))
      return( signature )
    },
    set = function(key, value, signature) {
      force(value)
      if ( missing(signature) ) {
        signature <- self$digest( value )
      } else {
        signature <- self$digest( signature )
      }
      private$exclusive({
        self$`@set`(key, value, signature = signature)
      })
      invisible(signature)
    },

    mset = function(..., .list = NULL) {
      .list <- c(list(...), .list)
      nms <- names(.list)
      lapply(nms, function(nm) {
        self$set(nm, .list[[nm]])
      })
    },



    `@get` = function(key) {
      not_implemented()
      return( private$map$get(key)$value )
    },
    get = function(key, missing_default) {
      if (missing(missing_default)) { missing_default <- self$missing_default }
      if ( self$has( key ) ) {
        self$`@get`(key)
      } else {
        missing_default
      }
    },

    mget = function(keys, missing_default) {
      if (missing(missing_default)) { missing_default <- self$missing_default }

      has_keys <- self$has( keys )

      re <- lapply(seq_along( keys ), function(ii) {
        if ( has_keys[[ii]] ) {
          self$`@get`(keys[[ ii ]])
        } else {
          missing_default
        }
      })
      names(re) <- keys
      re
    },



    as_list = function(sort = FALSE) {
      keys <- self$keys(include_signatures = FALSE)
      if (!length(keys)) {
        return(list())
      }
      if ( sort ) {
        keys <- sort(keys)
      }

      self$mget(keys)
    },

    `@validate` = function(...) {
      not_implemented()
    },
    validate = function(...) {
      stopifnot2(private$valid, msg = "Map is invalid/destroyed!")
      private$exclusive({
        self$`@validate`(...)
      })
    },

    # Usually should be called at the end of `initialization` to connect to
    # a database, a folder, or an existing queue
    # you should do checks whether the connection is new or it's an existing
    # queue
    `@connect` = function(...) {
      not_implemented()
      private$map <- fastmap::fastmap()
    },
    # thread-safe version. sometimes you need to override this function instead
    # of `@connect`, because `private$exclusive` requires lockfile to be locked
    # If you don't have lockers ready, or need to set lockers during the
    # connection, override this one
    connect = function(...) {
      private$exclusive({
        self$`@connect`(...)
      })
    },

    # will be called during Class$new(...), three tasks,
    # 1. set `get_locker` `free_locker` if lock type is not a file
    # 2. set lockfile (if using default lockers)
    # 3. call self$connect
    initialize = function(has_locker = TRUE, lockfile, ...) {
      if ( has_locker ) {
        self$lockfile <- lockfile
      }
      self$connect(...)
    },

    # destroy a queue, free up space
    # and call `delayedAssign('.lockfile', {stop(...)}, assign.env=private)`
    # to raise error if a destroyed queue is called again later.
    destroy = function() {
      locker_key(self$lockfile, set_default = FALSE, unset = TRUE)
      private$default_free_locker()
      private$valid <- FALSE
      delayedAssign(".lockfile", {
        cat2("Map is destroyed", level = "FATAL")
      }, assign.env = private)
    }
  ),
  active = list(

    # read-only version of self$id. It's safer than private$.id as the latter
    # one does not always exist
    id = function() {
      if (length(private$.id) != 1) {
        private$.id <- rand_string()
      }
      private$.id
    },

    # set/get lock file. Don't call private$.lockfile directly
    lockfile = function(v) {
      if (!missing(v)) {
        private$default_free_locker()
        private$.lockfile <- v
      } else if (!length(private$.lockfile)) {
        private$.lockfile <- rand_string()
      }
      private$.lockfile
    },

    is_valid = function() {
      private$valid
    }

  )
)