Skip to content

Commit 832f1b0

Browse files
artpol84artpol84
authored
Merge pull request #3790 from artpol84/orte/iof_sbatch
orte/iof: Address the case when output is a regular file
2 parents f996673 + 374c824 commit 832f1b0

File tree

10 files changed

+257
-121
lines changed

10 files changed

+257
-121
lines changed

‎opal/util/fd.c

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
/*
22
* Copyright (c) 2008-2014 Cisco Systems, Inc. All rights reserved.
33
* Copyright (c) 2009 Sandia National Laboratories. All rights reserved.
4+
* Copyright (c) 2017 Mellanox Technologies. All rights reserved.
45
*
56
* $COPYRIGHT$
67
*
@@ -11,6 +12,14 @@
1112

1213
#include "opal_config.h"
1314

15+
#ifdef HAVE_SYS_TYPES_H
16+
#include <sys/types.h>
17+
#endif
18+
#ifdef HAVE_SYS_STAT_H
19+
#include <sys/stat.h>
20+
#endif
21+
22+
1423
#ifdef HAVE_UNISTD_H
1524
#include <unistd.h>
1625
#endif
@@ -89,3 +98,31 @@ int opal_fd_set_cloexec(int fd)
8998

9099
return OPAL_SUCCESS;
91100
}
101+
102+
bool opal_fd_is_regular(int fd)
103+
{
104+
struct stat buf;
105+
if (fstat(fd, &buf)) {
106+
return false;
107+
}
108+
return S_ISREG(buf.st_mode);
109+
}
110+
111+
bool opal_fd_is_chardev(int fd)
112+
{
113+
struct stat buf;
114+
if (fstat(fd, &buf)) {
115+
return false;
116+
}
117+
return S_ISCHR(buf.st_mode);
118+
}
119+
120+
bool opal_fd_is_blkdev(int fd)
121+
{
122+
struct stat buf;
123+
if (fstat(fd, &buf)) {
124+
return false;
125+
}
126+
return S_ISBLK(buf.st_mode);
127+
}
128+

‎opal/util/fd.h

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
/*
22
* Copyright (c) 2008-2014 Cisco Systems, Inc. All rights reserved.
33
* Copyright (c) 2009 Sandia National Laboratories. All rights reserved.
4+
* Copyright (c) 2017 Mellanox Technologies. All rights reserved.
45
*
56
* $COPYRIGHT$
67
*
@@ -63,6 +64,37 @@ OPAL_DECLSPEC int opal_fd_write(int fd, int len, const void *buffer);
6364
*/
6465
OPAL_DECLSPEC int opal_fd_set_cloexec(int fd);
6566

67+
/**
68+
* Convenience function to check if fd point to an accessible regular file.
69+
*
70+
* @param fd File descriptor
71+
*
72+
* @returns true if "fd" points to a regular file.
73+
* @returns false otherwise.
74+
*/
75+
OPAL_DECLSPEC bool opal_fd_is_regular(int fd);
76+
77+
/**
78+
* Convenience function to check if fd point to an accessible character device.
79+
*
80+
* @param fd File descriptor
81+
*
82+
* @returns true if "fd" points to a regular file.
83+
* @returns false otherwise.
84+
*/
85+
OPAL_DECLSPEC bool opal_fd_is_chardev(int fd);
86+
87+
/**
88+
* Convenience function to check if fd point to an accessible block device.
89+
*
90+
* @param fd File descriptor
91+
*
92+
* @returns true if "fd" points to a regular file.
93+
* @returns false otherwise.
94+
*/
95+
OPAL_DECLSPEC bool opal_fd_is_blkdev(int fd);
96+
97+
6698
END_C_DECLS
6799

68100
#endif

‎orte/mca/iof/base/base.h

Lines changed: 110 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
* All rights reserved.
1515
* Copyright (c) 2015-2017 Intel, Inc. All rights reserved.
1616
* Copyright (c) 2017 IBM Corporation. All rights reserved.
17+
* Copyright (c) 2017 Mellanox Technologies. All rights reserved.
1718
* $COPYRIGHT$
1819
*
1920
* Additional copyrights may follow
@@ -48,11 +49,13 @@
4849
#include "opal/class/opal_bitmap.h"
4950
#include "orte/mca/mca.h"
5051
#include "opal/mca/event/event.h"
52+
#include "opal/util/fd.h"
5153

5254
#include "orte/mca/iof/iof.h"
5355
#include "orte/runtime/orte_globals.h"
5456
#include "orte/mca/rml/rml_types.h"
5557
#include "orte/util/threads.h"
58+
#include "orte/mca/errmgr/errmgr.h"
5659

5760
BEGIN_C_DECLS
5861

