Example for the Distributed Cache

[Previous] [Next] [Up]

This example demonstrates the distributed cache.

New Topics

• Implementation of serializables
• Implementation of object receivers
• Using the distributed cache

Detailed Description

Implementation of serializables

The objects to be handled by the distributed cache have to implement the mi::neuraylib::ISerializable interface. In this example we introduce the derived interface IMy_serializable which offers methods to set and to get a single mi::Uint64 value. To demonstrate the ability of the distributed cache to handle different classes two implementations of this interface called My_serializable_1 and My_serializable_2 are given.

The most important part of these implementations are the serialize() and deserialize() methods. A correct implementation of this methods is mandatory for proper functionality of the distributed cache.

Implementation of object receivers

The object receiver is responsible for creation and destruction of objects on the remote host. It is also responsible for communicating these events to the user application. In this example the printed messages take the role of such communication.

Note that this implementation of the object receiver stores a pointer to the created object in its internal map such that it can release the object later in the destroy_object() call.

Using the distributed cache

The example demonstrates a simple usage scenario for the distributed cache: hosts are either clients or workers. Clients create objects which are distributed to the workers. Later, these objects are destroyed again. Workers are passive and create and destroy objects as notified by the distributed cache.

Note that the distinction between client and worker in this examples has been made for the sake of simplicity. Any host can act as client and worker simultaneously.

To use a class with the distributed cache it is necessary to register it via mi::neuraylib::IDice_configuration::register_serializable_class(). Access to the distributed cache is obtained from mi::neuraylib::INeuray::get_api_component().

Clients can use the distributed cache without further setup steps. The method mi::neuraylib::IDistributed_cache::store_object() is used to store created objects in the cache. Likewise, the method mi::neuraylib::IDistributed_cache::remove_object() informs workers about the removal of an object.

Workers simply register object receivers for each class class ID to be handled. Note that the same object receiver instance can be used for several class IDs. Different instances are used in this example just for the illustration purposes.

Example Source

Source Code Location: examples/example_distributed_cache.h

/******************************************************************************
 * Copyright 2023 NVIDIA Corporation. All rights reserved.
 *****************************************************************************/
 
// examples/example_distributed_cache.cpp
//
// User-defined elements for the distributed cache shared by several examples.
 
#ifndef EXAMPLE_DISTRIBUTED_CACHE_H
#define EXAMPLE_DISTRIBUTED_CACHE_H
 
#include <iostream>
#include <map>
 
#include <mi/dice.h>
 
// Include code shared by all examples.
#include "example_shared.h"
 
extern mi::base::Handle<mi::base::ILogger> g_logger;
 
// The base interface for two serializable classes. The examples use several serializable classes
// to demonstrate the registration of several receivers (at most one per class ID).
class IMy_serializable : public
    mi::base::Interface_declare<0x693c4b52,0x593d,0x407b,0xab,0x2a,0x5a,0x82,0x1c,0x8c,0xd1,0xc5,
                                mi::neuraylib::ISerializable>
{
public:
    // Sets some data
    virtual void set_data( mi::Uint64 data) = 0;
    // Returns the data
    virtual mi::Uint64 get_data() const = 0;
};
 
// The first serializable class.
class My_serializable_1
  : public mi::neuraylib::Base<0xa07af7f8,0x9002,0x42fa,0x84,0x15,0x11,0x2a,0xae,0x10,0xca,0x3f,
                               IMy_serializable>
{
public:
    void destroy() { delete this; }
    mi::base::Uuid get_class_id() const { return IID(); }
    void serialize( mi::neuraylib::ISerializer* serializer) const { serializer->write( &m_data); }
    void deserialize( mi::neuraylib::IDeserializer* deserializer) { deserializer->read( &m_data); }
    void set_data( mi::Uint64 data) { m_data = data; }
    mi::Uint64 get_data() const { return m_data; }
 
private:
    mi::Uint64 m_data;
};
 
// The second serializable class.
class My_serializable_2
  : public mi::neuraylib::Base<0xe7ce6357,0xd8bb,0x4dfb,0xab,0xce,0x37,0x76,0x1c,0x80,0xba,0x70,
                               IMy_serializable>
{
public:
    void destroy() { delete this; }
    mi::base::Uuid get_class_id() const { return IID(); }
    void serialize( mi::neuraylib::ISerializer* serializer) const { serializer->write( &m_data); }
    void deserialize( mi::neuraylib::IDeserializer* deserializer) { deserializer->read( &m_data); }
    void set_data( mi::Uint64 data) { m_data = data; }
    mi::Uint64 get_data() const { return m_data; }
 
private:
    mi::Uint64 m_data;
};
 
