Example for Fragmented Jobs (locally)

[Previous] [Next] [Up]

This example demonstrates the implementation of fragmented jobs. This example focuses on a simple setting where fragments are only executed in different threads on the same host. The next example extends this example by distributing the execution of fragments to other hosts in the cluster.

New Topics

• Implementation of fragmented jobs
• Execution of fragmented jobs

Detailed Description

Implementation of fragmented jobs

In this example the fragmented job class computes the sum from 0 to some fixed large number. This computation is splitted into fragments such that each fragments computes the partial sum of a range of numbers. Fragments are executed concurrently in different threads. Note that the number of fragments as well as the size of each fragment are constants defined in the class itself. Of course, this is not a necessity and was done purely for simplicity.

The method execute_fragment() provides the necessary context for local execution of fragments. For reasons that will become clear with the next example the actual work is done in the do_execution() method. The method sum_results() combines the results of all fragments. There are several other methods for which the mi::neuraylib::Fragmented_job mixin provides default implementations.

Execution of fragmented jobs

A fragmented job is executed by passing it to the mi::neuraylib::IDice_transaction::execute_fragmented() method.

Example Source

Source Code Location: examples/example_fragmented_job.h

/******************************************************************************
 * Copyright 2024 NVIDIA Corporation. All rights reserved.
 *****************************************************************************/
 
// examples/example_fragmented_job.h
//
// An user-defined fragmented job shared by several examples.
 
#ifndef EXAMPLE_FRAGMENTED_JOB_H
#define EXAMPLE_FRAGMENTED_JOB_H
 
#include <iostream>
 
#include <mi/dice.h>
 
extern mi::base::Handle<mi::base::ILogger> g_logger;
 
// A simple fragmented job class that computes the sum from 0 to N_FRAGMENTS * FRAGMENT_SIZE - 1.
// Each fragment computes the partial sum of FRAGMENT_SIZE consecutive numbers. To avoid the need
// for synchronization primitives, the final sum is built at the end after all threads finished.
class My_fragmented_job : public
    mi::neuraylib::Fragmented_job<0x2f0d0ac7,0x53d6,0x4bc3,0x9c,0x21,0x85,0x4a,0xca,0xa3,0x97,0x7f>
{
public:
    static const mi::Size   N_FRAGMENTS   = 32;
    static const mi::Uint64 FRAGMENT_SIZE = 10000000;
 
    My_fragmented_job()
    {
        for( mi::Uint64 s = 0; s < N_FRAGMENTS; s++)
            m_results[s] = 0;
    }
 
    // The fragmented identified by "index" computes the sum from index * FRAGMENT_SIZE to
    // (index+1) * FRAGMENT_SIZE -1.
    mi::Uint64 do_execution( mi::Size index, mi::Size /*count*/)
    {
        mi::Uint64 result = 0;
        for( mi::Uint64 s = index * FRAGMENT_SIZE; s < (index+1) * FRAGMENT_SIZE; ++s)
            result += s;
        return result;
    }
 
    // If the fragment is executed locally the result is simply stored in the result array.
    void execute_fragment(
        mi::neuraylib::IDice_transaction* /*transaction*/,
        mi::Size index,
        mi::Size count,
        const mi::neuraylib::IJob_execution_context* /*context*/)
    {
        g_logger->printf( mi::base::MESSAGE_SEVERITY_INFO, "DICE:MAIN",
            "Executing fragment %llu of %llu on client.", index, count);
        m_results[index] = do_execution( index, count);
    }
 
    // If the fragment is executed remotely the result is written to the serializer.
    void execute_fragment_remote(
        mi::neuraylib::ISerializer* serializer,
        mi::neuraylib::IDice_transaction* /*transaction*/,
        mi::Size index,
        mi::Size count,
        const mi::neuraylib::IJob_execution_context* /*context*/)
    {
        g_logger->printf( mi::base::MESSAGE_SEVERITY_INFO, "DICE:MAIN",
            "Executing fragment %llu of %llu on a worker.", index, count);
        mi::Uint64 result = do_execution( index, count);
        serializer->write( &result);
    }
 
    // On the local host, the result from the fragment is read from the deserializer
    // and stored in the result array.
    void receive_remote_result(
        mi::neuraylib::IDeserializer* deserializer,
        mi::neuraylib::IDice_transaction* /*transaction*/,
        mi::Size index,
        mi::Size count)
    {
        g_logger->printf( mi::base::MESSAGE_SEVERITY_INFO, "DICE:MAIN",
            "Receiving fragment %llu of %llu from a worker.", index, count);
        mi::Uint64 result;
        deserializer->read( &result);
        m_results[index] = result;
    }
 
    mi::neuraylib::IRDMA_buffer* get_rdma_result_buffer(
        mi::neuraylib::IRDMA_context* rdma_context, mi::Size /*index*/)
    {
        return rdma_context->get_read_memory( sizeof( mi::Uint64));
    }
 
    mi::neuraylib::IRDMA_buffer* execute_fragment_remote_rdma(
        mi::neuraylib::IDice_transaction* /*transaction*/,
        mi::Size index,
        mi::Size count,
        mi::neuraylib::IRDMA_context* rdma_context,
        const mi::neuraylib::IJob_execution_context* /*context*/)
    {
        g_logger->printf( mi::base::MESSAGE_SEVERITY_INFO, "DICE:MAIN",
            "Executing fragment %llu of %llu on a worker (RDMA variant).", index, count);
        mi::Uint64 result = do_execution( index, count);
        mi::neuraylib::IRDMA_buffer* buffer = rdma_context->get_write_memory( sizeof( result));
        if( !buffer)
            return 0;
        memcpy( buffer->get_data(), &result, sizeof( result));
        return buffer;
    }
 
    void receive_remote_result_rdma(
        mi::neuraylib::IRDMA_buffer* buffer,
        mi::neuraylib::IDice_transaction* /*transaction*/,
        mi::Size index,
        mi::Size count)
    {
        g_logger->printf( mi::base::MESSAGE_SEVERITY_INFO, "DICE:MAIN",
            "Receiving fragment %llu of %llu from a worker (RDMA variant).", index, count);
        mi::Uint64 result;
        memcpy( &result, buffer->get_data(), sizeof( result));
        m_results[index] = result;
    }
 
    // Returns the sum of the results of all fragments.
    mi::Uint64 sum_results() const
    {
        mi::Uint64 result = 0;
        for( mi::Size s = 0; s < N_FRAGMENTS; ++s)
            result += m_results[s];
        return result;
    }
 
    // Since we want to distribute the fragmented jobs also to other hosts in the cluster we need
    // to override the default implementation (which returns LOCAL).
    mi::neuraylib::IFragmented_job::Scheduling_mode get_scheduling_mode() const
    {
        return mi::neuraylib::IFragmented_job::CLUSTER;
    }
 
private:
    // The result array has one slot for the result of each fragment.
    mi::Uint64 m_results[N_FRAGMENTS];
};
 
