Package {liteq}

liteq: Lightweight Portable Message Queue Using 'SQLite'

Description

Temporary and permanent message queues for R. Built on top of 'SQLite' databases. 'SQLite' provides locking, and makes it possible to detect crashed consumers. Crashed jobs can be automatically marked as "failed", or put in the queue again, potentially a limited number of times.

Author(s)

Maintainer: Gábor Csárdi csardi.gabor@gmail.com

Authors:

• Gábor Csárdi csardi.gabor@gmail.com

Other contributors:

• Posit Software, PBC [copyright holder, funder]

See Also

Useful links:

• https://github.com/r-lib/liteq#readme

• Report bugs at https://github.com/r-lib/liteq/issues

Acknowledge that the work on a message has finished successfully

Description

Acknowledge that the work on a message has finished successfully

Usage

ack(message)

Arguments

See Also

liteq for examples

Other liteq messages: consume(), is_empty(), list_failed_messages(), list_messages(), message_count(), publish(), remove_failed_messages(), requeue_failed_messages(), try_consume()

Consume a message from a queue

Description

Blocks and waits for a message if there isn't one to work on currently.

Usage

consume(queue, poll_interval = 500)

Arguments

Value

A message.

See Also

liteq for examples

Other liteq messages: ack(), is_empty(), list_failed_messages(), list_messages(), message_count(), publish(), remove_failed_messages(), requeue_failed_messages(), try_consume()

Create a queue in a database

Description

It also creates the database, if it does not exist.

Usage

create_queue(name = NULL, db = default_db(), crash_strategy = "fail")

Arguments

See Also

liteq for examples

Other liteq queues: delete_queue(), ensure_queue(), list_queues()

Positive or negative ackowledgement

Description

If positive, then we need to remove the message from the queue. If negative, we just set the status to FAILED.

Usage

db_ack(db, queue, id, lock, success)

Arguments

Consume a message from a message queue

Description

This is the blocking version of try_consume(). Currently it just polls twice a second, and sleeps between the polls. Each poll will also trigger a crash cleanup, if there are workers running.

Usage

db_consume(db, queue, poll_interval = 500)

Arguments

Create a queue

Description

The database columns:

• id Id of the message, it is generated automatically by the database.

• title The title of the message, can be empty. In the future, it can be used to filter messages.

• message The message, arbitrary text, can be empty.

• status Can be:

  • READY, ready to be consumed

  • WORKING, it is being consumed

  • FAILED, failed.

• requeued How many times the message was requeued.

Usage

db_create_queue(name, db, crash_strategy)

Arguments

Start a write transaction

Description

We use ⁠BEGIN IMMEDIATE⁠ for every write. A deferred transaction (plain BEGIN) that reads first and writes later might fail with "database is locked" right away, without waiting for the busy timeout, because SQLite cannot wait there without risking a deadlock. Unlike ⁠BEGIN EXCLUSIVE⁠, ⁠BEGIN IMMEDIATE⁠ does not block readers.

Usage

db_lock(con, retry = TRUE)

Arguments

Details

If retry is TRUE, then we keep trying until we get the lock. Each try waits for the busy timeout, so this does not spin.

Check if the worker holding a message lock is still alive

Description

A live worker holds an exclusive lock on the lock file, so reading it fails. We must not wait for the busy timeout here: we hold the write lock of the main DB, and every other client would have to wait, too.

Usage

db_probe_message_lock(lock)

Arguments

Value

"busy" if the worker is alive.

Try to consume a message from the queue

Description

If there is a message that it READY, it returns that. Otherwise it checks for crashed workers.

Usage

db_try_consume(db, queue, crashed = TRUE, con = NULL)

Arguments

Details of the implementation

The database must be locked for the whole operation, including checking on or creating the lock databases.

1. If there is a READY message, that one is taken.

2. Otherwise if there are WORKING messages, then we check them one by one. This might take a lot of time, and the DB must be locked for the whole search, so it is not ideal. But I don't have a better solution right now.

Taking a message means

1. Updating its row.status to WORKING.

2. Creating another database that serves as the lock for this message.

The name of the default database

Description

If the queue database is not specified explicitly, then liteq uses this file. Its location is determined via the rappdirs package, see rappdirs::user_data_dir().

Usage

default_db()

Value

A characater scalar, the name of the default database.

Delete a queue

Description

Delete a queue

Usage

delete_queue(queue, force = FALSE)

Arguments

See Also

liteq for examples

Other liteq queues: create_queue(), ensure_queue(), list_queues()

Ensure that the DB exists and has the right columns

Description

We try a query, and if it fails then we try to create the DB.

Usage

ensure_db(db)

Arguments

Make sure that a queue exists

Description

If it does not exist, then the queue will be created.

Usage

ensure_queue(name, db = default_db(), crash_strategy = "fail")

Arguments

Value

The queue object.

See Also

liteq for examples

Other liteq queues: create_queue(), delete_queue(), list_queues()

