Writing a C++ CoralSequencer Node

Bench.cpp

#include <jni.h>
#include <iostream>
#include<unistd.h> 
#include <cstdlib>
#include <ctime>
#include <sys/time.h>
#include <map>
#include <sstream>
#include <cmath>
#include <limits>
#include <iomanip>
#include "com_coralblocks_coralsequencer_mq_CNode.h"
using namespace std;

static const int ITERATIONS = 2000000;
static const int WARMUP = 1000000;
static const int MSG_SIZE = 256;

long get_nano_ts(timespec* ts) {
    clock_gettime(CLOCK_MONOTONIC, ts);
    return ts->tv_sec * 1000000000 + ts->tv_nsec;
}

struct mi {
   long value;
};

void add_perc(stringstream& ss, int size, double perc, map<int, mi*>* map) {
 
    if (map->empty()) return;
     
    int max = -1;
    long x = round(perc * size);
    long i = 0;
    long sum = 0;

    for(::map<int, mi*>::iterator iter = map->begin(); iter != map->end(); iter++) {
     
        int time = iter->first;
        long count = (iter->second)->value;
         
        for(int a = 0; a < count; a++) {
            i++;
            sum += time;
                
            if (i == x) {
                max = time;
                goto OUT;
            }
        }
    }

    OUT:
     
    ss << " | " << fixed << setprecision(5) << (perc * 100) << "%";
    ss << " = [avg: " << (sum / i);
    ss << ", max: " << max << "]";
}

char* createRandomCharArray(int size) {
    char* charArray = new char[size + 1];  // Add 1 for the null-terminating character
    srand(static_cast<unsigned int>(time(nullptr)));
    for (int i = 0; i < size; ++i) {
        charArray[i] = static_cast<char>(rand() % (126 - 32 + 1) + 32);
    }
    charArray[size] = '\0';
    return charArray;
}

int main(int argc, char **argv) {

    JavaVM *jvm;                       // Pointer to the JVM (Java Virtual Machine)
    JNIEnv *env;                       // Pointer to native interface
    JavaVMInitArgs vm_args;            // JVM initialization arguments
    JavaVMOption options[24];          // JVM options

    options[0].optionString = "-DshutdownHooksAddRemoveDebug=false";
    options[1].optionString = "-DshutdownHooksDebug=true"; 
    options[2].optionString = "-DerrorLogOnSmallSocketBufferSize=true"; 
    options[3].optionString = "-DnioReactorEnsureBootstrap=true"; 
    options[4].optionString = "-DnioReactorUsePatchedDatagramChannelImpl=true";
    options[5].optionString = "-verbose:gc";
    options[6].optionString = "-XX:+UseCompressedOops"; 
    options[7].optionString = "-XX:+OptimizeStringConcat"; 
    options[8].optionString = "-Xms2g"; 
    options[9].optionString = "-Xmx4g"; 
    options[10].optionString = "-XX:NewSize=512m"; 
    options[11].optionString = "-XX:MaxNewSize=1024m"; 
    options[12].optionString = "--enable-preview"; 
    options[13].optionString = "--patch-module=java.base=./lib/coralreactor-boot-jdk21.jar"; 
    options[14].optionString = "--add-exports=java.base/sun.nio.ch=ALL-UNNAMED"; 
    options[15].optionString = "--add-exports=java.base/jdk.internal.misc=ALL-UNNAMED"; 
    options[16].optionString = "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED"; 
    options[17].optionString = "--add-opens=java.base/java.nio=ALL-UNNAMED"; 
    options[18].optionString = "-Djava.class.path=../CoralBits/target/classes:../CoralThreads/target/classes:../CoralQueue/target/classes:../CoralLog/target/classes:../CoralStore/target/classes:../CoralReactor/target/classes:../CoralFIX/target/classes:target/classes:target/coralsequencer-all.jar:target/coralsequencer.jar:lib/jna-3.5.1.jar"; 
    options[19].optionString = "-DlogColors=true"; 
    options[20].optionString = "-DlogLevel=Info"; 
    options[21].optionString = "-DnioReactorPerformanceBoost=AUTO";
    options[22].optionString = "-DnioReactorProcsToBind=3";
    options[23].optionString = "-Djava.library.path=lib";
    
    vm_args.version = JNI_VERSION_1_6;                      // Set the JNI version
    vm_args.nOptions = 24;                                  // Set the number of options
    vm_args.options = options;                              // Set the options to the JVM
    
    // Load and initialize the JVM
    JNI_CreateJavaVM(&jvm, (void**)&env, &vm_args);

    cout << "JVM created!!!" << endl;

    jvm->AttachCurrentThread((void**)&env, NULL);

    jclass startJavaClass = env->FindClass("com/coralblocks/coralsequencer/Start");
    jmethodID findAppMethod = env->GetStaticMethodID(startJavaClass, "findApplication", "(Ljava/lang/String;)Lcom/coralblocks/coralsequencer/app/Application;");
    jclass nodeClass = env->FindClass("com/coralblocks/coralsequencer/mq/CNode");
    jmethodID sendCommandMethod = env->GetMethodID(nodeClass, "sendCommand", "(Ljava/lang/CharSequence;)Z");

    jstring str1 = env->NewStringUTF("mqs/cnode.mq"); // creates a NODE7 node
    jclass stringClass = env->FindClass("java/lang/String");
    jobject args = env->NewObjectArray(1, stringClass, str1);
    jmethodID mainMethod = env->GetStaticMethodID(startJavaClass, "main", "([Ljava/lang/String;)V");
    jmethodID isActiveMethod = env->GetMethodID(nodeClass, "isActive", "()Z");

    cout << "About to call Java main method..." << endl;

    env->CallStaticVoidMethod(startJavaClass, mainMethod, args);

    cout << "Returned from Java main method!" << endl;

    // Get the node (NODE7)
    jobject node = env->CallStaticObjectMethod(startJavaClass, findAppMethod, env->NewStringUTF("NODE7"));

    cout << "Waiting for node to become active..." << endl;

    // Sleep until node becomes active
    while(env->CallBooleanMethod(node, isActiveMethod) == JNI_FALSE) sleep(1);

    cout << "isActive() returned true!" << endl;

    jstring msgToSend = env->NewStringUTF(createRandomCharArray(MSG_SIZE));

    cout << "About to send first message!" << endl;

    env->CallObjectMethod(node, sendCommandMethod, msgToSend);

    cout << "First message sent!" << endl;

    jvm->DetachCurrentThread();

    // Release the JVM
    jvm->DestroyJavaVM(); // this will wait for Java threads to die...

    cout << "JVM Destroyed!!!" << endl;

    return 0;
}

