4fff584a68
The primary change that underlies all this is in the OOB. Specifically, the problem in the code until now has been that the OOB attempts to resolve an address when we call the "send" to an unknown recipient. The OOB would then wait forever if that recipient never actually started (and hence, never reported back its OOB contact info). In the case of an orted that failed to start, we would correctly detect that the orted hadn't started, but then we would attempt to order all orteds (including the one that failed to start) to die. This would cause the OOB to "hang" the system. Unfortunately, revising how the OOB resolves addresses introduced a number of additional problems. Specifically, and most troublesome, was the fact that comm_spawn involved the immediate transmission of the rendezvous point from parent-to-child after the child was spawned. The current code used the OOB address resolution as a "barrier" - basically, the parent would attempt to send the info to the child, and then "hold" there until the child's contact info had arrived (meaning the child had started) and the send could be completed. Note that this also caused comm_spawn to "hang" the entire system if the child never started... The app-failed-to-start helped improve that behavior - this code provides additional relief. With this change, the OOB will return an ADDRESSEE_UNKNOWN error if you attempt to send to a recipient whose contact info isn't already in the OOB's hash tables. To resolve comm_spawn issues, we also now force the cross-sharing of connection info between parent and child jobs during spawn. Finally, to aid in setting triggers to the right values, we introduce the "arith" API for the GPR. This function allows you to atomically change the value in a registry location (either divide, multiply, add, or subtract) by the provided operand. It is equivalent to first fetching the value using a "get", then modifying it, and then putting the result back into the registry via a "put". This commit was SVN r14711.
362 строки
12 KiB
C
362 строки
12 KiB
C
/*
|
|
* Copyright (c) 2004-2007 The Trustees of Indiana University and Indiana
|
|
* University Research and Technology
|
|
* Corporation. All rights reserved.
|
|
* Copyright (c) 2004-2006 The University of Tennessee and The University
|
|
* of Tennessee Research Foundation. All rights
|
|
* reserved.
|
|
* Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
|
|
* University of Stuttgart. All rights reserved.
|
|
* Copyright (c) 2004-2005 The Regents of the University of California.
|
|
* All rights reserved.
|
|
* Copyright (c) 2006-2007 Los Alamos National Security, LLC.
|
|
* All rights reserved.
|
|
* $COPYRIGHT$
|
|
*
|
|
* Additional copyrights may follow
|
|
*
|
|
* $HEADER$
|
|
*/
|
|
/** @file:
|
|
*
|
|
* Defines the functions for the tcp module.
|
|
*/
|
|
|
|
#ifndef _MCA_OOB_TCP_H_
|
|
#define _MCA_OOB_TCP_H_
|
|
|
|
#include "orte/mca/oob/oob.h"
|
|
#include "orte/mca/oob/base/base.h"
|
|
#include "opal/mca/base/base.h"
|
|
#include "orte/mca/ns/ns_types.h"
|
|
#include "opal/class/opal_free_list.h"
|
|
#include "opal/class/opal_hash_table.h"
|
|
#include "opal/runtime/opal_progress.h"
|
|
#include "opal/runtime/opal_cr.h"
|
|
#include "opal/threads/mutex.h"
|
|
#include "opal/threads/condition.h"
|
|
#include "orte/mca/oob/tcp/oob_tcp_peer.h"
|
|
#include "orte/mca/oob/tcp/oob_tcp_msg.h"
|
|
|
|
#include "opal/mca/timer/base/base.h"
|
|
|
|
|
|
#if defined(c_plusplus) || defined(__cplusplus)
|
|
extern "C" {
|
|
#endif
|
|
|
|
#define ORTE_OOB_TCP_KEY "oob-tcp"
|
|
|
|
#define OOB_TCP_DEBUG_CONNECT_FAIL 1 /* debug connection establishment failures */
|
|
#define OOB_TCP_DEBUG_CONNECT 2 /* other connection information */
|
|
#define OOB_TCP_DEBUG_INFO 3 /* information about startup, connection establish, etc. */
|
|
#define OOB_TCP_DEBUG_ALL 4 /* everything else */
|
|
|
|
|
|
/*
|
|
* standard component functions
|
|
*/
|
|
int mca_oob_tcp_component_open(void);
|
|
int mca_oob_tcp_component_close(void);
|
|
mca_oob_t* mca_oob_tcp_component_init(int* priority);
|
|
|
|
/**
|
|
* Hook function to allow the selected oob components
|
|
* to register their contact info with the registry
|
|
*/
|
|
|
|
int mca_oob_tcp_init(void);
|
|
|
|
/**
|
|
* Cleanup resources during shutdown.
|
|
*/
|
|
int mca_oob_tcp_fini(void);
|
|
|
|
/*
|
|
* Register my contact info with the General Purpose Registry
|
|
* This function causes the component to "put" its contact info
|
|
* on the registry.
|
|
*/
|
|
int mca_oob_tcp_register_contact_info(void);
|
|
|
|
/*
|
|
* Register a subscription to receive contact info on other processes
|
|
* This function will typically be called from within a GPR compound command
|
|
* to register a subscription against a stage gate trigger. When fired, this
|
|
* will return the OOB contact info for all processes in the specified job
|
|
*/
|
|
int mca_oob_tcp_register_subscription(orte_jobid_t job, char *trigger);
|
|
|
|
/*
|
|
* Get contact info for a process or job
|
|
* Returns contact info for the specified process. If the vpid in the process name
|
|
* is WILDCARD, then it returns the contact info for all processes in the specified
|
|
* job. If the jobid is WILDCARD, then it returns the contact info for processes
|
|
* of the specified vpid across all jobs. Obviously, combining the two WILDCARD
|
|
* values will return contact info for everyone!
|
|
*/
|
|
ORTE_DECLSPEC int mca_oob_tcp_get_contact_info(orte_process_name_t *name, orte_gpr_notify_data_t **data);
|
|
|
|
/**
|
|
* Compare two process names for equality.
|
|
*
|
|
* @param n1 Process name 1.
|
|
* @param n2 Process name 2.
|
|
* @return (-1 for n1<n2 0 for equality, 1 for n1>n2)
|
|
*
|
|
* Note that the definition of < or > is somewhat arbitrary -
|
|
* just needs to be consistently applied to maintain an ordering
|
|
* when process names are used as indices.
|
|
*/
|
|
int mca_oob_tcp_process_name_compare(const orte_process_name_t* n1, const orte_process_name_t* n2);
|
|
|
|
/**
|
|
* Obtain contact information for this host (e.g. <ipaddress>:<port>)
|
|
*/
|
|
|
|
char* mca_oob_tcp_get_addr(void);
|
|
|
|
/**
|
|
* Setup cached addresses for the peers.
|
|
*/
|
|
|
|
int mca_oob_tcp_set_addr(const orte_process_name_t*, const char*);
|
|
|
|
/**
|
|
* A routine to ping a given process name to determine if it is reachable.
|
|
*
|
|
* @param name The peer name.
|
|
* @param tv The length of time to wait on a connection/response.
|
|
*
|
|
* Note that this routine blocks up to the specified timeout waiting for a
|
|
* connection / response from the specified peer. If the peer is unavailable
|
|
* an error status is returned.
|
|
*/
|
|
|
|
int mca_oob_tcp_ping(const orte_process_name_t*, const char* uri, const struct timeval* tv);
|
|
|
|
/**
|
|
* Similiar to unix writev(2).
|
|
*
|
|
* @param peer (IN) Opaque name of peer process.
|
|
* @param msg (IN) Array of iovecs describing user buffers and lengths.
|
|
* @param count (IN) Number of elements in iovec array.
|
|
* @param tag (IN) User defined tag for matching send/recv.
|
|
* @param flags (IN) Currently unused.
|
|
* @return OMPI error code (<0) on error number of bytes actually sent.
|
|
*/
|
|
|
|
int mca_oob_tcp_send(
|
|
orte_process_name_t* peer,
|
|
struct iovec *msg,
|
|
int count,
|
|
int tag,
|
|
int flags);
|
|
|
|
/**
|
|
* Similiar to unix readv(2)
|
|
*
|
|
* @param peer (IN) Opaque name of peer process or ORTE_NAME_WILDCARD for wildcard receive.
|
|
* @param msg (IN) Array of iovecs describing user buffers and lengths.
|
|
* @param count (IN) Number of elements in iovec array.
|
|
* @param tag (IN) User defined tag for matching send/recv.
|
|
* @param flags (IN) May be MCA_OOB_PEEK to return up to the number of bytes provided in the
|
|
* iovec array without removing the message from the queue.
|
|
* @return OMPI error code (<0) on error or number of bytes actually received.
|
|
*/
|
|
|
|
int mca_oob_tcp_recv(
|
|
orte_process_name_t* peer,
|
|
struct iovec * msg,
|
|
int count,
|
|
int tag,
|
|
int flags);
|
|
|
|
|
|
/*
|
|
* Non-blocking versions of send/recv.
|
|
*/
|
|
|
|
/**
|
|
* Non-blocking version of mca_oob_send().
|
|
*
|
|
* @param peer (IN) Opaque name of peer process.
|
|
* @param msg (IN) Array of iovecs describing user buffers and lengths.
|
|
* @param count (IN) Number of elements in iovec array.
|
|
* @param tag (IN) User defined tag for matching send/recv.
|
|
* @param flags (IN) Currently unused.
|
|
* @param cbfunc (IN) Callback function on send completion.
|
|
* @param cbdata (IN) User data that is passed to callback function.
|
|
* @return OMPI error code (<0) on error number of bytes actually sent.
|
|
*
|
|
*/
|
|
|
|
int mca_oob_tcp_send_nb(
|
|
orte_process_name_t* peer,
|
|
struct iovec* msg,
|
|
int count,
|
|
int tag,
|
|
int flags,
|
|
mca_oob_callback_fn_t cbfunc,
|
|
void* cbdata);
|
|
|
|
/**
|
|
* Non-blocking version of mca_oob_recv().
|
|
*
|
|
* @param peer (IN) Opaque name of peer process or ORTE_NAME_WILDCARD for wildcard receive.
|
|
* @param msg (IN) Array of iovecs describing user buffers and lengths.
|
|
* @param count (IN) Number of elements in iovec array.
|
|
* @param tag (IN) User defined tag for matching send/recv.
|
|
* @param flags (IN) May be MCA_OOB_PEEK to return up to size bytes of msg w/out removing it from the queue,
|
|
* @param cbfunc (IN) Callback function on recv completion.
|
|
* @param cbdata (IN) User data that is passed to callback function.
|
|
* @return OMPI error code (<0) on error or number of bytes actually received.
|
|
*/
|
|
|
|
int mca_oob_tcp_recv_nb(
|
|
orte_process_name_t* peer,
|
|
struct iovec* msg,
|
|
int count,
|
|
int tag,
|
|
int flags,
|
|
mca_oob_callback_fn_t cbfunc,
|
|
void* cbdata);
|
|
|
|
/**
|
|
* Cancel non-blocking receive.
|
|
*
|
|
* @param peer (IN) Opaque name of peer process or ORTE_NAME_WILDCARD for wildcard receive.
|
|
* @param tag (IN) User defined tag for matching send/recv.
|
|
* @return OMPI error code (<0) on error or number of bytes actually received.
|
|
*/
|
|
|
|
int mca_oob_tcp_recv_cancel(
|
|
orte_process_name_t* peer,
|
|
int tag);
|
|
|
|
/**
|
|
* Attempt to map a peer name to its corresponding address.
|
|
*/
|
|
|
|
int mca_oob_tcp_resolve(mca_oob_tcp_peer_t*);
|
|
|
|
/**
|
|
* Parse a URI string into an IP address and port number.
|
|
*/
|
|
int mca_oob_tcp_parse_uri(
|
|
const char* uri,
|
|
#if OPAL_WANT_IPV6
|
|
struct sockaddr_in6* inaddr
|
|
#else
|
|
struct sockaddr_in* inaddr
|
|
#endif
|
|
);
|
|
|
|
/**
|
|
* Callback from registry on change to subscribed segments
|
|
*/
|
|
void mca_oob_tcp_registry_callback(
|
|
orte_gpr_notify_data_t* data,
|
|
void* cbdata);
|
|
|
|
/**
|
|
* Setup socket options
|
|
*/
|
|
|
|
void mca_oob_tcp_set_socket_options(int sd);
|
|
|
|
int mca_oob_tcp_ft_event(int state);
|
|
|
|
typedef enum { OOB_TCP_EVENT, OOB_TCP_LISTEN_THREAD } mca_oob_tcp_listen_type_t;
|
|
|
|
/**
|
|
* OOB TCP Component
|
|
*/
|
|
struct mca_oob_tcp_component_t {
|
|
mca_oob_base_component_1_0_0_t super; /**< base OOB component */
|
|
char* tcp_include; /**< list of ip interfaces to include */
|
|
char* tcp_exclude; /**< list of ip interfaces to exclude */
|
|
int tcp_listen_sd; /**< listen socket for incoming IPv4 connection requests */
|
|
unsigned short tcp_listen_port; /**< IPv4 listen port */
|
|
#if OPAL_WANT_IPV6
|
|
int tcp6_listen_sd; /**< listen socket for incoming IPv6 connection requests */
|
|
unsigned short tcp6_listen_port; /**< IPv6 listen port */
|
|
#endif
|
|
opal_list_t tcp_subscriptions; /**< list of registry subscriptions */
|
|
opal_list_t tcp_peer_list; /**< list of peers sorted in mru order */
|
|
opal_hash_table_t tcp_peers; /**< peers sorted by name */
|
|
opal_hash_table_t tcp_peer_names; /**< cache of peer contact info sorted by name */
|
|
opal_free_list_t tcp_peer_free; /**< free list of peers */
|
|
int tcp_peer_limit; /**< max size of tcp peer cache */
|
|
int tcp_peer_retries; /**< max number of retries before declaring peer gone */
|
|
int tcp_sndbuf; /**< socket send buffer size */
|
|
int tcp_rcvbuf; /**< socket recv buffer size */
|
|
opal_free_list_t tcp_msgs; /**< free list of messages */
|
|
opal_event_t tcp_send_event; /**< event structure for IPv4 sends */
|
|
opal_event_t tcp_recv_event; /**< event structure for IPv4 recvs */
|
|
#if OPAL_WANT_IPV6
|
|
opal_event_t tcp6_send_event; /**< event structure for IPv6 sends */
|
|
opal_event_t tcp6_recv_event; /**< event structure for IPv6 recvs */
|
|
#endif
|
|
opal_mutex_t tcp_lock; /**< lock for accessing module state */
|
|
opal_list_t tcp_events; /**< list of pending events (accepts) */
|
|
opal_list_t tcp_msg_post; /**< list of recieves user has posted */
|
|
opal_list_t tcp_msg_recv; /**< list of recieved messages */
|
|
opal_list_t tcp_msg_completed; /**< list of completed messages */
|
|
opal_mutex_t tcp_match_lock; /**< lock held while searching/posting messages */
|
|
opal_condition_t tcp_match_cond; /**< condition variable used in finalize */
|
|
int tcp_match_count; /**< number of matched recvs in progress */
|
|
int tcp_debug; /**< debug level */
|
|
|
|
bool tcp_shutdown;
|
|
mca_oob_tcp_listen_type_t tcp_listen_type;
|
|
|
|
opal_thread_t tcp_listen_thread;
|
|
opal_free_list_t tcp_pending_connections_fl;
|
|
opal_list_t tcp_pending_connections;
|
|
opal_list_t tcp_copy_out_connections;
|
|
opal_list_t tcp_copy_in_connections;
|
|
opal_list_t tcp_connections_return;
|
|
opal_list_t tcp_connections_return_copy;
|
|
opal_mutex_t tcp_pending_connections_lock;
|
|
|
|
opal_timer_t tcp_last_copy_time;
|
|
opal_timer_t tcp_copy_delta;
|
|
int tcp_copy_max_size;
|
|
int tcp_copy_spin_count;
|
|
int connect_sleep;
|
|
};
|
|
|
|
/**
|
|
* Convenience Typedef
|
|
*/
|
|
typedef struct mca_oob_tcp_component_t mca_oob_tcp_component_t;
|
|
|
|
ORTE_MODULE_DECLSPEC extern mca_oob_tcp_component_t mca_oob_tcp_component;
|
|
|
|
extern int mca_oob_tcp_output_handle;
|
|
|
|
#if defined(__WINDOWS__)
|
|
#define CLOSE_THE_SOCKET(socket) closesocket(socket)
|
|
#else
|
|
#define CLOSE_THE_SOCKET(socket) close(socket)
|
|
#endif /* defined(__WINDOWS__) */
|
|
|
|
|
|
struct mca_oob_tcp_pending_connection_t {
|
|
opal_free_list_item_t super;
|
|
int fd;
|
|
/* Bug, FIXME: Port to IPv6 */
|
|
struct sockaddr_in addr;
|
|
};
|
|
typedef struct mca_oob_tcp_pending_connection_t mca_oob_tcp_pending_connection_t;
|
|
OBJ_CLASS_DECLARATION(mca_oob_tcp_pending_connection_t);
|
|
|
|
|
|
#if defined(c_plusplus) || defined(__cplusplus)
|
|
}
|
|
#endif
|
|
|
|
#endif /* MCA_OOB_TCP_H_ */
|
|
|