Messages

RAJA::messages provides a portable interface and type-safe way to store messages that can be handled at a later time. In this context, a message will store function arguments and handled typically means passed to a function. For example, one use case could be logging error messages from the GPU to file. In this case, arguments from the GPU can be stored and passed to function that prints to a file on the CPU.

Warning

This capability is new. We would like users to try it out and give feedback to improve it.

How to manage messages?

All messages are handled via the message_manager, which is responsible for storing callbacks and a list of messages. For the purposes of RAJA::messages, a single message can be thought of:

  • MsgHeader: helper data internal to RAJA

  • MsgArgs: a tuple of arguments needed to pass to the function

To create the message_manager:

  auto host_allocator = RAJA::ResourceAllocator<char, decltype(res_host)>{res_host,
    RAJA::resources::MemoryAccess::Pinned};
  auto msg_manager    = RAJA::make_message_manager(buf_sz, res_host, host_allocator);

buf_sz is the size of the buffer that stores messages. res_host is the resource of the execution policy, which determines the memory space in which messages will be stored. For GPU resources, for example, this is PINNED memory.

Subscribing callbacks

To create a specific message type, callbacks must subscribe first. This can be done in two ways:

  • subscribe(Callable): Subscribing with just a callable will create a new type of message with the type depending on the parameters.

  • subscribe(msg_queue_id, Callable): Subscribing with msg_queue_id and a callable will append the new callback to the already existing callback list.

As an example for subscribing with both methods:

  auto cpu_msg_queue = msg_manager.subscribe<RAJA::mpsc_queue>(
    [](const my_string<128>& str, int* ptr, int idx, int value) {
      std::cout << "\n " << str.c_str() << " " << ptr << " a[" << idx << "] = " << value << "\n";
    }
  );

  auto err_callback = [] (const my_string<128>& str, int* ptr, int idx, int value) {
      std::cerr << "\n " << str.c_str() << " " << ptr << " a[" << idx << "] = " << value << "\n";
    };
  msg_manager.subscribe(cpu_msg_queue.get_id(), err_callback);

Unsubscribing callbacks

If a particular callback no longer needs to be subscribed to a message type, then the callback can be unsubcribed. This can be achieved in three ways:

  • unsubscribe(msg_queue_id, Callable): Looks for a specific callback that is subscribed to a particular message. If the callback is subscribed, remove from callback list. Otherwise, throws an exception.

  • unsubscribe_all(msg_queue_id): Removes all callbacks subscribed to a particular message

  • unsubscribe_all(): Removes all callbacks from map.

  • erase_all(msg_queue_id): Erases all callbacks and the map entry.

An example for unsubscribing a callback:

  msg_manager.unsubscribe(cpu_msg_queue.get_id(), err_callback);

Publishing messages

Messages can be published/stored in a MessageQueue. These are non-owning adapters to the MessageBus, which is responsible for storing all messages. The MessageQueue will contain additional type information as well as the msg_queue_id. A queue is created once a callback is subscribed to a new message type. Since MessageQueue is non-owning, these can be copied.

Here is how the MessageQueue can be used to publish messages:

  RAJA::forall<RAJA::seq_exec>(res_host, RAJA::RangeSegment(0, N), [=] (int i) { 
    if (a[i] < 0) { 
      cpu_msg_queue.try_post_message("message from RAJA seq_exec loop", a, i, a[i]); 
    }
    c[i] = a[i] + b[i]; 
  });

Handling messages

Lastly, there needs to be a way to direct messages to the corresponding callback(s). This is handled with the message_manager, which forces a synchronize on the resource provided.

  msg_manager.wait_all();

Handling messages with a GPU

Here is a complete example using the RAJA::messages with a GPU kernel.

  auto gpu_allocator = RAJA::ResourceAllocator<char, decltype(res)>{res,
    RAJA::resources::MemoryAccess::Pinned};
  auto msg_manager   = RAJA::make_message_manager<gpu_policy>(message_sz*10, gpu_allocator);

  auto log = [](const my_string<32>& str, int idx, int value) {
    std::cout << "[INFO]: " << str.c_str() << "[" << idx << "] = " << value << "\n";
  };

  // Create two types of messages:
  // queue1 stores one message type and prints with one callback
  // queue2 stores the other types of messages and forwards the message to multiple callbacks
  auto msg_queue1 = msg_manager.subscribe<RAJA::mpsc_queue>(log);
  msg_manager.subscribe(msg_queue1.get_id(),
    [](const my_string<32>& str, int idx, int value) {
      std::cout << "echo msg: " << str.c_str() << "[" << idx << "] = " << value << "\n";
    }
  );
  auto msg_queue2 = msg_manager.subscribe<RAJA::mpsc_queue>(log);

  RAJA::forall<gpu_policy>(RAJA::RangeSegment(0, N), 
    [=] RAJA_DEVICE (int i) { 
    if (d_a1[i] < 0 && i == 1) { 
      msg_queue1.try_post_message("d_a1", i, d_a1[i]); 
    }
    if (d_b1[i] > 0 && i == 1) { 
      msg_queue2.try_post_message("d_b1", i, d_b1[i]); 
    }
    d_c1[i] = d_a1[i] + d_b1[i]; 
  });    
  msg_manager.wait_all();