@@ -84,7 +87,9 @@ ORTE_DECLSPEC OBJ_CLASS_DECLARATION(orte_iof_job_t);
8487
typedef struct {
8588
opal_list_item_t super;
8689
bool pending;
90+
bool always_writable;
8791
opal_event_t *ev;
92+
struct timeval tv;
8893
int fd;
8994
opal_list_t outputs;
9095
} orte_iof_write_event_t;
@@ -106,9 +111,11 @@ typedef struct {
106111
opal_object_t super;
107112
struct orte_iof_proc_t *proc;
108113
opal_event_t *ev;
114+
struct timeval tv;
109115
int fd;
110116
orte_iof_tag_t tag;
111117
bool active;
118+
bool always_readable;
112119
orte_iof_sink_t *sink;
113120
} orte_iof_read_event_t;
114121
ORTE_DECLSPEC OBJ_CLASS_DECLARATION(orte_iof_read_event_t);
@@ -142,61 +149,120 @@ struct orte_iof_base_t {
142149
};
143150
typedef struct orte_iof_base_t orte_iof_base_t;
144151

152+
/* Write event macro's */
153+
154+
static inline bool
155+
orte_iof_base_fd_always_ready(int fd)
156+
{
157+
return opal_fd_is_regular(fd) ||
158+
(opal_fd_is_chardev(fd) && !isatty(fd)) ||
159+
opal_fd_is_blkdev(fd);
160+
}
161+
162+
#define ORTE_IOF_SINK_BLOCKSIZE (1024)
163+
164+
#define ORTE_IOF_SINK_ACTIVATE(wev) \
165+
do { \
166+
struct timeval *tv = NULL; \
167+
wev->pending = true; \
168+
ORTE_POST_OBJECT(wev); \
169+
if (wev->always_writable) { \
170+
/* Regular is always write ready. Use timer to activate */ \
171+
tv = &wev->tv; \
172+
} \
173+
if (opal_event_add(wev->ev, tv)) { \
174+
ORTE_ERROR_LOG(ORTE_ERR_BAD_PARAM); \
175+
} \
176+
} while(0);
177+
145178

