ARGOBOTS  29b0d4cc91ca184a8278ddf690fd0bcb2119000e
 All Data Structures Files Functions Variables Typedefs Enumerations Enumerator Macros Groups
pool.c
Go to the documentation of this file.
1 /* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
2 /*
3  * See COPYRIGHT in top-level directory.
4  */
5 
6 #include "abti.h"
7 
9  ABT_bool automatic, ABTI_pool **pp_newpool);
10 
32  ABT_pool *newpool)
33 {
34  ABTI_pool *p_newpool;
35  int abt_errno = pool_create(def, config, ABT_FALSE, &p_newpool);
36  ABTI_CHECK_ERROR(abt_errno);
37 
38  *newpool = ABTI_pool_get_handle(p_newpool);
39  return ABT_SUCCESS;
40 }
41 
57  ABT_bool automatic, ABT_pool *newpool)
58 {
59  ABTI_pool *p_newpool;
60  int abt_errno = ABTI_pool_create_basic(kind, access, automatic, &p_newpool);
61  ABTI_CHECK_ERROR(abt_errno);
62 
63  *newpool = ABTI_pool_get_handle(p_newpool);
64  return ABT_SUCCESS;
65 }
66 
76 {
77  ABT_pool h_pool = *pool;
78  ABTI_pool *p_pool = ABTI_pool_get_ptr(h_pool);
79 
80  ABTI_CHECK_TRUE(p_pool != NULL && h_pool != ABT_POOL_NULL,
82  ABTI_pool_free(p_pool);
83 
84  *pool = ABT_POOL_NULL;
85  return ABT_SUCCESS;
86 }
87 
98 {
99  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
100  ABTI_CHECK_NULL_POOL_PTR(p_pool);
101 
102  *access = p_pool->access;
103  return ABT_SUCCESS;
104 }
105 
119 int ABT_pool_get_total_size(ABT_pool pool, size_t *size)
120 {
121  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
122  ABTI_CHECK_NULL_POOL_PTR(p_pool);
123 
124  *size = ABTI_pool_get_total_size(p_pool);
125  return ABT_SUCCESS;
126 }
127 
140 int ABT_pool_get_size(ABT_pool pool, size_t *size)
141 {
142  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
143  ABTI_CHECK_NULL_POOL_PTR(p_pool);
144 
145  *size = ABTI_pool_get_size(p_pool);
146  return ABT_SUCCESS;
147 }
148 
158 int ABT_pool_pop(ABT_pool pool, ABT_unit *p_unit)
159 {
160  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
161  ABTI_CHECK_NULL_POOL_PTR(p_pool);
162 
163  *p_unit = ABTI_pool_pop(p_pool);
164  return ABT_SUCCESS;
165 }
166 
187 int ABT_pool_pop_wait(ABT_pool pool, ABT_unit *p_unit, double time_secs)
188 {
189  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
190  ABTI_CHECK_NULL_POOL_PTR(p_pool);
191 
192  *p_unit = ABTI_pool_pop_wait(p_pool, time_secs);
193  return ABT_SUCCESS;
194 }
195 
196 int ABT_pool_pop_timedwait(ABT_pool pool, ABT_unit *p_unit, double abstime_secs)
197 {
198  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
199  ABTI_CHECK_NULL_POOL_PTR(p_pool);
200 
201  *p_unit = ABTI_pool_pop_timedwait(p_pool, abstime_secs);
202  return ABT_SUCCESS;
203 }
204 
215 {
216  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
217  ABTI_CHECK_NULL_POOL_PTR(p_pool);
218 
220 
221  /* Save the producer ES information in the pool */
222  ABTI_pool_push(p_pool, unit);
223  return ABT_SUCCESS;
224 }
225 
236 {
237  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
238  ABTI_CHECK_NULL_POOL_PTR(p_pool);
239 
240  int abt_errno = ABTI_pool_remove(p_pool, unit);
241  ABTI_CHECK_ERROR(abt_errno);
242  return ABT_SUCCESS;
243 }
244 
264 int ABT_pool_print_all(ABT_pool pool, void *arg,
265  void (*print_fn)(void *, ABT_unit))
266 {
267  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
268  ABTI_CHECK_NULL_POOL_PTR(p_pool);
269  if (!p_pool->p_print_all) {
271  }
272  p_pool->p_print_all(pool, arg, print_fn);
273  return ABT_SUCCESS;
274 }
275 
289 {
290  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
291  ABTI_CHECK_NULL_POOL_PTR(p_pool);
292 
293  p_pool->data = data;
294  return ABT_SUCCESS;
295 }
296 
309 int ABT_pool_get_data(ABT_pool pool, void **data)
310 {
311  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
312  ABTI_CHECK_NULL_POOL_PTR(p_pool);
313 
314  *data = p_pool->data;
315  return ABT_SUCCESS;
316 }
317 
337 {
338  ABTI_local *p_local = ABTI_local_get_local();
339 
340  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
341  ABTI_CHECK_NULL_POOL_PTR(p_pool);
342 
343  ABTI_sched *p_sched = ABTI_sched_get_ptr(sched);
344  ABTI_CHECK_NULL_SCHED_PTR(p_sched);
345 
346  /* Mark the scheduler as it is used in pool */
348  p_sched->used = ABTI_SCHED_IN_POOL;
349 
350  /* In both ABT_SCHED_TYPE_ULT and ABT_SCHED_TYPE_TASK cases, we use ULT-type
351  * scheduler to reduce the code maintenance cost. ABT_SCHED_TYPE_TASK
352  * should be removed in the future. */
353  int abt_errno = ABTI_ythread_create_sched(p_local, p_pool, p_sched);
354  ABTI_CHECK_ERROR(abt_errno);
355  return ABT_SUCCESS;
356 }
357 
369 int ABT_pool_get_id(ABT_pool pool, int *id)
370 {
371  ABTI_pool *p_pool = ABTI_pool_get_ptr(pool);
372  ABTI_CHECK_NULL_POOL_PTR(p_pool);
373 
374  *id = (int)p_pool->id;
375  return ABT_SUCCESS;
376 }
377 
378 /*****************************************************************************/
379 /* Private APIs */
380 /*****************************************************************************/
381 
383  ABT_pool_access access,
384  ABT_bool automatic,
385  ABTI_pool **pp_newpool)
386 {
387  int abt_errno;
388  ABT_pool_def def;
389 
390  switch (kind) {
391  case ABT_POOL_FIFO:
392  abt_errno = ABTI_pool_get_fifo_def(access, &def);
393  break;
394  case ABT_POOL_FIFO_WAIT:
395  abt_errno = ABTI_pool_get_fifo_wait_def(access, &def);
396  break;
397  default:
398  abt_errno = ABT_ERR_INV_POOL_KIND;
399  break;
400  }
401  ABTI_CHECK_ERROR(abt_errno);
402 
403  abt_errno = pool_create(&def, ABT_POOL_CONFIG_NULL, automatic, pp_newpool);
404  ABTI_CHECK_ERROR(abt_errno);
405  return ABT_SUCCESS;
406 }
407 
409 {
410  LOG_DEBUG("[P%" PRIu64 "] freed\n", p_pool->id);
411  ABT_pool h_pool = ABTI_pool_get_handle(p_pool);
412  p_pool->p_free(h_pool);
413  ABTU_free(p_pool);
414 }
415 
416 void ABTI_pool_print(ABTI_pool *p_pool, FILE *p_os, int indent)
417 {
418  if (p_pool == NULL) {
419  fprintf(p_os, "%*s== NULL POOL ==\n", indent, "");
420  } else {
421  const char *access;
422 
423  switch (p_pool->access) {
425  access = "PRIV";
426  break;
428  access = "SPSC";
429  break;
431  access = "MPSC";
432  break;
434  access = "SPMC";
435  break;
437  access = "MPMC";
438  break;
439  default:
440  access = "UNKNOWN";
441  break;
442  }
443 
444  fprintf(p_os,
445  "%*s== POOL (%p) ==\n"
446  "%*sid : %" PRIu64 "\n"
447  "%*saccess : %s\n"
448  "%*sautomatic : %s\n"
449  "%*snum_scheds : %d\n"
450  "%*ssize : %zu\n"
451  "%*snum_blocked : %d\n"
452  "%*snum_migrations: %d\n"
453  "%*sdata : %p\n",
454  indent, "", (void *)p_pool, indent, "", p_pool->id, indent, "",
455  access, indent, "",
456  (p_pool->automatic == ABT_TRUE) ? "TRUE" : "FALSE", indent, "",
457  ABTD_atomic_acquire_load_int32(&p_pool->num_scheds), indent, "",
458  ABTI_pool_get_size(p_pool), indent, "",
461  indent, "", p_pool->data);
462  }
463  fflush(p_os);
464 }
465 
468 {
469  ABTD_atomic_release_store_uint64(&g_pool_id, 0);
470 }
471 
472 /*****************************************************************************/
473 /* Internal static functions */
474 /*****************************************************************************/
475 
476 static inline uint64_t pool_get_new_id(void);
478  ABT_bool automatic, ABTI_pool **pp_newpool)
479 {
480  int abt_errno;
481  ABTI_pool *p_pool;
482  abt_errno = ABTU_malloc(sizeof(ABTI_pool), (void **)&p_pool);
483  ABTI_CHECK_ERROR(abt_errno);
484 
485  p_pool->access = def->access;
486  p_pool->automatic = automatic;
490  p_pool->data = NULL;
491 
492  /* Set up the pool functions from def */
493  p_pool->u_get_type = def->u_get_type;
494  p_pool->u_get_thread = def->u_get_thread;
495  p_pool->u_get_task = def->u_get_task;
496  p_pool->u_is_in_pool = def->u_is_in_pool;
498  p_pool->u_create_from_task = def->u_create_from_task;
499  p_pool->u_free = def->u_free;
500  p_pool->p_init = def->p_init;
501  p_pool->p_get_size = def->p_get_size;
502  p_pool->p_push = def->p_push;
503  p_pool->p_pop = def->p_pop;
504  p_pool->p_pop_wait = def->p_pop_wait;
505  p_pool->p_pop_timedwait = def->p_pop_timedwait;
506  p_pool->p_remove = def->p_remove;
507  p_pool->p_free = def->p_free;
508  p_pool->p_print_all = def->p_print_all;
509  p_pool->id = pool_get_new_id();
510  LOG_DEBUG("[P%" PRIu64 "] created\n", p_pool->id);
511 
512  /* Configure the pool */
513  if (p_pool->p_init) {
514  abt_errno = p_pool->p_init(ABTI_pool_get_handle(p_pool), config);
515  if (abt_errno != ABT_SUCCESS) {
516  ABTU_free(p_pool);
517  return abt_errno;
518  }
519  }
520  *pp_newpool = p_pool;
521  return ABT_SUCCESS;
522 }
523 
524 static inline uint64_t pool_get_new_id(void)
525 {
526  return (uint64_t)ABTD_atomic_fetch_add_uint64(&g_pool_id, 1);
527 }
ABTD_atomic_int32 num_scheds
Definition: abti.h:285
ABT_pool_print_all_fn p_print_all
Definition: abti.h:309
static ABTD_atomic_uint64 g_pool_id
Definition: pool.c:466
struct ABT_unit_opaque * ABT_unit
Definition: abt.h:337
ABT_bool automatic
Definition: abti.h:283
int ABT_pool_pop(ABT_pool pool, ABT_unit *unit) ABT_API_PUBLIC
Pop a unit from the target pool.
Definition: pool.c:158
#define ABTI_CHECK_NULL_SCHED_PTR(p)
Definition: abti_error.h:184
static ABT_pool ABTI_pool_get_handle(ABTI_pool *p_pool)
Definition: abti_pool.h:26
ABT_unit_get_task_fn u_get_task
Definition: abt.h:494
static int32_t ABTD_atomic_acquire_load_int32(const ABTD_atomic_int32 *ptr)
Definition: abtd_atomic.h:780
#define ABT_POOL_NULL
Definition: abt.h:413
ABT_unit_get_type_fn u_get_type
Definition: abti.h:292
static ABT_unit ABTI_pool_pop(ABTI_pool *p_pool)
Definition: abti_pool.h:108
ABT_pool_init_fn p_init
Definition: abt.h:501
ABT_unit_create_from_thread_fn u_create_from_thread
Definition: abti.h:296
static void ABTD_atomic_release_store_uint64(ABTD_atomic_uint64 *ptr, uint64_t val)
Definition: abtd_atomic.h:971
#define ABTD_ATOMIC_UINT64_STATIC_INITIALIZER(val)
Definition: abtd_atomic.h:59
struct ABT_sched_opaque * ABT_sched
Definition: abt.h:319
ABT_pool_pop_timedwait_fn p_pop_timedwait
Definition: abti.h:306
void ABTI_pool_free(ABTI_pool *p_pool)
Definition: pool.c:408
int ABT_pool_get_size(ABT_pool pool, size_t *size) ABT_API_PUBLIC
Return the size of a pool.
Definition: pool.c:140
static ABTU_ret_err int pool_create(ABT_pool_def *def, ABT_pool_config config, ABT_bool automatic, ABTI_pool **pp_newpool)
Definition: pool.c:477
struct ABTI_local ABTI_local
Definition: abti.h:101
#define ABT_ERR_INV_POOL
Definition: abt.h:76
#define ABT_UNIT_NULL
Definition: abt.h:415
ABTU_ret_err int ABTI_ythread_create_sched(ABTI_local *p_local, ABTI_pool *p_pool, ABTI_sched *p_sched)
Definition: thread.c:1452
static ABT_unit ABTI_pool_pop_timedwait(ABTI_pool *p_pool, double abstime_secs)
Definition: abti_pool.h:97
int ABT_pool_pop_wait(ABT_pool pool, ABT_unit *unit, double time_secs) ABT_API_PUBLIC
Pop a unit from the target pool with wait.
Definition: pool.c:187
ABTU_ret_err int ABTI_pool_get_fifo_wait_def(ABT_pool_access access, ABT_pool_def *p_def)
Definition: fifo_wait.c:42
static ABT_unit ABTI_pool_pop_wait(ABTI_pool *p_pool, double time_secs)
Definition: abti_pool.h:87
int ABT_bool
Definition: abt.h:373
ABT_unit_get_thread_fn u_get_thread
Definition: abt.h:493
ABT_pool_pop_fn p_pop
Definition: abt.h:504
ABT_pool_access access
Definition: abt.h:489
ABT_pool_push_fn p_push
Definition: abti.h:303
ABT_unit_get_task_fn u_get_task
Definition: abti.h:294
struct ABT_pool_opaque * ABT_pool
Definition: abt.h:329
ABT_unit_is_in_pool_fn u_is_in_pool
Definition: abt.h:495
ABT_pool_push_fn p_push
Definition: abt.h:503
ABT_unit_get_thread_fn u_get_thread
Definition: abti.h:293
static uint64_t ABTD_atomic_fetch_add_uint64(ABTD_atomic_uint64 *ptr, uint64_t v)
Definition: abtd_atomic.h:411
ABT_pool_get_size_fn p_get_size
Definition: abti.h:302
ABT_unit_create_from_task_fn u_create_from_task
Definition: abti.h:297
ABT_unit_is_in_pool_fn u_is_in_pool
Definition: abti.h:295
int ABT_pool_get_data(ABT_pool pool, void **data) ABT_API_PUBLIC
Retrieve the specific data of the target user-defined pool.
Definition: pool.c:309
static ABTU_ret_err int ABTU_malloc(size_t size, void **p_ptr)
Definition: abtu.h:129
ABT_pool_pop_wait_fn p_pop_wait
Definition: abti.h:305
#define ABT_FALSE
Definition: abt.h:285
int ABT_pool_get_id(ABT_pool pool, int *id) ABT_API_PUBLIC
Get the ID of the target pool.
Definition: pool.c:369
#define ABTI_CHECK_TRUE(cond, abt_errno)
Definition: abti_error.h:137
ABT_pool_remove_fn p_remove
Definition: abti.h:307
ABT_pool_pop_fn p_pop
Definition: abti.h:304
static ABTI_pool * ABTI_pool_get_ptr(ABT_pool pool)
Definition: abti_pool.h:11
int ABT_pool_set_data(ABT_pool pool, void *data) ABT_API_PUBLIC
Set the specific data of the target user-defined pool.
Definition: pool.c:288
ABT_pool_free_fn p_free
Definition: abt.h:510
static ABTU_ret_err int ABTI_pool_remove(ABTI_pool *p_pool, ABT_unit unit)
Definition: abti_pool.h:80
ABT_unit_free_fn u_free
Definition: abti.h:298
#define ABT_ERR_INV_POOL_KIND
Definition: abt.h:77
int ABT_pool_free(ABT_pool *pool) ABT_API_PUBLIC
Free the given pool, and modify its value to ABT_POOL_NULL.
Definition: pool.c:75
int ABT_pool_print_all(ABT_pool pool, void *arg, void(*print_fn)(void *arg, ABT_unit)) ABT_API_PUBLIC
ABT_pool_init_fn p_init
Definition: abti.h:301
void * data
Definition: abti.h:288
#define ABT_SUCCESS
Definition: abt.h:64
ABT_unit_free_fn u_free
Definition: abt.h:498
ABT_pool_access
Definition: abt.h:161
#define ABT_TRUE
Definition: abt.h:284
static uint64_t pool_get_new_id(void)
Definition: pool.c:524
int ABT_pool_remove(ABT_pool pool, ABT_unit unit) ABT_API_PUBLIC
Remove a specified unit from the target pool.
Definition: pool.c:235
int ABT_pool_pop_timedwait(ABT_pool pool, ABT_unit *unit, double abstime_secs) ABT_DEPRECATED ABT_API_PUBLIC
Definition: pool.c:196
int ABT_pool_push(ABT_pool pool, ABT_unit unit) ABT_API_PUBLIC
Push a unit to the target pool.
Definition: pool.c:214
ABT_pool_access access
Definition: abti.h:282
int ABT_pool_add_sched(ABT_pool pool, ABT_sched sched) ABT_API_PUBLIC
Push a scheduler to a pool.
Definition: pool.c:336
ABT_pool_kind
Definition: abt.h:156
ABT_unit_create_from_task_fn u_create_from_task
Definition: abt.h:497
void ABTI_pool_print(ABTI_pool *p_pool, FILE *p_os, int indent)
Definition: pool.c:416
int ABT_pool_get_access(ABT_pool pool, ABT_pool_access *access) ABT_API_PUBLIC
Get the access type of target pool.
Definition: pool.c:97
int ABT_pool_create_basic(ABT_pool_kind kind, ABT_pool_access access, ABT_bool automatic, ABT_pool *newpool) ABT_API_PUBLIC
Create a new pool from a predefined type and return its handle through newpool.
Definition: pool.c:56
static void ABTI_pool_push(ABTI_pool *p_pool, ABT_unit unit)
Definition: abti_pool.h:65
#define ABTI_CHECK_NULL_POOL_PTR(p)
Definition: abti_error.h:175
ABTU_ret_err int ABTI_pool_get_fifo_def(ABT_pool_access access, ABT_pool_def *p_def)
Definition: fifo.c:47
ABTU_ret_err int ABTI_pool_create_basic(ABT_pool_kind kind, ABT_pool_access access, ABT_bool automatic, ABTI_pool **pp_newpool)
Definition: pool.c:382
static ABTI_sched * ABTI_sched_get_ptr(ABT_sched sched)
Definition: abti_sched.h:11
int ABT_pool_create(ABT_pool_def *def, ABT_pool_config config, ABT_pool *newpool) ABT_API_PUBLIC
Create a new pool and return its handle through newpool.
Definition: pool.c:31
#define LOG_DEBUG(fmt,...)
Definition: abti_log.h:26
ABT_unit_create_from_thread_fn u_create_from_thread
Definition: abt.h:496
ABT_pool_free_fn p_free
Definition: abti.h:308
#define ABTI_CHECK_ERROR(abt_errno)
Definition: abti_error.h:127
uint64_t id
Definition: abti.h:289
ABTD_atomic_int32 num_migrations
Definition: abti.h:287
#define ABT_ERR_UNIT
Definition: abt.h:100
#define ABT_ERR_INV_SCHED
Definition: abt.h:71
struct ABT_pool_config_opaque * ABT_pool_config
Definition: abt.h:331
ABTI_sched_used used
Definition: abti.h:260
ABT_unit_get_type_fn u_get_type
Definition: abt.h:492
ABT_pool_get_size_fn p_get_size
Definition: abt.h:502
void ABTI_pool_reset_id(void)
Definition: pool.c:467
ABT_pool_pop_wait_fn p_pop_wait
Definition: abt.h:505
static void ABTD_atomic_release_store_int32(ABTD_atomic_int32 *ptr, int32_t val)
Definition: abtd_atomic.h:935
static size_t ABTI_pool_get_total_size(ABTI_pool *p_pool)
Definition: abti_pool.h:138
int ABT_pool_get_total_size(ABT_pool pool, size_t *size) ABT_API_PUBLIC
Return the total size of a pool.
Definition: pool.c:119
static ABTI_local * ABTI_local_get_local(void)
Definition: abti_local.h:41
static void ABTU_free(void *ptr)
Definition: abtu.h:122
ABT_pool_remove_fn p_remove
Definition: abt.h:509
#define ABT_POOL_CONFIG_NULL
Definition: abt.h:414
ABT_pool_print_all_fn p_print_all
Definition: abt.h:511
#define ABT_ERR_POOL
Definition: abt.h:99
ABT_pool_pop_timedwait_fn p_pop_timedwait
Definition: abt.h:506
Definition: fifo.c:33
#define ABTU_ret_err
Definition: abtu.h:49
static size_t ABTI_pool_get_size(ABTI_pool *p_pool)
Definition: abti_pool.h:133
#define ABTI_HANDLE_ERROR(n)
Definition: abti_error.h:121
ABTD_atomic_int32 num_blocked
Definition: abti.h:286