/* * Copyright (C) 1994-2021 Altair Engineering, Inc. * For more information, contact Altair at www.altair.com. * * This file is part of both the OpenPBS software ("OpenPBS") * and the PBS Professional ("PBS Pro") software. * * Open Source License Information: * * OpenPBS is free software. You can redistribute it and/or modify it under * the terms of the GNU Affero General Public License as published by the * Free Software Foundation, either version 3 of the License, or (at your * option) any later version. * * OpenPBS is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public * License for more details. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . * * Commercial License Information: * * PBS Pro is commercially licensed software that shares a common core with * the OpenPBS software. For a copy of the commercial license terms and * conditions, go to: (http://www.pbspro.com/agreement.html) or contact the * Altair Legal Department. * * Altair's dual-license business model allows companies, individuals, and * organizations to create proprietary derivative works of OpenPBS and * distribute them - whether embedded or bundled with other software - * under a commercial license agreement. * * Use of Altair's trademarks, including but not limited to "PBS™", * "OpenPBS®", "PBS Professional®", and "PBS Pro™" and Altair's logos is * subject to Altair's trademark licensing policies. */ /* * qsub_sup.c * * Created on: Jul 3, 2020 * Author: bhagatr */ #include /* the master config generated by configure */ #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "pbs_ifl.h" #include "ifl_internal.h" #include "cmds.h" #include "libpbs.h" #include "net_connect.h" #include "dis.h" #include "port_forwarding.h" #include "credential.h" #include "libutil.h" #if defined(HAVE_SYS_IOCTL_H) #include #endif /* HAVE_SYS_IOCTL_H */ #define MAXPIPENAME sizeof(((struct sockaddr_un *) 0)->sun_path) #define QSUB_DMN_TIMEOUT_SHORT 5 #define QSUB_DMN_TIMEOUT_LONG 60 /* timeout for qsub background process */ #define DMN_REFUSE_EXIT 7 /* return code when daemon can't serve a job and exits */ #define CMDLINE 3 #define XAUTH_LEN 512 /* Max size of buffer to store Xauth cookie length */ #define XAUTH_ERR_REDIRECTION "2>&1" /* redirection string used for xauth command */ #define X11_PORT_LEN 8 /* Max size of buffer to store port information as string */ #if !defined(PBS_NO_POSIX_VIOLATION) char GETOPT_ARGS[] = "a:A:c:C:e:fhIj:J:k:l:m:M:N:o:p:q:r:R:S:u:v:VW:XzP:"; #else char GETOPT_ARGS[] = "a:A:c:C:e:fhj:J:k:l:m:M:N:o:p:q:r:R:S:u:v:VW:zP:"; #endif /* PBS_NO_POSIX_VIOLATION */ char usage[] = "usage: qsub [-a date_time] [-A account_string] [-c interval]\n" "\t[-C directive_prefix] [-e path] [-f ] [-h ] [-I [-X]] [-j oe|eo] [-J X-Y[:Z]]\n" "\t[-k keep] [-l resource_list] [-m mail_options] [-M user_list]\n" "\t[-N jobname] [-o path] [-p priority] [-P project] [-q queue] [-r y|n]\n" "\t[-R o|e|oe] [-S path] [-u user_list] [-W otherattributes=value...]\n" "\t[-v variable_list] [-V ] [-z] [script | -- command [arg1 ...]]\n"; char read_script_msg[] = "Job script will be read from standard input. Submit with CTRL+D.\n"; static struct termios oldtio; /* Terminal info */ static struct winsize wsz; /* Window size */ extern struct attrl *attrib; /* Attribute list */ char fl[MAXPIPENAME]; /* the filename used as the pipe name */ extern char *new_jobname; /* return from submit request */ extern char server_out[PBS_MAXSERVERNAME + PBS_MAXPORTNUM + 2]; /* Destination server, parsed from destination[] */ extern char *tmpdir; /* Path of temp directory in which to put the job script */ extern char cred_name[32]; /* space to hold small credential name */ extern char destination[PBS_MAXDEST]; /* Destination of the batch job, specified by q opt */ extern char script_tmp[MAXPATHLEN + 1]; /* name of script file copy */ extern char retmsg[MAXPATHLEN]; /* holds the message that background qsub process will send */ extern char qsub_cwd[MAXPATHLEN + 1]; /* buffer to pass cwd to background qsub */ extern char *basic_envlist; /* basic comma-separated environment variables list string */ extern char *qsub_envlist; /* comma-separated variables list string */ extern char *v_value; /* expanded variable list from v opt */ extern char *cred_buf; extern char *display; /* environment variable DISPLAY */ extern int comm_sock; /* Socket for interactive and block job */ int X11_comm_sock; /* Socket for x11 communication */ extern int Forwardx11_opt; extern int sd_svr; /* return from pbs_connect */ extern int is_background; extern size_t cred_len; extern int recv_attrl(void *s, struct attrl **attrib); extern int recv_string(void *s, char *str); extern int recv_dyn_string(void *s, char **strp); extern int send_string(void *s, char *str); extern int send_attrl(void *s, struct attrl *attrib); extern int send_opts(void *s); extern int recv_opts(void *s); extern int do_submit2(char *rmsg); extern int dosend(void *s, char *buf, int bufsize); extern int dorecv(void *s, char *buf, int bufsize); extern int do_dir(char *, int, char *, size_t); extern int check_qsub_daemon(char *); extern int Interact_opt; extern int sig_happened; extern void log_syslog(char *msg); extern void exit_qsub(int exitstatus); extern void qsub_free_attrl(struct attrl *attrib); extern void bailout(int ret); extern char *strdup_esc_commas(char *str_to_dup); static char *X11_get_authstring(void); extern void set_attr_error_exit(struct attrl **attrib, char *attrib_name, char *attrib_value); static char *port_X11(void); static void daemon_stuff(void); /** * @brief * Log a simple message to syslog * To be used from the qsub background daemon * * @param[in] msg - string to be logged * */ void log_syslog(char *msg) { openlog("qsub", LOG_PID | LOG_CONS | LOG_NOWAIT, LOG_USER); syslog(LOG_ERR, "%s", msg); closelog(); } /** * @brief * Helper function to get the pbs conf file path, and * convert it to a string, later added to the filename * (pipe or unix domain socket filename) that is used * for communications between the front-end and * background qsub processes. * The path to the pbs conf file is converted to a string * by replacing the slashes with underscore char. * If PBS_CONF_FILE is not set, then an empty string is returned. * * @return - The string representing path to the pbs conf file * */ static char * get_conf_path(void) { char *cnf = getenv("PBS_CONF_FILE"); /* static pointer so we can free heap memory from previous invocation of this function */ static char *dup_cnf_path = NULL; char *p; if (cnf) { p = strdup(cnf); if (p) { free(dup_cnf_path); dup_cnf_path = p; while (*p) { if (*p == '/' || *p == ' ' || *p == '.') *p = '_'; p++; } } return dup_cnf_path; } else if (dup_cnf_path) { return dup_cnf_path; } else { return ""; } } /** * @brief * Create a temporary file that will house the job script * * @param[in] file - Input file pointer * @param[out] script - Temp file location * @param[in] prefix - Prefix for PBS directives * * @return int * @retval -1 - Error processing qsub parameters * @retval 3 - Error writing script file * @retval 4 - Temp file creation failure * @retval 5 - Error reading input file * @retval 6 - Unexpected EOF on read */ int get_script(FILE *file, char *script, char *prefix) { char *sopt; int err = 0; int exec = FALSE; char tmp_name[MAXPATHLEN + 1]; FILE *TMP_FILE; char *in; char *s_in = NULL; int s_len = 0; char *extend; char *extend_in = NULL; int extend_len = 0; int extend_loc; static char tmp_template[] = "pbsscrptXXXXXX"; int fds; /* * Note: Need to populate script variable as soon as temp file is created so it * gets cleaned up in case of an error. */ snprintf(tmp_name, sizeof(tmp_name), "%s/%s", tmpdir, tmp_template); fds = mkstemp(tmp_name); /* returns file descriptor */ if (fds != -1) { snprintf(script, MAXPATHLEN + 1, "%s", tmp_name); if ((TMP_FILE = fdopen(fds, "w+")) == NULL) err = 1; } else { err = 1; } if (err != 0) { perror("mkstemp"); fprintf(stderr, "qsub: could not create/open tmp file %s for script\n", tmp_name); return (4); } while ((in = pbs_fgets(&s_in, &s_len, file)) != NULL) { if (!exec && ((sopt = pbs_ispbsdir(s_in, prefix)) != NULL)) { /* Check if this is a directive line that should be extended */ extend_loc = pbs_extendable_line(in); if (extend_loc >= 0) { in[extend_loc] = '\0'; /* remove the backslash (\) */ extend = pbs_fgets_extend(&extend_in, &extend_len, file); if (extend != NULL) { if (pbs_strcat(&s_in, &s_len, extend) == NULL) return (5); in = s_in; sopt = pbs_ispbsdir(s_in, prefix); } } /* * Setting options from the job script will not overwrite * options set on the command line. CMDLINE-1 means * "one less than CMDLINE priority" */ if (do_dir(sopt, CMDLINE - 1, retmsg, MAXPATHLEN) != 0) { fprintf(stderr, "%s", retmsg); free(extend_in); free(s_in); return (-1); } } else if (!exec && pbs_isexecutable(s_in)) { exec = TRUE; } if (fputs(in, TMP_FILE) < 0) { perror("fputs"); fprintf(stderr, "qsub: error writing copy of script, %s\n", tmp_name); fclose(TMP_FILE); free(extend_in); free(s_in); return (3); } } free(extend_in); free(s_in); if (fclose(TMP_FILE) != 0) { perror(" qsub: copy of script to tmp failed on close"); return (5); } if (ferror(file)) { fprintf(stderr, "qsub: error reading script file\n"); return (5); } return (0); } /** * @brief * signal handler to avoid race condition * * @param[in] sig - signal number * * @return Void * */ void blockint(int sig) { sig_happened = sig; } /** * @brief * Enable X11 Forwarding (on Unix) if specified. */ void enable_gui(void) { char *x11authstr = NULL; if (Forwardx11_opt) { if (!Interact_opt) { fprintf(stderr, "qsub: X11 Forwarding possible only for interactive jobs\n"); exit_qsub(1); } /* get the DISPLAY's auth protocol, hexdata, and screen number */ if ((x11authstr = X11_get_authstring()) != NULL) { set_attr_error_exit(&attrib, ATTR_X11_cookie, x11authstr); set_attr_error_exit(&attrib, ATTR_X11_port, port_X11()); #ifdef DEBUG fprintf(stderr, "x11auth string: %s\n", x11authstr); #endif } else { exit_qsub(1); } } } /** * @Brief * This function returns a string that consists of the protocol getting * used, the hex data and the screen number . This string will form the * basis of X authentication . It will be passed as a job attribute to * MOM. * @return char* * @retval authstring Success * @retval NULL Failure * */ static char * X11_get_authstring(void) { char line[XAUTH_LEN] = {'\0'}; char command[XAUTH_LEN] = {'\0'}; char protocol[XAUTH_LEN]; char hexdata[XAUTH_LEN]; char screen[XAUTH_LEN]; char format[XAUTH_LEN]; char *authstring = NULL; FILE *f; int got_data = 0, ret = 0; char *p; protocol[0] = '\0'; hexdata[0] = '\0'; screen[0] = '\0'; sprintf(format, " %%*s %%%ds %%%ds ", XAUTH_LEN - 1, XAUTH_LEN - 1); p = strchr(display, ':'); if (p == NULL) { fprintf(stderr, "qsub: Failed to get xauth data " "(check $DISPLAY variable)\n"); return NULL; } /* Try to get Xauthority information for the display. */ if (strncmp(display, "localhost", sizeof("localhost") - 1) == 0) { /* * Handle FamilyLocal case where $DISPLAY does * not match an authorization entry. For this we * just try "xauth list unix:displaynum.screennum". * "localhost" match to determine FamilyLocal * is not perfect. */ ret = snprintf(line, sizeof(line), "%s list unix:%s %s", XAUTH_BINARY, p + 1, XAUTH_ERR_REDIRECTION); if (ret >= sizeof(line)) { fprintf(stderr, " qsub: line overflow\n"); return NULL; } } else { ret = snprintf(line, sizeof(line), "%s list %.255s %s", XAUTH_BINARY, display, XAUTH_ERR_REDIRECTION); if (ret >= sizeof(line)) { fprintf(stderr, " qsub: line overflow\n"); return NULL; } } snprintf(command, sizeof(command), "%s", line); if (p != NULL) p = strchr(p, '.'); if (p != NULL) snprintf(screen, sizeof(screen), "%s", p + 1); else strcpy(screen, "0"); /* Should be safe because sizeof(screen) = XAUTH_LEN which is >= 2 */ #ifdef DEBUG fprintf(stderr, "X11_get_authstring: %s\n", line); #endif f = popen(line, "r"); if (f == NULL) { fprintf(stderr, "execution of '%s' failed, errno=%d \n", command, errno); } else if (fgets(line, sizeof(line), f) == 0) { fprintf(stderr, "cannot read data from '%s', errno=%d \n", command, errno); } else if (sscanf(line, format, protocol, hexdata) != 2) { fprintf(stderr, "cannot parse output from '%s'\n", command); } else { /* SUCCESS */ got_data = 1; } if (f != NULL) { /* * Check the return value of pclose to see if the command failed? * In that case, the "line" read from stdout is probably an * error message (since stderr is redirected to stdout) from the shell or xauth, * so display that to the user. */ if (pclose(f) != 0) { fprintf(stderr, "execution of xauth failed: %s", line); return NULL; } } if (!got_data) /* FAILURE */ return NULL; /** * Allocate 4 additional bytes for the terminating NULL character for * each of the strings inside malloc */ authstring = malloc(strlen(protocol) + strlen(hexdata) + strlen(screen) + 4); if (authstring == NULL) { /* FAILURE */ fprintf(stderr, " qsub: Malloc Failed\n"); return NULL; } sprintf(authstring, "%s:%s:%s", protocol, hexdata, screen); return (authstring); } /** * @brief * This function creates a socket to listen for "X11" data * and returns a port number where its listening for X data. * * @return char* * @retval portstring success * * @par Side Effects * If this function fails, it will exit the qsub process. * */ static char * port_X11(void) { pbs_socklen_t namelen; struct sockaddr_in myaddr; static char X11_port_str[X11_PORT_LEN]; unsigned short X11_port; X11_comm_sock = socket(AF_INET, SOCK_STREAM, 0); if (X11_comm_sock < 0) { perror("qsub: unable to create socket"); exit_qsub(1); } myaddr.sin_family = AF_INET; myaddr.sin_addr.s_addr = INADDR_ANY; myaddr.sin_port = 0; if (bind(X11_comm_sock, (struct sockaddr *) &myaddr, sizeof(myaddr)) < 0) { perror("qsub: unable to bind to socket"); exit_qsub(1); } /* get port number assigned */ namelen = sizeof(myaddr); if (getsockname(X11_comm_sock, (struct sockaddr *) &myaddr, &namelen) < 0) { perror("qsub: unable to get port number"); exit_qsub(1); } X11_port = ntohs(myaddr.sin_port); (void) sprintf(X11_port_str, "%u", (unsigned int) X11_port); if (listen(X11_comm_sock, 1) < 0) { perror("qsub: listening on X11 socket failed"); exit_qsub(1); } return (X11_port_str); } /** * @brief * Fork the current process. Call the daemon_stuff function in the * child process which starts listening on the unix domain socket etc. * The parent process continues out of this function and eventually * returns back control to the calling shell. * * @param[in] fname - The filename used for the communication pipe/socket for * the communication between background and forground qsub processes. * @param[in] handle - Handle to synchronization event between foreground and * background qsub processes. * @param[in] server - Target server name of NULL in case of default * * exits program on failure * */ void do_daemon_stuff() { int pid; pid = fork(); if (pid == 0) { /* * Try to become the session leader. * If that fails, exit with a syslog message */ if (setsid() == -1) { log_syslog("setsid failed"); exit(1); } /* * Just close standard files. We don't want to * be session leader or close all other files. */ (void) fclose(stdin); (void) fclose(stdout); (void) fclose(stderr); /* clear off all the attributes */ qsub_free_attrl(attrib); attrib = NULL; free(v_value); v_value = NULL; free(basic_envlist); basic_envlist = NULL; free(qsub_envlist); qsub_envlist = NULL; /* set single threaded mode */ pbs_client_thread_set_single_threaded_mode(); /* set when background qsub is running */ is_background = 1; daemon_stuff(); /* * Control should never reach here. * Still adding an exit, so it does not traverse parent code. */ exit(1); } } /** * @brief * Signal handler for SIGPIPE * @param[in] sig - signal number * @return void * */ void exit_on_sigpipe(int sig) { perror("qsub: SIGPIPE received, job submission interrupted."); exit_qsub(1); } /** * @brief * Set the signal handlers. */ void set_sig_handlers(void) { /* Catch SIGPIPE on write() failures. */ struct sigaction act; sigemptyset(&act.sa_mask); act.sa_handler = exit_on_sigpipe; act.sa_flags = 0; if (sigaction(SIGPIPE, &act, NULL) < 0) { perror("qsub: unable to catch SIGPIPE"); exit_qsub(1); } } /** * @brief * Send data of bufsize length to the peer. Used for communications * between the foreground and background qsub processes. * * @param[in] s - pointer to the windows PIPE or Unix domain socket * @param[in] buf - The buf to send data from * @param[in] bufsize - The amount of data to send * * @return int * @retval -1 - Failure * @retval 0 - Success * */ int dosend(void *s, char *buf, int bufsize) { int bytes = 0; int sock = (int) *((int *) s); int rc; char *p = buf; int remaining = bufsize; do { /* * For systems with MSG_NOSIGNAL defined (e.g. Linux 2.2 and later), * we use send() rather than write() in order to block the SIGPIPE * that qsub would receive if the remote side closes the stream. For * other systems, the exit_on_sigpipe() handler gets called. */ errno = 0; #ifdef MSG_NOSIGNAL rc = send(sock, p, remaining, MSG_NOSIGNAL); #else rc = write(sock, p, remaining); #endif if (rc == -1) return -1; if (rc == 0) break; bytes += rc; p += rc; remaining -= rc; } while (bytes < bufsize); if (bytes != bufsize) return -1; return 0; } /** * @brief * Receive data of bufsize length from the peer. Used for communications * between the foreground and background qsub processes. * * @param[in] s - pointer to the windows PIPE or Unix domain socket * @param[in] buf - The buf to receive data into * @param[in] bufsize - The amount of data to read * * @return int * @retval -1 - Failure * @retval 0 - Success * */ int dorecv(void *s, char *buf, int bufsize) { int bytes = 0; char *p = buf; int remaining = bufsize; int sock = *((int *) s); int rc; do { errno = 0; rc = read(sock, p, remaining); if (rc == -1) return -1; if (rc == 0) break; bytes += rc; p += rc; remaining -= rc; } while (bytes < bufsize); if (bytes != bufsize) return -1; return 0; } /** * @brief * Interactive Reader process: reads from the remote socket, * and writes that out to the stdout * * @param[in] s - socket (file descriptor) * * @return Error code * @retval -1 Failure * @retval 0 Success * */ int reader(int s) { char buf[4096]; int c; char *p; int wc; /* read from the socket, and write to stdout */ while (1) { c = CS_read(s, buf, sizeof(buf)); if (c > 0) { p = buf; while (c) { if ((wc = write(1, p, c)) < 0) { if (errno == EINTR) { continue; } else { perror("qsub: write error"); return (-1); } } c -= wc; p += wc; } } else if (c == 0) { return (0); /* EOF - all done */ } else { if (errno == EINTR) continue; else { perror("qsub: read error"); return (-1); } } } } /** * @brief This is a reader function which reads from the remote socket * when X forwarding is enabled and writes it back to stdout. * * @param[in] s - socket descriptor from where data is to be read. * * @return int * @retval 0 Success * @retval -1 Failure * @retval -2 Peer Closed connection * */ int reader_Xjob(int s) { static char buf[PF_BUF_SIZE]; int c = 0; char *p; int wc; int d = fileno(stdout); /* read from the socket and write to stdout */ c = CS_read(s, buf, sizeof(buf)); if (c > 0) { p = buf; while (c) { /*write data back to stdout*/ if ((wc = write(d, p, c)) < 0) { if (errno == EINTR) { continue; } else { perror("qsub: write error"); return (-1); } } c -= wc; p += wc; } } else if (c == 0) { /* * If control reaches here, then it means peer has closed the * connection. */ return (-2); } else if (errno == EINTR) { return (0); } else { perror("qsub: read error"); return (-1); } return (0); } /** * @brief * settermraw - set terminal into "raw" mode * * @param[in] ptio - pointer to termios structure * * @return None * @retval Void * */ void settermraw(struct termios *ptio) { struct termios tio; tio = *ptio; tio.c_lflag &= ~(ICANON | ISIG | ECHO | ECHOE | ECHOK); tio.c_iflag &= ~(IGNBRK | INLCR | ICRNL | IXON | IXOFF); tio.c_oflag = 0; tio.c_oflag |= (OPOST); /* TAB3 */ tio.c_cc[VMIN] = 1; tio.c_cc[VTIME] = 0; #if defined(TABDLY) && defined(TAB3) if ((tio.c_oflag & TABDLY) == TAB3) tio.c_oflag &= ~TABDLY; #endif tio.c_cc[VKILL] = -1; tio.c_cc[VERASE] = -1; if (tcsetattr(0, TCSANOW, &tio) < 0) perror("qsub: set terminal mode"); } /** * @brief * stopme - suspend process on ~^Z or ~^Y * on suspend, reset terminal to normal "cooked" mode; * when resumed, again set terminal to raw. * * @param[in] p - process id * * @return None * @retval Void * */ void stopme(pid_t p) { (void) tcsetattr(0, TCSANOW, &oldtio); /* reset terminal */ kill(p, SIGTSTP); settermraw(&oldtio); /* back to raw when we resume */ } /** * @brief * Writer process: reads from stdin, and writes * data out to the rem socket * * @param[in] s - file descriptor * * @return Void * */ void writer(int s) { char c; int i; int newline = 1; char tilde = '~'; int wi; /* read from stdin, and write to the socket */ while (1) { i = read(0, &c, 1); if (i > 0) { /* read data */ if (newline) { if (c == tilde) { /* maybe escape character */ /* read next character to check */ while ((i = read(0, &c, 1)) != 1) { if ((i == -1) && (errno == EINTR)) continue; else break; } if (i != 1) break; if (c == '.') /* termination character */ break; else if (c == oldtio.c_cc[VSUSP]) { stopme(0); /* ^Z suspend all */ continue; #ifdef VDSUSP } else if (c == oldtio.c_cc[VDSUSP]) { stopme(getpid()); continue; #endif /* VDSUSP */ } else { /* not escape, write out tilde */ while ((wi = CS_write(s, &tilde, 1)) != 1) { if ((wi == -1) && (errno == EINTR)) continue; else break; } if (wi != 1) break; } } newline = 0; /* no longer at start of line */ } else { /* reset to newline if \n \r kill or interrupt */ newline = (c == '\n') || (c == oldtio.c_cc[VKILL]) || (c == oldtio.c_cc[VINTR]) || (c == '\r'); } while ((wi = CS_write(s, &c, 1)) != 1) { /* write out character */ if ((wi == -1) && (errno == EINTR)) continue; else break; } if (wi != 1) break; } else if (i == 0) { /* EOF */ break; } else if (i < 0) { /* error */ if (errno == EINTR) continue; else { perror("qsub: read error"); return; } } } return; } /** * @brief * send_term - send the current TERM type and certain control characters * * @param[in] sock - file descriptor * * @return Void * */ void send_term(int sock) { char buf[PBS_TERM_BUF_SZ]; char *term; char cc_array[PBS_TERM_CCA]; term = getenv("TERM"); term = strdup_esc_commas(term); if (term == NULL) snprintf(buf, sizeof(buf), "TERM=unknown"); else { snprintf(buf, sizeof(buf), "TERM=%s", term); free(term); } (void) CS_write(sock, buf, PBS_TERM_BUF_SZ); cc_array[0] = oldtio.c_cc[VINTR]; cc_array[1] = oldtio.c_cc[VQUIT]; cc_array[2] = oldtio.c_cc[VERASE]; cc_array[3] = oldtio.c_cc[VKILL]; cc_array[4] = oldtio.c_cc[VEOF]; cc_array[5] = oldtio.c_cc[VSUSP]; CS_write(sock, cc_array, PBS_TERM_CCA); } /** * @brief * send_winsize = send the current tty's window size * * @param[in] sock - file descriptor * * @return Void * */ void send_winsize(int sock) { char buf[PBS_TERM_BUF_SZ]; (void) sprintf(buf, "WINSIZE %hu,%hu,%hu,%hu", wsz.ws_row, wsz.ws_col, wsz.ws_xpixel, wsz.ws_ypixel); (void) CS_write(sock, buf, PBS_TERM_BUF_SZ); return; } /** * @brief * getwinsize - get the current window size * * @param[in] pwsz - pointer to winsize structure * * @return Error code * @retval -1 Failure * @retval 0 Success * */ int getwinsize(struct winsize *pwsz) { if (ioctl(0, TIOCGWINSZ, &wsz) < 0) { perror("qsub: unable to get window size"); return (-1); } return (0); } /** * @brief * catchchild = signal handler for Death of Child * * @param[in] sig - signal number * * @return Void * */ void catchchild(int sig) { int status; int pid; while (1) { pid = waitpid(-1, &status, WNOHANG | WUNTRACED); if (pid == 0) return; if ((pid > 0) && (WIFSTOPPED(status) == 0)) break; if ((pid == -1) && (errno != EINTR)) { perror("qsub: bad status in catchchild: "); return; } } /* reset terminal to cooked mode */ (void) tcsetattr(0, TCSANOW, &oldtio); exit_qsub(0); } /** * @brief * prints can't suspend qsub process on arrival of signal causing suspension * * @param[in] sig - signal number * * @return Void * */ void no_suspend(int sig) { printf("Sorry, you cannot suspend qsub until the job is started\n"); fflush(stdout); } /** * @brief * signal handler function for interrupt signal * * @param[in] sig - signal number * * @return Void * */ void catchint(int sig) { int c; printf("Do you wish to terminate the job and exit (y|[n])? "); fflush(stdout); while (1) { alarm(60); /* give a minute to think about it */ c = getchar(); if ((c == 'n') || (c == 'N') || (c == '\n')) break; else if ((c == 'y') || (c == 'Y') || (c == EOF)) { bailout(0); } else { printf("yes or no please\n"); while ((c != '\n') && (c != EOF)) c = getchar(); } } alarm(0); /* reset alarm */ while ((c != '\n') && (c != EOF)) c = getchar(); return; } /** * @brief * signal handler for timeout scenario * * @param[in] sig - signal number * * @return Void * */ void toolong(int sig) { printf("Timeout -- deleting job\n"); bailout(0); } /** * @brief * Function used to log port forwarding messages. * * @param[in] msg - error message to be logged * * @return Void * */ static void log_cmds_portfw_msg(char *msg) { fprintf(stderr, "%s\n", msg); (void) fflush(stderr); (void) fflush(stdout); } /** * @brief * This function initializes pfwdsock structure and eventually * calls port_forwarder. * * @param[in] X_data_socket - socket descriptor used to read X data from mom * port forwarders. * @param[in] interactive_reader_socket - socket descriptor used to read * interactive job data coming from mom writer. * @return void * * @par Side Effects * On failure, the function will cause the qsub process to exit. * */ static void x11handler(int X_data_socket, int interactive_reader_socket) { int n; struct pfwdsock *socks; socks = calloc(sizeof(struct pfwdsock), NUM_SOCKS); if (!socks) { fprintf(stderr, "Calloc failed : out of memory\n"); exit_qsub(1); } for (n = 0; n < NUM_SOCKS; n++) { (socks + n)->active = 0; } socks->sock = X_data_socket; socks->active = 1; socks->listening = 1; /* Try to open a socket for the local X server. */ port_forwarder(socks, x11_connect_display, display, 0, interactive_reader_socket, reader_Xjob, log_cmds_portfw_msg); } /** * @brief * interactive - set up for interactive communication with job * * @return void * * @par Side Effects * On failure, the function will cause the qsub process to exit. * */ void interactive(void) { int amt; char cur_server[PBS_MAXSERVERNAME + PBS_MAXPORTNUM + 2]; pbs_socklen_t fromlen; char momjobid[PBS_MAXSVRJOBID + 1]; int news; int nsel; char *pc; fd_set selset; struct sigaction act; struct sockaddr_in from; struct timeval timeout; struct winsize wsz; int child; int ret; /* disallow ^Z which hangs up MOM starting an interactive job */ sigemptyset(&act.sa_mask); act.sa_handler = no_suspend; act.sa_flags = 0; if (sigaction(SIGTSTP, &act, NULL) < 0) { perror("sigaction(SIGTSTP)"); exit_qsub(1); } /* Catch SIGINT and SIGTERM, and setup to catch Death of child */ act.sa_handler = catchint; if ((sigaction(SIGINT, &act, NULL) < 0) || (sigaction(SIGTERM, &act, NULL) < 0)) { perror("unable to catch signals"); exit_qsub(1); } act.sa_handler = toolong; if ((sigaction(SIGALRM, &act, NULL) < 0)) { perror("cannot catch alarm"); exit_qsub(2); } /* save the old terminal setting */ if (tcgetattr(0, &oldtio) < 0) { perror("qsub: unable to get terminal settings"); exit_qsub(1); } /* Get the current window size, to be sent to MOM later */ if (getwinsize(&wsz)) { /* unable to get actual values, set defaults */ wsz.ws_row = 20; wsz.ws_col = 80; wsz.ws_xpixel = 0; wsz.ws_ypixel = 0; } printf("qsub: waiting for job %s to start\n", new_jobname); /* Accept connection on socket set up earlier */ nsel = 0; while (nsel == 0) { FD_ZERO(&selset); FD_SET(comm_sock, &selset); timeout.tv_usec = 0; timeout.tv_sec = 30; nsel = select(FD_SETSIZE, &selset, NULL, NULL, &timeout); if (nsel == -1) { if (errno == EINTR) nsel = 0; else { perror("qsub: select failed"); exit_qsub(1); } } if (nsel == 0) { /* connect to server, status job to see if still there */ if (!locate_job(new_jobname, server_out, cur_server)) { fprintf(stderr, "qsub: job %s apparently deleted\n", new_jobname); exit_qsub(1); } } } /* apparently someone is attempting to connect to us */ retry: fromlen = sizeof(from); if ((news = accept(comm_sock, (struct sockaddr *) &from, &fromlen)) < 0) { perror("qsub: accept error from Interactive socket "); exit_qsub(1); } /* * When Mom connects we expect: * first, to engage in an authentication activity * second, mom sends the job id for us to verify */ ret = CS_client_auth(news); if ((ret != CS_SUCCESS) && (ret != CS_AUTH_USE_IFF)) { fprintf(stderr, "qsub: failed authentication with execution host\n"); shutdown(news, 2); exit_qsub(1); } /* now verify the value of job id */ amt = PBS_MAXSVRJOBID + 1; pc = momjobid; while (amt > 0) { int len = CS_read(news, pc, amt); if (len <= 0) break; pc += len; if (*(pc - 1) == '\0') break; amt -= len; } if (pc == momjobid) { /* no data read */ shutdown(news, 2); close(news); goto retry; } if (strncmp(momjobid, new_jobname, PBS_MAXSVRJOBID) != 0) { fprintf(stderr, "qsub: invalid job name from execution server\n"); shutdown(news, 2); exit_qsub(1); } /* * got the right job, send: * terminal type as "TERM=xxxx" * window size as "WINSIZE=r,c,x,y" */ send_term(news); send_winsize(news); printf("qsub: job %s ready\n\n", new_jobname); /* set SIGINT, SIGTERM processing to default */ act.sa_handler = SIG_DFL; if ((sigaction(SIGINT, &act, NULL) < 0) || (sigaction(SIGTERM, &act, NULL) < 0) || (sigaction(SIGALRM, &act, NULL) < 0) || (sigaction(SIGTSTP, &act, NULL) < 0)) { perror("unable to reset signals"); exit_qsub(1); } child = fork(); if (child == 0) { /* child process - start the reader function set terminal into raw mode */ settermraw(&oldtio); if (Forwardx11_opt) { /* * if forwardx11_opt is set call x11handler which * will act as a reader as well as a port forwarder */ x11handler(X11_comm_sock, news); } else { /* call interactive job's reader */ (void) reader(news); } /* reset terminal */ tcsetattr(0, TCSANOW, &oldtio); printf("\nqsub: job %s completed\n", new_jobname); exit_qsub(0); } else if (child > 0) { /* parent - start the writer function */ act.sa_handler = catchchild; if (sigaction(SIGCHLD, &act, NULL) < 0) exit_qsub(1); writer(news); /* all done - make sure the child is gone and reset the terminal */ kill(child, SIGTERM); shutdown(comm_sock, SHUT_RDWR); close(comm_sock); tcsetattr(0, TCSANOW, &oldtio); printf("\nqsub: job %s completed\n", new_jobname); exit_qsub(0); } else { perror("qsub: unable to fork"); exit_qsub(1); } } /** * @brief * Sets the filename to be used for the unix domain socket based comm. * This is formed by appending the UID and the target server name to the * filename. The length of the string is restricted to the length of the * global variable fl. This is fairly small (108 characters) for Linux. * * @param[out] fname - The filename in tmpdir that is used as the unix domain socket * file. * */ void get_comm_filename(char *fname) { char *env_svr = getenv(PBS_CONF_SERVER_NAME); char *env_port = getenv(PBS_CONF_BATCH_SERVICE_PORT); int count = 0; char buf[LARGE_BUF_LEN]; int len; unsigned char hash[SHA_DIGEST_LENGTH]; int i; count = snprintf(fname, MAXPIPENAME, "%s/pbs_%.16s_%lu_%.8s_%.32s_%.16s_%.5s", tmpdir, ((server_out == NULL || server_out[0] == 0) ? "default" : server_out), (unsigned long int) getuid(), cred_name, get_conf_path(), env_svr ? env_svr : "", env_port ? env_port : ""); if (count >= MAXPIPENAME) { count = snprintf(fname, MAXPIPENAME, "%s/pbs_", TMP_DIR); len = snprintf(buf, MAXPIPENAME, "%.16s_%lu_%.8s_%.32s_%.16s_%.5s", ((server_out == NULL || server_out[0] == 0) ? "default" : server_out), (unsigned long int) getuid(), cred_name, get_conf_path(), (env_svr == NULL) ? "" : env_svr, (env_port == NULL) ? "" : env_port); if (len + count < MAXPIPENAME) { pbs_strncpy(fname + count, buf, MAXPIPENAME - count); } else { if (SHA1((const unsigned char *) buf, SHA_DIGEST_LENGTH, (unsigned char *) &hash)) { for (i = 0; i < SHA_DIGEST_LENGTH; i++) sprintf(buf + (i * 2), "%02x", hash[i]); buf[SHA_DIGEST_LENGTH * 2] = 0; } pbs_strncpy(fname + count, buf, MAXPIPENAME - count); } } } /** * @brief * Check whether a unix domain socket file is available. * That is an indication that a background qsub might already be running. * * @param[out] fname - The filename used for the communication pipe/socket for * the communication between background and forground * qsub processes. * * @return int * @retval 0 - Not available * @retval 1 - available * */ int check_qsub_daemon(char *fname) { get_comm_filename(fname); if (access(fname, F_OK) == 0) { /* check if file is usable */ return 1; } return 0; } /** * @brief * The daemon_stuff Unix counterpart. * It creates a unix domain socket server and starts listening on it. * The umask is set to 077 so that the domain socket file is owned and * accessible by the user executing qsub only. Once a client (foreground * qsub) connects, it receives all the data from the foreground qsub and * executes do_submit, on the pre-established connection to pbs_server. * The connection to server was estiblished by the caller of this function * by calling do_connect(). * This function also does a "select" wait on input of data from foreground * qsub processes, and a close notification on the socket with pbs_server. * The select breaks if foreground qsubs connect, the pbs_server dies, or * the timeout of 1 minutes expires. For the latter two cases, this function * does a silent exit of the background qsub daemon. * */ static void daemon_stuff(void) { int sock, bindfd; struct sockaddr_un s_un; struct sockaddr from; socklen_t fromlen; int rc; fd_set readset; fd_set workset; struct timeval timeout; int n, maxfd; mode_t cmask = 0077; time_t connect_time = time(0); sigset_t newsigmask, oldsigmask; char *err_op = ""; char log_buf[LOG_BUF_SIZE]; int cred_timeout = 0; /* set umask so socket file created is only accessible by same user */ umask(cmask); sigemptyset(&newsigmask); sigaddset(&newsigmask, SIGPIPE); sigprocmask(SIG_BLOCK, &newsigmask, NULL); /* start up a unix domain socket to listen */ if ((bindfd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) { err_op = "socket"; goto error; } s_un.sun_family = AF_UNIX; snprintf(s_un.sun_path, sizeof(s_un.sun_path), "%s", fl); if (bind(bindfd, (const struct sockaddr *) &s_un, sizeof(s_un)) == -1) exit(1); /* dont go to error */ FD_ZERO(&readset); if (listen(bindfd, 1) != 0) { err_op = "listen"; goto error; } FD_SET(bindfd, &readset); FD_SET(sd_svr, &readset); maxfd = (bindfd > sd_svr) ? bindfd : sd_svr; while (1) { err_op = ""; memcpy(&workset, &readset, sizeof(readset)); timeout.tv_usec = 0; /* since timeout gets reset on Linux */ if (cred_timeout == 1) timeout.tv_sec = QSUB_DMN_TIMEOUT_SHORT; /* Short timeout to allow any foreground process to finsih before exiting */ else timeout.tv_sec = QSUB_DMN_TIMEOUT_LONG; n = select(maxfd + 1, &workset, NULL, NULL, &timeout); if (n == 0) goto out; /* daemon timed out waiting for connect from foreground */ else if (n == -1) { err_op = "select failed"; goto error; } /* * check if we are past the credential timeout * Error out even if it is close to CREDENTIAL_LIFETIME, as * request could take a while to reach server and get processed * Qsub then does a regular submit (new connection) */ if (cred_timeout == 0 && ((time(0) - connect_time) > (CREDENTIAL_LIFETIME - QSUB_DMN_TIMEOUT_LONG))) { unlink(fl); cred_timeout = 1; } /* Shut the qsub daemon if the server had closed the connection */ if (FD_ISSET(sd_svr, &workset)) { if (recv(sd_svr, &rc, 1, MSG_OOB) < 1) goto out; } /* accept the connection */ fromlen = sizeof(from); if ((sock = accept(bindfd, &from, &fromlen)) == -1) { err_op = "accept"; goto error; } if ((recv_attrl(&sock, &attrib) != 0) || (recv_string(&sock, destination) != 0) || (recv_string(&sock, script_tmp) != 0) || (recv_string(&sock, cred_name) != 0) || (recv_dyn_string(&sock, &v_value) != 0) || (recv_dyn_string(&sock, &basic_envlist) != 0) || (recv_dyn_string(&sock, &qsub_envlist) != 0) || (recv_string(&sock, qsub_cwd) != 0) || (recv_opts(&sock) != 0)) { err_op = "recv data from foreground"; goto error; } /* * At this point the background qsub daemon has received all the data from the * foreground. Lets tell the foreground that we have received the data, so that * if the we crashed at any point after this the foreground should not end up * submitting a duplicate job. However, if the foreground did not get this intimation, * then it could go ahead and do a regular job submit. */ rc = 0; if (dosend(&sock, (char *) &rc, sizeof(int)) != 0) { err_op = "send data to foreground"; goto error; } /* set the current work directory by doing a chdir */ if (chdir(qsub_cwd) != 0) { err_op = "chdir"; goto error; } if (setenv("PWD", qsub_cwd, 1) != 0) { err_op = "setenv"; goto error; } sigemptyset(&newsigmask); sigaddset(&newsigmask, SIGXCPU); sigaddset(&newsigmask, SIGXFSZ); sigaddset(&newsigmask, SIGTSTP); sigaddset(&newsigmask, SIGINT); sigaddset(&newsigmask, SIGSTOP); sigaddset(&newsigmask, SIGTERM); sigaddset(&newsigmask, SIGTSTP); sigaddset(&newsigmask, SIGALRM); sigaddset(&newsigmask, SIGQUIT); sigaddset(&newsigmask, SIGUSR1); sigaddset(&newsigmask, SIGUSR2); sigprocmask(SIG_BLOCK, &newsigmask, &oldsigmask); rc = do_submit2(retmsg); if (send_string(&sock, retmsg) != 0) { err_op = "send data to foreground"; goto error; } if (dosend(&sock, (char *) &rc, sizeof(int)) != 0) { err_op = "send data to foreground"; goto error; } close(sock); sigprocmask(SIG_SETMASK, &oldsigmask, NULL); qsub_free_attrl(attrib); attrib = NULL; free(v_value); v_value = NULL; free(basic_envlist); basic_envlist = NULL; free(qsub_envlist); qsub_envlist = NULL; if (cred_buf != NULL) { memset(cred_buf, 0, cred_len); free(cred_buf); cred_buf = NULL; } /* Exit the daemon if it can't submit the job */ if (rc == DMN_REFUSE_EXIT) goto out; } out: close(bindfd); if (cred_timeout != 1) unlink(fl); exit(0); error: sprintf(log_buf, "Background qsub: Failed at %s, errno=%d", err_op, errno); log_syslog(log_buf); unlink(fl); close(bindfd); exit(1); } /* * @brief * Try to submit job through daemon. On Unix, the daemon would be created by * forking during a prior invocation of the qsub command. The foregound qsub * process tries to send the job to the daemon using Unix domain sockets. * * @param[out] daemon_up - Indicate whether daemon is running * @param[out] do_regular_submit - Indicate whether to do regular submit * @param[in] qsub_exe - Name of the qsub command * * @return rc - Error code */ int daemon_submit(int *daemon_up, int *do_regular_submit) { int sock; /* UNIX domain socket for talking to daemon */ struct sockaddr_un s_un; sigset_t newsigmask; int rc = 0; again: /* * In case of Unix, use fork. Foreground checks if connection is * possible with background daemon. The communication used is unix * domain sockets. Only the specified user can connect to this socket * since the domain socket is created with a 0600 permission. * * If connection fails, proceed with qsub in the normal flow, and at * the end fork and stay in the background, while the foreground * process returns control to the shell. Subsequent qsubs will be able * to connect to this forked background qsub. * */ *daemon_up = check_qsub_daemon(fl); if (*daemon_up == 1) { /* pass information to daemon */ /* wait for job-id or error string */ if ((sock = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) return rc; s_un.sun_family = AF_UNIX; snprintf(s_un.sun_path, sizeof(s_un.sun_path), "%s", fl); if (connect(sock, (const struct sockaddr *) &s_un, sizeof(s_un)) == -1) { int refused = (errno == ECONNREFUSED); close(sock); if (refused) { /* daemon unavailable, del temp file, restart */ if (unlink(fl) != 0) return rc; goto again; } return rc; } /* block SIGPIPE on write() failures. */ sigemptyset(&newsigmask); sigaddset(&newsigmask, SIGPIPE); sigprocmask(SIG_BLOCK, &newsigmask, NULL); if ((send_attrl(&sock, attrib) == 0) && (send_string(&sock, destination) == 0) && (send_string(&sock, script_tmp) == 0) && (send_string(&sock, cred_name) == 0) && (send_string(&sock, v_value ? v_value : "") == 0) && (send_string(&sock, basic_envlist) == 0) && (send_string(&sock, qsub_envlist ? qsub_envlist : "") == 0) && (send_string(&sock, qsub_cwd) == 0) && (send_opts(&sock) == 0)) { /* * Read back the first error code from the background, * which confirms whether the background received our data. */ if (dorecv(&sock, (char *) &rc, sizeof(int)) == 0) { /* * We were able to send data to the background daemon. * Now, even if we fail to read back response from * background, we do not want to submit again. */ *do_regular_submit = 0; } /* read back response from background daemon */ if ((recv_string(&sock, retmsg) != 0) || (dorecv(&sock, (char *) &rc, sizeof(int)) != 0) || rc == DMN_REFUSE_EXIT) { /* * Something bad happened, either background submitted * and failed to send us response, or it failed before * submitting. If background qsub detects -V option, then * submit the job through foreground. */ if (rc != DMN_REFUSE_EXIT) { rc = -1; sprintf(retmsg, "Failed to recv data from background qsub\n"); /* Error message will be printed in caller */ } else *do_regular_submit = 1; } } /* going down, no need to free stuff */ close(sock); } return rc; }