Note

In this example, gpu_policy depends on the build (i.e., CUDA, HIP), res is the default resource for gpu_policy, and d_* arrays are allocated for the device. These are removed above from the example to just show the RAJA::messages interface.

Handling messages across multiple streams

Here is a complete example using the RAJA::messages with multiple resources.

  auto allocator1     = RAJA::ResourceAllocator<char, decltype(res_gpu1)>{res_gpu1,
    RAJA::resources::MemoryAccess::Pinned};
  auto gpu_logger1    = RAJA::make_message_manager(buf_sz, res_gpu1, allocator1); 
  auto gpu_msg_queue1 = gpu_logger1.subscribe<RAJA::mpsc_queue>(
    [](int* ptr, int idx, int value) {
      std::cout << "\n gpu stream 1: pointer (" << ptr << ") d_array1[" << idx << "] = " << value << "\n";
    }
  );

  auto allocator2     = RAJA::ResourceAllocator<char, decltype(res_gpu2)>{res_gpu2,
    RAJA::resources::MemoryAccess::Pinned};
  auto gpu_logger2    = RAJA::make_message_manager(buf_sz, res_gpu2, allocator2);
  auto gpu_msg_queue2 = gpu_logger2.subscribe<RAJA::mpsc_queue>(
    [](int* ptr, int idx, int value) {
      std::cout << "\n gpu stream 2: pointer (" << ptr << ") d_array2[" << idx << "] = " << value << "\n";
    }
  );

  RAJA::forall<EXEC_POLICY>(res_gpu1, RAJA::RangeSegment(0,N),
    [=] RAJA_HOST_DEVICE (int i) {
      d_array1[i] = i;
      gpu_msg_queue1.try_post_message(d_array1, i, d_array1[i]);
    }
  );
   
  // Log message for stream 1 
  gpu_logger1.wait_all();   

  RAJA::forall<EXEC_POLICY>(res_gpu2, RAJA::RangeSegment(0,N),
    [=] RAJA_HOST_DEVICE (int i) {
      d_array2[i] = -1;
      gpu_msg_queue2.try_post_message(d_array2, i, d_array2[i]);
    }
  );

  RAJA::forall<EXEC_POLICY>(res_gpu1, RAJA::RangeSegment(0,N),
    [=] RAJA_HOST_DEVICE (int i) {
      d_array1[i] *= -1;
      gpu_msg_queue1.try_post_message(d_array1, i, d_array1[i]);
    }
  );

  // Log message for stream 2
  gpu_logger2.wait_all();   

  // Log message for stream 1 
  gpu_logger1.wait_all();

Note

In this example, res_gpu1 and res_gpu2 depend on the build (i.e., CUDA, HIP), EXEC_POLICY the exeuction policy for the loops (also depends on the build), and d_* arrays are allocated for the device while h_* are allocated for the host. These are removed from the example above to just show the RAJA::messages interface.

Message queue policies

Message queues can support various policies depending on the requirements of that queue, such as the number of producers/consumers or the type of atomic operations.

Message queue Policies

Brief description

spsc

Supports a single producer, single consumer; i.e., no atomic operations

mpsc

Supports multiple producers, a single consumer; i.e., requires atomic operations. Automatically determines which atomic operations to use.

Note

Producers and consumers can not operate at the same time

Building and running the example

The example examples/messages-forall.cpp is built from the RAJA source tree when:

  • ENABLE_EXAMPLES=On

To run the example:

./bin/messages-forall

This example will show how callbacks can be subscribed to various types of messages as well as how to publish messages on multiple different platforms. For the purposes of this example, output for messages will be printed using std::cout.

Application considerations

There are several things to consider when using RAJA::messages in an application.

  • The MessageQueue with the correct argument types is created when a callback subscribes. Certain patterns will cause this storage to slowly grow overtime. For example, creating a new MessageQueue every function call within a loop. Therefore, applications that use this pattern will want to unsubscribe at some point to avoid running out of memory.

  • Upon creation of the MessageManager, the MessageBus will be allocated with some size. This can be resized; however, resizing will force a synchronize and will loss any messages currently stored. Also, by default, the allocation is done through the resource, which can be less performant depending on the resource.

  • Since the MessageQueue is a fixed size, there is a chance of lossing messages. The try_post_message function will return a boolean. This will be true if the message is successfully added to the queue; otherwise, this is false.

Allocators

During the creation of the MessageManager, a custom allocator needs to be provided. This allocator should follow the C++ standard requirements for an allocator and will be used to allocate/deallocate the message bus. The message bus will always read messages on the host to forward the message to the respect callback(s). However, depending on the build and where messages are stored, messages may be written on either the host or the device. Therefore, in cases where messages are stored on the device, the allocator needs to allocate memory that is accessible on both the host and device.

Note

When using the move assignment operator, the MessageManager assumes that there are no currently pending messages.