Commit e1405dc9 authored by Marc Vef's avatar Marc Vef
Browse files

Added some configuration definitions to control Margo threading

parent 703ccde8

ifs/include/configure.hpp

+9 −2
Original line number Diff line number Diff line
@@ -36,8 +36,15 @@ 
// Optimize Key-Value store for tmpfs/ramdisk usage

//#define KV_OPTIMIZE_RAMDISK



// RPC configuration

#define RPCPORT 4433

// Buffer size for Rocksdb

#define KV_WRITE_BUFFER 16384



// Margo configuration



// Number of threads used for RPC and IPC handlers at the daemon

#define RPC_HANDLER_THREADS 16

#define IPC_HANDLER_THREADS 16

#define RPC_PORT 4433

#define RPC_TRIES 3

// rpc timeout to try again in milliseconds

#define RPC_TIMEOUT 180000

ifs/main.cpp

+1 −1
Original line number Diff line number Diff line
@@ -143,7 +143,7 @@ int main(int argc, const char* argv[]) { 
    }

    ADAFS_DATA->hosts(hostmap);

    ADAFS_DATA->host_size(hostmap.size());

    ADAFS_DATA->rpc_port(fmt::FormatInt(RPCPORT).str());

    ADAFS_DATA->rpc_port(fmt::FormatInt(RPC_PORT).str());

    ADAFS_DATA->hosts_raw(hosts_raw);

ifs/src/daemon/adafs_daemon.cpp

+3 −3
Original line number Diff line number Diff line
@@ -79,7 +79,7 @@ bool init_ipc_server() { 


    ADAFS_DATA->spdlogger()->debug("{}() Initializing Margo IPC server...", __func__);

    // Start Margo (this will also initialize Argobots and Mercury internally)

    auto mid = margo_init(protocol_port.c_str(), MARGO_SERVER_MODE, 1, 16);

    auto mid = margo_init(protocol_port.c_str(), MARGO_SERVER_MODE, 1, IPC_HANDLER_THREADS);



    if (mid == MARGO_INSTANCE_NULL) {

        ADAFS_DATA->spdlogger()->error("{}() margo_init() failed to initialize the Margo IPC server", __func__);
@@ -118,13 +118,13 @@ bool init_ipc_server() { 
}



bool init_rpc_server() {

    auto protocol_port = RPC_PROTOCOL + "://localhost:"s + to_string(RPCPORT);

    auto protocol_port = RPC_PROTOCOL + "://localhost:"s + to_string(RPC_PORT);

    hg_addr_t addr_self;

    hg_size_t addr_self_cstring_sz = 128;

    char addr_self_cstring[128];

    ADAFS_DATA->spdlogger()->debug("{}() Initializing Margo RPC server...", __func__);

    // Start Margo (this will also initialize Argobots and Mercury internally)

    auto mid = margo_init(protocol_port.c_str(), MARGO_SERVER_MODE, 1, 16);

    auto mid = margo_init(protocol_port.c_str(), MARGO_SERVER_MODE, 1, RPC_HANDLER_THREADS);

    if (mid == MARGO_INSTANCE_NULL) {

        ADAFS_DATA->spdlogger()->error("{}() margo_init failed to initialize the Margo RPC server", __func__);

        return false;

ifs/src/db/db_util.cpp

+8 −0
Original line number Diff line number Diff line
@@ -89,4 +89,12 @@ void optimize_rocksdb(rocksdb::Options& options) { 
//    options.arena_block_size = 1024 * 8;

//    options.compression = rocksdb::kNoCompression; // doesnt do anything

#endif

#if defined(KV_WRITE_BUFFER)

    // write_buffer_size is multiplied by the write_buffer_number to get the amount of data hold in memory.

    // at min_write_buffer_number_to_merge rocksdb starts to flush entries out to disk

    options.write_buffer_size = KV_WRITE_BUFFER << 20;

    // XXX experimental values. We only want one buffer, which is held in memory

    options.max_write_buffer_number = 1;

    options.min_write_buffer_number_to_merge = 1;

#endif

}
 
 No newline at end of file

ifs/src/preload/margo_ipc.cpp

+1 −1
Original line number Diff line number Diff line
@@ -82,7 +82,7 @@ bool ipc_send_get_fs_config() { 
            fs_config->gid = out.gid;

            fs_config->host_id = out.host_id;

            fs_config->host_size = out.host_size;

            fs_config->rpc_port = to_string(RPCPORT);

            fs_config->rpc_port = to_string(RPC_PORT);



            // split comma separated host string and create a hosts map

            string hosts_raw = out.hosts_raw;