ARGOBOTS  140a356fc09a44696eb3487150e459266f9b5405
abtd_stream.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 
8 static void *xstream_context_thread_func(void *arg)
9 {
11  void *(*thread_f)(void *) = p_ctx->thread_f;
12  void *p_arg = p_ctx->p_arg;
14  while (1) {
15  /* Execute a main execution stream function. */
16  thread_f(p_arg);
17  /* This thread has finished. */
18  ABT_bool restart;
19  pthread_mutex_lock(&p_ctx->state_lock);
20  /* If another execution stream is waiting for this thread completion,
21  * let's wake it up. */
23  pthread_cond_signal(&p_ctx->state_cond);
24  }
26  /* Wait for a request from ABTD_xstream_context_free() or
27  * ABTD_xstream_context_restart().
28  * The following loop is to deal with spurious wakeup. */
29  do {
30  pthread_cond_wait(&p_ctx->state_cond, &p_ctx->state_lock);
31  } while (p_ctx->state == ABTD_XSTREAM_CONTEXT_STATE_WAITING);
33  /* ABTD_xstream_context_free() terminates this thread. */
34  restart = ABT_FALSE;
35  } else {
36  /* ABTD_xstream_context_restart() restarts this thread */
39  restart = ABT_TRUE;
40  }
41  pthread_mutex_unlock(&p_ctx->state_lock);
42  if (!restart)
43  break;
44  }
45  return NULL;
46 }
47 
48 ABTU_ret_err int ABTD_xstream_context_create(void *(*f_xstream)(void *),
49  void *p_arg,
50  ABTD_xstream_context *p_ctx)
51 {
52  p_ctx->thread_f = f_xstream;
53  p_ctx->p_arg = p_arg;
55  pthread_mutex_init(&p_ctx->state_lock, NULL);
56  pthread_cond_init(&p_ctx->state_cond, NULL);
57  int ret = pthread_create(&p_ctx->native_thread, NULL,
59  ABTI_CHECK_TRUE(ret == 0, ABT_ERR_SYS);
60  return ABT_SUCCESS;
61 }
62 
64 {
65  /* Request termination */
66  pthread_mutex_lock(&p_ctx->state_lock);
69  pthread_cond_signal(&p_ctx->state_cond);
70  pthread_mutex_unlock(&p_ctx->state_lock);
71  /* Join the target thread. */
72  int ret = pthread_join(p_ctx->native_thread, NULL);
73  ABTI_ASSERT(ret == 0);
74  pthread_cond_destroy(&p_ctx->state_cond);
75  pthread_mutex_destroy(&p_ctx->state_lock);
76 }
77 
79 {
80  /* If not finished, sleep this thread. */
81  pthread_mutex_lock(&p_ctx->state_lock);
85  /* The following loop is to deal with spurious wakeup. */
86  do {
87  pthread_cond_wait(&p_ctx->state_cond, &p_ctx->state_lock);
88  } while (p_ctx->state == ABTD_XSTREAM_CONTEXT_STATE_REQ_JOIN);
89  }
91  pthread_mutex_unlock(&p_ctx->state_lock);
92 }
93 
95 {
96  /* Request restart */
97  pthread_mutex_lock(&p_ctx->state_lock);
100  pthread_cond_signal(&p_ctx->state_cond);
101  pthread_mutex_unlock(&p_ctx->state_lock);
102 }
103 
105 {
106  p_ctx->native_thread = pthread_self();
107 }
108 
110  int indent)
111 {
112  if (p_ctx == NULL) {
113  fprintf(p_os, "%*s== NULL XSTREAM CONTEXT ==\n", indent, "");
114  } else {
115  const char *state;
117  state = "RUNNING";
118  } else if (p_ctx->state == ABTD_XSTREAM_CONTEXT_STATE_WAITING) {
119  state = "WAITING";
120  } else if (p_ctx->state == ABTD_XSTREAM_CONTEXT_STATE_REQ_JOIN) {
121  state = "REQ_JOIN";
122  } else if (p_ctx->state == ABTD_XSTREAM_CONTEXT_STATE_REQ_TERMINATE) {
123  state = "REQ_TERMINATE";
124  } else {
125  state = "UNKNOWN";
126  }
127  fprintf(p_os,
128  "%*s== XSTREAM CONTEXT (%p) ==\n"
129  "%*sstate : %s\n",
130  indent, "", (void *)p_ctx, indent, "", state);
131  }
132  fflush(p_os);
133 }
ABTD_xstream_context::native_thread
pthread_t native_thread
Definition: abtd.h:22
ABT_ERR_SYS
#define ABT_ERR_SYS
Error code: an error related to system calls.
Definition: abt.h:385
ABT_bool
int ABT_bool
Boolean type.
Definition: abt.h:988
ABTD_XSTREAM_CONTEXT_STATE_WAITING
@ ABTD_XSTREAM_CONTEXT_STATE_WAITING
Definition: abtd.h:17
xstream_context_thread_func
static void * xstream_context_thread_func(void *arg)
Definition: abtd_stream.c:8
ABTD_xstream_context_revive
void ABTD_xstream_context_revive(ABTD_xstream_context *p_ctx)
Definition: abtd_stream.c:94
ABTD_XSTREAM_CONTEXT_STATE_RUNNING
@ ABTD_XSTREAM_CONTEXT_STATE_RUNNING
Definition: abtd.h:16
ABTD_xstream_context
Definition: abtd.h:21
ABTD_xstream_context_free
void ABTD_xstream_context_free(ABTD_xstream_context *p_ctx)
Definition: abtd_stream.c:63
ABTD_xstream_context_create
ABTU_ret_err int ABTD_xstream_context_create(void *(*f_xstream)(void *), void *p_arg, ABTD_xstream_context *p_ctx)
Definition: abtd_stream.c:48
ABTD_xstream_context_join
void ABTD_xstream_context_join(ABTD_xstream_context *p_ctx)
Definition: abtd_stream.c:78
ABTD_XSTREAM_CONTEXT_STATE_REQ_JOIN
@ ABTD_XSTREAM_CONTEXT_STATE_REQ_JOIN
Definition: abtd.h:18
ABTD_xstream_context::state_lock
pthread_mutex_t state_lock
Definition: abtd.h:26
abti.h
ABTD_XSTREAM_CONTEXT_STATE_REQ_TERMINATE
@ ABTD_XSTREAM_CONTEXT_STATE_REQ_TERMINATE
Definition: abtd.h:19
ABTD_xstream_context::state_cond
pthread_cond_t state_cond
Definition: abtd.h:27
ABTI_ASSERT
#define ABTI_ASSERT(cond)
Definition: abti_error.h:12
ABTD_xstream_context::p_arg
void * p_arg
Definition: abtd.h:24
ABT_SUCCESS
#define ABT_SUCCESS
Error code: the routine returns successfully.
Definition: abt.h:90
ABTU_ret_err
#define ABTU_ret_err
Definition: abtu.h:138
ABTD_xstream_context::state
ABTD_xstream_context_state state
Definition: abtd.h:25
ABT_TRUE
#define ABT_TRUE
True constant for ABT_bool.
Definition: abt.h:735
ABT_FALSE
#define ABT_FALSE
False constant for ABT_bool.
Definition: abt.h:737
ABTD_xstream_context::thread_f
void *(* thread_f)(void *)
Definition: abtd.h:23
ABTD_xstream_context_print
void ABTD_xstream_context_print(ABTD_xstream_context *p_ctx, FILE *p_os, int indent)
Definition: abtd_stream.c:109
ABTI_CHECK_TRUE
#define ABTI_CHECK_TRUE(cond, abt_errno)
Definition: abti_error.h:137
ABTD_xstream_context_set_self
void ABTD_xstream_context_set_self(ABTD_xstream_context *p_ctx)
Definition: abtd_stream.c:104