• 轻量级的低级线程和任务框架---Argobots

    Argobots是一个轻量级的低级线程和任务框架。

    一、安装

    1. #tar xzf argobots.tar.gz
    2. #cd argobots
    3. #./autogen.sh
    4.  
    5. //配置
    6. #./configure --prefix=/home/USERNAME/argobots-install 2>&1 | tee c.txt
    7.  
    8. //--prefix=/home/USERNAME/argobots-install 是安装路径
    9.  
    10. //构建
    11. #make 2>&1 | tee m.txt
    12.  
    13. //如果出现问题，请执行make clean，然后在V=1时再次运行make。
    14. #make V=1 2>&1 | tee m.txt
    15.  
    16. //安装
    17. #make install 2>&1 | tee mi.txt

    （1）可以将Argobot测试套件打包到ArgobotsDistribution中，然后执行：

    make check

    （2）还可以在示例目录中运行Argobots示例：

    make check

    （3）可以使用下列命令查看配置选项

    ./configure --help

    对于性能测试，建议使用以下标志：

    ./configure --enable-perf-opt --enable-affinity --disable-checks

    对于调试，建议使用以下标志：

    ./configure --enable-fast=O0 --enable-debug=most

    二、使用实例1(普通用法)

    1. #include 
    2. #include 
    3. #include 
    4. #include 
    5. #include 
    6.  
    7. #define DEFAULT_NUM_XSTREAMS 2
    8. #define DEFAULT_NUM_THREADS 8
    9.  
    10. typedef struct {
    11.     int tid;
    12. } thread_arg_t;
    13.  
    14. void hello_world(void *arg)
    15. {
    16.     int tid = ((thread_arg_t *)arg)->tid;
    17.     printf("Hello world! (thread = %d)\n", tid);
    18. }
    19.  
    20. int main(int argc, char **argv)
    21. {
    22.     int i;
    23.     /* Read arguments. */
    24.     int num_xstreams = DEFAULT_NUM_XSTREAMS;
    25.     int num_threads = DEFAULT_NUM_THREADS;
    26.     while (1) {
    27.         int opt = getopt(argc, argv, "he:n:");
    28.         if (opt == -1)
    29.             break;
    30.         switch (opt) {
    31.             case 'e':
    32.                 num_xstreams = atoi(optarg);
    33.                 break;
    34.             case 'n':
    35.                 num_threads = atoi(optarg);
    36.                 break;
    37.             case 'h':
    38.             default:
    39.                 printf("Usage: ./hello_world [-e NUM_XSTREAMS] "
    40.                        "[-n NUM_THREADS]\n");
    41.                 return -1;
    42.         }
    43.     }
    44.     if (num_xstreams <= 0)
    45.         num_xstreams = 1;
    46.     if (num_threads <= 0)
    47.         num_threads = 1;
    48.  
    49.     /* Allocate memory. */
    50.     ABT_xstream *xstreams =
    51.         (ABT_xstream *)malloc(sizeof(ABT_xstream) * num_xstreams);
    52.     ABT_pool *pools = (ABT_pool *)malloc(sizeof(ABT_pool) * num_xstreams);
    53.     ABT_thread *threads =
    54.         (ABT_thread *)malloc(sizeof(ABT_thread) * num_threads);
    55.     thread_arg_t *thread_args =
    56.         (thread_arg_t *)malloc(sizeof(thread_arg_t) * num_threads);
    57.  
    58.     /* Initialize Argobots. */
    59.     ABT_init(argc, argv);
    60.  
    61.     /* Get a primary execution stream. */
    62.     ABT_xstream_self(&xstreams[0]);
    63.  
    64.     /* Create secondary execution streams. */
    65.     for (i = 1; i < num_xstreams; i++) {
    66.         ABT_xstream_create(ABT_SCHED_NULL, &xstreams[i]);
    67.     }
    68.  
    69.     /* Get default pools. */
    70.     for (i = 0; i < num_xstreams; i++) {
    71.         ABT_xstream_get_main_pools(xstreams[i], 1, &pools[i]);
    72.     }
    73.  
    74.     /* Create ULTs. */
    75.     for (i = 0; i < num_threads; i++) {
    76.         int pool_id = i % num_xstreams;
    77.         thread_args[i].tid = i;
    78.         ABT_thread_create(pools[pool_id], hello_world, &thread_args[i],
    79.                           ABT_THREAD_ATTR_NULL, &threads[i]);
    80.     }
    81.  
    82.     /* Join and free ULTs. */
    83.     for (i = 0; i < num_threads; i++) {
    84.         ABT_thread_free(&threads[i]);
    85.     }
    86.  
    87.     /* Join and free secondary execution streams. */
    88.     for (i = 1; i < num_xstreams; i++) {
    89.         ABT_xstream_join(xstreams[i]);
    90.         ABT_xstream_free(&xstreams[i]);
    91.     }
    92.  
    93.     /* Finalize Argobots. */
    94.     ABT_finalize();
    95.  
    96.     /* Free allocated memory. */
    97.     free(xstreams);
    98.     free(pools);
    99.     free(threads);
    100.     free(thread_args);
    101.  
    102.     return 0;
    103. }

    使用实例2(带调度器和绑核)

    1. //入口函数
    2. int dispatch_request(crt_rpc_t* rpc) {
    3.   if (!g_umds_context->dispatcher.is_inited()) {
    4.     derr << "dispatcher is not inited" << dendl;
    5.     return -ERR_UMD_NOT_INITIAL;
    6.   }
    7.  
    8.   //生成请求req
    9.   req* request = new req(rpc);
    10.   //获取xstream执行流
    11.   XstreamInstance* xstream = g_umds_context->dispatcher.get_handler(request);
    12.  
    13.  
    14.   request->xstream = xstream;
    15.   request->cli_req_id = *((req_id_t*)rpc->cr_input);
    16.   //挂载执行函数
    17.   ABT_thread_create(request->xstream->task_pool, handle_request, request,
    18. 	            ABT_THREAD_ATTR_NULL, NULL);
    19.   return UMD_SUCCESS;
    20. }
    21.  
    22. XstreamInstance::XstreamInstance(SrvDispatcher* _dispatcher, int index) {
    23.   id = index;
    24.   total = 0;
    25.   dispatcher = _dispatcher;
    26.   step = 0;
    27.   //获取cpu id
    28.   int cpu_index = index % dispatcher->cpus.size();
    29.   //为绑核做准备
    30.   cpu = dispatcher->cpus[cpu_index];
    31.   pthread_spin_init(&data_lock, PTHREAD_PROCESS_PRIVATE);
    32. }
    33.  
    34. int SrvDispatcher::init() {
    35.   int ret = 0;
    36.   XstreamInstance* tmp_xstream;
    37.  
    38.   // init argobot
    39.   ret = ABT_init(0, NULL);
    40.   if (ret != ABT_SUCCESS) {
    41.     derr << "ABT init failed: " << ret << dendl;  
    42.     goto out_abt_init; 
    43.   }
    44.   dout(INFO) << "abt_init success" << dendl;
    45.  
    46.   // create xstream
    47.   for (uint32_t i =0; i_conf->xstream_num; i++) {
    48.     tmp_xstream = new XstreamInstance(this, i);
    49.     //xstream初始化，里面有创建pool，创建调度器、创建xstream
    50.     ret = tmp_xstream->init();
    51.     if (ret == UMD_SUCCESS) {
    52.     	xstreams.push_back(tmp_xstream);
    53.     } else {
    54. 	derr << "init xstream failed: " << get_err_desc(ret) << dendl;
    55.     	delete tmp_xstream;
    56.     	goto out_start_xstream;
    57.     }
    58.   }
    59.   dout(INFO) << "xstreams init success" << dendl;
    60.  
    61.   pthread_rwlock_init(&xstream_lock, NULL);
    62.   is_init = true;
    63.   return UMD_SUCCESS;
    64. }
    65.  
    66. int XstreamInstance::init_scheduler() {
    67.   ABT_sched_config config;
    68.   ABT_sched_config_var event_freq = {
    69.     .idx = 0,
    70.     .type = ABT_SCHED_CONFIG_INT
    71.   };
    72.   ABT_sched_config_var dx_ptr = {
    73.     .idx = 1,
    74.     .type = ABT_SCHED_CONFIG_PTR
    75.   };
    76.   ABT_sched_config_var timeout = {
    77.     .idx = 2,
    78.     .type = ABT_SCHED_CONFIG_INT
    79.   };
    80.   ABT_sched_def sched_def = {
    81.     .type = ABT_SCHED_TYPE_ULT,
    82.     .init = sched_init,
    83.     .run = sched_run,
    84.     .free = sched_free,
    85.     .get_migr_pool = NULL
    86.   };
    87.   int rc;
    88.   //创建pool
    89.   rc = ABT_pool_create_basic(ABT_POOL_FIFO_WAIT, ABT_POOL_ACCESS_MPMC,
    90.   				   ABT_TRUE, &task_pool);
    91.   dout(INFO) << "xstream " << id << " pool create success" << dendl;
    92.  
    93.   //创建调度器的配置项
    94.   rc = ABT_sched_config_create(&config, event_freq, g_cds_context->_conf->umds_event_check_interval, dx_ptr,
    95.   				this, timeout, g_cds_context->_conf->xstream_pool_timeout,
    96.   		   		ABT_sched_config_var_end);
    97.  
    98.   dout(INFO) << "xstream " << id << " scheduler config create success" << dendl;
    99.    //创建调度器
    100.   rc = ABT_sched_create(&sched_def, 1, &task_pool, config,
    101.   		      &scheduler);
    102.   ABT_sched_config_free(&config);
    103.  
    104.   dout(INFO) << "xstream " << id << " create scheduler success" << dendl;
    105.   return UMD_SUCCESS;	
    106. }
    107.  
    108. int XstreamInstance::init() {
    109.   int rc = init_scheduler();
    110.  
    111.   dout(INFO) << "scheduler init success" << dendl;
    112.   step = SRV_XSTREAM_STEP_SCHED;
    113.  
    114.   //创建xstream
    115.   rc = ABT_xstream_create_with_rank(scheduler, id + 1,
    116.   				  &xstream);
    117.  
    118.   dout(INFO) << "create xstream success" << dendl;
    119.   step = SRV_XSTREAM_STEP_XSTREAM;
    120.     //执行绑核操作
    121.   rc = ABT_thread_create(task_pool,
    122.   		       bind_xstream_cpu, this, ABT_THREAD_ATTR_NULL,
    123.   		       NULL);
    124.  
    125.   dout(INFO) << "create bind cpu task success" << dendl;
    126.   step = SRV_XSTREAM_STEP_BIND_CPU;
    127.   return UMD_SUCCESS;
    128. }
    129.  
    130.  
    131. static void bind_xstream_cpu(void *arg) {
    132.   XstreamInstance *srv_xstream = (XstreamInstance*)arg;
    133.   cds_assert(srv_xstream != NULL);
    134.   
    135.   int rc = hwloc_set_cpubind(srv_xstream->dispatcher->topology, srv_xstream->cpu, HWLOC_CPUBIND_THREAD);
    136.   if (rc) {
    137. 	derr << "xstream " << srv_xstream->id << " failed to set cpu affinity: " << rc << dendl;
    138.   	return;
    139.   }
    140.   dout(INFO) << "xstream " << srv_xstream->id << " bind cpu success" << dendl;
    141.   
    142.   // memory not need bind, because it will allocate memory from the bind cpu default
    143. }

  • 相关阅读:
    面试：HashMap
    为什么要考一级建造师，一建证书含金量有多高？
    超千万下载量的NPM包遭黑客攻击，美国监管机构紧急警告
    java基于Vue的社团管理系统
    服务器端优化、NIO、非阻塞的HTTP服务器
    Android桌面回到应用 重新打开 android.intent.action.MAIN 的界面
    极空间变身监控录像机，搭配Onvif摄像头，实现实时观看和录制视频回放功能
    java基于微信小程序的智能停车场管理系统+ssm+uinapp+Mysql+计算机毕业设计
    ftp传送文件脚本
    Maven配置MAVEN_OPTS
  • 原文地址：https://blog.csdn.net/cyq6239075/article/details/126745140