Check if a queue is empty

Description

Check if a queue is empty

Usage

is_empty(queue)

Arguments

Value

Logical, whether the queue is empty.

See Also

liteq for examples

Other liteq messages: ack(), consume(), list_failed_messages(), list_messages(), message_count(), publish(), remove_failed_messages(), requeue_failed_messages(), try_consume()

List failed messages in a queue

Description

List failed messages in a queue

Usage

list_failed_messages(queue)

Arguments

Value

Data frame with columns: id, title, status.

See Also

liteq for examples

Other liteq messages: ack(), consume(), is_empty(), list_messages(), message_count(), publish(), remove_failed_messages(), requeue_failed_messages(), try_consume()

List all messages in a queue

Description

List all messages in a queue

Usage

list_messages(queue)

Arguments

Value

Data frame with columns: id, title, status.

See Also

liteq for examples

Other liteq messages: ack(), consume(), is_empty(), list_failed_messages(), message_count(), publish(), remove_failed_messages(), requeue_failed_messages(), try_consume()

List all queues in a database

Description

List all queues in a database

Usage

list_queues(db = default_db())

Arguments

Value

A list of liteq_queue objects.

See Also

liteq for examples

Other liteq queues: create_queue(), delete_queue(), ensure_queue()

Lightweight Portable Message Queue Using 'SQLite'

Description

Message queues for R. Built on top of 'SQLite' databases.

Concurrency

liteq works with multiple producer and/or consumer processes accessing the same queue, via the locking mechanism of 'SQLite'. If a queue is locked by 'SQLite', the process that tries to access it, must wait until it is unlocked. The maximum amount of waiting time is by default 10 seconds, and it can be changed via the R_LITEQ_BUSY_TIMEOUT environment variable, in milliseconds. If you have many concurrent processes using the same liteq database, and see ⁠database locked⁠ errors, then you can try to increase the timeout value.

Examples

# We don't run this, because it writes to the cache directory
db <- tempfile()
q <- ensure_queue("jobs", db = db)
q
list_queues(db)

# Publish two messages
publish(q, title = "First message", message = "Hello world!")
publish(q, title = "Second message", message = "Hello again!")
is_empty(q)
message_count(q)
list_messages(q)

# Consume one
msg <- try_consume(q)
msg

ack(msg)
list_messages(q)
msg2 <- try_consume(q)
nack(msg2)
list_messages(q)

# No more messages
is_empty(q)
try_consume(q)

Author(s)

Maintainer: Gábor Csárdi csardi.gabor@gmail.com

Authors:

• Gábor Csárdi csardi.gabor@gmail.com

Other contributors:

• Posit Software, PBC [copyright holder, funder]

See Also

Useful links:

• https://github.com/r-lib/liteq#readme

• Report bugs at https://github.com/r-lib/liteq/issues

Examples

## See the manual page

Make a message object

Description

It creates the lock for the message as well.

Usage

make_message(id, title, message, db, queue, lockdir)

Arguments

Details

The message object contains the connection to the message lock. If the worker crashes, then there will be no reference to the connection, and the lock will be released. This is how we detect crashed workers.

Value

message object

Get the number of messages in a queue.

Description

Get the number of messages in a queue.

Usage

message_count(queue)

Arguments

Value

Number of messages in the queue.

See Also

liteq for examples

Other liteq messages: ack(), consume(), is_empty(), list_failed_messages(), list_messages(), publish(), remove_failed_messages(), requeue_failed_messages(), try_consume()

Report that the work on a message has failed

Description

Report that the work on a message has failed

Usage

nack(message)

Arguments

See Also

liteq for examples

Publish messages in a queue

Description

Publish messages in a queue

Usage

publish(queue, title = "", message = "")

Arguments

See Also

liteq for examples

Other liteq messages: ack(), consume(), is_empty(), list_failed_messages(), list_messages(), message_count(), remove_failed_messages(), requeue_failed_messages(), try_consume()

Remove failed messages from the queue

Description

Remove failed messages from the queue

Usage

remove_failed_messages(queue, id = NULL)

Arguments

See Also

liteq for examples

Other liteq messages: ack(), consume(), is_empty(), list_failed_messages(), list_messages(), message_count(), publish(), requeue_failed_messages(), try_consume()

Requeue failed messages

Description

Requeue failed messages

Usage

requeue_failed_messages(queue, id = NULL)

Arguments

See Also

liteq for examples

Other liteq messages: ack(), consume(), is_empty(), list_failed_messages(), list_messages(), message_count(), publish(), remove_failed_messages(), try_consume()

Consume a message if there is one available

Description

Consume a message if there is one available

Usage

try_consume(queue)

Arguments

Value

A message, or NULL if there is not message to work on.

See Also

liteq for examples

Other liteq messages: ack(), consume(), is_empty(), list_failed_messages(), list_messages(), message_count(), publish(), remove_failed_messages(), requeue_failed_messages()