make id active binding
make id active binding

mschubert/clustermq,mschubert/clustermq,mschubert/clustermq
#' Class for basic queuing system functions
#'
#' Provides the basic functions needed to communicate between machines
#' This should abstract most functions of rZMQ so the scheduler
#' implementations can rely on the higher level functionality
QSys = R6::R6Class("QSys",
    public = list(
        initialize = function() {
            private$job_num = 1
            private$zmq_context = rzmq::init.context()
        },

        # Submits one job to the queuing system
        #
        # @param memory      The amount of memory (megabytes) to request
        # @param log_worker  Create a log file for each worker
        submit_job = function(memory=NULL, log_worker=FALSE) {
            stop("Derived class needs to overwrite submit_job()")
        },

        # Send the data common to all workers, only serialize once
        send_common_data = function() {
            if (is.null(private$common_data))
                stop("Need to set_common_data() first")

            rzmq::send.socket(socket = private$socket,
                              data = private$common_data,
                              serialize = FALSE,
                              send.more = TRUE)
        },

        # Send iterated data to one worker
        send_job_data = function(...) {
            rzmq::send.socket(socket = private$socket, data = list(...))
        },

        # Read data from the socket
        receive_data = function() {
            rzmq::receive.socket(private$socket)
        },

        # Make sure all resources are closed properly
        cleanup = function() {
        }
    ),

    active = list(
        # We use the listening port as scheduler ID
        id = function() private$port
    ),

    private = list(
        zmq_context = NULL,
        socket = NULL,
        port = NA,
        master = NULL,
        job_num = NULL,
        common_data = NULL,

        set_common_data = function(fun, const, seed) {
            private$common_data = serialize(list(fun=fun, const=const, seed=seed), NULL)
        },

        # Create a socket and listen on a port in range
        #
        # @param fun    The function to be called
        # @param const  Constant arguments to the function call
        # @param seed   Common seed (to be used w/ job ID)
        # @return       Sets "port" and "master" attributes
        listen_socket = function(min_port, max_port=min_port, n_tries=100) {
            if (is.null(private$zmq_context))
                stop("QSys base class not initialized")

            private$socket = rzmq::init.socket(private$zmq_context, "ZMQ_REP")

            on.exit(sink())
            sink('/dev/null')
            for (i in 1:n_tries) {
                exec_socket = sample(min_port:max_port, size=1)
                addr = paste0("tcp://*:", exec_socket)
                port_found = rzmq::bind.socket(private$socket, addr)
                if (port_found)
                    break
            }
            sink()
            on.exit()

            if (!port_found)
                stop("Could not bind to port range (6000,8000) after 100 tries")

            private$port = exec_socket
            private$master = sprintf("tcp://%s:%i", Sys.info()[['nodename']], exec_socket)
        }
    ),

    cloneable = FALSE
)

#' Class for basic queuing system functions
#'
#' Provides the basic functions needed to communicate between machines
#' This should abstract most functions of rZMQ so the scheduler
#' implementations can rely on the higher level functionality
QSys = R6::R6Class("QSys",
    public = list(
        id = NA,

        initialize = function() {
            private$job_num = 1
            private$zmq_context = rzmq::init.context()
        },

        # Submits one job to the queuing system
        #
        # @param memory      The amount of memory (megabytes) to request
        # @param log_worker  Create a log file for each worker
        submit_job = function(memory=NULL, log_worker=FALSE) {
            stop("Derived class needs to overwrite submit_job()")
        },

        # Send the data common to all workers, only serialize once
        send_common_data = function() {
            if (is.null(private$common_data))
                stop("Need to set_common_data() first")

            rzmq::send.socket(socket = private$socket,
                              data = private$common_data,
                              serialize = FALSE,
                              send.more = TRUE)
        },

        # Send iterated data to one worker
        send_job_data = function(...) {
            rzmq::send.socket(socket = private$socket, data = list(...))
        },

        # Read data from the socket
        receive_data = function() {
            rzmq::receive.socket(private$socket)
        },

        # Make sure all resources are closed properly
        cleanup = function() {
        }
    ),

    private = list(
        zmq_context = NULL,
        socket = NULL,
        port = NULL,
        master = NULL,
        job_num = NULL,
        common_data = NULL,

        set_common_data = function(fun, const, seed) {
            private$common_data = serialize(list(fun=fun, const=const, seed=seed), NULL)
        },

        # Create a socket and listen on a port in range
        #
        # @param fun    The function to be called
        # @param const  Constant arguments to the function call
        # @param seed   Common seed (to be used w/ job ID)
        # @return       Sets "port" and "master" attributes
        listen_socket = function(min_port, max_port=min_port, n_tries=100) {
            if (is.null(private$zmq_context))
                stop("QSys base class not initialized")

            private$socket = rzmq::init.socket(private$zmq_context, "ZMQ_REP")

            on.exit(sink())
            sink('/dev/null')
            for (i in 1:n_tries) {
                exec_socket = sample(min_port:max_port, size=1)
                addr = paste0("tcp://*:", exec_socket)
                port_found = rzmq::bind.socket(private$socket, addr)
                if (port_found)
                    break
            }
            sink()
            on.exit()

            if (!port_found)
                stop("Could not bind to port range (6000,8000) after 100 tries")

            self$id = exec_socket
            private$port = exec_socket
            private$master = sprintf("tcp://%s:%i", Sys.info()[['nodename']], exec_socket)
        }
    ),

    cloneable = FALSE
)