// A simple object receiver implementation which can be used for different (fixed) classes.
class Object_receiver : public mi::base::Interface_implement<mi::neuraylib::IObject_receiver>
{
public:
    Object_receiver( mi::Uint64 receiver_id) : m_next_id( 0), m_receiver_id( receiver_id) { }
 
    mi::Uint64 create_object( mi::base::Uuid class_id, mi::neuraylib::IDeserializer* deserializer)
    {
        IMy_serializable* object;
        if( class_id == My_serializable_1::IID())
             object = new My_serializable_1();
        else if( class_id == My_serializable_2::IID())
            object = new My_serializable_2();
        else {
            check_success( false); // line should never be reached
            return 0;
        }
        object->deserialize( deserializer);
        mi::Uint64 id = m_next_id++;
        m_objects[id] = object;
        g_logger->printf( mi::base::MESSAGE_SEVERITY_INFO, "DICE:MAIN",
            "Receiver with ID %llu created object with ID %llu at address %p with data %llu.",
            m_receiver_id, id, object, object->get_data());
        return id;
    }
 
    void destroy_object( mi::Uint64 id, bool owner_died)
    {
        IMy_serializable* object = m_objects[id];
        m_objects.erase( id);
        g_logger->printf( mi::base::MESSAGE_SEVERITY_INFO, "DICE:MAIN",
            "Receiver with ID %llu destroyed object with ID %llu at address %p with data %llu "
            "(owner died: %s)",
            m_receiver_id, id, object, object->get_data(), owner_died ? "yes" : "no");
        object->release();
    }
 
private:
    std::map<mi::Uint64, IMy_serializable*> m_objects;
    mi::Uint64 m_next_id;
    mi::Uint64 m_receiver_id;
};
 
#endif // EXAMPLE_DISTRIBUTED_CACHE_H

Source Code Location: examples/example_distributed_cache.cpp

/******************************************************************************
 * Copyright 2023 NVIDIA Corporation. All rights reserved.
 *****************************************************************************/
 
// examples/example_distributed_cache.cpp
//
// Demonstrates usage of the distributed cache
//
// The example expects the following command line arguments:
//
//   example_distributed_cache <mode> <multicast_address> <N>
//
// mode                 either "client" or "worker"
// multicast_address    UDP multicast address
// N                    number of objects being created on each client and distributed to each
//                      worker via the distributed cache
 
#include <iostream>
#include <map>
 
#include <mi/dice.h>
 
// Include code shared by all examples.
#include "example_shared.h"
// Include definition of My_serializable_1, My_serializable_2, and Object_receiver.
#include "example_distributed_cache.h"
 
mi::base::Handle<mi::base::ILogger> g_logger;
 
// Set up network and register the serializable classes.
void configuration( mi::neuraylib::INeuray* neuray, const char* multicast_address)
{
    // Set UDP networking mode and multicast address 224.1.1.1
    mi::base::Handle<mi::neuraylib::INetwork_configuration> network_configuration(
        neuray->get_api_component<mi::neuraylib::INetwork_configuration>());
    check_success( network_configuration->set_mode(
        mi::neuraylib::INetwork_configuration::MODE_UDP) == 0);
    check_success( network_configuration->set_multicast_address( multicast_address) == 0);
 
    // Register our class serializable classes
    mi::base::Handle<mi::neuraylib::IDice_configuration> dice_configuration(
        neuray->get_api_component<mi::neuraylib::IDice_configuration>());
    check_success( dice_configuration.is_valid_interface());
    check_success( dice_configuration->register_serializable_class<My_serializable_1>());
    check_success( dice_configuration->register_serializable_class<My_serializable_2>());
}
 
