Shared Memory Atomic Operations

Implements named semaphores from the 'boost' 'C++' library < https://www.boost.org/> for interprocess communication. Multiple 'R' sessions on the same host can block (with optional timeout) on a semaphore until it becomes positive, then atomically decrement it and unblock. Any session can increment the semaphore.

semaphore

The goal of semaphore is to enable synchronization of concurrent R processes.

Implements named semaphores from the Boost C++ library for interprocess communication. Multiple R sessions on the same host can block (with optional timeout) on a semaphore until it becomes positive, then atomically decrement it and unblock. Any session can increment the semaphore.

Semaphores are managed by the operating system, and can be used to synchronize a mixture of R sessions and other types of processes if needed.

Works cross-platform, including Windows, MacOS, and Linux.

Installation

# Install the latest stable version from CRAN:
install.packages("semaphore")

# Or the development version from GitHub:
install.packages("pak")
pak::pak("cmmr/semaphore")

Usage

library(semaphore)

s <- create_semaphore()
print(s)
#> [1] "uUkKpNMbTVgaborHG4rH"

increment_semaphore(s)

decrement_semaphore(s, wait = 10)      # wait up to ten seconds
#> [1] TRUE
decrement_semaphore(s, wait = FALSE)   # return immediately
#> [1] FALSE

remove_semaphore(s)
#> [1] TRUE

Example: Producer/Consumer

Open two separate R sessions on the same machine.

Session 1 - Producer

library(semaphore)
s <- 'mySemaphore'

create_semaphore(s)

# enable session 2 to output 'unblocked!' three times
increment_semaphore(s)
increment_semaphore(s)
increment_semaphore(s)

remove_semaphore(s)

Session 2 - Consumer

library(semaphore)
s <- 'mySemaphore'

for (i in 1:3) {

  # Block until session 1 increments the semaphore
  decrement_semaphore(s, wait = TRUE)
  
  # Do some work
  message('unblocked!')
}

message('finished')