smpd_connect.c

fortran并行计算包

/* -*- Mode: C; c-basic-offset:4 ; -*- */
/*
 *  (C) 2001 by Argonne National Laboratory.
 *      See COPYRIGHT in top-level directory.
 */

#include <stdio.h>
#include "smpd.h"
#include <stdlib.h>
#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif
#ifdef HAVE_STDARG_H
#include <stdarg.h>
#endif
#ifdef HAVE_TERMIOS_H
#include <termios.h>
#endif
#ifdef HAVE_SYS_TYPES_H
#include <sys/types.h>
#endif
#ifdef HAVE_SYS_STAT_H
#include <sys/stat.h>
#endif
#ifdef HAVE_STRING_H
#include <string.h>
#endif

#ifdef HAVE_SIGACTION
#ifdef HAVE_SIGNAL_H
#include <signal.h>
#endif
#ifdef HAVE_SYS_WAIT_H
#include <sys/wait.h>
#endif
#ifdef HAVE_ERRNO_H
#include <errno.h>
#endif
#endif

smpd_global_t smpd_process = 
    { SMPD_IDLE,        /* state                  */
      -1,               /* id                     */
      -1,               /* parent_id              */
      -1,               /* level                  */
      NULL,             /* left_context           */
      NULL,             /* right_context          */
      NULL,             /* parent_context         */
      NULL,             /* context_list           */
      NULL,             /* listener_context       */
      NULL,             /* process_list           */
      SMPD_FALSE,       /* closing                */
      SMPD_FALSE,       /* root_smpd              */
      MPIDU_SOCK_INVALID_SET, /* set                    */
      "",               /* host                   */
      "",               /* pszExe                 */
      SMPD_FALSE,       /* bService               */
      SMPD_TRUE,        /* bNoTTY                 */
      SMPD_FALSE,       /* bPasswordProtect       */
      "",               /* SMPDPassword           */
      "",               /* passphrase             */
      SMPD_FALSE,       /* logon                  */
      "",               /* UserAccount            */
      "",               /* UserPassword           */
      0,                /* cur_tag                */
      SMPD_DBG_STATE_ERROUT, /* dbg_state         */
      /*NULL,*/             /* dbg_fout               */
      "",               /* dbg_filename           */
      SMPD_DBG_FILE_SIZE,/*dbg_file_size          */
      SMPD_FALSE,       /* have_dbs               */
      "",               /* kvs_name               */
      "",               /* domain_name            */
#ifdef HAVE_WINDOWS_H
      NULL,             /* hCloseStdinThreadEvent */
      NULL,             /* hStdinThread           */
#endif
#ifdef USE_PTHREAD_STDIN_REDIRECTION
      0,                /* stdin_thread           */
      0,                /* stdin_read             */
      0,                /* stdin_write            */
#endif
      SMPD_FALSE,       /* do_console             */
      SMPD_LISTENER_PORT, /* smpd port            */
      SMPD_FALSE,       /* Is singleton client ? */
      -1,               /* Port to connect back to a singleton client*/
      "",               /* console_host           */
      NULL,             /* host_list              */
      NULL,             /* launch_list            */
      SMPD_TRUE,        /* credentials_prompt     */
      SMPD_TRUE,        /* do_multi_color_output  */
      SMPD_FALSE,       /* no_mpi                 */
      SMPD_FALSE,       /* output_exit_codes      */
      SMPD_FALSE,       /* local_root             */
      SMPD_FALSE,       /* use_iproot             */
      SMPD_FALSE,       /* use_process_session    */
      0,                /* nproc                  */
      0,                /* nproc_launched         */
      0,                /* nproc_exited           */
      SMPD_FALSE,       /* verbose                */
      /*SMPD_FALSE,*/       /* shutdown               */
      /*SMPD_FALSE,*/       /* restart                */
      /*SMPD_FALSE,*/       /* validate               */
      /*SMPD_FALSE,*/       /* do_status              */
      SMPD_CMD_NONE,    /* builtin_cmd            */
#ifdef HAVE_WINDOWS_H
      FALSE,            /* bOutputInitialized     */
      NULL,             /* hOutputMutex           */
      NULL,             /* hLaunchProcessMutex    */
#endif
#ifdef USE_WIN_MUTEX_PROTECT
      NULL,             /* hDBSMutext             */
#endif
      NULL,             /* pDatabase              */
      NULL,             /* pDatabaseIter          */
      /*0,*/                /* nNextAvailableDBSID    */
      0,                /* nInitDBSRefCount       */
      NULL,             /* barrier_list           */
#ifdef HAVE_WINDOWS_H
      {                 /* ssStatus                    */
	  SERVICE_WIN32_OWN_PROCESS, /* dwServiceType  */
	  SERVICE_STOPPED,           /* dwCurrentState */
	  SERVICE_ACCEPT_STOP,   /* dwControlsAccepted */
	  NO_ERROR,     /* dwWin32ExitCode             */
	  0,            /* dwServiceSpecificExitCode   */
	  0,            /* dwCheckPoint                */
	  3000,         /* dwWaitHint                  */
      },
      NULL,             /* sshStatusHandle         */
      NULL,             /* hBombDiffuseEvent       */
      NULL,             /* hBombThread             */
#endif
      SMPD_FALSE,       /* service_stop            */
      SMPD_FALSE,       /* noprompt                */
      "",               /* smpd_filename           */
      SMPD_FALSE,       /* stdin_toall             */
      SMPD_FALSE,       /* stdin_redirecting       */
      NULL,             /* default_host_list       */
      NULL,             /* cur_default_host        */
      0,                /* cur_default_iproc       */
#ifdef HAVE_WINDOWS_H
      NULL,             /* hSMPDDataMutex          */
#endif
      "",               /* printf_buffer           */
      SMPD_SUCCESS,     /* state_machine_ret_val   */
      SMPD_FALSE,       /* exit_on_done            */
      0,                /* tree_parent             */
      1,                /* tree_id                 */
      NULL,             /* s_host_list             */
      NULL,             /* s_cur_host              */
      0,                /* s_cur_count             */
      SMPD_FALSE,       /* use_inherited_handles   */
      NULL,             /* pg_list                 */
      SMPD_FALSE,       /* use_abort_exit_code     */
      0,                /* abort_exit_code         */
      SMPD_TRUE,        /* verbose_abort_output    */
      0,                /* mpiexec_exit_code       */
      SMPD_FALSE,       /* map0to1                 */
      SMPD_FALSE,       /* rsh_mpiexec             */
      SMPD_FALSE,       /* mpiexec_inorder_launch  */
      SMPD_FALSE,       /* mpiexec_run_local       */
#ifdef HAVE_WINDOWS_H
      NULL,             /* timeout_thread          */
#else
#ifdef HAVE_PTHREADS_H
      NULL,             /* timeout_thread          */
#endif
#endif
      -1,               /* timeout                 */
      MPIDU_SOCK_INVALID_SOCK, /* timeout_sock     */
      MPIDU_SOCK_INVALID_SOCK, /* mpiexec_abort_sock */
      SMPD_TRUE,        /* use_pmi_server          */
      NULL,             /* mpiexec_argv0           */
      "dummy",          /* encrypt_prefix          */
      SMPD_FALSE,       /* plaintext               */
      SMPD_FALSE,       /* use_sspi                */
      SMPD_FALSE,       /* use_delegation          */
      SMPD_FALSE,       /* use_sspi_job_key        */
#ifdef HAVE_WINDOWS_H
      NULL,             /* sec_fn                  */
#endif
      NULL,             /* sspi_context_list       */
      "",               /* job_key                 */
      "",               /* job_key_account         */
      "",               /* job_key_password        */
      "",               /* key                     */
      "",               /* val                     */
      SMPD_FALSE,       /* do_console_returns      */
      "",               /* env_channel             */
      "",               /* env_dll                 */
      "",               /* env_wrap_dll            */
      NULL,             /* delayed_spawn_queue     */
      SMPD_FALSE,       /* spawning                */
      0,                /* user_index              */
      SMPD_FALSE        /* prefix_output           */
    };

