diff --git a/.github/workflows/startup-fault-tests.yml b/.github/workflows/startup-fault-tests.yml new file mode 100644 index 00000000..e82d2c72 --- /dev/null +++ b/.github/workflows/startup-fault-tests.yml @@ -0,0 +1,238 @@ +name: startup-fault-tests + +on: + workflow_dispatch: + +permissions: + contents: read + packages: read + +concurrency: + group: startup-fault-tests-${{github.ref}} + cancel-in-progress: true + +jobs: + startup-fault-serial-ubuntu_x86: + name: Startup fault tests (Serial, Ubuntu) + environment: Spindle CI + runs-on: ubuntu-latest + timeout-minutes: 25 + steps: + - name: Check out Spindle + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + + - name: Setup Docker Compose + uses: docker/setup-compose-action@54042514f505b273907334ae2b9cdbb9a0213c1a + with: + version: latest + + - name: Build spindle-serial-ubuntu image + run: | + cd containers/spindle-serial-ubuntu + docker compose --progress=plain build + + - name: Bring spindle-serial-ubuntu up + run: | + cd containers/spindle-serial-ubuntu + docker compose up -d + + - name: Verify munge works in spindle-serial-ubuntu + run: | + docker exec spindlenode bash -c 'munge -n | unmunge' + + - name: Run spindle-serial-ubuntu startup fault tests + run: | + docker exec spindlenode bash -c 'cd Spindle-build/testsuite && ./run_startup_fault_tests.sh --launcher=serial' + + - name: Bring spindle-serial-ubuntu down + if: ${{ always() }} + continue-on-error: true + run: | + cd containers/spindle-serial-ubuntu + docker compose down + + startup-fault-serial-ubuntu_arm64: + name: Startup fault tests (Serial, Ubuntu, arm64) + environment: Spindle CI + runs-on: ubuntu-24.04-arm + timeout-minutes: 35 + steps: + - name: Check out Spindle + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + + - name: Setup Docker Compose + uses: docker/setup-compose-action@54042514f505b273907334ae2b9cdbb9a0213c1a + with: + version: latest + + - name: Build spindle-serial-ubuntu image + run: | + cd containers/spindle-serial-ubuntu + docker compose --progress=plain build + + - name: Bring spindle-serial-ubuntu up + run: | + cd containers/spindle-serial-ubuntu + docker compose up -d + + - name: Verify munge works in spindle-serial-ubuntu + run: | + docker exec spindlenode bash -c 'munge -n | unmunge' + + - name: Run spindle-serial-ubuntu startup fault tests + run: | + docker exec spindlenode bash -c 'cd Spindle-build/testsuite && ./run_startup_fault_tests.sh --launcher=serial' + + - name: Bring spindle-serial-ubuntu down + if: ${{ always() }} + continue-on-error: true + run: | + cd containers/spindle-serial-ubuntu + docker compose down + + startup-fault-flux-ubuntu: + name: Startup fault tests (Flux, Ubuntu) + environment: Spindle CI + runs-on: ubuntu-latest + timeout-minutes: 35 + steps: + - name: Check out Spindle + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + + - name: Setup Docker Compose + uses: docker/setup-compose-action@54042514f505b273907334ae2b9cdbb9a0213c1a + with: + version: latest + + - name: Build spindle-flux-ubuntu image + run: | + cd containers/spindle-flux-ubuntu + docker compose --progress=plain build + + - name: Bring spindle-flux-ubuntu up + run: | + cd containers/spindle-flux-ubuntu + docker compose up -d --wait --wait-timeout 60 + + - name: Verify munge works in spindle-flux-ubuntu + run: | + docker exec node-1 bash -c 'munge -n | unmunge' + + - name: Run spindle-flux-ubuntu startup fault tests + run: | + docker exec node-1 bash -c 'cd Spindle-build/testsuite && flux alloc --nodes=${workers} ./run_startup_fault_tests.sh --launcher=flux --nodes=${workers}' + + - name: Run spindle-flux-ubuntu session startup fault tests + run: | + docker exec node-1 bash -c 'cd Spindle-build/testsuite && flux alloc --nodes=${workers} ./run_startup_fault_tests.sh --launcher=flux --nodes=${workers} --session' + + - name: Bring spindle-flux-ubuntu down + if: ${{ always() }} + continue-on-error: true + run: | + cd containers/spindle-flux-ubuntu + docker compose down + + startup-fault-slurm-srun-ubuntu: + name: Startup fault tests (Slurm srun, Ubuntu) + environment: Spindle CI + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - name: Check out Spindle + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + + - name: Setup Docker Compose + uses: docker/setup-compose-action@54042514f505b273907334ae2b9cdbb9a0213c1a + with: + version: latest + + - name: Login to GitHub Container Registry + if: ${{ !env.ACT }} + uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Generate MariaDB configuration + run: | + cd containers/spindle-slurm-ubuntu/testing-srun + ./generate_config.sh + + - name: Build spindle-slurm-srun-ubuntu image + run: | + cd containers/spindle-slurm-ubuntu/testing-srun + docker compose --progress=plain build + + - name: Bring spindle-slurm-srun-ubuntu up + run: | + cd containers/spindle-slurm-ubuntu/testing-srun + docker compose up -d --wait --wait-timeout 120 + + - name: Verify munge works in spindle-slurm-srun-ubuntu + run: | + docker exec slurm-srun-head bash -c 'munge -n | unmunge' + + - name: Run spindle-slurm-srun-ubuntu startup fault tests + run: | + docker exec slurm-srun-head bash -c 'cd Spindle-build/testsuite && ./run_startup_fault_tests.sh --launcher=slurm --nodes=${workers}' + + - name: Bring spindle-slurm-srun-ubuntu down + if: ${{ always() }} + continue-on-error: true + run: | + cd containers/spindle-slurm-ubuntu/testing-srun + docker compose down + + startup-fault-slurm-plugin-ubuntu: + name: Startup fault tests (Slurm Plugin, Ubuntu) + environment: Spindle CI + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - name: Check out Spindle + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + + - name: Setup Docker Compose + uses: docker/setup-compose-action@54042514f505b273907334ae2b9cdbb9a0213c1a + with: + version: latest + + - name: Login to GitHub Container Registry + if: ${{ !env.ACT }} + uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Generate MariaDB configuration + run: | + cd containers/spindle-slurm-ubuntu/testing-plugin + ./generate_config.sh + + - name: Build spindle-slurm-plugin-ubuntu image + run: | + cd containers/spindle-slurm-ubuntu/testing-plugin + docker compose --progress=plain build + + - name: Bring spindle-slurm-plugin-ubuntu up + run: | + cd containers/spindle-slurm-ubuntu/testing-plugin + docker compose up -d --wait --wait-timeout 120 + + - name: Verify munge works in spindle-slurm-plugin-ubuntu + run: | + docker exec slurm-plugin-head bash -c 'munge -n | unmunge' + + - name: Run spindle-slurm-plugin-ubuntu startup fault tests + run: | + docker exec slurm-plugin-head bash -c 'cd Spindle-build/testsuite && ./run_startup_fault_tests.sh --launcher=slurm-plugin --nodes=${workers}' + + - name: Bring spindle-slurm-plugin-ubuntu down + if: ${{ always() }} + continue-on-error: true + run: | + cd containers/spindle-slurm-ubuntu/testing-plugin + docker compose down diff --git a/src/client/beboot/spindle_bootstrap.c b/src/client/beboot/spindle_bootstrap.c index dabd8516..50a312f6 100644 --- a/src/client/beboot/spindle_bootstrap.c +++ b/src/client/beboot/spindle_bootstrap.c @@ -78,7 +78,10 @@ static int establish_connection() return -1; send_pid(ldcsid); - send_rankinfo_query(ldcsid, &rankinfo[0], &rankinfo[1], &rankinfo[2], &rankinfo[3]); + if (send_rankinfo_query(ldcsid, &rankinfo[0], &rankinfo[1], &rankinfo[2], &rankinfo[3]) == -1) { + err_printf("Failed to get rank info from server\n"); + return -1; + } if (opts & OPT_NUMA) send_cpu(ldcsid, get_cur_cpu()); @@ -406,7 +409,10 @@ int main(int argc, char *argv[]) /** * Exec the user's application. **/ - setup_environment(); + if (setup_environment() != 0) { + err_printf("spindle_bootstrap could not get the cache path from the server\n"); + return -1; + } execvp(executable, cmdline); diff --git a/src/client/client_comlib/client_api_pipe.c b/src/client/client_comlib/client_api_pipe.c index 7f3df457..b9577764 100644 --- a/src/client/client_comlib/client_api_pipe.c +++ b/src/client/client_comlib/client_api_pipe.c @@ -26,6 +26,7 @@ Place, Suite 330, Boston, MA 02111-1307 USA #include #include #include +#include #include "client_heap.h" #include "ldcs_api_pipe.h" @@ -72,6 +73,8 @@ static int write_pipe(int fd, const void *data, int bytes) bwrite = left; bwrote = write(fd, dataptr, bwrite); if (bwrote == -1) { + if (errno == EINTR) + continue; err_printf("Failed to wrote bytes to pipe: %s\n", strerror(errno)); return -1; } @@ -95,9 +98,15 @@ static int read_pipe(int fd, void *data, int bytes) btoread = left; bread = read(fd, dataptr, btoread); if (bread == -1) { + if (errno == EINTR) + continue; err_printf("Error reading data from pipe: %s\n", strerror(errno)); return -1; } + if (bread == 0) { + err_printf("Server closed the pipe connection\n"); + return -1; + } left -= bread; dataptr += bread; } @@ -188,6 +197,33 @@ static int remap_to_high_fd(int fd) #endif } +static long long now_ms(void) +{ + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (long long) ts.tv_sec * 1000 + ts.tv_nsec / 1000000; +} + +static long long env_timelimit_ms(const char *name, long long default_ms) +{ + char *value = getenv(name); + if (value && atoi(value) > 0) + return 1000LL * atoi(value); + return default_ms; +} + +/* Timeout used for connecting to a server after the ready file has appeared */ +static long long connect_timeout_ms(void) +{ + return env_timelimit_ms("COBO_CONNECT_TIMELIMIT", 60 * 1000); +} + +/* Timeout used when waiting for the server's ready file to appear */ +static long long ready_timeout_ms(void) +{ + return env_timelimit_ms("SPINDLE_READY_TIMELIMIT", 4 * connect_timeout_ms()); +} + int client_open_connection_pipe(char* location, number_t number) { (void)number; @@ -197,6 +233,7 @@ int client_open_connection_pipe(char* location, number_t number) char fifo[MAX_PATH_LEN]; char ready[MAX_PATH_LEN]; int find_r_fd = 0, find_w_fd = 0; + int opened_in = 0, opened_out = 0; debug_printf("Client creating pipe for connection to server\n"); fd = get_new_fd_pipe(); @@ -205,20 +242,26 @@ int client_open_connection_pipe(char* location, number_t number) fdlist_pipe[fd].type = LDCS_PIPE_FD_TYPE_CONN; - /* wait for directory (at most one minute) */ + long long deadline = now_ms() + ready_timeout_ms(); + + /* wait for the server to be ready */ stat_cnt = 0; snprintf(ready, MAX_PATH_LEN, "%s/spindle_comm/ready", location); memset(&st, 0, sizeof(st)); - - while (((stat(ready, &st) == -1) || ((st.st_mode & (S_IRUSR | S_IWUSR)) == 0)) && - (stat_cnt<600)) { + while ((stat(ready, &st) == -1) || ((st.st_mode & (S_IRUSR | S_IWUSR)) == 0)) { + if (now_ms() >= deadline) { + err_printf("Timed out waiting for server ready file %s\n", ready); + return -1; + } if (stat_cnt % 10 == 0) debug_printf3("waiting: location %s does not exists (after %d seconds)\n", ready, stat_cnt/10); usleep(100000); /* .1 seconds */ stat_cnt++; } - - /* create incomming fifo */ + + /* create incoming fifo; open the read end first */ + fdlist_pipe[fd].in_fd = fdlist_pipe[fd].out_fd = 0; + fdlist_pipe[fd].out_fn = NULL; sprintf(fifo, "%s/spindle_comm/fifo-%d-0", location, getpid()); result = ldcs_mkfifo(fifo); if (result == -2) @@ -227,6 +270,20 @@ int client_open_connection_pipe(char* location, number_t number) return -1; fdlist_pipe[fd].in_fn = spindle_strdup(fifo); assert(fdlist_pipe[fd].in_fn); + if (find_r_fd) { + /* Fifo already exists. Find an existing open fd. */ + debug_printf3("Finding existing fd for %s\n", fdlist_pipe[fd].in_fn); + find_existing_fds(fdlist_pipe[fd].in_fn, NULL, &fdlist_pipe[fd].in_fd, &fdlist_pipe[fd].out_fd); + } + if (fdlist_pipe[fd].in_fd == 0) { + debug_printf3("Opening input fifo %s\n", fdlist_pipe[fd].in_fn); + fdlist_pipe[fd].in_fd = open(fdlist_pipe[fd].in_fn, O_RDONLY | O_NONBLOCK); + if (fdlist_pipe[fd].in_fd == -1) { + err_printf("Error opening input fifo %s: %s\n", fdlist_pipe[fd].in_fn, strerror(errno)); + goto fail; + } + opened_in = 1; + } /* create outgoing fifo */ sprintf(fifo, "%s/spindle_comm/fifo-%d-1", location, getpid()); @@ -234,43 +291,58 @@ int client_open_connection_pipe(char* location, number_t number) if (result == -2) find_w_fd = 1; else if (result == -1) - return -1; + goto fail; fdlist_pipe[fd].out_fn = spindle_strdup(fifo); - /* open fifos */ - fdlist_pipe[fd].in_fd = fdlist_pipe[fd].out_fd = 0; - if (find_w_fd || find_r_fd) { - /* Fifos already exist. Find existing open fds. */ - debug_printf3("Finding existing fds for %s and/or %s\n", - fdlist_pipe[fd].out_fn, fdlist_pipe[fd].in_fn); - find_existing_fds(find_r_fd ? fdlist_pipe[fd].in_fn : NULL, - find_w_fd ? fdlist_pipe[fd].out_fn : NULL, - &fdlist_pipe[fd].in_fd, &fdlist_pipe[fd].out_fd); - } - if (fdlist_pipe[fd].in_fd == 0) { - debug_printf3("Opening input fifo %s\n", fdlist_pipe[fd].in_fn); - fdlist_pipe[fd].in_fd = open(fdlist_pipe[fd].in_fn, O_RDONLY); - fdlist_pipe[fd].in_fd = remap_to_high_fd(fdlist_pipe[fd].in_fd); + if (find_w_fd) { + /* Fifo already exists. Find an existing open fd. */ + debug_printf3("Finding existing fd for %s\n", fdlist_pipe[fd].out_fn); + find_existing_fds(NULL, fdlist_pipe[fd].out_fn, &fdlist_pipe[fd].in_fd, &fdlist_pipe[fd].out_fd); } - debug_printf3("Opened input fifo %s = %d\n", fdlist_pipe[fd].in_fn, fdlist_pipe[fd].in_fd); if (fdlist_pipe[fd].out_fd == 0) { + /* will get ENXIO until the server has opened its read end */ debug_printf3("Opening output fifo %s\n", fdlist_pipe[fd].out_fn); - fdlist_pipe[fd].out_fd = open(fdlist_pipe[fd].out_fn, O_WRONLY); - fdlist_pipe[fd].out_fd = remap_to_high_fd(fdlist_pipe[fd].out_fd); + deadline = now_ms() + connect_timeout_ms(); + for (;;) { + fdlist_pipe[fd].out_fd = open(fdlist_pipe[fd].out_fn, O_WRONLY | O_NONBLOCK); + if (fdlist_pipe[fd].out_fd != -1 || errno != ENXIO) + break; + if (now_ms() >= deadline) { + err_printf("Timed out waiting for server to open %s\n", fdlist_pipe[fd].out_fn); + goto fail; + } + usleep(10000); + } + if (fdlist_pipe[fd].out_fd == -1) { + err_printf("Error opening output fifo %s: %s\n", fdlist_pipe[fd].out_fn, strerror(errno)); + goto fail; + } + opened_out = 1; } - debug_printf3("Opened output fifo %s = %d\n", fdlist_pipe[fd].out_fn, fdlist_pipe[fd].out_fd); + /* Turn nonblocking IO on the fifos off so as to not alter behavior after startup */ + fcntl(fdlist_pipe[fd].in_fd, F_SETFL, fcntl(fdlist_pipe[fd].in_fd, F_GETFL) & ~O_NONBLOCK); + fcntl(fdlist_pipe[fd].out_fd, F_SETFL, fcntl(fdlist_pipe[fd].out_fd, F_GETFL) & ~O_NONBLOCK); + if (opened_in) + fdlist_pipe[fd].in_fd = remap_to_high_fd(fdlist_pipe[fd].in_fd); + if (opened_out) + fdlist_pipe[fd].out_fd = remap_to_high_fd(fdlist_pipe[fd].out_fd); + debug_printf3("Opened fifos %s = %d, %s = %d\n", fdlist_pipe[fd].in_fn, fdlist_pipe[fd].in_fd, + fdlist_pipe[fd].out_fn, fdlist_pipe[fd].out_fd); - /* Check opened fifos */ - if (fdlist_pipe[fd].in_fd == -1) { - err_printf("Error opening input fifo %s: %s\n", fdlist_pipe[fd].in_fn, strerror(errno)); - return -1; - } - if (fdlist_pipe[fd].out_fd == -1) { - err_printf("Error opening output fifo %s: %s\n", fdlist_pipe[fd].out_fn, strerror(errno)); - return -1; - } + return fd; - return fd; + fail: + /* on failure, close our fds/fifos so the server will + drop this connection if it tries to use them */ + if (opened_in) + close(fdlist_pipe[fd].in_fd); + if (opened_out) + close(fdlist_pipe[fd].out_fd); + if (!find_r_fd) + unlink(fdlist_pipe[fd].in_fn); + if (!find_w_fd && fdlist_pipe[fd].out_fn) + unlink(fdlist_pipe[fd].out_fn); + return -1; } int client_register_connection_pipe(char *connection_str) diff --git a/src/cobo/cobo.c b/src/cobo/cobo.c index 0b4280bd..ac121bf0 100644 --- a/src/cobo/cobo.c +++ b/src/cobo/cobo.c @@ -30,11 +30,13 @@ Place, Suite 330, Boston, MA 02111-1307 USA #include #include #include +#include #include #include #include #include "ldcs_cobo.h" +#include "cobo_fault_inject.h" #include "spindle_debug.h" /* Reads environment variable, bails if not set */ @@ -130,6 +132,18 @@ static struct timeval tree_start, tree_end; #include "sion_debug.h" #endif +long long cobo_now_ms(void) +{ + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (long long) ts.tv_sec * 1000 + ts.tv_nsec / 1000000; +} + +int cobo_get_startup_timeout_ms(void) +{ + return (int) (cobo_connect_timelimit * 1000); +} + static cobo_preconnect_cb_t preconnect_cb = NULL; void cobo_register_preconnect_cb(cobo_preconnect_cb_t f) @@ -229,12 +243,15 @@ static int* cobo_int_dup(int* src, int n) } /* write size bytes from buf into fd, retry if necessary */ -static int cobo_write_fd_w_suppress(int fd, void* buf, int size, int suppress) +int cobo_write_fd_w_suppress(int fd, void* buf, int size, int suppress) { int rc; int n = 0; char* offset = (char*) buf; + if (fd < 0) + return -1; + while (n < size) { rc = write(fd, offset, size - n); @@ -253,7 +270,7 @@ static int cobo_write_fd_w_suppress(int fd, void* buf, int size, int suppress) err_printf("Unexpected return code of 0 from write to file descriptor (write(fd=%d,offset=%p,size=%d))\n", fd, offset, size-n); } else { - err_printf("Unexpected return code of 0 from write to file descriptor (write(fd=%d,offset=%p,size=%d))\n", + debug_printf3("Unexpected return code of 0 from write to file descriptor (write(fd=%d,offset=%p,size=%d))\n", fd, offset, size-n); } return -1; @@ -272,12 +289,18 @@ static int cobo_write_fd(int fd, void* buf, int size) return cobo_write_fd_w_suppress(fd, buf, size, 0); } -/* read size bytes into buf from fd, retry if necessary */ -static int cobo_read_fd_w_timeout(int fd, void* buf, int size, int usecs) +/* read size bytes into buf from fd, within timeout_ms, no limit if negative */ +int cobo_read_fd_w_timeout(int fd, void* buf, int size, int timeout_ms) { int rc; int n = 0; char* offset = (char*) buf; + long long deadline = cobo_now_ms() + timeout_ms; + + if (fd < 0) { + err_printf("Invalid file descriptor %d for read\n", fd); + return -1; + } struct pollfd fds; fds.fd = fd; @@ -285,13 +308,20 @@ static int cobo_read_fd_w_timeout(int fd, void* buf, int size, int usecs) fds.revents = 0x0; while (n < size) { - /* poll the connection with a timeout value */ - int poll_rc = poll(&fds, 1, usecs); + int wait_ms = -1; + if (timeout_ms >= 0) { + long long remaining = deadline - cobo_now_ms(); + wait_ms = remaining > 0 ? (int) remaining : 0; + } + int poll_rc = poll(&fds, 1, wait_ms); if (poll_rc < 0) { + if (errno == EINTR) + continue; err_printf("Polling file descriptor for read (read(fd=%d,offset=%p,size=%d) %m errno=%d)\n", fd, offset, size-n, errno); return -1; } else if (poll_rc == 0) { + debug_printf3("Timed out reading %d bytes from fd %d\n", size-n, fd); return -1; } @@ -341,10 +371,10 @@ static int cobo_read_fd_w_timeout(int fd, void* buf, int size, int usecs) return n; } -/* read size bytes into buf from fd, retry if necessary */ +/* read size bytes into buf from fd, within the startup time limit */ static int cobo_read_fd(int fd, void* buf, int size) { - return cobo_read_fd_w_timeout(fd, buf, size, -1); + return cobo_read_fd_w_timeout(fd, buf, size, cobo_get_startup_timeout_ms()); } /* Open a connection on socket FD to peer at ADDR (which LEN bytes long). @@ -518,6 +548,7 @@ static int cobo_connect_hostname(char* hostname, int rank) case HSHAKE_DROP_CONNECTION: debug_printf3("Handshake said to drop connection\n"); close(s); + s = -1; continue; case HSHAKE_ABORT: handle_security_error(spindle_handshake_last_error_str()); @@ -572,6 +603,7 @@ static int cobo_connect_hostname(char* hostname, int rank) /* if the connection test failed, close the socket, otherwise we've got a good connection */ if (test_failed) { close(s); + s = -1; } else { connected = 1; break; @@ -822,11 +854,42 @@ static int cobo_open_tree() /* accept a connection from parent and receive socket table */ int reply_timeout = cobo_connect_timeout * 100; + long long accept_deadline = cobo_now_ms() + cobo_get_startup_timeout_ms(); int have_parent = 0; + /* nonblocking as otherwiseo a connection reset between poll and accept would block */ + fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFL) | O_NONBLOCK); while (!have_parent) { struct sockaddr parent_addr; socklen_t parent_len = sizeof(parent_addr); + struct pollfd pfd; + long long remaining = accept_deadline - cobo_now_ms(); + if (remaining <= 0) { + err_printf("Timed out waiting for parent to connect\n"); + cobo_parent_fd = -1; + close(sockfd); + return -1; + } + pfd.fd = sockfd; + pfd.events = POLLIN; + pfd.revents = 0; + int prc = poll(&pfd, 1, (int) remaining); + if (prc < 0 && errno != EINTR) { + err_printf("Polling listen socket failed (%m errno=%d)\n", errno); + cobo_parent_fd = -1; + close(sockfd); + return -1; + } + if (prc <= 0) + continue; cobo_parent_fd = accept(sockfd, (struct sockaddr *) &parent_addr, &parent_len); + if (cobo_parent_fd < 0) { + if (errno == EINTR || errno == ECONNABORTED || + errno == EAGAIN || errno == EWOULDBLOCK) + continue; + err_printf("Accepting parent connection failed (%m errno=%d)\n", errno); + close(sockfd); + return -1; + } _cobo_opt_socket(cobo_parent_fd); @@ -929,6 +992,8 @@ static int cobo_open_tree() return -1; } + cobo_test_fault_point("tree", cobo_me); + /* if (cobo_me == 0) { for (i=0; i < cobo_nprocs; i++) { @@ -1102,7 +1167,8 @@ static int cobo_allreduce_max_int_tree(int* sendbuf, int* recvbuf) } /* broadcast result of reduction from rank 0 to all tasks */ - cobo_bcast_tree(recvbuf, sizeof(int)); + if (cobo_bcast_tree(recvbuf, sizeof(int)) != COBO_SUCCESS) + return -1; return rc; } @@ -1232,7 +1298,10 @@ int cobo_barrier() /* use allreduce of an int for our barrier */ int dummy; int myint = 1; - cobo_allreduce_max_int_tree(&myint, &dummy); + if (cobo_allreduce_max_int_tree(&myint, &dummy) != COBO_SUCCESS) { + err_printf("Barrier failed\n"); + return -1; + } cobo_gettimeofday(&end); debug_printf3("Exiting cobo_barrier(), took %f seconds for %d procs\n", cobo_getsecs(&end,&start), cobo_nprocs); @@ -1371,7 +1440,8 @@ static int cobo_allreduce_max_int(int* sendint, int* recvint) debug_printf3("Starting cobo_allreducemaxint()"); /* compute allreduce via tree */ - cobo_allreduce_max_int_tree(sendint, recvint); + if (cobo_allreduce_max_int_tree(sendint, recvint) != COBO_SUCCESS) + return -1; cobo_gettimeofday(&end); debug_printf3("Exiting cobo_allreducemaxint(), took %f seconds for %d procs\n", cobo_getsecs(&end,&start), cobo_nprocs); @@ -1471,19 +1541,42 @@ int cobo_allreduce( uint64_t *pval, cobo_op_t op ){ } /* broadcast result of reduction from rank 0 to all tasks */ - cobo_bcast_tree(pval, sizeof(int64_t)); + if (cobo_bcast_tree(pval, sizeof(int64_t)) != COBO_SUCCESS) + return -1; return COBO_SUCCESS; } +/* read COBO_CONNECT_* settings from the environment */ +static void cobo_read_env_settings(void) +{ + char *value; + + /* milliseconds */ + if ((value = cobo_getenv("COBO_CONNECT_TIMEOUT", ENV_OPTIONAL)) && atoi(value) > 0) + cobo_connect_timeout = atoi(value); + + /* exponential backoff factor for connect */ + if ((value = cobo_getenv("COBO_CONNECT_BACKOFF", ENV_OPTIONAL)) && atoi(value) > 0) + cobo_connect_backoff = atoi(value); + + /* milliseconds to sleep before rescanning ports */ + if ((value = cobo_getenv("COBO_CONNECT_SLEEP", ENV_OPTIONAL)) && atoi(value) > 0) + cobo_connect_sleep = atoi(value); + + /* seconds */ + if ((value = cobo_getenv("COBO_CONNECT_TIMELIMIT", ENV_OPTIONAL)) && atoi(value) > 0) + cobo_connect_timelimit = (double) atoi(value); +} + /* provide list of ports and number of ports as input, get number of tasks and my rank as output */ int cobo_open(uint64_t sessionid, int* portlist, int num_ports, int* rank, int* num_ranks) { setvbuf(stdout, NULL, _IONBF, 0); - char *value; struct timeval start, end; cobo_gettimeofday(&start); + cobo_test_exit_hosts(); /* we now know this process is a client, although we don't know what our rank is yet */ cobo_me = -1; @@ -1495,25 +1588,7 @@ int cobo_open(uint64_t sessionid, int* portlist, int num_ports, int* rank, int* * Get information from environment * ======================================================= */ - /* milliseconds */ - if ((value = cobo_getenv("COBO_CONNECT_TIMEOUT\n", ENV_OPTIONAL))) { - cobo_connect_timeout = atoi(value); - } - - /* exponential backoff factor for connect */ - if ((value = cobo_getenv("COBO_CONNECT_BACKOFF\n", ENV_OPTIONAL))) { - cobo_connect_backoff = atoi(value); - } - - /* milliseconds to sleep before rescanning ports */ - if ((value = cobo_getenv("COBO_CONNECT_SLEEP\n", ENV_OPTIONAL))) { - cobo_connect_sleep = atoi(value); - } - - /* seconds */ - if ((value = cobo_getenv("COBO_CONNECT_TIMELIMIT\n", ENV_OPTIONAL))) { - cobo_connect_timelimit = (double) atoi(value); - } + cobo_read_env_settings(); debug_printf3("In cobo_init():\n" \ "COBO_CONNECT_TIMEOUT: %d, COBO_CONNECT_BACKOFF: %d, COBO_CONNECT_SLEEP: %d, COBO_CONNECT_TIMELIMIT: %d\n", @@ -1528,7 +1603,10 @@ int cobo_open(uint64_t sessionid, int* portlist, int num_ports, int* rank, int* } /* open the tree */ - cobo_open_tree(); + if (cobo_open_tree() != COBO_SUCCESS) { + err_printf("Failed to connect the tree\n"); + return -1; + } /* need to check that tree opened successfully before returning, so do a barrier */ if (cobo_barrier() != COBO_SUCCESS) { @@ -1596,6 +1674,8 @@ int cobo_server_get_root_socket(int* fd) /* given a hostlist and portlist where clients are running, open the tree and assign ranks to clients */ int cobo_server_open(uint64_t sessionid, char** hostlist, int num_hosts, int* portlist, int num_ports) { + cobo_read_env_settings(); + /* at this point, we know this process is the server, so set its rank */ cobo_me = -2; cobo_nprocs = num_hosts; diff --git a/src/cobo/cobo_comm.c b/src/cobo/cobo_comm.c index 7f2084b2..34fbd442 100644 --- a/src/cobo/cobo_comm.c +++ b/src/cobo/cobo_comm.c @@ -36,123 +36,21 @@ Place, Suite 330, Boston, MA 02111-1307 USA #include "cobo_comm.h" #include "config.h" -/* read size bytes into buf from fd, retry if necessary */ -static int ldcs_cobo_read_fd_w_timeout(int fd, void* buf, int size, int usecs) -{ - int rc; - int n = 0; - char* offset = (char*) buf; - - struct pollfd fds; - fds.fd = fd; - fds.events = POLLIN; - fds.revents = 0x0; - - while (n < size) { - /* poll the connection with a timeout value */ - int poll_rc = poll(&fds, 1, usecs); - if (poll_rc < 0) { - err_printf("Polling file descriptor for read (read(fd=%d,offset=%p,size=%d) %m errno=%d)\n", - fd, offset, size-n, errno); - return -1; - } else if (poll_rc == 0) { - return -1; - } - - /* check the revents field for errors */ - if (fds.revents & POLLHUP) { - debug_printf3("Hang up error on poll for read(fd=%d,offset=%p,size=%d)\n", - fd, offset, size-n); - return -1; - } - - if (fds.revents & POLLERR) { - debug_printf3("Error on poll for read(fd=%d,offset=%p,size=%d)\n", - fd, offset, size-n); - return -1; - } - - if (fds.revents & POLLNVAL) { - err_printf("Invalid request on poll for read(fd=%d,offset=%p,size=%d)\n", - fd, offset, size-n); - return -1; - } - - if (!(fds.revents & POLLIN)) { - err_printf("No errors found, but POLLIN is not set for read(fd=%d,offset=%p,size=%d)\n", - fd, offset, size-n); - return -1; - } - - /* poll returned that fd is ready for reading */ - rc = read(fd, offset, size - n); - - if (rc < 0) { - if(errno == EINTR || errno == EAGAIN) { continue; } - err_printf("Reading from file descriptor (read(fd=%d,offset=%p,size=%d) %m errno=%d)\n", - fd, offset, size-n, errno); - return rc; - } else if(rc == 0) { - err_printf("Unexpected return code of 0 from read from file descriptor (read(fd=%d,offset=%p,size=%d) revents=%x)\n", - fd, offset, size-n, fds.revents); - return -1; - } - - offset += rc; - n += rc; - } - - return n; -} - /* read size bytes into buf from fd, retry if necessary */ int ldcs_cobo_read_fd(int fd, void* buf, int size) { - return ldcs_cobo_read_fd_w_timeout(fd, buf, size, -1); + return cobo_read_fd_w_timeout(fd, buf, size, -1); } -/* write size bytes from buf into fd, retry if necessary */ -static int ldcs_cobo_write_fd_w_suppress(int fd, void* buf, int size, int suppress) +int ldcs_cobo_read_fd_timeout(int fd, void* buf, int size, int timeout_ms) { - int rc; - int n = 0; - char* offset = (char*) buf; - - while (n < size) { - rc = write(fd, offset, size - n); - - if (rc < 0) { - if(errno == EINTR || errno == EAGAIN) { continue; } - if (!suppress) { - err_printf("Writing to file descriptor (write(fd=%d,offset=%p,size=%d) %m errno=%d)\n", - fd, offset, size-n, errno); - } else { - debug_printf3("Writing to file descriptor (write(fd=%d,offset=%p,size=%d) %m errno=%d)\n", - fd, offset, size-n, errno); - } - return rc; - } else if(rc == 0) { - if (!suppress) { - err_printf("Unexpected return code of 0 from write to file descriptor (write(fd=%d,offset=%p,size=%d))\n", - fd, offset, size-n); - } else { - debug_printf3("Unexpected return code of 0 from write to file descriptor (write(fd=%d,offset=%p,size=%d))\n", - fd, offset, size-n); - } - return -1; - } - - offset += rc; - n += rc; - } - - return n; + return cobo_read_fd_w_timeout(fd, buf, size, timeout_ms); } /* write size bytes from buf into fd, retry if necessary */ int ldcs_cobo_write_fd(int fd, void* buf, int size) { - return ldcs_cobo_write_fd_w_suppress(fd, buf, size, 0); + return cobo_write_fd_w_suppress(fd, buf, size, 0); } int ll_read(int fd, void *buf, size_t count) @@ -299,3 +197,19 @@ int write_msg_nosignal(int fd, ldcs_message_t *msg) return 0; } + +int read_msg_timeout(int fd, ldcs_message_t *msg, int timeout_ms) +{ + if (ldcs_cobo_read_fd_timeout(fd, msg, sizeof(*msg), timeout_ms) < 0) + return -1; + msg->data = NULL; + if (msg->header.len) { + msg->data = (char *) malloc(msg->header.len); + if (ldcs_cobo_read_fd_timeout(fd, msg->data, msg->header.len, timeout_ms) < 0) { + free(msg->data); + msg->data = NULL; + return -1; + } + } + return 0; +} diff --git a/src/cobo/cobo_comm.h b/src/cobo/cobo_comm.h index f6612ad9..c9c1ddd6 100644 --- a/src/cobo/cobo_comm.h +++ b/src/cobo/cobo_comm.h @@ -19,14 +19,19 @@ Place, Suite 330, Boston, MA 02111-1307 USA #define COBO_SUCCESS (0) +/* Value rank 0 sends the FE once every server has joined the tree */ +#define COBO_READY_ACK 13 + #include "ldcs_api.h" int ldcs_cobo_read_fd(int fd, void* buf, int size); +int ldcs_cobo_read_fd_timeout(int fd, void* buf, int size, int timeout_ms); int ldcs_cobo_write_fd(int fd, void* buf, int size); int ll_write(int fd, void *buf, size_t count); int ll_send_nosignal(int fd, void *buf, size_t count); int ll_read(int fd, void *buf, size_t count); int write_msg(int fd, ldcs_message_t *msg); int write_msg_nosignal(int fd, ldcs_message_t *msg); +int read_msg_timeout(int fd, ldcs_message_t *msg, int timeout_ms); #endif /* _COBO_COMM_H */ diff --git a/src/cobo/cobo_fault_inject.c b/src/cobo/cobo_fault_inject.c new file mode 100644 index 00000000..565d38ee --- /dev/null +++ b/src/cobo/cobo_fault_inject.c @@ -0,0 +1,116 @@ +/* +This file is part of Spindle. For copyright information see the COPYRIGHT +file in the top level directory, or at +https://github.com/hpc/Spindle/blob/master/COPYRIGHT + +This program is free software; you can redistribute it and/or modify it under +the terms of the GNU Lesser General Public License (as published by the Free Software +Foundation) version 2.1 dated February 1999. This program 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 terms +and conditions of the GNU Lesser General Public License for more details. You should +have received a copy of the GNU Lesser General Public License along with this +program; if not, write to the Free Software Foundation, Inc., 59 Temple +Place, Suite 330, Boston, MA 02111-1307 USA +*/ + +/* COBO fault injection for startup resiliency testing */ + +#if !defined(_GNU_SOURCE) +#define _GNU_SOURCE +#endif + +#include +#include +#include +#include +#include +#include +#include "cobo_fault_inject.h" + +/* Check whether comma-separated list contains item. */ +static int list_contains(const char *list, const char *item) +{ + size_t len = strlen(item); + const char *p = list; + while (p && *p) { + const char *end = strchr(p, ','); + size_t n = end ? (size_t) (end - p) : strlen(p); + if (n == len && strncmp(p, item, len) == 0) + return 1; + p = end ? end + 1 : NULL; + } + return 0; +} + +/* Stall until the parent closes or the stall time elapses */ +static void stall(void) +{ + const char *secs_s = getenv("SPINDLE_TEST_BE_STALL_SECS"); + long long deadline = cobo_now_ms() + 1000LL * (secs_s ? atoi(secs_s) : 300); + struct pollfd pfd; + int parent_fd = -1; + cobo_get_parent_socket(&parent_fd); + pfd.fd = parent_fd; + pfd.events = POLLRDHUP; + for (;;) { + long long remaining = deadline - cobo_now_ms(); + if (remaining <= 0) + return; + pfd.revents = 0; + int rc = poll(&pfd, 1, (int) remaining); + if (rc < 0 && errno == EINTR) + continue; + if (rc != 0) + _exit(1); + } +} + +/* Handles fault points in cobo_open_tree and just before READY. + * phase is either "tree" or "ready" to select the desired point. */ +void cobo_test_fault_point(const char *phase, int rank) +{ + char rank_s[16]; + const char *selected_phase = getenv("SPINDLE_TEST_BE_FAULT_PHASE"); + const char *exit_list = getenv("SPINDLE_TEST_BE_EXIT"); + const char *stall_list = getenv("SPINDLE_TEST_BE_STALL"); + + /* Fault injection is disabled unless SPINDLE_TEST_BE_FAULT_PHASE is set and + * at least one of SPINDLE_TEST_BE_EXIT or SPINDLE_TEST_BE_STALL is set */ + if (!selected_phase || !*selected_phase) + return; + if ((!exit_list || !*exit_list) && (!stall_list || !*stall_list)) + return; + + /* Check if this is the right phase for the fault point. */ + if (strcmp(selected_phase, phase) != 0) + return; + + /* Check if this is the right rank for the fault point. */ + snprintf(rank_s, sizeof(rank_s), "%d", rank); + + /* If this rank is listed in SPINDLE_TEST_BE_EXIT, + * exit at the fault point. */ + if (exit_list && list_contains(exit_list, rank_s)) + _exit(1); + + /* If this rank is listed in SPINDLE_TEST_BE_STALL, + * stall at the fault point. */ + if (stall_list && list_contains(stall_list, rank_s)) + stall(); +} + +/* The fault point at the beginning of cobo_open occurs before + * ranks have been assigned and so uses hostname rather than + * rank. Later fault points use rank so we can make some ranks + * within an node exit or stall while others proceed normally. */ +void cobo_test_exit_hosts(void) +{ + char host[256]; + const char *list = getenv("SPINDLE_TEST_BE_EXIT_HOSTS"); + if (!list || gethostname(host, sizeof(host)) != 0) + return; + host[sizeof(host) - 1] = '\0'; + if (list_contains(list, host)) + _exit(1); +} diff --git a/src/cobo/cobo_fault_inject.h b/src/cobo/cobo_fault_inject.h new file mode 100644 index 00000000..d859d2a6 --- /dev/null +++ b/src/cobo/cobo_fault_inject.h @@ -0,0 +1,31 @@ +/* +This file is part of Spindle. For copyright information see the COPYRIGHT +file in the top level directory, or at +https://github.com/hpc/Spindle/blob/master/COPYRIGHT + +This program is free software; you can redistribute it and/or modify it under +the terms of the GNU Lesser General Public License (as published by the Free Software +Foundation) version 2.1 dated February 1999. This program 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 terms +and conditions of the GNU Lesser General Public License for more details. You should +have received a copy of the GNU Lesser General Public License along with this +program; if not, write to the Free Software Foundation, Inc., 59 Temple +Place, Suite 330, Boston, MA 02111-1307 USA +*/ + +#ifndef _COBO_FAULT_INJECT_H +#define _COBO_FAULT_INJECT_H + +#include "ldcs_cobo.h" + +#if defined(COBO_NAMESPACE) +#define cobo_test_fault_point COMBINE(COBO_NAMESPACE, cobo_test_fault_point) +#define cobo_test_exit_hosts COMBINE(COBO_NAMESPACE, cobo_test_exit_hosts) +#endif + +/* Fault injection for startup resiliency testing */ +void cobo_test_fault_point(const char *phase, int rank); +void cobo_test_exit_hosts(void); + +#endif diff --git a/src/cobo/ldcs_cobo.h b/src/cobo/ldcs_cobo.h index fafbda6a..2b6ac512 100644 --- a/src/cobo/ldcs_cobo.h +++ b/src/cobo/ldcs_cobo.h @@ -64,6 +64,10 @@ extern "C" { #define cobo_bcast_down COMBINE(COBO_NAMESPACE, cobo_bcast_down) #define cobo_get_child_socket COMBINE(COBO_NAMESPACE, cobo_get_child_socket) #define cobo_set_handshake COMBINE(COBO_NAMESPACE, cobo_set_handshake) +#define cobo_get_startup_timeout_ms COMBINE(COBO_NAMESPACE, cobo_get_startup_timeout_ms) +#define cobo_now_ms COMBINE(COBO_NAMESPACE, cobo_now_ms) +#define cobo_read_fd_w_timeout COMBINE(COBO_NAMESPACE, cobo_read_fd_w_timeout) +#define cobo_write_fd_w_suppress COMBINE(COBO_NAMESPACE, cobo_write_fd_w_suppress) #define cobo_preconnect_cb_t COMBINE(COBO_NAMESPACE, cobo_preconnect_cb_t) #define cobo_register_preconnect_cb COMBINE(COBO_NAMESPACE, cobo_register_preconnect_cb) #endif @@ -173,6 +177,18 @@ int cobo_get_child_socket(int num, int *fd); void cobo_set_handshake(handshake_protocol_t *hs); +/* Startup time limit in milliseconds (COBO_CONNECT_TIMELIMIT, default 60s) */ +int cobo_get_startup_timeout_ms(void); + +/* Monotonic clock in milliseconds */ +long long cobo_now_ms(void); + +/* Read size bytes from fd within timeout_ms overall, no limit if negative */ +int cobo_read_fd_w_timeout(int fd, void* buf, int size, int timeout_ms); + +/* Write size bytes to fd; with suppress, errors are written as debug logs instead */ +int cobo_write_fd_w_suppress(int fd, void* buf, int size, int suppress); + void handle_security_error(const char *msg); int initialize_handshake_security(handshake_protocol_t *protocol); diff --git a/src/fe/cobo/Makefile.am b/src/fe/cobo/Makefile.am index c9138843..e7ed23a2 100644 --- a/src/fe/cobo/Makefile.am +++ b/src/fe/cobo/Makefile.am @@ -1,3 +1,3 @@ noinst_LTLIBRARIES = libldcs_cobo.la -libldcs_cobo_la_SOURCES = $(top_srcdir)/../cobo/cobo.c $(top_srcdir)/../cobo/handshake.c $(top_srcdir)/../cobo/cobo_comm.c $(top_srcdir)/../cobo/cobo_handshake.c +libldcs_cobo_la_SOURCES = $(top_srcdir)/../cobo/cobo.c $(top_srcdir)/../cobo/handshake.c $(top_srcdir)/../cobo/cobo_comm.c $(top_srcdir)/../cobo/cobo_fault_inject.c $(top_srcdir)/../cobo/cobo_handshake.c AM_CPPFLAGS = -I$(top_srcdir)/../logging -I$(top_srcdir)/../cobo -I$(top_srcdir)/../include $(MUNGE_CFLAGS) $(GCRYPT_CFLAGS) diff --git a/src/fe/cobo/Makefile.in b/src/fe/cobo/Makefile.in index f204598a..0aeaf508 100644 --- a/src/fe/cobo/Makefile.in +++ b/src/fe/cobo/Makefile.in @@ -114,6 +114,7 @@ am__dirstamp = $(am__leading_dot)dirstamp am_libldcs_cobo_la_OBJECTS = $(top_builddir)/../cobo/cobo.lo \ $(top_builddir)/../cobo/handshake.lo \ $(top_builddir)/../cobo/cobo_comm.lo \ + $(top_builddir)/../cobo/cobo_fault_inject.lo \ $(top_builddir)/../cobo/cobo_handshake.lo libldcs_cobo_la_OBJECTS = $(am_libldcs_cobo_la_OBJECTS) AM_V_lt = $(am__v_lt_@AM_V@) @@ -137,6 +138,7 @@ depcomp = $(SHELL) $(top_srcdir)/../../scripts/depcomp am__maybe_remake_depfiles = depfiles am__depfiles_remade = $(top_builddir)/../cobo/$(DEPDIR)/cobo.Plo \ $(top_builddir)/../cobo/$(DEPDIR)/cobo_comm.Plo \ + $(top_builddir)/../cobo/$(DEPDIR)/cobo_fault_inject.Plo \ $(top_builddir)/../cobo/$(DEPDIR)/cobo_handshake.Plo \ $(top_builddir)/../cobo/$(DEPDIR)/handshake.Plo am__mv = mv -f @@ -337,7 +339,7 @@ top_build_prefix = @top_build_prefix@ top_builddir = @top_builddir@ top_srcdir = @top_srcdir@ noinst_LTLIBRARIES = libldcs_cobo.la -libldcs_cobo_la_SOURCES = $(top_srcdir)/../cobo/cobo.c $(top_srcdir)/../cobo/handshake.c $(top_srcdir)/../cobo/cobo_comm.c $(top_srcdir)/../cobo/cobo_handshake.c +libldcs_cobo_la_SOURCES = $(top_srcdir)/../cobo/cobo.c $(top_srcdir)/../cobo/handshake.c $(top_srcdir)/../cobo/cobo_comm.c $(top_srcdir)/../cobo/cobo_fault_inject.c $(top_srcdir)/../cobo/cobo_handshake.c AM_CPPFLAGS = -I$(top_srcdir)/../logging -I$(top_srcdir)/../cobo -I$(top_srcdir)/../include $(MUNGE_CFLAGS) $(GCRYPT_CFLAGS) all: all-am @@ -398,6 +400,9 @@ $(top_builddir)/../cobo/handshake.lo: \ $(top_builddir)/../cobo/cobo_comm.lo: \ $(top_builddir)/../cobo/$(am__dirstamp) \ $(top_builddir)/../cobo/$(DEPDIR)/$(am__dirstamp) +$(top_builddir)/../cobo/cobo_fault_inject.lo: \ + $(top_builddir)/../cobo/$(am__dirstamp) \ + $(top_builddir)/../cobo/$(DEPDIR)/$(am__dirstamp) $(top_builddir)/../cobo/cobo_handshake.lo: \ $(top_builddir)/../cobo/$(am__dirstamp) \ $(top_builddir)/../cobo/$(DEPDIR)/$(am__dirstamp) @@ -415,6 +420,7 @@ distclean-compile: @AMDEP_TRUE@@am__include@ @am__quote@$(top_builddir)/../cobo/$(DEPDIR)/cobo.Plo@am__quote@ # am--include-marker @AMDEP_TRUE@@am__include@ @am__quote@$(top_builddir)/../cobo/$(DEPDIR)/cobo_comm.Plo@am__quote@ # am--include-marker +@AMDEP_TRUE@@am__include@ @am__quote@$(top_builddir)/../cobo/$(DEPDIR)/cobo_fault_inject.Plo@am__quote@ # am--include-marker @AMDEP_TRUE@@am__include@ @am__quote@$(top_builddir)/../cobo/$(DEPDIR)/cobo_handshake.Plo@am__quote@ # am--include-marker @AMDEP_TRUE@@am__include@ @am__quote@$(top_builddir)/../cobo/$(DEPDIR)/handshake.Plo@am__quote@ # am--include-marker @@ -584,6 +590,7 @@ clean-am: clean-generic clean-libtool clean-noinstLTLIBRARIES \ distclean: distclean-am -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_comm.Plo + -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_fault_inject.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_handshake.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/handshake.Plo -rm -f Makefile @@ -633,6 +640,7 @@ installcheck-am: maintainer-clean: maintainer-clean-am -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_comm.Plo + -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_fault_inject.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_handshake.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/handshake.Plo -rm -f Makefile diff --git a/src/fe/comlib/cobo_fe_comm.c b/src/fe/comlib/cobo_fe_comm.c index 0df792e8..817e9d01 100644 --- a/src/fe/comlib/cobo_fe_comm.c +++ b/src/fe/comlib/cobo_fe_comm.c @@ -68,14 +68,28 @@ int ldcs_audit_server_fe_md_open ( char **hostlist, int numhosts, unsigned int p portlist[num_ports] = 0; debug_printf2("Opening with port %d - %d\n", portlist[0], portlist[num_ports-1]); - cobo_server_open(unique_id, hostlist, numhosts, portlist, num_ports); + rc = cobo_server_open(unique_id, hostlist, numhosts, portlist, num_ports); free(portlist); + if (rc != COBO_SUCCESS) { + err_printf("Could not connect to the root Spindle server\n"); + return -1; + } - cobo_server_get_root_socket(&root_fd); - - ldcs_cobo_read_fd(root_fd, &ack, sizeof(ack)); + if (cobo_server_get_root_socket(&root_fd) != COBO_SUCCESS) { + err_printf("Failed to establish connection to the root Spindle server\n"); + return -1; + } + + if (ldcs_cobo_read_fd_timeout(root_fd, &ack, sizeof(ack), 2 * cobo_get_startup_timeout_ms()) < 0) { + err_printf("Spindle servers did not acknowledge startup\n"); + return -1; + } + if (ack != COBO_READY_ACK) { + err_printf("Unexpected startup ack message %d from Spindle servers\n", ack); + return -1; + } - return(rc); + return 0; } int ldcs_audit_server_fe_md_waitfor_close() @@ -99,69 +113,41 @@ int ldcs_audit_server_fe_md_waitfor_close() } } -static int wait_for_read_or_timeout(int fd, int timeout_seconds) +int ldcs_audit_server_fe_md_waitfor_alive() { - int nfds, result, error; - fd_set rset; - struct timeval timeout; + int fd; + ldcs_message_t out_msg; - timeout.tv_sec = timeout_seconds; - timeout.tv_usec = 0; - - for (;;) { - FD_ZERO(&rset); - FD_SET(fd, &rset); - nfds = fd+1; + debug_printf2("Blocking while waiting for spindle alive\n"); - result = select(nfds, &rset, NULL, NULL, &timeout); - if (result == -1 && errno == EINTR) { - continue; - } - else if (result == -1) { - error = errno; - err_printf("Failure during select call: %s\n", strerror(error)); - return -1; - } - else if (result == 0) { - err_printf("Timeout during select call\n"); - return -2; - } - else if (result == 1 && FD_ISSET(fd, &rset)) { - return 0; - } - else { - err_printf("Unexpected return code %d during select\n", result); - return -1; - } + cobo_server_get_root_socket(&fd); + + memset(&out_msg, 0, sizeof(out_msg)); + if (read_msg_timeout(fd, &out_msg, cobo_get_startup_timeout_ms()) == -1) { + err_printf("No alive response from Spindle servers within %d seconds\n", cobo_get_startup_timeout_ms() / 1000); + return -1; } + if (out_msg.header.type == LDCS_MSG_ALIVE_RESP) + return 0; + err_printf("Unexpected message of type %d\n", (int) out_msg.header.type); + return -1; } - -int ldcs_audit_server_fe_md_waitfor_alive(int timeout_seconds) + +int ldcs_audit_server_fe_md_waitfor_preload() { - int fd, result; + int fd; ldcs_message_t out_msg; - debug_printf2("Blocking while waiting for spindle alive\n"); + debug_printf2("Blocking while waiting for spindle preload\n"); cobo_server_get_root_socket(&fd); - result = wait_for_read_or_timeout(fd, timeout_seconds); - if (result == -1) { - err_printf("Error waiting for alive message to be ready\n"); - return -1; - } - else if (result == -2) { - err_printf("Timeout waiting for alive message\n"); - return -1; - } - memset(&out_msg, 0, sizeof(out_msg)); - result = read_msg(fd, &out_msg); - if (result == -1) { - err_printf("ERROR reading message while waiting for server alive\n"); + if (read_msg(fd, &out_msg) == -1) { + err_printf("ERROR reading message while waiting for server preload\n"); return -1; } - if (out_msg.header.type == LDCS_MSG_ALIVE_RESP) + if (out_msg.header.type == LDCS_MSG_PRELOAD_DONE) return 0; err_printf("Unexpected message of type %d\n", (int) out_msg.header.type); return -1; @@ -192,7 +178,7 @@ int ldcs_audit_server_fe_broadcast(ldcs_message_t *msg, void *data) debug_printf("Broadcasting message %s to daemons\n",_message_type_to_str( msg->header.type ) ); cobo_server_get_root_socket(&root_fd); - return write_msg(root_fd, msg); + return write_msg_nosignal(root_fd, msg); } diff --git a/src/fe/comlib/fe_comm.h b/src/fe/comlib/fe_comm.h index 39b139e8..1fffb337 100644 --- a/src/fe/comlib/fe_comm.h +++ b/src/fe/comlib/fe_comm.h @@ -28,7 +28,8 @@ int ldcs_audit_server_fe_md_open(char **hostlist, int numhosts, unsigned int por unique_id_t unique_id, void **data); int ldcs_audit_server_fe_md_close(void *data); int ldcs_audit_server_fe_md_waitfor_close(); -int ldcs_audit_server_fe_md_waitfor_alive(int timeout_seconds); +int ldcs_audit_server_fe_md_waitfor_alive(); +int ldcs_audit_server_fe_md_waitfor_preload(); int ldcs_audit_server_fe_broadcast(ldcs_message_t *msg, void *data); #if defined(__cplusplus) diff --git a/src/fe/startup/spindle_fe.cc b/src/fe/startup/spindle_fe.cc index 86c535f1..8c2ab4f5 100644 --- a/src/fe/startup/spindle_fe.cc +++ b/src/fe/startup/spindle_fe.cc @@ -42,8 +42,6 @@ static const char *logging_file = NULL; static const char spindle_bootstrap[] = LIBEXECDIR "/spindle_bootstrap"; static bool sendAndWaitForAlive(); -#define STARTUP_TIMEOUT 60 - using namespace std; template @@ -413,9 +411,12 @@ int spindleInitFE(const char **hosts, spindle_args_t *params) params->bundle_cachesize_kb, params->crash_log); printSpindleFlags(params->opts); debug_printf("Starting FE servers with hostlist of size %u on port %u\n", hosts_size, params->port); - ldcs_audit_server_fe_md_open(const_cast(hosts), hosts_size, - params->port, params->num_ports, params->unique_id, - &md_data_ptr); + if (ldcs_audit_server_fe_md_open(const_cast(hosts), hosts_size, + params->port, params->num_ports, params->unique_id, + &md_data_ptr) == -1) { + err_printf("Could not start the Spindle server tree\n"); + goto fail; + } /* Broadcast parameters */ debug_printf("Sending parameters to servers\n"); @@ -426,20 +427,44 @@ int spindleInitFE(const char **hosts, spindle_args_t *params) msg.header.type = LDCS_MSG_SETTINGS; msg.header.len = param_buffer_size; msg.data = static_cast(param_buffer); - ldcs_audit_server_fe_broadcast(&msg, md_data_ptr); + int result; + result = ldcs_audit_server_fe_broadcast(&msg, md_data_ptr); free(param_buffer); + if (result == -1) { + err_printf("Could not send settings to Spindle servers\n"); + goto fail; + } + + /* Wait for servers to indicate startup */ + if (!sendAndWaitForAlive()) { + err_printf("Spindle servers did not report ready\n"); + goto fail; + } /* Broadcast preload contents */ if (preload_msg) { debug_printf("Sending message with preload information\n"); - ldcs_audit_server_fe_broadcast(preload_msg, md_data_ptr); + result = ldcs_audit_server_fe_broadcast(preload_msg, md_data_ptr); cleanPreloadMsg(preload_msg); + if (result == -1) { + err_printf("Could not send preload list to Spindle servers\n"); + return -1; + } + + /* The root server accepts no client connections while it reads the + * preload files, so the job must not start until it is done. */ + if (ldcs_audit_server_fe_md_waitfor_preload() == -1) { + err_printf("Spindle servers did not finish the preload\n"); + return -1; + } } - /* Wait for servers to indicate startup */ - sendAndWaitForAlive(); + return 0; - return 0; + fail: + if (preload_msg) + cleanPreloadMsg(preload_msg); + return -1; } int spindleWaitForCloseFE(spindle_args_t *params) @@ -508,7 +533,7 @@ static bool sendAndWaitForAlive() return false; } - result = ldcs_audit_server_fe_md_waitfor_alive(STARTUP_TIMEOUT); + result = ldcs_audit_server_fe_md_waitfor_alive(); if (result == -1) { debug_printf("Failure waiting for alive message\n"); return false; diff --git a/src/fe/startup/spindle_fe_main.cc b/src/fe/startup/spindle_fe_main.cc index 530f269e..fed534c8 100644 --- a/src/fe/startup/spindle_fe_main.cc +++ b/src/fe/startup/spindle_fe_main.cc @@ -156,6 +156,7 @@ int main(int argc, char *argv[]) int iresult = spindleInitFE(hosts, params); if (iresult == -1) { err_printf("spindleInitFE returned an error\n"); + launcher->handleShutdown(); return -1; } debug_printf("Done calling spindleInitFE\n"); @@ -240,6 +241,7 @@ int main(int argc, char *argv[]) int iresult = spindleInitFE(hosts, params); if (iresult == -1) { err_printf("[LMON FE] spindleInitFE returned an error\n"); + launcher->handleShutdown(); return -1; } debug_printf("Done calling spindleInitFE\n"); diff --git a/src/fe/startup/spindle_session.cc b/src/fe/startup/spindle_session.cc index 4b61ab55..54f2e577 100644 --- a/src/fe/startup/spindle_session.cc +++ b/src/fe/startup/spindle_session.cc @@ -56,10 +56,10 @@ static int sock = -1; #define RET_END_SESSION 3 void signal_init_done(); -static void setup_init_done_pipe(); +static int setup_init_done_pipe(); static void setup_init_done_pipe_parent(); static void setup_init_done_pipe_child(); -static void waitfor_init_done(); +static bool waitfor_init_done(); extern bool getRandom(void *bytes, size_t bytes_size); @@ -457,8 +457,9 @@ int init_session(spindle_args_t *args, const ConfigMap &config, Launcher *launch (void)! write(1, "\n", 1); } - setup_init_done_pipe(); - + if (setup_init_done_pipe() == -1) + exit(-1); + close(0); open("/dev/null", O_RDONLY); close(1); @@ -473,9 +474,10 @@ int init_session(spindle_args_t *args, const ConfigMap &config, Launcher *launch exit(-1); } else if (pid > 0) { - debug_printf("Session daemon startup was successful. Exiting\n"); setup_init_done_pipe_parent(); - waitfor_init_done(); + if (!waitfor_init_done()) + exit(-1); + debug_printf("Session daemon startup was successful. Exiting\n"); exit(0); } else { @@ -614,16 +616,17 @@ int return_session_cmd(app_id_t appid, int app_argc, char **app_argv) #define PIPE_RD 0 #define PIPE_WR 1 static int init_done_pipe[2] = { -1, -1 }; -void setup_init_done_pipe() +static int setup_init_done_pipe() { int result, error; - result = pipe(init_done_pipe); + result = pipe2(init_done_pipe, O_CLOEXEC); if (result == -1) { error = errno; err_printf("Could not setup init_done_pipe: %s\n", strerror(error)); init_done_pipe[PIPE_RD] = init_done_pipe[PIPE_WR] = -1; - return; + return -1; } + return 0; } void setup_init_done_pipe_parent() { @@ -657,14 +660,11 @@ void signal_init_done() { init_done_pipe[PIPE_WR] = -1; } -void waitfor_init_done() +static bool waitfor_init_done() { int result, error; char x = '0'; - - if (init_done_pipe[PIPE_RD] == -1) - return; - + debug_printf2("Waiting for session start to finish before exiting launcher exec\n"); do { result = read(init_done_pipe[PIPE_RD], &x, 1); @@ -673,9 +673,13 @@ void waitfor_init_done() error = errno; err_printf("Could not read from init_done pipe: %s\n", strerror(error)); } - if (x != 'x') { + else if (result == 0) { + err_printf("Session daemon exited before finishing startup\n"); + } + else if (x != 'x') { err_printf("Unexpected character from %c pipe_rd\n", x); } close(init_done_pipe[PIPE_RD]); init_done_pipe[PIPE_RD] = -1; + return result == 1 && x == 'x'; } diff --git a/src/flux/flux-spindle.c b/src/flux/flux-spindle.c index 138137df..a3651f97 100644 --- a/src/flux/flux-spindle.c +++ b/src/flux/flux-spindle.c @@ -598,6 +598,22 @@ static int sp_init (flux_plugin_t *p, else unsetenv("SPINDLE_NO_CRASH_DEDUP"); + /* Pass through the startup time limit and testsuite fault injection */ + { + static const char *passthrough[] = { + "COBO_CONNECT_TIMELIMIT", "SPINDLE_TEST_BE_EXIT", "SPINDLE_TEST_BE_EXIT_HOSTS", + "SPINDLE_TEST_BE_STALL", "SPINDLE_TEST_BE_STALL_SECS", "SPINDLE_TEST_BE_FAULT_PHASE", NULL + }; + const char *val; + int j; + for (j = 0; passthrough[j]; j++) { + if ((val = flux_shell_getenv(shell, passthrough[j]))) + setenv(passthrough[j], val, 1); + else + unsetenv(passthrough[j]); + } + } + debug_printf(1, "initializing spindle flux plugin\n"); /* Spindle requires that TMPDIR is set. Propagate TMPDIR from job diff --git a/src/server/auditserver/ldcs_audit_server_handlers.c b/src/server/auditserver/ldcs_audit_server_handlers.c index 47e1e8d7..8020e8ee 100644 --- a/src/server/auditserver/ldcs_audit_server_handlers.c +++ b/src/server/auditserver/ldcs_audit_server_handlers.c @@ -2100,6 +2100,7 @@ static int handle_preload_filelist(ldcs_process_data_t *procdata, ldcs_message_t int num_dirs, num_files, i, global_result = 0, result; char *data = (char *) msg->data; char *pathname; + ldcs_message_t done_msg; debug_printf2("At top of handle_preload_filelist\n"); @@ -2160,6 +2161,15 @@ static int handle_preload_filelist(ldcs_process_data_t *procdata, ldcs_message_t global_result = -1; } + done_msg.header.type = LDCS_MSG_PRELOAD_DONE; + done_msg.header.len = 0; + done_msg.data = NULL; + result = ldcs_audit_server_md_to_frontend(procdata, &done_msg); + if (result == -1) { + err_printf("Failure sending preload done to FE\n"); + global_result = -1; + } + return global_result; } @@ -3027,7 +3037,10 @@ int handle_cachepath_consensus(ldcs_process_data_t *procdata) debug_printf3( " procdata->commpath = %s\n", procdata->commpath ); debug_printf3( " num_children = %d\n", num_children ); - ldcs_audit_server_md_allreduce_AND( &procdata->cachepath_bitidx ); + if (ldcs_audit_server_md_allreduce_AND( &procdata->cachepath_bitidx ) == -1) { + err_printf("Cachepath consensus reduction failed\n"); + return -1; + } debug_printf3( "The consensus value for procdata->cachepath_bitidx is: %#"PRIx64"\n", procdata->cachepath_bitidx ); if( procdata->cachepath_bitidx == 0 ) { diff --git a/src/server/auditserver/ldcs_audit_server_md.h b/src/server/auditserver/ldcs_audit_server_md.h index a4640370..203182b4 100644 --- a/src/server/auditserver/ldcs_audit_server_md.h +++ b/src/server/auditserver/ldcs_audit_server_md.h @@ -108,7 +108,8 @@ int ldcs_audit_server_md_get_num_children(ldcs_process_data_t *procdata); int ldcs_audit_server_md_is_parent(node_peer_t peer); -void ldcs_audit_server_md_allreduce_AND( uint64_t *val ); +int ldcs_audit_server_md_allreduce_AND( uint64_t *val ); +void ldcs_audit_server_md_test_fault_point(const char *phase, int md_rank); #if defined(__cplusplus) diff --git a/src/server/auditserver/ldcs_audit_server_md_cobo.c b/src/server/auditserver/ldcs_audit_server_md_cobo.c index 27393a55..9bcd2bbd 100644 --- a/src/server/auditserver/ldcs_audit_server_md_cobo.c +++ b/src/server/auditserver/ldcs_audit_server_md_cobo.c @@ -30,6 +30,7 @@ #include "ldcs_audit_server_handlers.h" #include "ldcs_cache.h" #include "ldcs_cobo.h" +#include "cobo_fault_inject.h" #include "cobo_comm.h" #include "config.h" @@ -93,8 +94,9 @@ int ldcs_audit_server_md_init(unsigned int port, unsigned int num_ports, /* initialize the client (read environment variables) */ debug_printf2("Opening cobo with port %d - %d\n", portlist[0], portlist[num_ports-1]); if (cobo_open(unique_id, (int *) portlist, num_ports, &my_rank, &ranks) != COBO_SUCCESS) { - printf("Failed to init\n"); - exit(1); + err_printf("Failed to open the server tree\n"); + free(portlist); + return -1; } debug_printf2("cobo_open complete. Cobo rank %d/%d\n", my_rank, ranks); free(portlist); @@ -106,19 +108,24 @@ int ldcs_audit_server_md_init(unsigned int port, unsigned int num_ports, cobo_get_num_childs(&fanout); data->server_stat.md_fan_out = data->md_fan_out = fanout; - cobo_barrier(); + if (cobo_barrier() != COBO_SUCCESS) { + err_printf("Server tree barrier failed\n"); + return -1; + } /* send ack about being ready */ - if (data->md_rank == 0) { - int root_fd, ack=13; + if (data->md_rank == 0) { + int root_fd, ack = COBO_READY_ACK; /* send fe client signal to stop (ack) */ - cobo_get_parent_socket(&root_fd); - - ldcs_cobo_write_fd(root_fd, &ack, sizeof(ack)); + if (cobo_get_parent_socket(&root_fd) != COBO_SUCCESS || + ldcs_cobo_write_fd(root_fd, &ack, sizeof(ack)) < 0) { + err_printf("Could not send ready ack to FE\n"); + return -1; + } debug_printf3("sent FE client signal that server are ready %d\n",ack); } - + return(rc); } @@ -261,13 +268,12 @@ int ldcs_audit_server_md_trash_bytes(node_peer_t peer, size_t size) int ldcs_audit_server_md_recv_from_parent(ldcs_message_t *msg) { - int fd, rc; - node_peer_t peer; + int fd = -1; - cobo_get_parent_socket(&fd); - rc = read_msg(fd, &peer, msg); - if (rc == -1) { - debug_printf2("Error return during recv from parent.\n"); + if (cobo_get_parent_socket(&fd) != COBO_SUCCESS) + return -1; + if (read_msg_timeout(fd, msg, cobo_get_startup_timeout_ms()) == -1) { + err_printf("No message from parent within the startup time limit\n"); return -1; } return 0; @@ -298,7 +304,10 @@ int ldcs_audit_server_md_cobo_CB(int fd, int nc, void *data) return -1; } + /* A handler failure is not a failure of the *connection*. */ rc = handle_server_message(ldcs_process_data, peer, &msg); + if (rc == -1) + err_printf("Error handling %s from fd %d\n", _message_type_to_str(msg.header.type), fd); ldcs_process_data->server_stat.md_cb.cnt++; ldcs_process_data->server_stat.md_cb.time+=(ldcs_get_time() - starttime); @@ -306,7 +315,7 @@ int ldcs_audit_server_md_cobo_CB(int fd, int nc, void *data) if (msg.data) free(msg.data); - return(rc); + return 0; } int ldcs_audit_server_md_send(ldcs_process_data_t *ldcs_process_data, ldcs_message_t *msg, node_peer_t peer) @@ -402,6 +411,11 @@ int ldcs_audit_server_md_get_num_children(ldcs_process_data_t *procdata) return num_childs; } -void ldcs_audit_server_md_allreduce_AND( uint64_t *val ){ - cobo_allreduce( val, COBO_OP_BITWISE_AND ); +int ldcs_audit_server_md_allreduce_AND( uint64_t *val ){ + return cobo_allreduce( val, COBO_OP_BITWISE_AND ) == COBO_SUCCESS ? 0 : -1; +} + +void ldcs_audit_server_md_test_fault_point(const char *phase, int md_rank) +{ + cobo_test_fault_point(phase, md_rank); } diff --git a/src/server/auditserver/ldcs_audit_server_process.c b/src/server/auditserver/ldcs_audit_server_process.c index 246d931c..40121ca4 100644 --- a/src/server/auditserver/ldcs_audit_server_process.c +++ b/src/server/auditserver/ldcs_audit_server_process.c @@ -81,7 +81,10 @@ int ldcs_audit_server_network_setup(unsigned int port, unsigned int num_ports, u memset(&ldcs_process_data, 0, sizeof(ldcs_process_data)); /* Initialize server->server network */ - ldcs_audit_server_md_init(port, num_ports, unique_id, &ldcs_process_data); + if (ldcs_audit_server_md_init(port, num_ports, unique_id, &ldcs_process_data) == -1) { + err_printf("Error initializing server network\n"); + return -1; + } /* Use network to broadcast configuration parameters */ ldcs_message_t msg; @@ -226,6 +229,7 @@ int ldcs_audit_server_process(spindle_args_t *args) return -1; } ldcs_process_data.serverid = serverid; + ldcs_audit_server_md_test_fault_point("ready", ldcs_process_data.md_rank); fd = ldcs_get_fd(serverid); ldcs_process_data.serverfd = fd; diff --git a/src/server/auditserver/ldcs_audit_server_server_cb.c b/src/server/auditserver/ldcs_audit_server_server_cb.c index 547cf883..f844abcf 100644 --- a/src/server/auditserver/ldcs_audit_server_server_cb.c +++ b/src/server/auditserver/ldcs_audit_server_server_cb.c @@ -30,7 +30,7 @@ int _ldcs_server_CB ( int infd, int serverid, void *data ) { (void)infd; int rc=0; ldcs_process_data_t *ldcs_process_data = (ldcs_process_data_t *) data ; - int nc, fd, more_avail; + int nc, fd, more_avail, connid; double cb_starttime; cb_starttime=ldcs_get_time(); @@ -52,7 +52,12 @@ int _ldcs_server_CB ( int infd, int serverid, void *data ) { } if(nc==ldcs_process_data->client_table_size) _error("internal error with client table (table full)"); - ldcs_process_data->client_table[nc].connid = ldcs_open_server_connections(serverid, nc, &more_avail); + connid = ldcs_open_server_connections(serverid, nc, &more_avail); + if (connid == -1) { + debug_printf("No new client connection accepted\n"); + continue; + } + ldcs_process_data->client_table[nc].connid = connid; ldcs_process_data->client_table[nc].state = LDCS_CLIENT_STATUS_ACTIVE; ldcs_process_data->client_table[nc].null_msg_cnt = 0; ldcs_process_data->client_table[nc].query_open = 0; diff --git a/src/server/cobo/Makefile.am b/src/server/cobo/Makefile.am index 9b643ae1..533c9054 100644 --- a/src/server/cobo/Makefile.am +++ b/src/server/cobo/Makefile.am @@ -1,4 +1,4 @@ noinst_LTLIBRARIES = libldcs_cobo.la -libldcs_cobo_la_SOURCES = $(top_srcdir)/../cobo/cobo.c $(top_srcdir)/../cobo/handshake.c $(top_srcdir)/../cobo/cobo_comm.c $(top_srcdir)/../cobo/cobo_handshake.c +libldcs_cobo_la_SOURCES = $(top_srcdir)/../cobo/cobo.c $(top_srcdir)/../cobo/handshake.c $(top_srcdir)/../cobo/cobo_comm.c $(top_srcdir)/../cobo/cobo_fault_inject.c $(top_srcdir)/../cobo/cobo_handshake.c AM_CPPFLAGS = -I$(top_srcdir)/../logging -I$(top_srcdir)/../cobo -I$(top_srcdir)/../include $(MUNGE_CFLAGS) $(GCRYPT_CFLAGS) diff --git a/src/server/cobo/Makefile.in b/src/server/cobo/Makefile.in index fb25289e..50fcfbe2 100644 --- a/src/server/cobo/Makefile.in +++ b/src/server/cobo/Makefile.in @@ -113,6 +113,7 @@ am__dirstamp = $(am__leading_dot)dirstamp am_libldcs_cobo_la_OBJECTS = $(top_builddir)/../cobo/cobo.lo \ $(top_builddir)/../cobo/handshake.lo \ $(top_builddir)/../cobo/cobo_comm.lo \ + $(top_builddir)/../cobo/cobo_fault_inject.lo \ $(top_builddir)/../cobo/cobo_handshake.lo libldcs_cobo_la_OBJECTS = $(am_libldcs_cobo_la_OBJECTS) AM_V_lt = $(am__v_lt_@AM_V@) @@ -136,6 +137,7 @@ depcomp = $(SHELL) $(top_srcdir)/../../scripts/depcomp am__maybe_remake_depfiles = depfiles am__depfiles_remade = $(top_builddir)/../cobo/$(DEPDIR)/cobo.Plo \ $(top_builddir)/../cobo/$(DEPDIR)/cobo_comm.Plo \ + $(top_builddir)/../cobo/$(DEPDIR)/cobo_fault_inject.Plo \ $(top_builddir)/../cobo/$(DEPDIR)/cobo_handshake.Plo \ $(top_builddir)/../cobo/$(DEPDIR)/handshake.Plo am__mv = mv -f @@ -339,7 +341,7 @@ top_build_prefix = @top_build_prefix@ top_builddir = @top_builddir@ top_srcdir = @top_srcdir@ noinst_LTLIBRARIES = libldcs_cobo.la -libldcs_cobo_la_SOURCES = $(top_srcdir)/../cobo/cobo.c $(top_srcdir)/../cobo/handshake.c $(top_srcdir)/../cobo/cobo_comm.c $(top_srcdir)/../cobo/cobo_handshake.c +libldcs_cobo_la_SOURCES = $(top_srcdir)/../cobo/cobo.c $(top_srcdir)/../cobo/handshake.c $(top_srcdir)/../cobo/cobo_comm.c $(top_srcdir)/../cobo/cobo_fault_inject.c $(top_srcdir)/../cobo/cobo_handshake.c AM_CPPFLAGS = -I$(top_srcdir)/../logging -I$(top_srcdir)/../cobo -I$(top_srcdir)/../include $(MUNGE_CFLAGS) $(GCRYPT_CFLAGS) all: all-am @@ -400,6 +402,9 @@ $(top_builddir)/../cobo/handshake.lo: \ $(top_builddir)/../cobo/cobo_comm.lo: \ $(top_builddir)/../cobo/$(am__dirstamp) \ $(top_builddir)/../cobo/$(DEPDIR)/$(am__dirstamp) +$(top_builddir)/../cobo/cobo_fault_inject.lo: \ + $(top_builddir)/../cobo/$(am__dirstamp) \ + $(top_builddir)/../cobo/$(DEPDIR)/$(am__dirstamp) $(top_builddir)/../cobo/cobo_handshake.lo: \ $(top_builddir)/../cobo/$(am__dirstamp) \ $(top_builddir)/../cobo/$(DEPDIR)/$(am__dirstamp) @@ -417,6 +422,7 @@ distclean-compile: @AMDEP_TRUE@@am__include@ @am__quote@$(top_builddir)/../cobo/$(DEPDIR)/cobo.Plo@am__quote@ # am--include-marker @AMDEP_TRUE@@am__include@ @am__quote@$(top_builddir)/../cobo/$(DEPDIR)/cobo_comm.Plo@am__quote@ # am--include-marker +@AMDEP_TRUE@@am__include@ @am__quote@$(top_builddir)/../cobo/$(DEPDIR)/cobo_fault_inject.Plo@am__quote@ # am--include-marker @AMDEP_TRUE@@am__include@ @am__quote@$(top_builddir)/../cobo/$(DEPDIR)/cobo_handshake.Plo@am__quote@ # am--include-marker @AMDEP_TRUE@@am__include@ @am__quote@$(top_builddir)/../cobo/$(DEPDIR)/handshake.Plo@am__quote@ # am--include-marker @@ -586,6 +592,7 @@ clean-am: clean-generic clean-libtool clean-noinstLTLIBRARIES \ distclean: distclean-am -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_comm.Plo + -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_fault_inject.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_handshake.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/handshake.Plo -rm -f Makefile @@ -635,6 +642,7 @@ installcheck-am: maintainer-clean: maintainer-clean-am -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_comm.Plo + -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_fault_inject.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/cobo_handshake.Plo -rm -f $(top_builddir)/../cobo/$(DEPDIR)/handshake.Plo -rm -f Makefile diff --git a/src/server/comlib/ldcs_api_pipe.c b/src/server/comlib/ldcs_api_pipe.c index ca233788..eb9a7cbc 100644 --- a/src/server/comlib/ldcs_api_pipe.c +++ b/src/server/comlib/ldcs_api_pipe.c @@ -164,10 +164,17 @@ int ldcs_open_server_connections_pipe(int fd, int nc, int *more_avail) { break; } } + if (!ldcs_notify_more_avail(fdlist_pipe[fd].notify_fd)) { + *more_avail = 0; + return -1; + } } connfd=get_new_fd_pipe(); - if(connfd<0) return(-1); + if(connfd<0) { + *more_avail=ldcs_notify_more_avail(fdlist_pipe[fd].notify_fd); + return(-1); + } fdlist_pipe[connfd].serverfd=fd; fdlist_pipe[connfd].type=LDCS_PIPE_FD_TYPE_CONN; @@ -178,12 +185,32 @@ int ldcs_open_server_connections_pipe(int fd, int nc, int *more_avail) { fdlist_pipe[connfd].in_fn = strdup(fifo); debug_printf3("before open fifo '%s'\n",fdlist_pipe[connfd].out_fn); - if (-1 == (fifoid = open(fdlist_pipe[connfd].out_fn, O_WRONLY))) _error("open fifo failed"); + /* the client opens its read end before creating -1. + * If we get ENXIO that means it has died.*/ + fifoid = open(fdlist_pipe[connfd].out_fn, O_WRONLY | O_NONBLOCK); + if (fifoid == -1) { + debug_printf("Dropping connection from pid %d: %s\n", pid, strerror(errno)); + unlink(fdlist_pipe[connfd].out_fn); + unlink(fdlist_pipe[connfd].in_fn); + free_fd_pipe(connfd); + *more_avail = ldcs_notify_more_avail(fdlist_pipe[fd].notify_fd); + return -1; + } + fcntl(fifoid, F_SETFL, fcntl(fifoid, F_GETFL) & ~O_NONBLOCK); debug_printf3("after open fifo (out): -> fifoid=%d\n",fifoid); fdlist_pipe[connfd].out_fd=fifoid; debug_printf3("before open fifo '%s'\n",fdlist_pipe[connfd].in_fn); - if (-1 == (fifoid = open(fdlist_pipe[connfd].in_fn, O_RDONLY|O_NONBLOCK))) _error("open fifo failed"); + fifoid = open(fdlist_pipe[connfd].in_fn, O_RDONLY | O_NONBLOCK); + if (fifoid == -1) { + debug_printf("Dropping connection from pid %d: %s\n", pid, strerror(errno)); + close(fdlist_pipe[connfd].out_fd); + unlink(fdlist_pipe[connfd].out_fn); + unlink(fdlist_pipe[connfd].in_fn); + free_fd_pipe(connfd); + *more_avail = ldcs_notify_more_avail(fdlist_pipe[fd].notify_fd); + return -1; + } /* if (-1 == (fifoid = open(fdlist_pipe[connfd].in_fn, O_RDONLY))) _error("open fifo failed"); */ debug_printf3("after open fifo (in) : -> fifoid=%d\n",fifoid); fdlist_pipe[connfd].in_fd=fifoid; diff --git a/src/server/comlib/ldcs_api_pipe_notify.c b/src/server/comlib/ldcs_api_pipe_notify.c index 115696e7..b835c160 100644 --- a/src/server/comlib/ldcs_api_pipe_notify.c +++ b/src/server/comlib/ldcs_api_pipe_notify.c @@ -138,7 +138,8 @@ char *ldcs_notify_get_next_file(int fd) { if (ldcs_fdlist_nt[fd].buffer_enddata < 0) { if (errno == EINTR) { - /* need to reissue system call */ + ldcs_fdlist_nt[fd].buffer_enddata = 0; + return NULL; } else { _error ("read on notify failed"); } diff --git a/src/server/startup/spindle_be.cc b/src/server/startup/spindle_be.cc index a2716a65..9e3ebe2a 100644 --- a/src/server/startup/spindle_be.cc +++ b/src/server/startup/spindle_be.cc @@ -121,7 +121,7 @@ int spindleRunBE(unsigned int port, unsigned int num_ports, unique_id_t unique_i int (*post_setup)(spindle_args_t *)) { int result; - spindle_args_t args; + spindle_args_t args = {0}; LOGGING_INIT(const_cast("Server")); @@ -137,8 +137,7 @@ int spindleRunBE(unsigned int port, unsigned int num_ports, unique_id_t unique_i result = ldcs_audit_server_network_setup(port, num_ports, unique_id, &setup_data, &setup_data_size); if (result == -1) { err_printf("Error setting up network in spindleRunBE\n"); - if (args.startup_type == startup_external) - LOGGING_FINI; + LOGGING_FINI; return -1; } unpack_data(&args, setup_data, setup_data_size); diff --git a/src/slurm_plugin/slurm_plugin.c b/src/slurm_plugin/slurm_plugin.c index 7f82cce4..ea7be905 100644 --- a/src/slurm_plugin/slurm_plugin.c +++ b/src/slurm_plugin/slurm_plugin.c @@ -376,6 +376,21 @@ static int forward_environment_to_slurmstepd(spank_t spank) free(envVal); } + { + static const char *passthrough[] = { + "COBO_CONNECT_TIMELIMIT", "SPINDLE_TEST_BE_EXIT", "SPINDLE_TEST_BE_EXIT_HOSTS", + "SPINDLE_TEST_BE_STALL", "SPINDLE_TEST_BE_STALL_SECS", "SPINDLE_TEST_BE_FAULT_PHASE", NULL + }; + int i; + for (i = 0; passthrough[i]; i++) { + envVal = readSpankEnv(spank, passthrough[i]); + if (envVal) { + setenv(passthrough[i], envVal, 1); + free(envVal); + } + } + } + return 0; } diff --git a/testsuite/Makefile.am b/testsuite/Makefile.am index 43a0d299..2cff5ec9 100644 --- a/testsuite/Makefile.am +++ b/testsuite/Makefile.am @@ -3,7 +3,7 @@ noinst_PROGRAMS = libgenerator ABS_TEST_DIR = $(abspath $(top_builddir)/testsuite) BUILT_SOURCES = libtest10.so libtest11.so libtest12.so libtest13.so libtest14.so libtest15.so libtest16.so libtest17.so libtest18.so libtest19.so libtest20.so libtest50.so libtest100.so libtest500.so libtest1000.so libtest2000.so libtest4000.so libtest6000.so libtest8000.so libtest10000.so libtls1.c libtls2.c libtls3.c libtls4.c libtls5.c libtls6.c libtls7.c libtls8.c libtls9.c libtls10.c libtls11.c libtls12.c libtls13.c libtls14.c libtls15.c libtls16.c libtls17.c libtls18.c libtls19.c libtls20.c libsymlink.so libdepC.so libdepB.so libdepA.so libcxxexceptB.so libcxxexceptA.so origin_dir/liboriginlib.so origin_dir/origin_subdir/liborigintarget.so libtestoutput.so libfuncdict.so runTests run_driver run_driver_rm spindle.rc preload_file_list test_driver test_driver_libs retzero_rx retzero_r retzero_x retzero_ badinterp hello_r.py hello_x.py hello_rx.py hello_.py hello_l.py badlink.py spindle_exec_test spindle_deactivated.sh liblocal.so symbind_test interpreter_test interpreter_test_dir/interpreter_test_perl alias/aliastest.py exec_shell.sh exec_shell.tcsh exec_shell_expected_output exec_shell exec_shell_env.sh libbigtls1024.so libbigtls204800.so libbigtls1048576.so close_range_test libtagmalloc.so liblongdlopendep.so liblongdlopen.so long_needed_path_test long_needed_app_malloc_test long_needed_lib_malloc_test realpath_free_test realpath_free_lib_test -BUILT_SOURCES += crash_test crash_test_fixedaddr crash_test_pie libcrashfuncs.so libcrashctor.so libcrashfixed.so run_crash_tests.sh +BUILT_SOURCES += crash_test crash_test_fixedaddr crash_test_pie libcrashfuncs.so libcrashctor.so libcrashfixed.so run_crash_tests.sh run_startup_fault_tests.sh if BGQ_BLD DYNAMIC_FLAG=-dynamic @@ -468,7 +468,10 @@ libcrashfixed.so: $(srcdir)/libcrashfixed.c run_crash_tests.sh: $(srcdir)/run_crash_tests_template.sh $(top_builddir)/Makefile $(AM_V_GEN)$(SED) -e s,TEST_RESOURCE_MANAGER,$(TESTRM),g\;s,SPINDLE_EXEC,$(bindir)/spindle,g\;s,SPINDLE_RC_PATH,$(sysconfdir)/spindle/spindle.rc,g\;s,TEST_RUN_DIR,$(ABS_TEST_DIR),g < $< > $@; chmod 700 $@ -EXTRA_DIST = run_crash_tests_template.sh crash_test.c crash_functions.c crash_functions.h libcrashfuncs.c libcrashctor.c libcrashfixed.c +run_startup_fault_tests.sh: $(srcdir)/run_startup_fault_tests_template.sh $(top_builddir)/Makefile + $(AM_V_GEN)$(SED) -e s,TEST_RESOURCE_MANAGER,$(TESTRM),g\;s,SPINDLE_EXEC,$(bindir)/spindle,g\;s,SPINDLE_RC_PATH,$(sysconfdir)/spindle/spindle.rc,g\;s,TEST_RUN_DIR,$(ABS_TEST_DIR),g < $< > $@; chmod 700 $@ + +EXTRA_DIST = run_crash_tests_template.sh run_startup_fault_tests_template.sh crash_test.c crash_functions.c crash_functions.h libcrashfuncs.c libcrashctor.c libcrashfixed.c exec_shell.sh: $(srcdir)/exec_shell_template.sh $(AM_V_GEN)cp $< $@; chmod 700 $@ @@ -518,5 +521,5 @@ realpath_free_test: $(srcdir)/realpath_free_test.c $(srcdir)/tagmalloc.c realpath_free_lib_test: $(srcdir)/realpath_free_test.c libtagmalloc.so $(AM_V_CCLD)$(CC) $(CFLAGS) -o $@ $(srcdir)/realpath_free_test.c $(TAGMALLOC_LIBS) -CLEANFILES = libtest10.c libtest11.c libtest12.c libtest13.c libtest14.c libtest15.c libtest16.c libtest17.c libtest18.c libtest19.c libtest20.c libtest10.so libtest50.c libtest50.so libtest100.c libtest100.so libtest500.c libtest500.so libtest1000.c libtest1000.so libtest2000.c libtest2000.so libtest4000.c libtest4000.so libtest6000.c libtest6000.so libtest8000.c libtest8000.so libtest10000.c libtest10000.so libsymlink.so libdepA.so libdepB.so libdepC.so libcxxexceptA.so libcxxexceptB.so libtestoutput.so libfuncdict.so runTests run_driver run_driver_rm spindle.rc test_driver test_driver_libs preload_file_list retzero_rx retzero_r retzero_x retzero_ badinterp hello_r.py hello_x.py hello_rx.py hello_.py hello_l.py badlink.py libtls1.c libtls2.c libtls3.c libtls4.c libtls5.c libtls6.c libtls7.c libtls8.c libtls9.c libtls10.c libtls11.c libtls12.c libtls13.c libtls14.c libtls15.c libtls16.c libtls17.c libtls18.c libtls19.c libtls20.c libtls1.so libtls2.so libtls3.so libtls4.so libtls5.so libtls6.so libtls7.so libtls8.so libtls9.so libtls10.so libtls11.so libtls12.so libtls13.so libtls14.so libtls15.so libtls16.so libtls17.so libtls18.so libtls19.so libtls20.so symbind_test libsymbind_a.so libsymbind_b.so libsymbind_c.so libsymbind_d.so libsymbind_e.so libsymbind_f.so libsymbind_g.so interpreter_test interpreter_test_dir/interpreter_test_perl alias exec_shell.sh exec_shell.tcsh exec_shell_env.sh exec_shell_expected_output exec_shell crash_test crash_test_fixedaddr crash_test_pie libcrashfuncs.so libcrashctor.so libcrashfixed.so run_crash_tests.sh libbigtls204800.so libbigtls1048576.so close_range_test libtagmalloc.so liblongdlopendep.so liblongdlopen.so long_needed_path_test long_needed_app_malloc_test long_needed_lib_malloc_test realpath_free_test realpath_free_lib_test +CLEANFILES = libtest10.c libtest11.c libtest12.c libtest13.c libtest14.c libtest15.c libtest16.c libtest17.c libtest18.c libtest19.c libtest20.c libtest10.so libtest50.c libtest50.so libtest100.c libtest100.so libtest500.c libtest500.so libtest1000.c libtest1000.so libtest2000.c libtest2000.so libtest4000.c libtest4000.so libtest6000.c libtest6000.so libtest8000.c libtest8000.so libtest10000.c libtest10000.so libsymlink.so libdepA.so libdepB.so libdepC.so libcxxexceptA.so libcxxexceptB.so libtestoutput.so libfuncdict.so runTests run_driver run_driver_rm spindle.rc test_driver test_driver_libs preload_file_list retzero_rx retzero_r retzero_x retzero_ badinterp hello_r.py hello_x.py hello_rx.py hello_.py hello_l.py badlink.py libtls1.c libtls2.c libtls3.c libtls4.c libtls5.c libtls6.c libtls7.c libtls8.c libtls9.c libtls10.c libtls11.c libtls12.c libtls13.c libtls14.c libtls15.c libtls16.c libtls17.c libtls18.c libtls19.c libtls20.c libtls1.so libtls2.so libtls3.so libtls4.so libtls5.so libtls6.so libtls7.so libtls8.so libtls9.so libtls10.so libtls11.so libtls12.so libtls13.so libtls14.so libtls15.so libtls16.so libtls17.so libtls18.so libtls19.so libtls20.so symbind_test libsymbind_a.so libsymbind_b.so libsymbind_c.so libsymbind_d.so libsymbind_e.so libsymbind_f.so libsymbind_g.so interpreter_test interpreter_test_dir/interpreter_test_perl alias exec_shell.sh exec_shell.tcsh exec_shell_env.sh exec_shell_expected_output exec_shell crash_test crash_test_fixedaddr crash_test_pie libcrashfuncs.so libcrashctor.so libcrashfixed.so run_crash_tests.sh run_startup_fault_tests.sh libbigtls204800.so libbigtls1048576.so close_range_test libtagmalloc.so liblongdlopendep.so liblongdlopen.so long_needed_path_test long_needed_app_malloc_test long_needed_lib_malloc_test realpath_free_test realpath_free_lib_test diff --git a/testsuite/Makefile.in b/testsuite/Makefile.in index ad2439e8..918b7c9a 100644 --- a/testsuite/Makefile.in +++ b/testsuite/Makefile.in @@ -365,7 +365,8 @@ BUILT_SOURCES = libtest10.so libtest11.so libtest12.so libtest13.so \ long_needed_app_malloc_test long_needed_lib_malloc_test \ realpath_free_test realpath_free_lib_test crash_test \ crash_test_fixedaddr crash_test_pie libcrashfuncs.so \ - libcrashctor.so libcrashfixed.so run_crash_tests.sh + libcrashctor.so libcrashfixed.so run_crash_tests.sh \ + run_startup_fault_tests.sh @BGQ_BLD_FALSE@DYNAMIC_FLAG = @BGQ_BLD_TRUE@DYNAMIC_FLAG = -dynamic @BGQ_BLD_FALSE@IS_BLUEGENE = false @@ -381,7 +382,7 @@ test_driver_libsLDADD = -ltest10 -ltest11 -ltest12 -ltest13 -ltest14 -ltest15 -l test_driver_libsLDFLAGS = -Wl,-E -L$(top_builddir)/testsuite $(MPI_CLDFLAGS) -L$(top_builddir)/src/client/spindle_api -L. $(DYNAMIC_FLAG) -no-install $(LDFLAGS) -Wl,-rpath,$(PWD)/origin_dir -L$(PWD)/origin_dir -Wl,-rpath-link,$(PWD)/origin_dir/origin_subdir -L$(PWD)/exeorigin_dir '-Wl,-rpath,$$ORIGIN/exeorigin_dir' REGLIB_SRC = $(srcdir)/registerlib.c LD_FUNCDICT = -L$(top_builddir)/testsuite -lfuncdict -EXTRA_DIST = run_crash_tests_template.sh crash_test.c crash_functions.c crash_functions.h libcrashfuncs.c libcrashctor.c libcrashfixed.c +EXTRA_DIST = run_crash_tests_template.sh run_startup_fault_tests_template.sh crash_test.c crash_functions.c crash_functions.h libcrashfuncs.c libcrashctor.c libcrashfixed.c LONG_NEEDED_PAD = $$(printf '/%.0s' $$(seq 300)) LONG_NEEDED_TARGET = origin_dir/origin_subdir/liborigintarget.so LONG_NEEDED_LIBS = -Wl,--no-as-needed -lc "$(LONG_NEEDED_PAD)$(abs_builddir)/$(LONG_NEEDED_TARGET)" -ldl @@ -414,11 +415,12 @@ CLEANFILES = libtest10.c libtest11.c libtest12.c libtest13.c \ exec_shell.tcsh exec_shell_env.sh exec_shell_expected_output \ exec_shell crash_test crash_test_fixedaddr crash_test_pie \ libcrashfuncs.so libcrashctor.so libcrashfixed.so \ - run_crash_tests.sh libbigtls204800.so libbigtls1048576.so \ - close_range_test libtagmalloc.so liblongdlopendep.so \ - liblongdlopen.so long_needed_path_test \ - long_needed_app_malloc_test long_needed_lib_malloc_test \ - realpath_free_test realpath_free_lib_test + run_crash_tests.sh run_startup_fault_tests.sh \ + libbigtls204800.so libbigtls1048576.so close_range_test \ + libtagmalloc.so liblongdlopendep.so liblongdlopen.so \ + long_needed_path_test long_needed_app_malloc_test \ + long_needed_lib_malloc_test realpath_free_test \ + realpath_free_lib_test all: $(BUILT_SOURCES) $(MAKE) $(AM_MAKEFLAGS) all-am @@ -1164,6 +1166,9 @@ libcrashfixed.so: $(srcdir)/libcrashfixed.c run_crash_tests.sh: $(srcdir)/run_crash_tests_template.sh $(top_builddir)/Makefile $(AM_V_GEN)$(SED) -e s,TEST_RESOURCE_MANAGER,$(TESTRM),g\;s,SPINDLE_EXEC,$(bindir)/spindle,g\;s,SPINDLE_RC_PATH,$(sysconfdir)/spindle/spindle.rc,g\;s,TEST_RUN_DIR,$(ABS_TEST_DIR),g < $< > $@; chmod 700 $@ +run_startup_fault_tests.sh: $(srcdir)/run_startup_fault_tests_template.sh $(top_builddir)/Makefile + $(AM_V_GEN)$(SED) -e s,TEST_RESOURCE_MANAGER,$(TESTRM),g\;s,SPINDLE_EXEC,$(bindir)/spindle,g\;s,SPINDLE_RC_PATH,$(sysconfdir)/spindle/spindle.rc,g\;s,TEST_RUN_DIR,$(ABS_TEST_DIR),g < $< > $@; chmod 700 $@ + exec_shell.sh: $(srcdir)/exec_shell_template.sh $(AM_V_GEN)cp $< $@; chmod 700 $@ diff --git a/testsuite/run_startup_fault_tests_template.sh b/testsuite/run_startup_fault_tests_template.sh new file mode 100755 index 00000000..e2b71d4d --- /dev/null +++ b/testsuite/run_startup_fault_tests_template.sh @@ -0,0 +1,342 @@ +#!/usr/bin/env bash + +# Usage: +# ./run_startup_fault_tests.sh [--launcher=serial|flux|slurm|slurm-plugin] +# [--nodes=N] [--cases=LIST] [--session] +# +# --launcher: specify launcher to use (default TEST_RESOURCE_MANAGER) +# --nodes=N: run on N nodes (required for launchers other than serial) +# --cases=list: list of tests to run (default all) +# --session: run session tests instead of standard tests (requires flux) +# +# Injects startup faults into the Spindle servers and checks that startup +# ends with a nonzero status within a deadline instead of hanging. + +set -u + +LAUNCHER="TEST_RESOURCE_MANAGER" +NODES="" +CASES="" +SESSION=0 +SPINDLE="${SPINDLE:-SPINDLE_EXEC}" +SPINDLE_RC="${SPINDLE_RC:-SPINDLE_RC_PATH}" +TESTDIR="${TESTDIR:-TEST_RUN_DIR}" + +# Tests set TIMELIMIT as COBO_CONNECT_TIMELIMIT (limit on each server startup wait) +TIMELIMIT="${STARTUP_FAULT_TIMELIMIT:-20}" +# Tests set READY_WAIT as SPINDLE_READY_TIMELIMIT (how long client waits for ready file) +READY_WAIT=$((4 * TIMELIMIT)) + +# How long to keep checking that servers and clients have exited after the launcher returns +LEFTOVER_WAIT=$((READY_WAIT + 5)) + +# Port range to search for leftover listeners +PORT_LO="${SPINDLE_TEST_PORT_LO:-21940}" +PORT_HI="${SPINDLE_TEST_PORT_HI:-22189}" + +PASSED=0 +FAILED=0 + +die() { echo "ERROR: $*" >&2; exit 1; } +pass() { echo "PASS: $1"; PASSED=$((PASSED + 1)); } +fail() { echo "FAIL: $1"; FAILED=$((FAILED + 1)); } + +# Table of test cases to run +# Special values in environment variables: +# @MID@: an interior rank that has children +# @LAST@: the last rank +# @LAST_HOST@: the hostname of the last node +# @HALF@: half the time limit +CASES_TABLE=( +# name ; environment variables ; expected result + 'baseline ; ; success' + 'delay-rank-tree ; SPINDLE_TEST_BE_STALL=@MID@ SPINDLE_TEST_BE_STALL_SECS=@HALF@ SPINDLE_TEST_BE_FAULT_PHASE=tree ; success' + 'exit-host-early ; SPINDLE_TEST_BE_EXIT_HOSTS=@LAST_HOST@ ; failure' + 'exit-rank-tree ; SPINDLE_TEST_BE_EXIT=@MID@ SPINDLE_TEST_BE_FAULT_PHASE=tree ; failure' + 'stall-rank-tree ; SPINDLE_TEST_BE_STALL=@MID@ SPINDLE_TEST_BE_FAULT_PHASE=tree ; failure' + 'exit-rank-ready ; SPINDLE_TEST_BE_EXIT=@LAST@ SPINDLE_TEST_BE_FAULT_PHASE=ready ; failure' + 'stall-rank-ready ; SPINDLE_TEST_BE_STALL=@LAST@ SPINDLE_TEST_BE_FAULT_PHASE=ready ; failure' +) + +trim() { local s="$1"; s="${s#"${s%%[![:space:]]*}"}"; echo "${s%"${s##*[![:space:]]}"}"; } + +parse_args() { + for a in "$@"; do + case "$a" in + --launcher=*) LAUNCHER="${a#*=}" ;; + --nodes=*) NODES="${a#*=}" ;; + --cases=*) CASES="${a#*=}" ;; + --session) SESSION=1 ;; + --help|-h) sed -n '3,/^$/s/^# \{0,1\}//p' "$0"; exit 0 ;; + *) die "unknown argument '$a'" ;; + esac + done + case "$LAUNCHER" in + serial) NODES=1 ;; + flux|slurm|slurm-plugin) + [[ "$NODES" =~ ^[1-9][0-9]*$ ]] || die "--nodes=N required" + [ "$NODES" -ge 3 ] || die "--nodes must be at least 3 so that @MID@ has children" ;; + *) die "unknown launcher '$LAUNCHER'" ;; + esac + if (( SESSION )) && [ "$LAUNCHER" != flux ]; then + die "--session requires --launcher=flux" + fi +} + +# True if case was explicitly selected or all cases are enabled +selected() { [ -z "$CASES" ] || [[ ",$CASES," == *",$1,"* ]]; } + +# True if the exit code indicates the process was killed by timeout +hung() { [ "$1" -eq 124 ] || [ "$1" -eq 137 ]; } + +# Run a bash command on every node of the allocation +on_all_nodes() { + case "$LAUNCHER" in + serial) bash -c "$1" ;; + flux) flux exec -r all bash -c "$1" ;; + slurm|slurm-plugin) salloc -N"$NODES" -n"$NODES" srun bash -c "$1" 2>/dev/null ;; + esac +} + +# Get hostname of last node of the allocation +last_host() { + case "$LAUNCHER" in + serial) hostname ;; + flux) flux exec -r $((NODES - 1)) hostname ;; + slurm|slurm-plugin) salloc -N"$NODES" -n"$NODES" \ + srun -N1 -n1 --relative=$((NODES - 1)) hostname 2>/dev/null ;; + esac +} + +# Count TCP sockets listening on port range +count_listeners() { + local n=0 local_addr st port + while read -r _ local_addr _ st _; do + [ "$st" = 0A ] || continue + port=$((16#${local_addr##*:})) + [ "$port" -ge "$1" ] && [ "$port" -le "$2" ] && n=$((n + 1)) + done < <(tail -n +2 /proc/net/tcp; tail -n +2 /proc/net/tcp6 2>/dev/null) + echo "$n" +} + +# Count spindle_bootstrap processes +count_bootstraps() { + pgrep -c -f "[s]pindle_bootstrap" || true +} + +# Count Spindle server processes +count_servers() { + pgrep -c -x spindle_be || true +} + +# True if a Spindle session is published in the Flux KVS +session_published() { + flux kvs namespace list | grep -qw spindle +} + +# True if the session is still published and has a server on every node. +# A job run after the session has gone falls back to a non-session launch +# and succeeds. +session_alive() { + local n + session_published || return 1 + n=$(on_all_nodes "$(declare -f count_servers); count_servers" | sum_lines) + [ "$n" -eq "$NODES" ] +} + +# Sleep after ending session because subsequent session won't start if +# a session exists but is being deleted. +wait_for_session_removal() { + sleep 3 +} + +end_session() { + "$SPINDLE" "--end-session${1:+=$1}" >/dev/null 2>&1 + wait_for_session_removal +} + +# Create FIFOs for a client that never connects: make_stale_fifos PID SUFFIX... +make_stale_fifos() { + local d pid=$1 s + shift + find "${TMPDIR:-/tmp}" -maxdepth 4 -type d -name spindle_comm 2>/dev/null | + while read -r d; do + for s in "$@"; do mkfifo "$d/fifo-$pid-$s"; done + done +} + +# Sum the numbers read one per line; other lines (launcher noise) are ignored +sum_lines() { + local line sum=0 + while read -r line || [ -n "$line" ]; do + [[ "$line" =~ ^[0-9]+$ ]] && sum=$((sum + 10#$line)) + done + echo "$sum" +} + +# Tests should fail within FAIL_WITHIN seconds, and are killed after BOUND +set_time_bounds() { + FAIL_WITHIN=$((READY_WAIT + 15)) + # The SPANK plugin's task_exit waits an extra 30s for the exit socket + if [ "$LAUNCHER" = slurm-plugin ]; then + FAIL_WITHIN=$((FAIL_WITHIN + 30)) + fi + BOUND=$((FAIL_WITHIN + 60)) +} + +set_launch_cmd() { + case "$LAUNCHER" in + serial) CMD=("$SPINDLE" --no-mpi -- /bin/true) ;; + flux) CMD=(flux run -o userrc="$SPINDLE_RC" -o spindle -N"$NODES" -n"$NODES" -- /bin/true) ;; + slurm) CMD=(salloc -N"$NODES" -n"$NODES" "$SPINDLE" -- srun /bin/true) ;; + slurm-plugin) CMD=(salloc -N"$NODES" -n"$NODES" srun --spindle /bin/true) ;; + esac +} + +substitute() { + local s="$1" + s="${s//@MID@/$MID}" + s="${s//@LAST@/$LAST}" + s="${s//@HALF@/$((TIMELIMIT / 2))}" + s="${s//@LAST_HOST@/$LAST_HOST}" + echo "$s" +} + +check_leftovers() { + local name="$1" listeners bootstraps deadline=$((SECONDS + LEFTOVER_WAIT)) + while :; do + listeners=$(on_all_nodes "$(declare -f count_listeners); count_listeners $PORT_LO $PORT_HI" | sum_lines) + bootstraps=$(on_all_nodes "$(declare -f count_bootstraps); count_bootstraps" | sum_lines) + [ "$listeners" -eq 0 ] && [ "$bootstraps" -eq 0 ] && return 0 + [ $SECONDS -lt $deadline ] || break + sleep 5 + done + if [ "$listeners" -ne 0 ]; then + fail "$name: $listeners server(s) still waiting for a parent" + else + fail "$name: $bootstraps spindle_bootstrap process(es) left running" + fi + return 1 +} + +run_case() { + local name="$1" envs="$2" expect="$3" out rc start elapsed + local -a case_env + read -ra case_env <<< "$envs" + out="$SCRATCH/$name.log" + start=$SECONDS + timeout -k 10 "$BOUND" env SPINDLE_TEST_BE_STALL_SECS="$BOUND" "${case_env[@]}" "${CMD[@]}" >"$out" 2>&1 + rc=$? + elapsed=$((SECONDS - start)) + if hung $rc; then + fail "$name: hung (killed after ${BOUND}s, log $out)" + elif [ "$expect" = success ]; then + if [ $rc -eq 0 ]; then + pass "$name (${elapsed}s)" + else + fail "$name: expected success, got rc=$rc (log $out)" + fi + elif [ $rc -eq 0 ]; then + fail "$name: startup failure unexpectedly returned 0 (log $out)" + elif [ $elapsed -gt $FAIL_WITHIN ]; then + fail "$name: failure took ${elapsed}s, longer than limit ${FAIL_WITHIN}s (log $out)" + elif [ "$LAUNCHER" = flux ] && ! grep -q "spindleInitFE error" "$out"; then + fail "$name: FE did not report an expected failure (log $out)" + elif check_leftovers "$name"; then + pass "$name (rc=$rc, ${elapsed}s)" + fi +} + +# A session whose server on the last node exits before listening must fail to start +session_exit_host_early() { + local name=session-exit-host-early out rc sid + out="$SCRATCH/$name.log" + SPINDLE_TEST_BE_EXIT_HOSTS="$LAST_HOST" \ + timeout -k 10 "$BOUND" "$SPINDLE" --start-session >"$out" 2>"$out.err" + rc=$? + if hung $rc; then + fail "$name: --start-session hung (log $out.err)" + elif [ $rc -eq 0 ]; then + sid=$(head -n 1 "$out") + end_session "$sid" + fail "$name: --start-session reported success (log $out.err)" + elif session_published; then + fail "$name: failed session left published in the Flux KVS" + else + check_leftovers "$name" && pass "$name (rc=$rc)" + fi + wait_for_session_removal +} + +# A client that died mid-connect must not block the session's servers: +# session_stale_fifo NAME PID SUFFIX... +session_stale_fifo() { + local name=$1 out rc sid alive + shift + out="$SCRATCH/$name.log" + sid=$("$SPINDLE" --start-session 2>"$SCRATCH/$name-start.log") || { + fail "$name: could not start session"; return; } + on_all_nodes "$(declare -f make_stale_fifos); make_stale_fifos $*" >/dev/null 2>&1 + timeout -k 10 "$BOUND" "${CMD[@]}" >"$out" 2>&1 + rc=$? + session_alive; alive=$? + end_session "$sid" + if [ $rc -ne 0 ]; then + fail "$name: rc=$rc (log $out)" + elif [ $alive -ne 0 ]; then + fail "$name: session servers did not survive (log $out)" + else + pass "$name" + fi +} + +# A session left idle past every startup wait must still run a job +session_idle() { + local name=session-idle out rc sid alive + local idle=$((TIMELIMIT + 10)) + out="$SCRATCH/$name.log" + sid=$("$SPINDLE" --start-session 2>"$SCRATCH/$name-start.log") || { + fail "$name: could not start session"; return; } + sleep "$idle" + timeout -k 10 "$BOUND" "${CMD[@]}" >"$out" 2>&1 + rc=$? + session_alive; alive=$? + end_session "$sid" + if [ $rc -ne 0 ]; then + fail "$name: rc=$rc after ${idle}s idle (log $out)" + elif [ $alive -ne 0 ]; then + fail "$name: session servers did not survive ${idle}s idle (log $out)" + else + pass "$name (idle ${idle}s)" + fi +} + +parse_args "$@" +set_time_bounds +set_launch_cmd +MID=$((NODES / 2)) +LAST=$((NODES - 1)) +LAST_HOST=$(last_host) +[ -n "$LAST_HOST" ] || die "could not determine the last node's hostname" +SCRATCH=$(mktemp -d "${TMPDIR:-/tmp}/startup_fault.XXXXXX") || die "can't create scratch dir" +cd "$TESTDIR" || die "can't cd to $TESTDIR" +export COBO_CONNECT_TIMELIMIT="$TIMELIMIT" SPINDLE_READY_TIMELIMIT="$READY_WAIT" + +if (( SESSION )); then + selected session-exit-host-early && session_exit_host_early + # Client died after creating both FIFOs: the server's open of -0 gets ENXIO + selected session-stale-fifo && session_stale_fifo session-stale-fifo 999999 0 1 + # Only -1 was created: the server's open of -0 gets ENOENT + selected session-stale-fifo1 && session_stale_fifo session-stale-fifo1 999998 1 + selected session-idle && session_idle +else + for entry in "${CASES_TABLE[@]}"; do + IFS=';' read -r name envs expect <<< "$entry" + name=$(trim "$name"); envs=$(trim "$envs"); expect=$(trim "$expect") + selected "$name" || continue + run_case "$name" "$(substitute "$envs")" "$expect" + done +fi + +echo "$PASSED passed, $FAILED failed" +[ $FAILED -eq 0 ] && [ $PASSED -gt 0 ]