• 高性能网络通信框架--Mercury

    Mercury是一个专门设计用于HPC系统的RPC框架，允许异步传输参数和执行请求，以及直接支持大型数据参数。网络实现是抽象的，允许轻松移植到未来系统，并有效使用现有的本地传输机制。Mercury的接口是通用的，允许对任何函数调用进行序列化。汞是微服务Mochi生态系统的核心组成部分。

    一、支持的架构

    MPI实现支持的架构通常由网络抽象层支持。

    网络抽象（NA）层在内部由RPC层和批量层使用。NA层使用插件机制，以便在运行时轻松添加和选择对各种网络协议的支持。

    NA支持不同的后端实现。推荐OFI/libfabric用于节点间通信的插件，而SM（共享内存）则用于节点内通信。

     

    在libfabric不可用或不建议使用的平台上，UCX插件也可用作替代传输，目前支持的协议是tcp和verbs。
    MPI和BMI（tcp）插件仍然受支持，但逐渐被移动为不推荐使用的插件。

     

    二、软件需求

    要使用OFI libfabric插件，参考后面的libfabrib构建说明。

    要使用UCX插件，参考后面的UCX构建说明。

    要在Linux上使用本机NA SM（共享内存）插件，需要内核v3.2中引入的跨内存连接（CMA）功能。

    要使用BMI插件，最方便的方法是通过spack安装，也可以：

    1. git clone https://github.com/radix-io/bmi.git && cd bmi
    2. ./prepare && ./configure --enable-shared --enable-bmi-only
    3. make && make install

    要使用MPI插件，Mercury需要一个配置良好的MPI实现（MPICH2 v1.4.1或更高版本/OpenMPI v1.6或更高），并在接受远程连接的目标上提供MPI_THREAD_MULTIPLE。不接受传入连接的进程不需要具有多线程执行级别。

    三、构建

    1. #bzip2 -dc mercury-X.tar.bz2 | tar xvf -
    2.  
    3. #cd mercury-X
    4.  
    5. #mkdir build
    6.  
    7. #cd build
    8.  
    9. #ccmake .. //(".." 是mercury-X目录的相对路径)

    多次按“c”，然后选择合适的选项。建议的选项有：

    1. BUILD_SHARED_LIBS                ON (or OFF if the library you link
    2.                                  against requires static libraries)
    3. BUILD_TESTING                    ON/OFF
    4. Boost_INCLUDE_DIR                /path/to/include/directory
    5. CMAKE_INSTALL_PREFIX             /path/to/install/directory
    6. MERCURY_ENABLE_DEBUG             ON/OFF
    7. MERCURY_TESTING_ENABLE_PARALLEL  ON/OFF
    8. MERCURY_USE_BOOST_PP             ON
    9. MERCURY_USE_CHECKSUMS            ON/OFF
    10. MERCURY_USE_SYSTEM_BOOST         ON/OFF
    11. MERCURY_USE_SYSTEM_MCHECKSUM     ON/OFF
    12. MERCURY_USE_XDR                  OFF
    13. NA_USE_BMI                       ON/OFF
    14. NA_USE_MPI                       ON/OFF
    15. NA_USE_OFI                       ON/OFF
    16. NA_USE_PSM                       ON/OFF
    17. NA_USE_PSM2                      ON/OFF
    18. NA_USE_SM                        ON/OFF
    19. NA_USE_UCX                       ON/OFF

    设置包含目录和库路径可能需要您通过键入“t”切换到高级模式。完成后，如果没有看到任何错误，请键入“g”以生成生成生成文件。退出CMake配置屏幕并准备构建目标后，请执行以下操作：

    #make

    详细编译/生成输出是通过在make命令中插入VERBOSE=1来实现的

    make VERBOSE=1

    四、安装

    假设已设置CMAKE_INSTALL_PREFIX（见上一步），并且对目标目录具有写入权限，请从构建目录执行：

    make install

    可以运行测试以检查基本RPC功能（请求和批量数据传输）是否正常工作。CTest用于运行测试，只需从构建目录运行：

    ctest .

    详细测试是通过在ctest命令中插入-V来完成的：

    ctest -V .

    通过插入-VV可以显示额外的详细信息：

    ctest -VV .

    五、实例分析

    客户端

    1. int
    2. main(void)
    3. {
    4.     const char *info_string = NULL;
    5.  
    6.     char target_addr_string[PATH_MAX], *p;
    7.     FILE *na_config = NULL;
    8.  
    9.     hg_class_t *hg_class;
    10.     hg_context_t *hg_context;
    11.     hg_addr_t hg_target_addr;
    12.  
    13.     hg_return_t hg_ret;
    14.  
    15.     /* Get info string */
    16.     info_string = getenv("HG_PORT_NAME");
    17.     if (!info_string) {
    18.         fprintf(stderr, "HG_PORT_NAME environment variable must be set\n");
    19.         exit(0);
    20.     }
    21.     printf("Using %s\n", info_string);
    22.  
    23.     HG_Set_log_level("warning");
    24.  
    25.     /* 使用所需的网络抽象类初始化Mercury */
    26.     hg_class = HG_Init(info_string, HG_FALSE);
    27.  
    28.     /* Create HG context */
    29.     hg_context = HG_Context_create(hg_class);
    30.  
    31.     /* 连接字符串是在NA_Addr_self()/NA_Addr_to_string()后生成, 得到字符串并将其传递给  NA_Addr_lookup() */
    32.     na_config = fopen(TEMP_DIRECTORY CONFIG_FILE_NAME, "r");
    33.     if (!na_config) {
    34.         fprintf(stderr, "Could not open config file from: %s\n",
    35.             TEMP_DIRECTORY CONFIG_FILE_NAME);
    36.         exit(0);
    37.     }
    38.     fgets(target_addr_string, PATH_MAX, na_config);
    39.     p = strrchr(target_addr_string, '\n');
    40.     if (p != NULL)
    41.         *p = '\0';
    42.     printf("Target address is: %s\n", target_addr_string);
    43.     fclose(na_config);
    44.  
    45.     /* 查找 target address */
    46.     HG_Addr_lookup2(hg_class, target_addr_string, &hg_target_addr);
    47.  
    48.     /* Register RPC */
    49.     snappy_compress_id_g = snappy_compress_register(hg_class);
    50.  
    51.     /* Send RPC to target */
    52.     snappy_compress_rpc(hg_class, hg_context, hg_target_addr);
    53.  
    54.     /* Poke progress engine and check for events */
    55.     do {
    56.         unsigned int actual_count = 0;
    57.         do {
    58.             hg_ret = HG_Trigger(
    59.                 hg_context, 0 /* timeout */, 1 /* max count */, &actual_count);
    60.         } while ((hg_ret == HG_SUCCESS) && actual_count);
    61.  
    62.         /* Do not try to make progress anymore if we're done */
    63.         if (snappy_compress_done_g)
    64.             break;
    65.  
    66.         hg_ret = HG_Progress(hg_context, HG_MAX_IDLE_TIME);
    67.     } while (hg_ret == HG_SUCCESS);
    68.  
    69.     /* Finalize */
    70.     HG_Addr_free(hg_class, hg_target_addr);
    71.  
    72.     HG_Context_destroy(hg_context);
    73.     HG_Finalize(hg_class);
    74.  
    75.     return EXIT_SUCCESS;
    76. }
    77.  
    78. static int
    79. snappy_compress_rpc(
    80.     hg_class_t *hg_class, hg_context_t *hg_context, hg_addr_t hg_target_addr)
    81. {
    82.     int *input;
    83.     size_t source_length = NR_ITEMS * sizeof(int);
    84.     hg_bulk_t input_bulk_handle;
    85.  
    86.     void *compressed;
    87.     size_t max_compressed_length;
    88.     hg_bulk_t compressed_bulk_handle;
    89.  
    90.     snappy_compress_in_t snappy_compress_input;
    91.     struct snappy_compress_rpc_args *snappy_compress_rpc_args;
    92.     hg_handle_t handle;
    93.     int i;
    94.  
    95.     /**
    96.      * 我们将取一个缓冲区并将其发送到服务器进行压缩。
    97.      */
    98.     max_compressed_length = snappy_max_compressed_length(source_length);
    99.     printf("Input buffer length is: %zu\n", source_length);
    100.     printf("Max compressed length is: %zu\n", max_compressed_length);
    101.  
    102.     /* 生成 input buffer */
    103.     input = (int *) malloc(source_length);
    104.     for (i = 0; i < NR_ITEMS; i++) {
    105.         input[i] = rand() % 10;
    106.     }
    107.     print_buf(20, input);
    108.  
    109.     /* Allocate compressed buffer */
    110.     compressed = malloc(max_compressed_length);
    111.     memset(compressed, '\0', max_compressed_length);
    112.  
    113.     /* Create HG handle bound to target */
    114.     HG_Create(hg_context, hg_target_addr, snappy_compress_id_g, &handle);
    115.  
    116.     /**
    117.      * 将“handle”与内存区域关联。Mercury的批量传输将从该区域获取/输入数据
    118.      */
    119.     HG_Bulk_create(hg_class, 1, (void **) &input, &source_length,
    120.         HG_BULK_READ_ONLY, &input_bulk_handle);
    121.     HG_Bulk_create(hg_class, 1, &compressed, &max_compressed_length,
    122.         HG_BULK_READWRITE, &compressed_bulk_handle);
    123.  
    124.     /* 创建结构以保存参数，因为调用将异步执行 */
    125.     snappy_compress_rpc_args = (struct snappy_compress_rpc_args *) malloc(
    126.         sizeof(struct snappy_compress_rpc_args));
    127.     snappy_compress_rpc_args->input = input;
    128.     snappy_compress_rpc_args->input_length = source_length;
    129.     snappy_compress_rpc_args->input_bulk_handle = input_bulk_handle;
    130.     snappy_compress_rpc_args->compressed = compressed;
    131.     snappy_compress_rpc_args->compressed_bulk_handle = compressed_bulk_handle;
    132.  
    133.     /* Set input arguments that will be passed to HG_Forward */
    134.     snappy_compress_input.input_bulk_handle = input_bulk_handle;
    135.     snappy_compress_input.compressed_bulk_handle = compressed_bulk_handle;
    136.  
    137.     /* Forward the call */
    138.     printf("Sending input to target\n");
    139.     HG_Forward(handle, snappy_compress_rpc_cb, snappy_compress_rpc_args,
    140.         &snappy_compress_input);
    141.  
    142.     /* Handle will be destroyed when call completes (reference count) */
    143.     HG_Destroy(handle);
    144.  
    145.     return 0;
    146. }
    147.  
    148. /* 该例程在调用HG_Trigger和RPC完成后执行 */
    149. static hg_return_t
    150. snappy_compress_rpc_cb(const struct hg_cb_info *callback_info)
    151. {
    152.     struct snappy_compress_rpc_args *snappy_compress_rpc_args =
    153.         (struct snappy_compress_rpc_args *) callback_info->arg;
    154.     hg_handle_t handle = callback_info->info.forward.handle;
    155.  
    156.     int *input;
    157.     size_t source_length;
    158.  
    159.     void *compressed;
    160.     size_t compressed_length;
    161.  
    162.     int *uncompressed;
    163.     size_t uncompressed_length;
    164.  
    165.     snappy_compress_out_t snappy_compress_output;
    166.     snappy_status ret;
    167.  
    168.     /* 获取 output */
    169.     printf("Received output from target\n");
    170.     HG_Get_output(handle, &snappy_compress_output);
    171.  
    172.     /*获取  output parameters */
    173.     ret = snappy_compress_output.ret;
    174.     compressed_length = snappy_compress_output.compressed_length;
    175.     compressed = snappy_compress_rpc_args->compressed;
    176.     input = snappy_compress_rpc_args->input;
    177.     source_length = snappy_compress_rpc_args->input_length;
    178.  
    179.     /* Check ret */
    180.     if (ret != SNAPPY_OK) {
    181.         fprintf(stderr, "Error: snappy_compressed failed with ret %d\n", ret);
    182.     }
    183.  
    184.     /*  输出数据现在在bulk缓冲区中  */
    185.     printf("Compressed buffer length is: %zu\n", compressed_length);
    186.     print_buf(5, (int *) compressed);
    187.     if (snappy_validate_compressed_buffer(compressed, compressed_length) ==
    188.         SNAPPY_OK) {
    189.         printf("Compressed buffer validated: compressed successfully\n");
    190.     }
    191.  
    192.     uncompressed_length = source_length * sizeof(int);
    193.     uncompressed = (int *) malloc(uncompressed_length);
    194.  
    195.     /* Uncompress data and check uncompressed_length */
    196.     printf("Uncompressing buffer...\n");
    197.     snappy_uncompress(compressed, compressed_length, (char *) uncompressed,
    198.         &uncompressed_length);
    199.     printf("Uncompressed buffer length is: %zu\n", uncompressed_length);
    200.     print_buf(20, uncompressed);
    201.  
    202.     /* Free output and handles */
    203.     HG_Free_output(handle, &snappy_compress_output);
    204.     HG_Bulk_free(snappy_compress_rpc_args->input_bulk_handle);
    205.     HG_Bulk_free(snappy_compress_rpc_args->compressed_bulk_handle);
    206.  
    207.     /* Free data */
    208.     free(uncompressed);
    209.     free(compressed);
    210.     free(input);
    211.     free(snappy_compress_rpc_args);
    212.  
    213.     /* We're done */
    214.     snappy_compress_done_g = HG_TRUE;
    215.  
    216.     return HG_SUCCESS;
    217. }
    218.

    服务端

    1. int
    2. main(void)
    3. {
    4.     const char *info_string = NULL;
    5.  
    6.     char self_addr_string[PATH_MAX];
    7.     hg_addr_t self_addr;
    8.     FILE *na_config = NULL;
    9.  
    10.     hg_class_t *hg_class;
    11.     hg_context_t *hg_context;
    12.     unsigned major;
    13.     unsigned minor;
    14.     unsigned patch;
    15.     hg_return_t hg_ret;
    16.     hg_size_t self_addr_string_size = PATH_MAX;
    17.  
    18.     HG_Version_get(&major, &minor, &patch);
    19.  
    20.     printf("Server running mercury version %u.%u.%u\n", major, minor, patch);
    21.  
    22.     /* Get info string */
    23.     /* bmi+tcp://localhost:port */
    24.     info_string = getenv("HG_PORT_NAME");
    25.     if (!info_string) {
    26.         fprintf(stderr, "HG_PORT_NAME environment variable must be set, "
    27.                         "e.g.:\nHG_PORT_NAME=\"tcp://127.0.0.1:22222\"\n");
    28.         exit(0);
    29.     }
    30.  
    31.     HG_Set_log_level("warning");
    32.  
    33.     /* Initialize Mercury with the desired network abstraction class */
    34.     hg_class = HG_Init(info_string, HG_TRUE);
    35.  
    36.     /* Get self addr to tell client about */
    37.     HG_Addr_self(hg_class, &self_addr);
    38.     HG_Addr_to_string(
    39.         hg_class, self_addr_string, &self_addr_string_size, self_addr);
    40.     HG_Addr_free(hg_class, self_addr);
    41.     printf("Server address is: %s\n", self_addr_string);
    42.  
    43.     /* Write addr to a file */
    44.     na_config = fopen(TEMP_DIRECTORY CONFIG_FILE_NAME, "w+");
    45.     if (!na_config) {
    46.         fprintf(stderr, "Could not open config file from: %s\n",
    47.             TEMP_DIRECTORY CONFIG_FILE_NAME);
    48.         exit(0);
    49.     }
    50.     fprintf(na_config, "%s\n", self_addr_string);
    51.     fclose(na_config);
    52.  
    53.     /* Create HG context */
    54.     hg_context = HG_Context_create(hg_class);
    55.  
    56.     /* Register RPC */
    57.     snappy_compress_register(hg_class);
    58.  
    59.     /* Poke progress engine and check for events */
    60.     do {
    61.         unsigned int actual_count = 0;
    62.         do {
    63.             hg_ret = HG_Trigger(
    64.                 hg_context, 0 /* timeout */, 1 /* max count */, &actual_count);
    65.         } while ((hg_ret == HG_SUCCESS) && actual_count);
    66.  
    67.         /* Do not try to make progress anymore if we're done */
    68.         if (snappy_compress_done_target_g)
    69.             break;
    70.  
    71.         hg_ret = HG_Progress(hg_context, HG_MAX_IDLE_TIME);
    72.  
    73.     } while (hg_ret == HG_SUCCESS);
    74.  
    75.     /* Finalize */
    76.     HG_Context_destroy(hg_context);
    77.     HG_Finalize(hg_class);
    78.  
    79.     return EXIT_SUCCESS;
    80. }
    81.  
    82. hg_id_t
    83. snappy_compress_register(hg_class_t *hg_class)
    84. {
    85.     return MERCURY_REGISTER(hg_class, "snappy_compress", snappy_compress_in_t,
    86.         snappy_compress_out_t, snappy_compress_cb);
    87. }
    88.  
    89. /**
    90.  * 设置实际执行工作的例程的例程.
    91.  * 这个“handle”参数是传递给这个回调的唯一值，但Mercury例程允许我们查询有关调用上下文的信息. */
    92. static hg_return_t
    93. snappy_compress_cb(hg_handle_t handle)
    94. {
    95.     struct snappy_transfer_args *snappy_transfer_args;
    96.     size_t input_length;
    97.  
    98.     snappy_transfer_args = (struct snappy_transfer_args *) malloc(
    99.         sizeof(struct snappy_transfer_args));
    100.     snappy_transfer_args->handle = handle;
    101.  
    102.     /* 获取从源经过HG_Forward() 发送过来的输入参数 */
    103.     HG_Get_input(handle, &snappy_transfer_args->snappy_compress_input);
    104.  
    105.     /* Now set up the bulk transfer and get the input length */
    106.     input_length = HG_Bulk_get_size(
    107.         snappy_transfer_args->snappy_compress_input.input_bulk_handle);
    108.  
    109.     /* bulk handle 基本上是一个指针，另外“handle”可以引用多个内存区域. */
    110.     HG_Bulk_create(HG_Get_info(handle)->hg_class, 1, NULL, &input_length,
    111.         HG_BULK_READWRITE, &snappy_transfer_args->local_input_bulk_handle);
    112.  
    113.     /* 将数据从origin的内存中拉到本端的内存中 */
    114.     /* 另一种方法是通过HG_Bulk_access，这将允许mercury在“co-resident”的情况下避免复制数据 */
    115.     HG_Bulk_transfer(HG_Get_info(handle)->context, snappy_pull_cb,
    116.         snappy_transfer_args, HG_BULK_PULL, HG_Get_info(handle)->addr,
    117.         snappy_transfer_args->snappy_compress_input.input_bulk_handle,
    118.         0,                                                /* origin */
    119.         snappy_transfer_args->local_input_bulk_handle, 0, /* local */
    120.         input_length, HG_OP_ID_IGNORE);
    121.  
    122.     return HG_SUCCESS;
    123. }
    124.  
    125. static hg_return_t
    126. snappy_pull_cb(const struct hg_cb_info *hg_cb_info)
    127. {
    128.     struct snappy_transfer_args *snappy_transfer_args =
    129.         (struct snappy_transfer_args *) hg_cb_info->arg;
    130.     hg_return_t ret = HG_SUCCESS;
    131.     void *input;
    132.     size_t input_length;
    133.     size_t source_length =
    134.         HG_Bulk_get_size(snappy_transfer_args->local_input_bulk_handle);
    135.  
    136.     /* 从本地handle获取指向输入缓冲区的指针input */
    137.     HG_Bulk_access(hg_cb_info->info.bulk.local_handle, 0, source_length,
    138.         HG_BULK_READ_ONLY, 1, &input, &input_length, NULL);
    139.     printf("Transferred input buffer of length: %zu\n", input_length);
    140.     print_buf(20, (int *) input);
    141.  
    142.     /* 为压缩输入数据分配压缩缓冲区 */
    143.     snappy_transfer_args->compressed_length =
    144.         snappy_max_compressed_length(input_length);
    145.     snappy_transfer_args->compressed =
    146.         malloc(snappy_transfer_args->compressed_length);
    147.  
    148.     /* Compress data */
    149.     printf("Compressing buffer...\n");
    150.     snappy_transfer_args->ret =
    151.         snappy_compress(input, input_length, snappy_transfer_args->compressed,
    152.             &snappy_transfer_args->compressed_length);
    153.     printf(
    154.         "Return value of snappy_compress is: %d\n", snappy_transfer_args->ret);
    155.     printf("Compressed buffer length is: %zu\n",
    156.         snappy_transfer_args->compressed_length);
    157.     print_buf(5, (int *) snappy_transfer_args->compressed);
    158.  
    159.     /* Free bulk handles */
    160.     HG_Bulk_free(snappy_transfer_args->local_input_bulk_handle);
    161.  
    162.     if (snappy_validate_compressed_buffer(snappy_transfer_args->compressed,
    163.             snappy_transfer_args->compressed_length) == SNAPPY_OK) {
    164.         printf("Compressed buffer validated: compressed successfully\n");
    165.     }
    166.  
    167.     /* 将压缩后的数据push回源节点 */
    168.     HG_Bulk_create(HG_Get_info(snappy_transfer_args->handle)->hg_class, 1,
    169.         &snappy_transfer_args->compressed,
    170.         &snappy_transfer_args->compressed_length, HG_BULK_READ_ONLY,
    171.         &snappy_transfer_args->local_compressed_bulk_handle);
    172.  
    173.     HG_Bulk_transfer(HG_Get_info(snappy_transfer_args->handle)->context,
    174.         snappy_push_cb, snappy_transfer_args, HG_BULK_PUSH,
    175.         HG_Get_info(snappy_transfer_args->handle)->addr,
    176.         snappy_transfer_args->snappy_compress_input.compressed_bulk_handle,
    177.         0,                                                     /* origin */
    178.         snappy_transfer_args->local_compressed_bulk_handle, 0, /* local */
    179.         snappy_transfer_args->compressed_length, HG_OP_ID_IGNORE);
    180.  
    181.     return ret;
    182. }
    183.  
    184.  
    185. /* 压缩数据推送到源节点后回调，给源节点HG_Forward请求回消息*/
    186.  
    187. static hg_return_t
    188. snappy_push_cb(const struct hg_cb_info *hg_cb_info)
    189. {
    190.     struct snappy_transfer_args *snappy_transfer_args =
    191.         (struct snappy_transfer_args *) hg_cb_info->arg;
    192.     hg_return_t ret = HG_SUCCESS;
    193.     snappy_compress_out_t snappy_compress_output;
    194.  
    195.     /* 设置输出参数以通知源节点 */
    196.     snappy_compress_output.ret = snappy_transfer_args->ret;
    197.     snappy_compress_output.compressed_length =
    198.         snappy_transfer_args->compressed_length;
    199.     printf("Transferred compressed buffer of length %zu\n",
    200.         snappy_transfer_args->compressed_length);
    201.  
    202.     printf("Sending output parameters back to origin\n");
    203.     HG_Respond(snappy_transfer_args->handle, snappy_compress_done_cb, NULL,
    204.         &snappy_compress_output);
    205.  
    206.     /* Free bulk handles */
    207.     printf("Freeing resources\n");
    208.     HG_Bulk_free(snappy_transfer_args->local_compressed_bulk_handle);
    209.     free(snappy_transfer_args->compressed);
    210.  
    211.     /* Free input */
    212.     HG_Free_input(snappy_transfer_args->handle,
    213.         &snappy_transfer_args->snappy_compress_input);
    214.  
    215.     /* Destroy handle (no longer need it, safe because of reference count) */
    216.     HG_Destroy(snappy_transfer_args->handle);
    217.     free(snappy_transfer_args);
    218.  
    219.     return ret;
    220. }
    221.  
    222. static hg_return_t
    223. snappy_compress_done_cb(const struct hg_cb_info *callback_info)
    224. {
    225.     /* We're done */
    226.     snappy_compress_done_target_g = HG_TRUE;
    227.  
    228.     return callback_info->ret;
    229. }

  • 相关阅读:
    7、DVWA——SQL盲注
    【MySQL —— 数据库约束】
    功能测试进阶建议，学习思路讲解
    shopee知虾数据：提升Shopee店铺运营效果必备工具—知虾数据工具
    【翻译】Raft 共识算法：集群成员变更
    Codeforces Round #779 (Div. 2)
    MySQL底层知识总结
    轻松学会结构栈
    1069 The Black Hole of Numbers
    【目标检测】Faster R-CNN 论文复现代码（含源代码）
  • 原文地址：https://blog.csdn.net/cyq6239075/article/details/126752268