#undef FCNAME
#define FCNAME "smpd_post_abort_command"
int smpd_post_abort_command(char *fmt, ...)
{
    int result;
    char error_str[2048] = "";
    smpd_command_t *cmd_ptr;
    smpd_context_t *context;
    va_list list;

    smpd_enter_fn(FCNAME);

    va_start(list, fmt);
    vsnprintf(error_str, 2048, fmt, list);
    va_end(list);

    result = smpd_create_command("abort", smpd_process.id, 0, SMPD_FALSE, &cmd_ptr);
    if (result != SMPD_SUCCESS)
    {
	smpd_err_printf("unable to create an abort command.\n");
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    result = smpd_add_command_arg(cmd_ptr, "error", error_str);
    if (result != SMPD_SUCCESS)
    {
	smpd_err_printf("Unable to add the error string to the abort command.\n");
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    smpd_command_destination(0, &context);
    if (context == NULL)
    {
	if (smpd_process.left_context == NULL)
	{
	    printf("Aborting: %s\n", error_str);
	    fflush(stdout);
	    smpd_exit_fn(FCNAME);
	    smpd_exit(-1);
	}

	smpd_process.closing = SMPD_TRUE;
	result = smpd_create_command("close", 0, 1, SMPD_FALSE, &cmd_ptr);
	if (result != SMPD_SUCCESS)
	{
	    smpd_err_printf("unable to create the close command to tear down the job tree.\n");
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}
	result = smpd_post_write_command(smpd_process.left_context, cmd_ptr);
	if (result != SMPD_SUCCESS)
	{
	    smpd_err_printf("unable to post a write of the close command to tear down the job tree as part of the abort process.\n");
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}
    }
    else
    {
	smpd_dbg_printf("sending abort command to %s context: \"%s\"\n", smpd_get_context_str(context), cmd_ptr->cmd);
	result = smpd_post_write_command(context, cmd_ptr);
	if (result != SMPD_SUCCESS)
	{
	    smpd_err_printf("unable to post a write of the abort command to the %s context.\n", smpd_get_context_str(context));
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}
    }
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#ifdef HAVE_SIGACTION
void smpd_child_handler(int code)
{
    int status;
    int pid;

    if (smpd_process.root_smpd && code == SIGCHLD)
    {
	/*pid = waitpid(-1, &status, WNOHANG);*/
	pid = waitpid(-1, &status, 0);
	if (pid < 0)
	{
	    fprintf(stderr, "waitpid failed, error %d\n", errno);
	}
	/*
	else
	{
	    printf("process %d exited with code: %d\n", pid, WIFEXITED(status) ? WEXITSTATUS(status) : -1);
	    fflush(stdout);
	}
	*/
    }
}
#endif

#ifdef HAVE_WINDOWS_H
#undef FCNAME
#define FCNAME "smpd_make_socket_loop"
int smpd_make_socket_loop(SOCKET *pRead, SOCKET *pWrite)
{
    SOCKET sock;
    char host[100];
    int port;
    int len;
    /*LINGER linger;*/
    BOOL b;
    SOCKADDR_IN sockAddr;
    int error;

    smpd_enter_fn(FCNAME);

    /* Create a listener */

    /* create the socket */
    sock = WSASocket(PF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
    if (sock == INVALID_SOCKET)
    {
	*pRead = INVALID_SOCKET;
	*pWrite = INVALID_SOCKET;
	smpd_exit_fn(FCNAME);
	return WSAGetLastError();
    }

    memset(&sockAddr,0,sizeof(sockAddr));
    
    sockAddr.sin_family = AF_INET;
    sockAddr.sin_addr.s_addr = INADDR_ANY;
    sockAddr.sin_port = htons((unsigned short)ADDR_ANY);
    
    if (bind(sock, (SOCKADDR*)&sockAddr, sizeof(sockAddr)) == SOCKET_ERROR)
    {
	error = WSAGetLastError();
	smpd_err_printf("bind failed: error %d\n", error);
	*pRead = INVALID_SOCKET;
	*pWrite = INVALID_SOCKET;
	smpd_exit_fn(FCNAME);
	return error;
    }
    
    /* listen */
    listen(sock, 2);

    /* get the host and port where we're listening */
    len = sizeof(sockAddr);
    getsockname(sock, (struct sockaddr*)&sockAddr, &len);
    port = ntohs(sockAddr.sin_port);
    smpd_get_hostname(host, 100);

    /* Connect to myself */

    /* create the socket */
    *pWrite = WSASocket(PF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
    if (*pWrite == INVALID_SOCKET)
    {
	error = WSAGetLastError();
	smpd_err_printf("WSASocket failed, error %d\n", error);
	if (closesocket(sock) == SOCKET_ERROR)
	{
	    smpd_err_printf("closesocket failed, sock %d, error %d\n", sock, WSAGetLastError());
	}
	*pRead = INVALID_SOCKET;
	*pWrite = INVALID_SOCKET;
	smpd_exit_fn(FCNAME);
	return error;
    }

    /* set the nodelay option */
    /*
    b = TRUE;
    setsockopt(*pWrite, IPPROTO_TCP, TCP_NODELAY, (char*)&b, sizeof(BOOL));
    */

    /* Set the linger on close option */
    /*
    linger.l_onoff = 1 ;
    linger.l_linger = 60;
    setsockopt(*pWrite, SOL_SOCKET, SO_LINGER, (char*)&linger, sizeof(linger));
    */

    sockAddr.sin_addr.s_addr = inet_addr("127.0.0.1");

    /* connect to myself */
    if (connect(*pWrite, (SOCKADDR*)&sockAddr, sizeof(sockAddr)) == SOCKET_ERROR)
    {
	error = WSAGetLastError();
	if (closesocket(*pWrite) ==  SOCKET_ERROR)
	{
	    smpd_err_printf("closesocket failed, sock %d, error %d\n", *pWrite, WSAGetLastError());
	}
	if (closesocket(sock) == SOCKET_ERROR)
	{
	    smpd_err_printf("closesocket failed, sock %d, error %d\n", sock, WSAGetLastError());
	}
	*pRead = INVALID_SOCKET;
	*pWrite = INVALID_SOCKET;
	smpd_exit_fn(FCNAME);
	return error;
    }

    /* Accept the connection from myself */
    len = sizeof(sockAddr);
    *pRead = accept(sock, (SOCKADDR*)&sockAddr, &len);

    /* set the nodelay option */
    b = TRUE;
    setsockopt(*pWrite, IPPROTO_TCP, TCP_NODELAY, (char*)&b, sizeof(BOOL));
    /* set the nodelay option */
    b = TRUE;
    setsockopt(*pRead, IPPROTO_TCP, TCP_NODELAY, (char*)&b, sizeof(BOOL));

    if (closesocket(sock) == SOCKET_ERROR)
    {
	smpd_err_printf("closesocket failed, sock %d, error %d\n", sock, WSAGetLastError());
    }
    smpd_exit_fn(FCNAME);
    return 0;
}

#undef FCNAME
#define FCNAME "smpd_make_socket_loop_choose"
int smpd_make_socket_loop_choose(SOCKET *pRead, int read_overlapped, SOCKET *pWrite, int write_overlapped)
{
    SOCKET sock;
    char host[100];
    int port;
    int len;
    /*LINGER linger;*/
    BOOL b;
    SOCKADDR_IN sockAddr;
    int error;
    DWORD flag;

    smpd_enter_fn(FCNAME);

    /* Create a listener */

    /* create the socket */
    flag = read_overlapped ? WSA_FLAG_OVERLAPPED : 0;
    sock = WSASocket(PF_INET, SOCK_STREAM, 0, NULL, 0, flag);
    if (sock == INVALID_SOCKET)
    {
	*pRead = INVALID_SOCKET;
	*pWrite = INVALID_SOCKET;
	smpd_exit_fn(FCNAME);
	return WSAGetLastError();
    }

    memset(&sockAddr,0,sizeof(sockAddr));
    
    sockAddr.sin_family = AF_INET;
    sockAddr.sin_addr.s_addr = INADDR_ANY;
    sockAddr.sin_port = htons((unsigned short)ADDR_ANY);

    if (bind(sock, (SOCKADDR*)&sockAddr, sizeof(sockAddr)) == SOCKET_ERROR)
    {
	error = WSAGetLastError();
	smpd_err_printf("bind failed: error %d\n", error);
	*pRead = INVALID_SOCKET;
	*pWrite = INVALID_SOCKET;
	smpd_exit_fn(FCNAME);
	return error;
    }
    
    /* listen */
    listen(sock, 2);

    /* get the host and port where we're listening */
    len = sizeof(sockAddr);
    getsockname(sock, (struct sockaddr*)&sockAddr, &len);
    port = ntohs(sockAddr.sin_port);
    smpd_get_hostname(host, 100);

    /* Connect to myself */

    /* create the socket */
    flag = write_overlapped ? WSA_FLAG_OVERLAPPED : 0;
    *pWrite = WSASocket(PF_INET, SOCK_STREAM, 0, NULL, 0, flag);
    if (*pWrite == INVALID_SOCKET)
    {
	error = WSAGetLastError();
	smpd_err_printf("WSASocket failed, error %d\n", error);