demux.c 6.76 KB
Newer Older
1
/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
2
3
4
5
6
7
8
9
10
11
12
/*
 *  (C) 2008 by Argonne National Laboratory.
 *      See COPYRIGHT in top-level directory.
 */

#include "demux.h"
#include "demux_internal.h"

int HYDT_dmxu_num_cb_fds = 0;
struct HYDT_dmxu_callback *HYDT_dmxu_cb_list = NULL;
struct HYDT_dmxu_fns HYDT_dmxu_fns = { 0 };
13

14
static int got_sigttin = 0;
15
16
17
18
19
20

#if defined(SIGTTIN) && defined(HAVE_ISATTY)
static void signal_cb(int sig)
{
    HYDU_FUNC_ENTER();

21
22
    if (sig == SIGTTIN)
        got_sigttin = 1;
23
24
25
26
27
28
    /* Ignore all other signals */

    HYDU_FUNC_EXIT();
    return;
}
#endif /* SIGTTIN and HAVE_ISATTY */
29
30
31
32
33
34
35

HYD_status HYDT_dmx_init(char **demux)
{
    HYD_status status = HYD_SUCCESS;

    HYDU_FUNC_ENTER();

36
    if (!(*demux)) {    /* user didn't specify anything */
37
38
#if defined HAVE_POLL
        HYDT_dmxu_fns.wait_for_event = HYDT_dmxu_poll_wait_for_event;
39
        HYDT_dmxu_fns.stdin_valid = HYDT_dmxu_poll_stdin_valid;
40
41
42
        *demux = HYDU_strdup("poll");
#elif defined HAVE_SELECT
        HYDT_dmxu_fns.wait_for_event = HYDT_dmxu_select_wait_for_event;
43
        HYDT_dmxu_fns.stdin_valid = HYDT_dmxu_select_stdin_valid;
44
45
46
47
48
49
        *demux = HYDU_strdup("select");
#endif /* HAVE_SELECT */
    }
    else if (!strcmp(*demux, "poll")) { /* user wants to use poll */
#if defined HAVE_POLL
        HYDT_dmxu_fns.wait_for_event = HYDT_dmxu_poll_wait_for_event;
50
        HYDT_dmxu_fns.stdin_valid = HYDT_dmxu_poll_stdin_valid;
51
52
#endif /* HAVE_POLL */
    }
53
    else if (!strcmp(*demux, "select")) {       /* user wants to use select */
54
55
#if defined HAVE_SELECT
        HYDT_dmxu_fns.wait_for_event = HYDT_dmxu_select_wait_for_event;
56
        HYDT_dmxu_fns.stdin_valid = HYDT_dmxu_select_stdin_valid;
57
58
59
#endif /* HAVE_SELECT */
    }

60
    if (HYDT_dmxu_fns.wait_for_event == NULL || HYDT_dmxu_fns.stdin_valid == NULL) {
61
        /* We couldn't find anything; return an error */
62
63
        HYDU_ERR_SETANDJUMP(status, HYD_INTERNAL_ERROR,
                            "cannot find an appropriate demux engine\n");
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
    }

  fn_exit:
    HYDU_FUNC_EXIT();
    return status;

  fn_fail:
    goto fn_exit;
}

HYD_status HYDT_dmx_register_fd(int num_fds, int *fd, HYD_event_t events, void *userp,
                                HYD_status(*callback) (int fd, HYD_event_t events,
                                                       void *userp))
{
    struct HYDT_dmxu_callback *cb_element, *run;
#if defined HAVE_ERROR_CHECKING
    int i, j;
#endif /* HAVE_ERROR_CHECKING */
    HYD_status status = HYD_SUCCESS;

    HYDU_FUNC_ENTER();

    HYDU_ASSERT(events, status);

#if defined HAVE_ERROR_CHECKING
    for (i = 0; i < num_fds; i++) {
        if (fd[i] < 0)
91
            HYDU_ERR_SETANDJUMP(status, HYD_INTERNAL_ERROR, "registering bad fd %d\n", fd[i]);
92
93
94
95
96

        cb_element = HYDT_dmxu_cb_list;
        while (cb_element) {
            for (j = 0; j < cb_element->num_fds; j++) {
                if (cb_element->fd[j] == fd[i]) {
97
98
                    HYDU_ERR_SETANDJUMP(status, HYD_INTERNAL_ERROR,
                                        "registering duplicate fd %d\n", fd[i]);
99
100
101
102
103
104
105
                }
            }
            cb_element = cb_element->next;
        }
    }
#endif /* HAVE_ERROR_CHECKING */

106
107
    HYDU_MALLOC(cb_element, struct HYDT_dmxu_callback *, sizeof(struct HYDT_dmxu_callback),
                status);
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
    cb_element->num_fds = num_fds;
    HYDU_MALLOC(cb_element->fd, int *, num_fds * sizeof(int), status);
    memcpy(cb_element->fd, fd, num_fds * sizeof(int));
    cb_element->events = events;
    cb_element->userp = userp;
    cb_element->callback = callback;
    cb_element->next = NULL;

    if (HYDT_dmxu_cb_list == NULL) {
        HYDT_dmxu_cb_list = cb_element;
    }
    else {
        run = HYDT_dmxu_cb_list;
        while (run->next)
            run = run->next;
        run->next = cb_element;
    }

    HYDT_dmxu_num_cb_fds += num_fds;

  fn_exit:
    HYDU_FUNC_EXIT();
    return status;

  fn_fail:
    goto fn_exit;
}

