test_dvs_server.cpp | Ansys Developer Portal

Go to the documentation of this file.
/*  *************************************************************
 *   Copyright 2017-2023 ANSYS, Inc.
 *   All Rights Reserved.
 *
 *        Restricted Rights Legend
 *
 *   Use, duplication, or disclosure of this
 *   software and its documentation by the
 *   Government is subject to restrictions as
 *   set forth in subdivision [(b)(3)(ii)] of
 *   the Rights in Technical Data and Computer
 *   Software clause at 52.227-7013.
 *  *************************************************************
 */
 
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <stdarg.h>
 
#include "dvs_server_interface.h"
#include "dynamic_visualization_store_enums.h"
#include "logger.h"
 
#include <atomic>
#include <chrono>
#include <mutex>
#include <thread>
#include <vector>
#include <signal.h>
 
static std::atomic<bool> s_terminate{false};
static std::mutex s_server_mutex;
static DVS::IServer* s_server{nullptr};
 
void signal_callback_handler(int signum)
{
    fprintf(stdout, "Caught Signal: %i\n", signum);
    s_terminate = true;
    std::lock_guard<std::mutex> lock(s_server_mutex);
    if (s_server)
    {
        s_server->terminating();
    }
}
 
int main(int argc, char** argv)
{
    //Register a signal handler instead of just killing the app so
    //we can shutdown gracefully
    signal(SIGINT, signal_callback_handler);
 
    uint32_t port = 50055;
    uint32_t delay_ms = 500;
    char host[512] = {0};
    strcpy(host, "127.0.0.1");
    char protocol[10] = {0};
    strcpy(protocol, "grpc");
 
    char secret[255] = {0};
    char cache_uri[512] = {0};
    char dvs_file_loc[1024] = {0};
    uint32_t debug_wait_sec = 0;
    uint32_t server_number = 0;
    uint32_t local_ranks = 1;
    uint32_t server_verbosity = 3;
    uint32_t timeout_sec = 15;
 
    uint32_t i = 1;
    while (i < argc)
    {
        if ((strcmp(argv[i], "-p") == 0 ) && (i < argc - 1))
        {
            i++;
            port = atoi(argv[i]);
        }
        else if ((strcmp(argv[i], "-h") == 0) && (i < argc - 1))
        {
            i++;
            strncpy(host, argv[i], 255);
        }
        else if ((strcmp(argv[i], "-protocol") == 0) && (i < argc - 1))
        {
            i++;
            strncpy(protocol, argv[i], 10);
            if (strcmp(protocol, "grpc") != 0 && strcmp(protocol, "null") != 0 )
            {
                fprintf(stderr, "protocol: %s invalid\n", protocol);
                exit(1);
            }
        }
        else if ((strcmp(argv[i], "-d") == 0) && (i < argc - 1))
        {
            i++;
            delay_ms = atoi(argv[i]);
        }
        else if (strcmp(argv[i], "-secret") == 0)
        {
            i++;
            strncpy(secret, argv[i], 255);
        }
        else if (strcmp(argv[i], "-server") == 0 && (i < argc - 3))
        {
            i++;
            server_number = atoi(argv[i++]);
            local_ranks = atoi(argv[i++]);
            server_verbosity = atoi(argv[i]);
        }
        else if (strcmp(argv[i], "-cache_uri") == 0)
        {
            i++;
            strncpy(cache_uri, argv[i], 512);
        }
        else if (strcmp(argv[i], "-debug_wait") == 0)
        {
            i++;
            debug_wait_sec = atoi(argv[i]);
        }
        else if (strcmp(argv[i], "-timeout") == 0)
        {
            i++;
            timeout_sec = atoi(argv[i]);
        }
        else if (strcmp(argv[i], "-dvs_file") == 0)
        {
            i++;
            strncpy(dvs_file_loc, argv[i], sizeof(dvs_file_loc)-1);
        }
        else
        {
            fprintf(stderr, "Unknown option: %s\n", argv[i]);
            fprintf(stderr, "Usage: %s [-p port] [-h host] [-d ms delay] [-s dx dy dz] \n", argv[0]);
            fprintf(stderr, "Options:\n");
            fprintf(stderr, "  -p port              Server port to connect to.  Default: %u\n", port);
            fprintf(stderr, "  -h host              Server hostname to connect to.  Default: %s\n", host);
            fprintf(stderr, "  -protocol str        Server protocol. I.E. grpc or null. Default: grpc");
            fprintf(stderr, "  -d delay(ms)         Milliseconds to delay between server updates.  Default: %u\n", delay_ms);
            fprintf(stderr, "  -secret string       Shared secret to use when talking with server Default: no shared secret\n");
            fprintf(stderr, "  -server n r v        Start a server using server number [n], expecting [r] local ranks for DVS connections to connect to with verbosity [v]. Local Rank Min/Max: 1/1000\n");
            fprintf(stderr, "  -cache_uri str       The URI for the server to use for the cache. Default: No cache\n");
            fprintf(stderr, "  -dvs_file str        Set a location for the dvs files to be created Default: ./\n");
            fprintf(stderr, "  -timeout sec         Shutdown the server if no new data has been received within timeout. Default: %u sec\n", timeout_sec);
            fprintf(stderr, "  -debug_wait s        Wait for [s] for debugging\n");
            exit(1);
        }
        i++;
    }
 
    //This is just used for debugging/attaching to the app if some problem arises
    if (debug_wait_sec > 0) {
        std::this_thread::sleep_for(std::chrono::seconds(debug_wait_sec));
    }
 
    char uri[512] = {0};
    sprintf(uri, "%s://%s:%u", protocol, host, port);
    //The server URI is used to setup the protocol host and port of the server
    //(where and how clients need to connect). It will be something like:
    //grpc:://localhost:50055
    fprintf(stderr, "Server URI: %s\n", uri);
    fprintf(stderr, "Update Delay(ms): %u\n", delay_ms);
 
    {
        std::lock_guard<std::mutex> lock(s_server_mutex);
        s_server = DVS::CREATE_SERVER_INSTANCE(uri);
    }
    if (s_server == nullptr)
    {
        fprintf(stderr, "Failed to create server\n");
        exit(1);
    }
    char temp[10];
    snprintf(temp, 10, "%u", server_verbosity);
    s_server->set_option("VERBOSE",temp);
    if (strlen(cache_uri) > 0)
    {
        //If not using a cache_uri the cache will be all in memory, not
        //really useful for this specific test server app but showing a simple example.
        //In the future you will be able to use the DVS Reader API to read data from
        //memory using this instead of needing to read data from the cache.
 
        //The cache URI should be something like: hdf5://localhost/D:/my_cache
        s_server->set_option("CACHE_URI",cache_uri);
    }
 
    if (strlen(dvs_file_loc) > 0)
    {
        //The DVS file location let's you point the location of the auto created
        //.dvs files to another location, absolute or relative to the cache_uri location
        s_server->set_option("DVS_FILE_LOCATION", dvs_file_loc);
    }
 
    if (strlen(secret) > 0)
    {
        //This is needed if wanting to use a simple secret to allow clients
        //to connect to you. The clients will also need this secret.
        s_server->set_option("SERVER_SECURITY_SECRET", secret);
    }
 
    //The startup_unthreaded call will start/initialize the server and block until
    //a all clients have initialized the dataset and sent a single timestep. A call
    //to s_server->terminating() can be seen above in the signal handler to allow you
    //to exit this call early.
    s_server->startup_unthreaded(server_number, local_ranks);
    uint32_t size;
    auto _uri = s_server->get_uri();
    if (_uri){
        fprintf(stderr, "Actual Server URI: %s\n", _uri);
    }
    //For this example we are going to quit early if number pending or number complete have not
    //changed within timeout_sec seconds.
    uint32_t num_pending = 0;
    uint32_t num_complete = 0;
    auto start_time = std::chrono::system_clock::now();
 
    while (!s_terminate) {
        //This is the main control loop of the server. DVS::IServer::update() just
        //needs to be called in a loop to handle the server state until the server
        //should be terminated.
        //Note: update() will flush data to the cache so shutting down / exiting mid
        //update will likely cause cache corruption.
        dvs_ret ret_val = s_server->update();
        // Avoid burning up a CPU waiting for the I/O to complete
        std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
 
        uint32_t cur_num_pending = 0;
        uint32_t cur_num_complete = 0;
        s_server->get_timestep_count(cur_num_pending, cur_num_complete);
        if (cur_num_pending == num_pending && cur_num_complete == num_complete) {
            //If our values haven't changed check to see if we should time out
            std::chrono::duration<double> elapsed_seconds = std::chrono::system_clock::now() - start_time;
            if (elapsed_seconds.count() > timeout_sec) {
                //Terminate the server and quit out
                s_server->terminating();
                s_terminate = true;
            }
        }
        else {
            //We are different so reset our counters
            start_time = std::chrono::system_clock::now();
            num_pending = cur_num_pending;
            num_complete = cur_num_complete;
        }
    }
 
    num_pending = 1;
    num_complete = 0;
    while (num_pending > 0) {
        //This loop may or may not need to be done by a application. But is here
        //to show potential cleanup when terminating early. If terminating you may
        //want to finish the current in flight timestep. Since the server has been
        //told it is terminating in the signal handler above it will not accept new
        //timesteps but will allow pending timesteps to still complete.
 
        //Note: A user probably will want to do more due dilligence than this when
        //shutting down. A timestep that is waiting on a client that is hung or crashed
        //will never flip from pending->complete currently. This could cause this loop
        //to be infinite waiting on a client who will never send its data.
        dvs_ret error = s_server->update();
        if (error != DVS_NONE) {
            break;
        }
        //If the update() call above isn't called in this loop num_pending will
        //never change also calling this loop to be infinite. We must update the server
        //to get timestep state changes.
        error = s_server->get_timestep_count(num_pending, num_complete);
        if (error != DVS_NONE) {
            break;
        }
        // Avoid burning up a CPU waiting for the I/O to complete
        std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
    }
 
    //Destroy the server on exit, guarding against the s_server destruction here
    //in case the signal handler above is triggered.
    std::lock_guard<std::mutex> lock(s_server_mutex);
    DVS::DESTROY_SERVER_INSTANCE(s_server);
    s_server = nullptr;
 
    return 0;
}
Post-processing tools 2025 R1

test_dvs_server

Connect with Ansys