// On the client add some objects to the distributed cache and remove them again.
void run_client( mi::neuraylib::INeuray* neuray, mi::Uint64 N)
{
    g_logger->message(
        mi::base::MESSAGE_SEVERITY_INFO, "DICE:MAIN", "This host acts as client.");
 
    // Get the distributed cache
    mi::base::Handle<mi::neuraylib::IDice_configuration> dice_configuration(
        neuray->get_api_component<mi::neuraylib::IDice_configuration>());
    mi::base::Handle<mi::neuraylib::IDistributed_cache> distributed_cache(
        neuray->get_api_component<mi::neuraylib::IDistributed_cache>());
 
    // Generate N objects and store them in the distributed cache.
    // Alternate between both serializable classes. We use the index squared as example data.
    IMy_serializable** objects = new IMy_serializable*[N];
    for( mi::Uint64 i = 0; i < N; ++i) {
        IMy_serializable* object = ( (i+1)%2 )
            ? static_cast<IMy_serializable*>( new My_serializable_1)
            : static_cast<IMy_serializable*>( new My_serializable_2);
        object->set_data( i*i);
        g_logger->printf( mi::base::MESSAGE_SEVERITY_INFO, "DICE:MAIN",
            "Storing object with index %llu at address %p with data %llu.",
            i, object, object->get_data());
        distributed_cache->store_object( object);
        objects[i] = object;
        sleep_seconds( 1);
    }
 
    // Remove objects from the distributed cache and release them.
    for( mi::Uint64 i = 0; i < N; ++i) {
        IMy_serializable* object = objects[i];
        objects[i] = 0;
        g_logger->printf( mi::base::MESSAGE_SEVERITY_INFO, "DICE:MAIN",
            "Removing object with index %llu at address %p with data %llu.",
            i, object, object->get_data());
        distributed_cache->remove_object( object);
        object->release();
        sleep_seconds( 1);
    }
    delete[] objects;
}
 
// On the workers monitor the objects added to and removed from the distributed cache.
void run_worker( mi::neuraylib::INeuray* neuray, mi::Uint64 N)
{
    g_logger->message(
        mi::base::MESSAGE_SEVERITY_INFO, "DICE:MAIN", "This host acts as worker.");
 
    // Get the distributed cache
    mi::base::Handle<mi::neuraylib::IDice_configuration> dice_configuration(
        neuray->get_api_component<mi::neuraylib::IDice_configuration>());
    mi::base::Handle<mi::neuraylib::IDistributed_cache> distributed_cache(
        neuray->get_api_component<mi::neuraylib::IDistributed_cache>());
 
    // Create a receiver instance for each serializable class and register them with
    // the corresponding class IDs.
    Object_receiver object_receiver_1( 1);
    Object_receiver object_receiver_2( 2);
    distributed_cache->register_receiver( My_serializable_1::IID(), &object_receiver_1);
    distributed_cache->register_receiver( My_serializable_2::IID(), &object_receiver_2);
 
    // Wait for some time
    sleep_seconds( static_cast<mi::Float32>( 3*N));
 
    // Unregister receiver instances
    distributed_cache->unregister_receiver( My_serializable_1::IID());
    distributed_cache->unregister_receiver( My_serializable_2::IID());
}
 
int main( int argc, char* argv[])
{
    // Check command line parameters
    if( argc != 4 || ((strcmp( argv[1], "client") != 0) && (strcmp( argv[1], "worker") != 0))) {
        std::cerr << "Usage: example_distributed_cache <mode> <multicast_address> <N>" << std::endl;
        keep_console_open();
        return EXIT_FAILURE;
    }
    bool is_worker                = strcmp( argv[1], "worker") == 0;
    const char* multicast_address = argv[2];
    mi::Uint64 N                  = static_cast<mi::Uint64>( atoi( argv[3]));
 
    // Access the DiCE library
    mi::base::Handle<mi::neuraylib::INeuray> neuray( load_and_get_ineuray());
    check_success( neuray.is_valid_interface());
 
    // Configure the DiCE library
    configuration( neuray.get(), multicast_address);
 
    mi::base::Handle<mi::neuraylib::ILogging_configuration> logging_configuration(
        neuray->get_api_component<mi::neuraylib::ILogging_configuration>());
    g_logger = logging_configuration->get_forwarding_logger();
    logging_configuration = 0;
 
    // Start the DiCE library
    mi::Sint32 result = neuray->start();
    check_start_success( result);
 
    // Test the distributed cache
    is_worker ? run_worker( neuray.get(), N) : run_client( neuray.get(), N);
 
    // Shut down the DiCE library
    check_success( neuray->shutdown() == 0);
    g_logger = 0;
    neuray = 0;
 
    // Unload the DiCE library
    check_success( unload());
 
    keep_console_open();
    return EXIT_SUCCESS;
}

[Previous] [Next] [Up]