Получить PID для подпроцессов для асинхронного фьючерса в R блестящий

Question

server <- function(input, output, session) {
  out1_rows <- reactiveVal()

   observeEvent(input$run1, {
   prog <- Progress$new(session)
   prog$set(message = "Analysis in progress",
         detail = "This may take a while...",
         value = NULL)

  fut1 = future({
  system(paste("Command1" , input$file ">", "out1.txt"))

  system(paste("Command2" , out1.txt ">", "out2.txt"))
  head_rows <- read.delim("out2.txt")
    return(head_rows)
    }) %...>%
     out1_rows() %>%
  finally( ~ prog$close())
NULL
})


 observeEvent(req(out1_rows()), {
 output$out_table <-
  DT::renderDataTable(DT::datatable(
    out1_rows(),
    )
  ))

observeEvent(input$cancel, {
    async_pid <- fut1$job$pid  ##this is empty
    #async_pid <- Sys.getpid()  ##this return PID for main process and kills "/opt/shiny-server/R/SockJSAdapter.R"  but not for subprocesses inside future()
    system(paste("kill -15", async_pid))
  })
}

Здесь мне нужно будет убить процесс, выполняющий команды внутри future ().Я попытался описанным выше способом получить PID, выполняющий процесс future (), и завершить работу при срабатывании input$cancel.Однако fut1$job$pid не возвращает никакого значения PID, и, следовательно, операция уничтожения не выполнена.

Эта ссылка из будущих виньеток показывает, как получить PID для будущих () заданий.Однако в моем случае я не могу использовать Sys.getpid() внутри future (), поскольку я не уверен, как сохранить значение PID, поскольку процесс уже возвращает некоторые выходные данные из моих системных команд.

Эта страница future GIT показывает альтернативный способ External Kill с синтаксисом fut1$job$pid.Но это не в состоянии получить PID.

Я не мог понять это после попытки различных способов или ослеплен синтаксисом.Может ли кто-нибудь намекнуть, как это сделать.

Answer 1

Не могли бы вы предоставить нам полный воспроизводимый пример?

Возможно, вы захотите взглянуть на библиотеку ( future.callr ):

, используя plan(callr) вы можете получить pid и завершить процесс следующим образом:

library(future)
library(promises)
library(future.callr)

plan(callr)

myFuture <- future({
  Sys.sleep(5)
  return(runif(1))
})

myFuture$process$get_pid()
myFuture$process$is_alive()

# myFuture$process$kill()
# myFuture$process$is_alive()

then(myFuture, onFulfilled = function(value){
print(value)
}, onRejected = NULL)

Редактировать - адаптировано из вашего кода:

library(shiny)
library(DT)
library(future)
library(promises)
library(future.callr)
library(shinyjs)
library(V8)

plan(callr)

ui <- fluidPage(
  useShinyjs(),
  titlePanel("Trigger & kill future"),
  sidebarLayout(
    sidebarPanel(
      actionButton(inputId="run1", label="run future"),
      actionButton(inputId="cancel", label="kill future")
    ),
    mainPanel(
      dataTableOutput('out_table')
    )
  )
)

server <- function(input, output, session) {

  disable("cancel") 
  out1 <- reactiveValues(rows=NULL)

  observeEvent(input$run1, {

    disable("run1")
    enable("cancel")
    out1$rows <- NULL

    prog <- Progress$new(session)
    prog$set(message = "Analysis in progress",
             detail = "This may take a while...",
             value = NULL)

    fut1 <<- future({
      # system(paste("Command1" , input$file, ">", "out1.txt"))
      # system(paste("Command2" , out1.txt, ">", "out2.txt"))
      # head_rows <- read.delim("out2.txt")
      head_rows <- data.frame(replicate(5, sample(runif(20, 0, 1), 20, rep=TRUE)))
      Sys.sleep(5)
      return(head_rows)
    })

    print(paste("Running async process with PID:", fut1$process$get_pid()))

    then(fut1, onFulfilled = function(value){
      out1$rows <<- value
    }, onRejected = function(error){NULL})

    finally(fut1, function(){
      prog$close()
      disable("cancel")
      enable("run1")
    })

    return(NULL)
  }, ignoreInit = TRUE)


  observeEvent(req(out1$rows), {
    output$out_table <- DT::renderDataTable(DT::datatable(out1$rows))
  })

  observeEvent(input$cancel, {
    async_pid <- fut1$process$get_pid()
    print(paste("Killing PID:", async_pid))
    # system(paste("kill -9", async_pid)) # Linux - kill
    # system(paste("taskkill /f /pid", async_pid)) # Windows - kill
    fut1$process$kill() # library(future.callr) - kill
    out1$rows <- NULL
    disable("cancel")
    enable("run1")
  }, ignoreInit = TRUE)

}

shinyApp(ui = ui, server = server)

Время от времени возникают ошибки:

Unhandled promise error: callr failed, could not start R, exited with non-zero status, has crashed or was killed 
Warning: Error in : callr failed, could not start R, exited with non-zero status, has crashed or was killed 
  95: <Anonymous>

Может быть, это заявление от @HenrikB говорит нам о том, с чем мы сталкиваемся:

Однако для правильной работы вам, вероятно, также необходимобудущие выражения / будущие кодовые прерывания с помощью withCallingHandlers () и т. д. Мне также неизвестно, что произойдет, если вы подадите слишком много прерываний подряд - возможно, вам удастся прервать основной R-цикл рабочего, чтозатем заставит R-работника завершиться.Это приведет к отсутствию R-работника, и у вас возникнет проблема, о которой вы упомянули в начале.

Ошибка также упоминается здесь , но в настоящее время в будущем.-контекст Я не знаю, как обойти это.

2-е редактирование: К настоящему времени я получил еще один отзыв от Хенрика Бенгтссона.Он еще раз упоминает, что основное будущее API в настоящее время не поддерживает прекращение фьючерсов .Таким образом, в конце концов, независимо от того, какой бэкэнд мы используем, у нас могут возникнуть проблемы.

Не обращая внимания на эту информацию, я бы еще раз взглянул на library(ipc) виньетка , в которой приведены два примера.по теме убить долго работающий процесс .Но я уже указал на это здесь - что, вероятно, привело к этому вопросу .

В конце концов, все это может оказаться бесполезным в отношении вашего сценария, поскольку вы используете system() вызовов, которые создают свои собственные подпроцессы (и соответственно имеют свой собственный pid).Поэтому, почему бы вам не использовать wait = FALSE в вашей системной команде (как я уже упоминал в комментариях здесь уже), чтобы получить асинхронное поведение и поймать их pid, используя что-то вроде myCommand & echo $! (см. это ).Таким образом, вам не нужно никакого будущего.