Class CudaMessageQueue<T>

Namespace

DotCompute.Backends.CUDA.RingKernels

Assembly

DotCompute.Backends.CUDA.dll

CUDA-based lock-free message queue for inter-kernel communication.

public sealed class CudaMessageQueue<T> : IMessageQueue<T>, IMessageQueue, IAsyncDisposable where T : unmanaged

Type Parameters

T

Message payload type (must be unmanaged).

Inheritance

object

CudaMessageQueue<T>

Implements

IMessageQueue<T>

IMessageQueue

IAsyncDisposable

Inherited Members

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.ReferenceEquals(object, object)

object.ToString()

Remarks

Implements a GPU-resident lock-free ring buffer using CUDA atomics. The queue is optimized for high throughput concurrent access from multiple GPU threads.

Constructors

CudaMessageQueue(int, ILogger<CudaMessageQueue<T>>)

Initializes a new instance of the CudaMessageQueue<T> class.

public CudaMessageQueue(int capacity, ILogger<CudaMessageQueue<T>> logger)

Parameters

capacity int

Queue capacity (must be power of 2).

logger ILogger<CudaMessageQueue<T>>

Logger instance.

Properties

Capacity

Gets the queue capacity (maximum number of messages).

public int Capacity { get; }

Property Value

int

Count

Gets the approximate current size of the queue.

public int Count { get; }

Property Value

int

IsEmpty

Gets whether the queue is empty.

public bool IsEmpty { get; }

Property Value

bool

IsFull

Gets whether the queue is full.

public bool IsFull { get; }

Property Value

bool

Methods

ClearAsync(CancellationToken)

Clears all messages from the queue.

public Task ClearAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken

Cancellation token.

Returns

Task

A task representing the clear operation.

DequeueAsync(TimeSpan, CancellationToken)

Dequeues a message, blocking until one is available.

public Task<KernelMessage<T>> DequeueAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default)

Parameters

timeout TimeSpan

Maximum wait time (default: infinite).

cancellationToken CancellationToken

Cancellation token.

Returns

Task<KernelMessage<T>>

The dequeued message.

Exceptions

TimeoutException

Thrown if the timeout expires before a message arrives.

DisposeAsync()

Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources asynchronously.

public ValueTask DisposeAsync()

Returns

ValueTask

A task that represents the asynchronous dispose operation.

EnqueueAsync(KernelMessage<T>, TimeSpan, CancellationToken)

Enqueues a message, blocking until space is available.

public Task EnqueueAsync(KernelMessage<T> message, TimeSpan timeout = default, CancellationToken cancellationToken = default)

Parameters

message KernelMessage<T>

The message to enqueue.

timeout TimeSpan

Maximum wait time (default: infinite).

cancellationToken CancellationToken

Cancellation token.

Returns

Task

A task that completes when the message is enqueued.

Exceptions

TimeoutException

Thrown if the timeout expires before space becomes available.

GetBuffer()

Gets the memory buffer handle for GPU access.

public IUnifiedMemoryBuffer GetBuffer()

Returns

IUnifiedMemoryBuffer

GetHeadPtr()

Gets the head pointer buffer (atomic counter for dequeue position).

public IUnifiedMemoryBuffer GetHeadPtr()

Returns

IUnifiedMemoryBuffer

GetStatisticsAsync()

Gets queue statistics for monitoring.

public Task<MessageQueueStatistics> GetStatisticsAsync()

Returns

Task<MessageQueueStatistics>

Queue statistics including message counts and throughput.

GetTailPtr()

Gets the tail pointer buffer (atomic counter for enqueue position).

public IUnifiedMemoryBuffer GetTailPtr()

Returns

IUnifiedMemoryBuffer

InitializeAsync(CancellationToken)

Initializes the message queue and allocates GPU buffers.

public Task InitializeAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken

Cancellation token.

Returns

Task

A task representing the initialization operation.

TryDequeueAsync(CancellationToken)

Attempts to dequeue a message (non-blocking).

public Task<KernelMessage<T>?> TryDequeueAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken

Cancellation token.

Returns

Task<KernelMessage<T>?>

The dequeued message if available; null if the queue was empty.

Remarks

This is a non-blocking operation that returns null if the queue is empty. Use BlockingDequeue if you need to wait for messages.

TryEnqueueAsync(KernelMessage<T>, CancellationToken)

Attempts to enqueue a message (non-blocking).

public Task<bool> TryEnqueueAsync(KernelMessage<T> message, CancellationToken cancellationToken = default)

Parameters

message KernelMessage<T>

The message to enqueue.

cancellationToken CancellationToken

Cancellation token.

Returns

Task<bool>

True if the message was enqueued; false if the queue was full.

Remarks

This is a non-blocking operation that returns false if the queue is full. Use BlockingEnqueue if you need guaranteed delivery with backpressure.