#endif // EXAMPLE_FRAGMENTED_JOB_H

Source Code Location: examples/example_fragmented_job.cpp

/******************************************************************************
 * Copyright 2023 NVIDIA Corporation. All rights reserved.
 *****************************************************************************/
 
// examples/example_fragmented_job.cpp
 
#include <iostream>
 
#include <mi/dice.h>
 
// Include code shared by all examples.
#include "example_shared.h"
// Include definition of My_fragmented_job.
#include "example_fragmented_job.h"
 
mi::base::Handle<mi::base::ILogger> g_logger;
 
// Test the My_fragmented_job class.
void test_my_fragmented_job( mi::neuraylib::IDice_transaction* transaction)
{
    // Execute an instance of the job in n fragments which are delegated to other threads.
    My_fragmented_job fragmented_job;
    mi::Size n = My_fragmented_job::N_FRAGMENTS;
    transaction->execute_fragmented( &fragmented_job, n);
 
    // Compare the result with the expected result.
    mi::Uint64 s = My_fragmented_job::FRAGMENT_SIZE;
    mi::Uint64 expected_result = n*s/2 * (n*s-1);
    check_success( fragmented_job.sum_results() == expected_result);
}
 
// Obtain a DiCE transaction and run the test for the My_fragmented_job class.
void run( mi::neuraylib::INeuray* neuray)
{
    // Get the database, the global scope of the database and create a transaction in the global
    // scope.
    mi::base::Handle<mi::neuraylib::IDatabase> database(
        neuray->get_api_component<mi::neuraylib::IDatabase>());
    check_success( database.is_valid_interface());
    mi::base::Handle<mi::neuraylib::IScope> scope( database->get_global_scope());
    mi::base::Handle<mi::neuraylib::IDice_transaction> transaction(
        scope->create_transaction<mi::neuraylib::IDice_transaction>());
    check_success( transaction.is_valid_interface());
 
    // Test the My_fragmented_job class.
    test_my_fragmented_job( transaction.get());
 
    transaction->commit();
}
 
int main( int /*argc*/, char* /*argv*/[])
{
    // Access the DiCE library
    mi::base::Handle<mi::neuraylib::INeuray> neuray( load_and_get_ineuray());
    check_success( neuray.is_valid_interface());
 
    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);
 
    // Run the tests for My_fragmented_job
    run( neuray.get());
 
    // 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]