iof/base/base.h

0001 /*
0002  * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
0003  *                         University Research and Technology
0004  *                         Corporation.  All rights reserved.
0005  * Copyright (c) 2004-2011 The University of Tennessee and The University
0006  *                         of Tennessee Research Foundation.  All rights
0007  *                         reserved.
0008  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
0009  *                         University of Stuttgart.  All rights reserved.
0010  * Copyright (c) 2004-2005 The Regents of the University of California.
0011  *                         All rights reserved.
0012  * Copyright (c) 2008-2020 Cisco Systems, Inc.  All rights reserved
0013  * Copyright (c) 2012-2013 Los Alamos National Security, LLC.
0014  *                         All rights reserved.
0015  * Copyright (c) 2015-2019 Intel, Inc.  All rights reserved.
0016  * Copyright (c) 2017      IBM Corporation.  All rights reserved.
0017  * Copyright (c) 2017      Mellanox Technologies. All rights reserved.
0018  * Copyright (c) 2018      Research Organization for Information Science
0019  *                         and Technology (RIST).  All rights reserved.
0020  * Copyright (c) 2021-2023 Nanook Consulting.  All rights reserved.
0021  * $COPYRIGHT$
0022  *
0023  * Additional copyrights may follow
0024  *
0025  * $HEADER$
0026  */
0027 /**
0028  * @file
0029  *
0030  * I/O Forwarding Service
0031  */
0032
0033 #ifndef MCA_IOF_BASE_H
0034 #define MCA_IOF_BASE_H
0035
0036 #include "prte_config.h"
0037 #ifdef HAVE_SYS_TYPES_H
0038 #    include <sys/types.h>
0039 #endif
0040 #ifdef HAVE_SYS_UIO_H
0041 #    include <sys/uio.h>
0042 #endif
0043 #ifdef HAVE_NET_UIO_H
0044 #    include <net/uio.h>
0045 #endif
0046 #ifdef HAVE_UNISTD_H
0047 #    include <unistd.h>
0048 #endif
0049 #include <signal.h>
0050
0051 #include "src/class/pmix_bitmap.h"
0052 #include "src/class/pmix_list.h"
0053 #include "src/event/event-internal.h"
0054 #include "src/mca/mca.h"
0055 #include "src/pmix/pmix-internal.h"
0056 #include "src/util/pmix_fd.h"
0057
0058 #include "src/mca/errmgr/errmgr.h"
0059 #include "src/mca/iof/iof.h"
0060 #include "src/rml/rml_types.h"
0061 #include "src/runtime/prte_globals.h"
0062 #include "src/threads/pmix_threads.h"
0063
0064 BEGIN_C_DECLS
0065
0066 /*
0067  * MCA framework
0068  */
0069 PRTE_EXPORT extern pmix_mca_base_framework_t prte_iof_base_framework;
0070 /*
0071  * Select an available component.
0072  */
0073 PRTE_EXPORT int prte_iof_base_select(void);
0074
0075 /*
0076  * Maximum size of single msg
0077  */
0078 #define PRTE_IOF_BASE_MSG_MAX        4096
0079 #define PRTE_IOF_BASE_TAG_MAX        1024
0080 #define PRTE_IOF_BASE_TAGGED_OUT_MAX 8192
0081 #define PRTE_IOF_MAX_INPUT_BUFFERS   50
0082
0083 typedef struct {
0084     pmix_list_item_t super;
0085     bool pending;
0086     bool always_writable;
0087     prte_event_t *ev;
0088     struct timeval tv;
0089     int fd;
0090     pmix_list_t outputs;
0091 } prte_iof_write_event_t;
0092 PRTE_EXPORT PMIX_CLASS_DECLARATION(prte_iof_write_event_t);
0093
0094 typedef struct {
0095     pmix_list_item_t super;
0096     pmix_proc_t name;
0097     pmix_proc_t daemon;
0098     prte_iof_tag_t tag;
0099     prte_iof_write_event_t *wev;
0100     bool xoff;
0101     bool exclusive;
0102     bool closed;
0103 } prte_iof_sink_t;
0104 PRTE_EXPORT PMIX_CLASS_DECLARATION(prte_iof_sink_t);
0105
0106 struct prte_iof_proc_t;
0107 typedef struct {
0108     pmix_object_t super;
0109     struct prte_iof_proc_t *proc;
0110     prte_event_t *ev;
0111     struct timeval tv;
0112     int fd;
0113     prte_iof_tag_t tag;
0114     bool active;
0115     bool activated;
0116     bool always_readable;
0117     prte_iof_sink_t *sink;
0118 } prte_iof_read_event_t;
0119 PRTE_EXPORT PMIX_CLASS_DECLARATION(prte_iof_read_event_t);
0120
0121 typedef struct {
0122     pmix_list_item_t super;
0123     pmix_proc_t name;
0124     prte_iof_sink_t *stdinev;
0125     prte_iof_read_event_t *revstdout;
0126     prte_iof_read_event_t *revstderr;
0127 } prte_iof_proc_t;
0128 PRTE_EXPORT PMIX_CLASS_DECLARATION(prte_iof_proc_t);
0129
0130 typedef struct {
0131     pmix_list_item_t super;
0132     char data[PRTE_IOF_BASE_TAGGED_OUT_MAX];
0133     int numbytes;
0134 } prte_iof_write_output_t;
0135 PRTE_EXPORT PMIX_CLASS_DECLARATION(prte_iof_write_output_t);
0136
0137 typedef struct{
0138     pmix_object_t super;
0139     pmix_proc_t source;
0140     pmix_byte_object_t bo;
0141 } prte_iof_deliver_t;
0142 PRTE_EXPORT PMIX_CLASS_DECLARATION(prte_iof_deliver_t);
0143
0144 /* Write event macro's */
0145
0146 static inline bool prte_iof_base_fd_always_ready(int fd)
0147 {
0148     return pmix_fd_is_regular(fd) || (pmix_fd_is_chardev(fd) && !isatty(fd))
0149            || pmix_fd_is_blkdev(fd);
0150 }
0151
0152 #define PRTE_IOF_SINK_BLOCKSIZE (1024)
0153
0154 #define PRTE_IOF_SINK_ACTIVATE(wev)                                    \
0155     do {                                                               \
0156         struct timeval *tv = NULL;                                     \
0157         wev->pending = true;                                           \
0158         PMIX_POST_OBJECT(wev);                                         \
0159         if (wev->always_writable) {                                    \
0160             /* Regular is always write ready. Use timer to activate */ \
0161             tv = &wev->tv;                                             \
0162         }                                                              \
0163         if (prte_event_add(wev->ev, tv)) {                             \
0164             PRTE_ERROR_LOG(PRTE_ERR_BAD_PARAM);                        \
0165         }                                                              \
0166     } while (0);
0167
0168 /* define an output "sink", adding it to the provided
0169  * endpoint list for this proc */
0170 #define PRTE_IOF_SINK_DEFINE(snk, nm, fid, tg, wrthndlr)                                           \
0171     do {                                                                                           \
0172         prte_iof_sink_t *ep;                                                                       \
0173         PMIX_OUTPUT_VERBOSE((1, prte_iof_base_framework.framework_output,                          \
0174                              "defining endpt: file %s line %d fd %d", __FILE__, __LINE__, (fid))); \
0175         ep = PMIX_NEW(prte_iof_sink_t);                                                            \
0176         PMIX_LOAD_PROCID(&ep->name, (nm)->nspace, (nm)->rank);                                     \
0177         ep->tag = (tg);                                                                            \
0178         if (0 <= (fid)) {                                                                          \
0179             ep->wev->fd = (fid);                                                                   \
0180             ep->wev->always_writable = prte_iof_base_fd_always_ready(fid);                         \
0181             if (ep->wev->always_writable) {                                                        \
0182                 prte_event_evtimer_set(prte_event_base, ep->wev->ev, wrthndlr, ep);                \
0183             } else {                                                                               \
0184                 prte_event_set(prte_event_base, ep->wev->ev, ep->wev->fd, PRTE_EV_WRITE, wrthndlr, \
0185                                ep);                                                                \
0186             }                                                                                      \
0187         }                                                                                          \
0188         *(snk) = ep;                                                                               \
0189         PMIX_POST_OBJECT(ep);                                                                      \
0190     } while (0);
0191
0192 /* Read event macro's */
0193 #define PRTE_IOF_READ_ADDEV(rev)                \
0194     do {                                        \
0195         struct timeval *tv = NULL;              \
0196         if (rev->always_readable) {             \
0197             tv = &rev->tv;                      \
0198         }                                       \
0199         if (prte_event_add(rev->ev, tv)) {      \
0200             PRTE_ERROR_LOG(PRTE_ERR_BAD_PARAM); \
0201         }                                       \
0202     } while (0);
0203
0204 #define PRTE_IOF_READ_ACTIVATE(rev) \
0205     do {                            \
0206         rev->active = true;         \
0207         PMIX_POST_OBJECT(rev);      \
0208         PRTE_IOF_READ_ADDEV(rev);   \
0209     } while (0);
0210
0211 /* add list of structs that has name of proc + prte_iof_tag_t - when
0212  * defining a read event, search list for proc, add flag to the tag.
0213  * when closing a read fd, find proc on list and zero out that flag
0214  * when all flags = 0, then iof is complete - set message event to
0215  * daemon processor indicating proc iof is terminated
0216  */
0217 #define PRTE_IOF_READ_EVENT(rv, p, fid, tg, cbfunc, actv)                                     \
0218     do {                                                                                      \
0219         prte_iof_read_event_t *rev;                                                           \
0220         PMIX_OUTPUT_VERBOSE((1, prte_iof_base_framework.framework_output,                     \
0221                              "%s defining read event for %s: %s %d",                          \
0222                              PRTE_NAME_PRINT(PRTE_PROC_MY_NAME), PRTE_NAME_PRINT(&(p)->name), \
0223                              __FILE__, __LINE__));                                            \
0224         rev = PMIX_NEW(prte_iof_read_event_t);                                                \
0225         PMIX_RETAIN((p));                                                                     \
0226         rev->proc = (struct prte_iof_proc_t *) (p);                                           \
0227         rev->tag = (tg);                                                                      \
0228         rev->fd = (fid);                                                                      \
0229         rev->always_readable = prte_iof_base_fd_always_ready(fid);                            \
0230         *(rv) = rev;                                                                          \
0231         if (rev->always_readable) {                                                           \
0232             prte_event_evtimer_set(prte_event_base, rev->ev, (cbfunc), rev);                  \
0233         } else {                                                                              \
0234             prte_event_set(prte_event_base, rev->ev, (fid), PRTE_EV_READ, (cbfunc), rev);     \
0235         }                                                                                     \
0236         if ((actv)) {                                                                         \
0237             PRTE_IOF_READ_ACTIVATE(rev)                                                       \
0238         }                                                                                     \
0239     } while (0);
0240
0241 PRTE_EXPORT int prte_iof_base_flush(void);
0242
0243 PRTE_EXPORT extern int prte_iof_base_output_limit;
0244
0245 /* base functions */
0246 PRTE_EXPORT int prte_iof_base_write_output(const pmix_proc_t *name, prte_iof_tag_t stream,
0247                                            const unsigned char *data, int numbytes,
0248                                            prte_iof_write_event_t *channel);
0249 PRTE_EXPORT void prte_iof_base_write_handler(int fd, short event, void *cbdata);
0250
0251 PRTE_EXPORT void prte_iof_base_output(const pmix_proc_t *source,
0252                                       pmix_iof_channel_t channel,
0253                                       char *string);
0254
0255 END_C_DECLS
0256
0257 #endif /* MCA_IOF_BASE_H */