146179
/* define an output "sink", adding it to the provided
147180
* endpoint list for this proc */
148-
#define ORTE_IOF_SINK_DEFINE(snk, nm, fid, tg, wrthndlr) \
149-
do { \
150-
orte_iof_sink_t *ep; \
151-
OPAL_OUTPUT_VERBOSE((1, orte_iof_base_framework.framework_output, \
152-
"defining endpt: file %s line %d fd %d",\
153-
__FILE__, __LINE__, (fid))); \
154-
ep = OBJ_NEW(orte_iof_sink_t); \
155-
ep->name.jobid = (nm)->jobid; \
156-
ep->name.vpid = (nm)->vpid; \
157-
ep->tag = (tg); \
158-
if (0 <= (fid)) { \
159-
ep->wev->fd = (fid); \
160-
opal_event_set(orte_event_base, \
161-
ep->wev->ev, ep->wev->fd, \
162-
OPAL_EV_WRITE, \
163-
wrthndlr, ep); \
164-
opal_event_set_priority(ep->wev->ev, ORTE_MSG_PRI); \
165-
} \
166-
*(snk) = ep; \
167-
ORTE_POST_OBJECT(ep); \
181+
#define ORTE_IOF_SINK_DEFINE(snk, nm, fid, tg, wrthndlr) \
182+
do { \
183+
orte_iof_sink_t *ep; \
184+
OPAL_OUTPUT_VERBOSE((1, \
185+
orte_iof_base_framework.framework_output, \
186+
"defining endpt: file %s line %d fd %d", \
187+
__FILE__, __LINE__, (fid))); \
188+
ep = OBJ_NEW(orte_iof_sink_t); \
189+
ep->name.jobid = (nm)->jobid; \
190+
ep->name.vpid = (nm)->vpid; \
191+
ep->tag = (tg); \
192+
if (0 <= (fid)) { \
193+
ep->wev->fd = (fid); \
194+
ep->wev->always_writable = \
195+
orte_iof_base_fd_always_ready(fid); \
196+
if(ep->wev->always_writable) { \
197+
opal_event_evtimer_set(orte_event_base, \
198+
ep->wev->ev, wrthndlr, ep); \
199+
} else { \
200+
opal_event_set(orte_event_base, \
201+
ep->wev->ev, ep->wev->fd, \
202+
OPAL_EV_WRITE, \
203+
wrthndlr, ep); \
204+
} \
205+
opal_event_set_priority(ep->wev->ev, ORTE_MSG_PRI); \
206+
} \
207+
*(snk) = ep; \
208+
ORTE_POST_OBJECT(ep); \
168209
} while(0);
169210

211+
/* Read event macro's */
212+
#define ORTE_IOF_READ_ADDEV(rev) \
213+
do { \
214+
struct timeval *tv = NULL; \
215+
if (rev->always_readable) { \
216+
tv = &rev->tv; \
217+
} \
218+
if (opal_event_add(rev->ev, tv)) { \
219+
ORTE_ERROR_LOG(ORTE_ERR_BAD_PARAM); \
220+
} \
221+
} while(0);
222+
223+
#define ORTE_IOF_READ_ACTIVATE(rev) \
224+
do { \
225+
rev->active = true; \
226+
ORTE_POST_OBJECT(rev); \
227+
ORTE_IOF_READ_ADDEV(rev); \
228+
} while(0);
229+
230+
170231
/* add list of structs that has name of proc + orte_iof_tag_t - when
171232
* defining a read event, search list for proc, add flag to the tag.
172233
* when closing a read fd, find proc on list and zero out that flag
173234
* when all flags = 0, then iof is complete - set message event to
174235
* daemon processor indicating proc iof is terminated
175236
*/
176-
#define ORTE_IOF_READ_EVENT(rv, p, fid, tg, cbfunc, actv) \
177-
do { \
178-
orte_iof_read_event_t *rev; \
179-
OPAL_OUTPUT_VERBOSE((1, orte_iof_base_framework.framework_output, \
180-
"%s defining read event for %s: %s %d", \
181-
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), \
182-
ORTE_NAME_PRINT(&(p)->name), \
183-
__FILE__, __LINE__)); \
184-
rev = OBJ_NEW(orte_iof_read_event_t); \
185-
OBJ_RETAIN((p)); \
186-
rev->proc = (struct orte_iof_proc_t*)(p); \
187-
rev->tag = (tg); \
188-
rev->fd = (fid); \
189-
*(rv) = rev; \
190-
opal_event_set(orte_event_base, \
191-
rev->ev, (fid), \
192-
OPAL_EV_READ, \
193-
(cbfunc), rev); \
194-
opal_event_set_priority(rev->ev, ORTE_MSG_PRI); \
195-
if ((actv)) { \
196-
rev->active = true; \
197-
ORTE_POST_OBJECT(rev); \
198-
opal_event_add(rev->ev, 0); \
199-
} \
237+
#define ORTE_IOF_READ_EVENT(rv, p, fid, tg, cbfunc, actv) \
238+
do { \
239+
orte_iof_read_event_t *rev; \
240+
OPAL_OUTPUT_VERBOSE((1, \
241+
orte_iof_base_framework.framework_output, \
242+
"%s defining read event for %s: %s %d", \
243+
ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), \
244+
ORTE_NAME_PRINT(&(p)->name), \
245+
__FILE__, __LINE__)); \
246+
rev = OBJ_NEW(orte_iof_read_event_t); \
247+
OBJ_RETAIN((p)); \
248+
rev->proc = (struct orte_iof_proc_t*)(p); \
249+
rev->tag = (tg); \
250+
rev->fd = (fid); \
251+
rev->always_readable = orte_iof_base_fd_always_ready(fid); \
252+
*(rv) = rev; \
253+
if(rev->always_readable) { \
254+
opal_event_evtimer_set(orte_event_base, \
255+
rev->ev, (cbfunc), rev); \
256+
} else { \
257+
opal_event_set(orte_event_base, \
258+
rev->ev, (fid), \
259+
OPAL_EV_READ, \
260+
(cbfunc), rev); \
261+
} \
262+
opal_event_set_priority(rev->ev, ORTE_MSG_PRI); \
263+
if ((actv)) { \
264+
ORTE_IOF_READ_ACTIVATE(rev) \
265+
} \
200266
} while(0);
201267

202268

‎orte/mca/iof/base/iof_base_frame.c

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
* Copyright (c) 2015-2017 Research Organization for Information Science
1616
* and Technology (RIST). All rights reserved.
1717
* Copyright (c) 2017 IBM Corporation. All rights reserved.
18+
* Copyright (c) 2017 Mellanox Technologies. All rights reserved.
1819
* $COPYRIGHT$
1920
*
2021
* Additional copyrights may follow
@@ -269,6 +270,8 @@ static void orte_iof_base_read_event_construct(orte_iof_read_event_t* rev)
269270
rev->active = false;
270271
rev->ev = opal_event_alloc();
271272
rev->sink = NULL;
273+
rev->tv.tv_sec = 0;
274+
rev->tv.tv_usec = 0;
272275
}
273276
static void orte_iof_base_read_event_destruct(orte_iof_read_event_t* rev)
274277
{
@@ -298,9 +301,12 @@ OBJ_CLASS_INSTANCE(orte_iof_read_event_t,
298301
static void orte_iof_base_write_event_construct(orte_iof_write_event_t* wev)
299302
{
300303
wev->pending = false;
304+
wev->always_writable = false;
301305
wev->fd = -1;
302306
OBJ_CONSTRUCT(&wev->outputs, opal_list_t);
303307
wev->ev = opal_event_alloc();
308+
wev->tv.tv_sec = 0;
309+
wev->tv.tv_usec = 0;
304310
}
305311
static void orte_iof_base_write_event_destruct(orte_iof_write_event_t* wev)
306312
{

0 commit comments

Comments
 (0)