HYD_status HYDT_dmx_deregister_fd(int fd)
{
    int i;
    struct HYDT_dmxu_callback *cb_element;
    HYD_status status = HYD_SUCCESS;

    HYDU_FUNC_ENTER();

    cb_element = HYDT_dmxu_cb_list;
    while (cb_element) {
        for (i = 0; i < cb_element->num_fds; i++) {
            if (cb_element->fd[i] == fd) {
148
                cb_element->fd[i] = HYD_FD_UNSET;
149
150
151
152
153
154
155
156
                HYDT_dmxu_num_cb_fds--;
                goto fn_exit;
            }
        }
        cb_element = cb_element->next;
    }

    /* FD is not found */
157
158
    HYDU_ERR_SETANDJUMP(status, HYD_INTERNAL_ERROR,
                        "could not find fd to deregister: %d\n", fd);
159
160
161
162
163
164
165
166
167
168
169
170
171
172

  fn_exit:
    HYDU_FUNC_EXIT();
    return status;

  fn_fail:
    goto fn_exit;
}

HYD_status HYDT_dmx_wait_for_event(int wtime)
{
    return HYDT_dmxu_fns.wait_for_event(wtime);
}

Pavan Balaji's avatar
Pavan Balaji committed
173
int HYDT_dmx_query_fd_registration(int fd)
174
175
{
    struct HYDT_dmxu_callback *run;
Pavan Balaji's avatar
Pavan Balaji committed
176
    int i, ret;
177
178
179

    HYDU_FUNC_ENTER();

Pavan Balaji's avatar
Pavan Balaji committed
180
    ret = 0;
181
182
183
    for (run = HYDT_dmxu_cb_list; run; run = run->next) {
        for (i = 0; i < run->num_fds; i++) {
            if (run->fd[i] == fd) {     /* found it */
Pavan Balaji's avatar
Pavan Balaji committed
184
                ret = 1;
185
186
187
                break;
            }
        }
Pavan Balaji's avatar
Pavan Balaji committed
188
        if (ret)
189
190
191
192
193
            break;
    }

    HYDU_FUNC_EXIT();

Pavan Balaji's avatar
Pavan Balaji committed
194
    return ret;
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
}

HYD_status HYDT_dmx_finalize(void)
{
    struct HYDT_dmxu_callback *run1, *run2;
    HYD_status status = HYD_SUCCESS;

    HYDU_FUNC_ENTER();

    run1 = HYDT_dmxu_cb_list;
    while (run1) {
        run2 = run1->next;
        if (run1->fd)
            HYDU_FREE(run1->fd);
        HYDU_FREE(run1);
        run1 = run2;
    }
    HYDT_dmxu_cb_list = NULL;

    HYDU_FUNC_EXIT();
    return status;
}

218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
HYD_status HYDT_dmxi_stdin_valid(int *out)
{
    int ret;
    HYD_status status = HYD_SUCCESS;

    HYDU_FUNC_ENTER();

    /* This is an extremely round-about way of solving a simple
     * problem. isatty(STDIN_FILENO) seems to return 1, even when
     * mpiexec is run in the background. So, instead of relying on
     * that, we catch SIGTTIN and ignore it. But that causes the
     * read() call to return an error (with errno == EINTR) when we
     * are not attached to the terminal. */
#if defined(SIGTTIN) && defined(HAVE_ISATTY)
    if (isatty(STDIN_FILENO)) {
        status = HYDU_set_signal(SIGTTIN, signal_cb);
        HYDU_ERR_POP(status, "unable to set SIGTTIN\n");
    }
#endif /* SIGTTIN and HAVE_ISATTY */

    ret = read(STDIN_FILENO, NULL, 0);
    if (ret < 0 && errno == EINTR && got_sigttin)
        *out = 0;
    else
        *out = 1;

#if defined(SIGTTIN) && defined(HAVE_ISATTY)
    if (isatty(STDIN_FILENO)) {
        status = HYDU_set_signal(SIGTTIN, SIG_IGN);
        HYDU_ERR_POP(status, "unable to set SIGTTIN\n");
    }
#endif /* SIGTTIN and HAVE_ISATTY */

  fn_exit:
    HYDU_FUNC_EXIT();
    return status;

  fn_fail:
    goto fn_exit;
}

259
260
HYD_status HYDT_dmx_stdin_valid(int *out)
{
261
    return HYDT_dmxu_fns.stdin_valid(out);
262
}