struct timespec ts;
long long x = 0;
long long totalTime = 0;
int minTime = numeric_limits<int>::max();
int maxTime = numeric_limits<int>::min();
long startTime = 0;
long endTime = 0;
map<int, mi*>* results;
int iterations = 0;

jobject node;
jmethodID sendCommandMethod;
jmethodID isRewindingMethod;
jstring msgToSend;

JNIEXPORT void JNICALL Java_com_coralblocks_coralsequencer_mq_CNode_handleMessageC
  (JNIEnv *env, jobject obj, jboolean isMine, jobject msg) {

    endTime = get_nano_ts(&ts);

    if (node == NULL) {
        jclass startJavaClass = env->FindClass("com/coralblocks/coralsequencer/Start");
        jmethodID findAppMethod = env->GetStaticMethodID(startJavaClass, "findApplication", "(Ljava/lang/String;)Lcom/coralblocks/coralsequencer/app/Application;");
        jclass nodeClass = env->FindClass("com/coralblocks/coralsequencer/mq/CNode");
        sendCommandMethod = env->GetMethodID(nodeClass, "sendCommand", "(Ljava/lang/CharSequence;)Z");
        isRewindingMethod = env->GetMethodID(nodeClass, "isRewinding", "()Z");

        node = env->CallStaticObjectMethod(startJavaClass, findAppMethod, env->NewStringUTF("NODE7"));

        node = env->NewGlobalRef(node);

        msgToSend = env->NewStringUTF(createRandomCharArray(MSG_SIZE));

        results = new map<int, mi*>();
    }

    if (env->CallBooleanMethod(node, isRewindingMethod) == JNI_TRUE) return;

    int res = startTime > 0 ? (endTime - startTime) : 1; // 1 only for first message/pass

    if (res <= 0) res = 1;

    if (iterations++ >= WARMUP) {
        totalTime += res;
        minTime = min(minTime, res);
        maxTime = max(maxTime, res);
            
        ::map<int, mi*>::iterator iter = results->find(res);
            
        if (iter != results->end()) {
            
            (iter->second)->value = (iter->second)->value + 1;
                
        } else {
            
            mi* elem = new mi();
            elem->value = 1;
            (*results)[res] = elem;
        }
    }

    if (iterations == ITERATIONS) {

        int count = ITERATIONS - WARMUP;
        
        double avg = totalTime / count;
        
        stringstream ss;
        
        ss << "Iterations: " << count << " | Avg Time: " << avg;
    
        if (count > 0) {
            ss << " | Min Time: " << minTime << " | Max Time: " << maxTime;
        }
        
        add_perc(ss, count, 0.75, results);
        add_perc(ss, count, 0.90, results);
        add_perc(ss, count, 0.99, results);
        add_perc(ss, count, 0.999, results);
        add_perc(ss, count, 0.9999, results);
        add_perc(ss, count, 0.99999, results);
        
        cout << ss.str() << endl << endl;

        delete results;

    } else {

        startTime = get_nano_ts(&ts);

        env->CallObjectMethod(node, sendCommandMethod, msgToSend);

    }

}