自学教程：C++ DIS_tcp_setup函数代码示例

intPBSD_msg_put(int c, char *jobid, int fileopt, char *msg, char *extend, int rpp, char **msgid){	int rc = 0;	int sock;	if (!rpp) {		sock = connection[c].ch_socket;		DIS_tcp_setup(sock);	} else {		sock = c;		if ((rc = is_compose_cmd(sock, IS_CMD, msgid)) != DIS_SUCCESS)			return rc;	}	if ((rc = encode_DIS_ReqHdr(sock, PBS_BATCH_MessJob,		pbs_current_user)) ||		(rc = encode_DIS_MessageJob(sock, jobid, fileopt, msg)) ||		(rc = encode_DIS_ReqExtend(sock, extend))) {		return (pbs_errno = PBSE_PROTOCOL);	}	if (DIS_wflush(sock, rpp)) {		pbs_errno = PBSE_PROTOCOL;		rc	  = pbs_errno;	}	return rc;}

/** * @brief * 	-PBS_resc() - internal common code for sending resource requests * * @par Functionality: *	Formats and sends the requests for pbs_rescquery(), pbs_rescreserve(), *	and pbs_rescfree().   Note, while the request is overloaded for all *	three, each has its own expected reply format. * * @param[in] c - communication handle * @param[in] reqtype - request type * @param[in] rescl- pointer to resource list * @param[in] ct - count of query strings * @param[in] rh - resource handle * * @return      int * @retval      0       success * @retval      !0      error * */static intPBS_resc(int c, int reqtype, char **rescl, int ct, pbs_resource_t rh){	int rc;	int sock;	sock = connection[c].ch_socket;	/* setup DIS support routines for following DIS calls */	DIS_tcp_setup(sock);	if ((rc = encode_DIS_ReqHdr(sock, reqtype, pbs_current_user)) ||		(rc = encode_DIS_Resc(sock, rescl, ct, rh)) ||		(rc = encode_DIS_ReqExtend(sock, (char *)0))) {		connection[c].ch_errtxt = strdup(dis_emsg[rc]);		if (connection[c].ch_errtxt == NULL) {			pbs_errno = PBSE_SYSTEM;		} else {			pbs_errno = PBSE_PROTOCOL;		}		return (pbs_errno);	}	if (DIS_tcp_wflush(sock)) {		return (pbs_errno = PBSE_PROTOCOL);	}	return (0);}

static int dis_reply_write(  int                 sfds,    /* I */  struct batch_reply *preply)  /* I */  {  int              rc = PBSE_NONE;  char             log_buf[LOCAL_LOG_BUF_SIZE];  struct tcp_chan *chan = NULL;  /* setup for DIS over tcp */  if ((chan = DIS_tcp_setup(sfds)) == NULL)    {    }  /* send message to remote client */  else if ((rc = encode_DIS_reply(chan, preply)) ||           (rc = DIS_tcp_wflush(chan)))    {    sprintf(log_buf, "DIS reply failure, %d", rc);    log_event(PBSEVENT_SYSTEM, PBS_EVENTCLASS_REQUEST, __func__, log_buf);    /* don't need to get the lock here because we already have it from process request */    close_conn(sfds, FALSE);    }  if (chan != NULL)    DIS_tcp_cleanup(chan);  return(rc);  }  /* END dis_reply_write() */

intPBSD_py_spawn_put(int c, char *jobid, char **argv, char **envp, int rpp, char **msgid){	int rc = 0;	int sock;	if (!rpp) {		sock = connection[c].ch_socket;		DIS_tcp_setup(sock);	} else {		sock = c;		if ((rc = is_compose_cmd(sock, IS_CMD, msgid)) != DIS_SUCCESS)			return rc;	}	if ((rc = encode_DIS_ReqHdr(sock, PBS_BATCH_PySpawn,		pbs_current_user)) ||		(rc = encode_DIS_PySpawn(sock, jobid, argv, envp)) ||		(rc = encode_DIS_ReqExtend(sock, NULL))) {			return (pbs_errno = PBSE_PROTOCOL);	}	if (DIS_wflush(sock, rpp)) {		pbs_errno = PBSE_PROTOCOL;		rc = pbs_errno;	}	return rc;}

/** * @brief read a batch reply from the given socket * * @param[in] sock - The socket fd to read from * @param[out] rc  - Return DIS error code * * @return Batch reply structure * @retval  !NULL - Success * @retval   NULL - Failure * */struct batch_reply *PBSD_rdrpy_sock(int sock, int *rc){	struct batch_reply *reply;	time_t old_timeout;	*rc = DIS_SUCCESS;	/* clear any prior error message */	if ((reply = (struct batch_reply *)malloc(sizeof(struct batch_reply))) == 0) {		pbs_errno = PBSE_SYSTEM;		return ((struct batch_reply *)0);	}	(void)memset(reply, 0, sizeof(struct batch_reply));	DIS_tcp_setup(sock);	old_timeout = pbs_tcp_timeout;	if (pbs_tcp_timeout < PBS_DIS_TCP_TIMEOUT_LONG)		pbs_tcp_timeout = PBS_DIS_TCP_TIMEOUT_LONG;	if ((*rc = decode_DIS_replyCmd(sock, reply)) != 0) {		(void)free(reply);		pbs_errno = PBSE_PROTOCOL;		return (struct batch_reply *)NULL;	}	DIS_tcp_reset(sock, 0);		/* reset DIS read buffer */	pbs_tcp_timeout = old_timeout;	pbs_errno = reply->brp_code;	return reply;}

int pbs_disconnect(  int connect)  /* I (socket descriptor) */  {  int  sock;  /* send close-connection message */  sock = connection[connect].ch_socket;  DIS_tcp_setup(sock);  if (encode_DIS_ReqHdr(sock, PBS_BATCH_Disconnect, pbs_current_user) == 0)    {    DIS_tcp_wflush(sock);    }  close(sock);  DIS_tcp_release(sock);  if (connection[connect].ch_errtxt != (char *)NULL)    free(connection[connect].ch_errtxt);  connection[connect].ch_errno = 0;  connection[connect].ch_inuse = 0;  return(0);  }  /* END pbs_disconnect() */

static int addrm(  int stream)  /* I */  {  struct out *op, **head;  if ((op = (struct out *)calloc(1, sizeof(struct out))) == NULL)    {    return(errno);    }  else if ((op->chan = DIS_tcp_setup(stream)) == NULL)    {    free(op);    return errno;    }  head = &outs[stream % HASHOUT];  op->len = -1;  op->next = *head;  *head = op;  return 0;  }

static int dis_reply_write(  int                 sfds,    /* I */  struct batch_reply *preply)  /* I */  {  int rc;  /* setup for DIS over tcp */  DIS_tcp_setup(sfds);  /* send message to remote client */  if ((rc = encode_DIS_reply(sfds, preply)) ||      (rc = DIS_tcp_wflush(sfds)))    {    sprintf(log_buffer, "DIS reply failure, %d",            rc);    LOG_EVENT(      PBSEVENT_SYSTEM,      PBS_EVENTCLASS_REQUEST,      "dis_reply_write",      log_buffer);    close_conn(sfds);    }  return(rc);  }  /* END dis_reply_write() */

int__pbs_orderjob(int c, char *job1, char *job2, char *extend){	struct batch_reply *reply;	int rc;	int sock;	if ((job1 == NULL) || (*job1 == '/0') ||		(job2 == NULL) || (*job2 == '/0'))		return (pbs_errno = PBSE_IVALREQ);	sock = connection[c].ch_socket;	/* initialize the thread context data, if not already initialized */	if (pbs_client_thread_init_thread_context() != 0)		return pbs_errno;	/* lock pthread mutex here for this connection */	/* blocking call, waits for mutex release */	if (pbs_client_thread_lock_connection(c) != 0)		return pbs_errno;	/* setup DIS support routines for following DIS calls */	DIS_tcp_setup(sock);	if ((rc = encode_DIS_ReqHdr(sock, PBS_BATCH_OrderJob, pbs_current_user)) ||		(rc = encode_DIS_MoveJob(sock, job1, job2)) ||		(rc = encode_DIS_ReqExtend(sock, extend))) {		connection[c].ch_errtxt = strdup(dis_emsg[rc]);		if (connection[c].ch_errtxt == NULL) {			pbs_errno = PBSE_SYSTEM;		} else {			pbs_errno = PBSE_PROTOCOL;		}		(void)pbs_client_thread_unlock_connection(c);		return pbs_errno;	}	if (DIS_tcp_wflush(sock)) {		pbs_errno = PBSE_PROTOCOL;		(void)pbs_client_thread_unlock_connection(c);		return pbs_errno;	}	/* read reply */	reply = PBSD_rdrpy(c);	PBSD_FreeReply(reply);	rc = connection[c].ch_errno;	/* unlock the thread lock and update the thread context data */	if (pbs_client_thread_unlock_connection(c) != 0)		return pbs_errno;	return rc;}

int parse_response_svr(    int sock,    char **err_msg)  {  /*   * PBS_BATCH_PROT_TYPE   * PBS_BATCH_PROT_VER   * reply->brp_code   * reply->brp_auxcode   * reply->brp_choice   * if reply->brp_choice == BATCH_REPLY_CHOICE_Text also read:   * preq->rq_reply.brp_un.brp_txt.brp_str   * using   * preq->rq_reply.brp_un.brp_txt.brp_txtlen   */  int rc = PBSE_NONE;  struct batch_reply *reply = NULL;  char *tmp_val = NULL;  struct tcp_chan *chan = NULL;  if ((chan = DIS_tcp_setup(sock)) == NULL)    {    }  else if ((reply = (struct batch_reply *)calloc(1, sizeof(struct batch_reply))) == NULL)    {    }  else if ((rc = decode_DIS_replyCmd(chan, reply)))    {    free(reply);    if (chan->IsTimeout == TRUE)      {      rc = PBSE_TIMEOUT;      }    else      {      rc = PBSE_PROTOCOL;      }    if ((tmp_val = pbs_strerror(rc)) == NULL)      {      char err_buf[80];      snprintf(err_buf, 79, "Error creating error message for code %d", rc);      *err_msg = strdup(err_buf);      }    else      *err_msg = strdup(tmp_val);    }  else    {    rc = reply->brp_code;    if (reply->brp_code != PBSE_NONE)      {      *err_msg = strdup(reply->brp_un.brp_txt.brp_str);      }    free(reply);    }  DIS_tcp_cleanup(chan);  return rc;  }

void *start_process_request(void *vp)  {  int sock = *(int *)vp;  struct tcp_chan *chan = NULL;  if ((chan = DIS_tcp_setup(sock)) == NULL)    return NULL;  process_request(chan);  DIS_tcp_cleanup(chan);  return(NULL);  }

int PBSD_selectattr_put(  int             c,  int             type,  struct attropl *attropl,  struct attrl   *attrib,  char           *extend)  {  int rc = PBSE_NONE;  int sock;  struct tcp_chan *chan = NULL;    if ((c < 0) ||       (c >= PBS_NET_MAX_CONNECTIONS))    {    return(PBSE_IVALREQ);    }  pthread_mutex_lock(connection[c].ch_mutex);  sock = connection[c].ch_socket;  /* setup DIS support routines for following DIS calls */  if ((chan = DIS_tcp_setup(sock)) == NULL)    {    pthread_mutex_unlock(connection[c].ch_mutex);    rc = PBSE_PROTOCOL;    return rc;    }  else if ((rc = encode_DIS_ReqHdr(chan, type, pbs_current_user)) ||      (rc = encode_DIS_attropl(chan, attropl)) ||      (rc = encode_DIS_attrl(chan, attrib)) ||      (rc = encode_DIS_ReqExtend(chan, extend)))    {    connection[c].ch_errtxt = strdup(dis_emsg[rc]);    pthread_mutex_unlock(connection[c].ch_mutex);    DIS_tcp_cleanup(chan);    return (PBSE_PROTOCOL);    }  pthread_mutex_unlock(connection[c].ch_mutex);  /* write data */  if (DIS_tcp_wflush(chan))    {    rc = PBSE_PROTOCOL;    }  DIS_tcp_cleanup(chan);  return rc;  } /* END PBSD_selectattr_put() */

int PBSD_rdytocmt(  int   connect,  char *jobid)  {  int                 rc;  struct batch_reply *reply;  int                 sock;  struct tcp_chan *chan = NULL;    if ((connect < 0) ||       (connect >= PBS_NET_MAX_CONNECTIONS))    {    return(PBSE_IVALREQ);    }  pthread_mutex_lock(connection[connect].ch_mutex);  sock = connection[connect].ch_socket;  pthread_mutex_unlock(connection[connect].ch_mutex);  if ((chan = DIS_tcp_setup(sock)) == NULL)    {    return(PBSE_MEM_MALLOC);    }  else if ((rc = encode_DIS_ReqHdr(chan, PBS_BATCH_RdytoCommit, pbs_current_user)) ||      (rc = encode_DIS_JobId(chan, jobid)) ||      (rc = encode_DIS_ReqExtend(chan, NULL)))    {    pthread_mutex_lock(connection[connect].ch_mutex);    connection[connect].ch_errtxt = strdup(dis_emsg[rc]);    pthread_mutex_unlock(connection[connect].ch_mutex);    DIS_tcp_cleanup(chan);    return(PBSE_PROTOCOL);    }  if (DIS_tcp_wflush(chan))    {    DIS_tcp_cleanup(chan);    return(PBSE_PROTOCOL);    }  DIS_tcp_cleanup(chan);  /* read reply */  reply = PBSD_rdrpy(&rc, connect);  PBSD_FreeReply(reply);  return(rc);  }

intPBSD_ucred(int c, char *user, int type, char *buf, int len){	int			rc;	struct batch_reply	*reply = NULL;	int			sock;	sock = connection[c].ch_socket;	/* initialize the thread context data, if not already initialized */	if (pbs_client_thread_init_thread_context() != 0)		return pbs_errno;	/* lock pthread mutex here for this connection */	/* blocking call, waits for mutex release */	if (pbs_client_thread_lock_connection(c) != 0)		return pbs_errno;	DIS_tcp_setup(sock);	if ((rc =encode_DIS_ReqHdr(sock, PBS_BATCH_UserCred, pbs_current_user)) ||		(rc = encode_DIS_UserCred(sock, user, type, buf, len)) ||		(rc = encode_DIS_ReqExtend(sock, (char *)0))) {		connection[c].ch_errtxt = strdup(dis_emsg[rc]);		if (connection[c].ch_errtxt == NULL) {			pbs_errno = PBSE_SYSTEM;		} else {			pbs_errno = PBSE_PROTOCOL;		}		(void)pbs_client_thread_unlock_connection(c);		return pbs_errno;	}	if (DIS_tcp_wflush(sock)) {		pbs_errno = PBSE_PROTOCOL;		(void)pbs_client_thread_unlock_connection(c);		return pbs_errno;	}	reply = PBSD_rdrpy(c);	PBSD_FreeReply(reply);	rc = connection[c].ch_errno;	/* unlock the thread lock and update the thread context data */	if (pbs_client_thread_unlock_connection(c) != 0)		return pbs_errno;	return rc;}

int PBSD_status_put(    int           c,    int           function,    char         *id,    struct attrl *attrib,    char         *extend){    int rc = 0;    int sock;    struct tcp_chan *chan = NULL;    if ((c < 0) ||            (c >= PBS_NET_MAX_CONNECTIONS))    {        return(PBSE_IVALREQ);    }    mutex_mgr ch_mutex = mutex_mgr(connection[c].ch_mutex, false);    sock = connection[c].ch_socket;    if ((chan = DIS_tcp_setup(sock)) == NULL)    {        rc = PBSE_MEM_MALLOC;        return rc;    }    else if ((rc = encode_DIS_ReqHdr(chan, function, pbs_current_user)) ||             (rc = encode_DIS_Status(chan, id, attrib)) ||             (rc = encode_DIS_ReqExtend(chan, extend)))    {        connection[c].ch_errtxt = strdup(dis_emsg[rc]);        DIS_tcp_cleanup(chan);        return(PBSE_PROTOCOL);    }    ch_mutex.unlock();    if (DIS_tcp_wflush(chan))    {        rc = PBSE_PROTOCOL;    }    /* success */    DIS_tcp_cleanup(chan);    return(PBSE_NONE);}  /* END PBSD_status_put() */

int pbs_lock(  int   c,      /* I */  int   action, /* I */  char *extend) /* I */  {  struct batch_reply *reply;  int rc = 0;  int sock;  /* send request */  sock = connection[c].ch_socket;  /* setup DIS support routines for following DIS calls */  DIS_tcp_setup(sock);  if ((rc = encode_DIS_ReqHdr(sock, PBS_BATCH_SchedulerLock, pbs_current_user)) ||      (rc = encode_DIS_SchedLock(sock, action)) ||      (rc = encode_DIS_ReqExtend(sock, extend)))    {    connection[c].ch_errtxt = strdup(dis_emsg[rc]);    pbs_errno = PBSE_PROTOCOL;    return(pbs_errno);    }  if (DIS_tcp_wflush(sock))    {    pbs_errno = PBSE_PROTOCOL;    return(pbs_errno);    }  /* read in reply */  reply = PBSD_rdrpy(c);  rc = connection[c].ch_errno;  PBSD_FreeReply(reply);  return(rc);  }  /* END pbs_terminate() */

int PBSD_mgr_put(  int             c,        /* I */  int             function, /* I */  int             command,  /* I */  int             objtype,  /* I */  char           *objname,  /* I */  struct attropl *aoplp,    /* I */  char           *extend)   /* I */  {  int rc = PBSE_NONE;  int sock;  struct tcp_chan *chan = NULL;  pthread_mutex_lock(connection[c].ch_mutex);  sock = connection[c].ch_socket;  if ((chan = DIS_tcp_setup(sock)) == NULL)    {    pthread_mutex_unlock(connection[c].ch_mutex);    rc = PBSE_PROTOCOL;    return(rc);    }  else if ((rc = encode_DIS_ReqHdr(chan, function, pbs_current_user)) ||           (rc = encode_DIS_Manage(chan, command, objtype, objname, aoplp)) ||           (rc = encode_DIS_ReqExtend(chan, extend)))    {    connection[c].ch_errtxt = strdup(dis_emsg[rc]);    pthread_mutex_unlock(connection[c].ch_mutex);    DIS_tcp_cleanup(chan);    rc = PBSE_PROTOCOL; /* We shouldn't be overridding this error!!! */    return rc;    }  if (DIS_tcp_wflush(chan))    {    pthread_mutex_unlock(connection[c].ch_mutex);    rc = PBSE_PROTOCOL;    }  pthread_mutex_unlock(connection[c].ch_mutex);  DIS_tcp_cleanup(chan);  return rc;  }  /* END PBSD_mgr_put() */

int pbs_disconnect_socket(  int sock)  /* I (socket descriptor) */  {  char tmp_buf[THE_BUF_SIZE / 4];  struct tcp_chan *chan = NULL;  if ((chan = DIS_tcp_setup(sock)) == NULL)    {    }  else if ((encode_DIS_ReqHdr(chan,PBS_BATCH_Disconnect, pbs_current_user) == 0)      && (DIS_tcp_wflush(chan) == 0))    {    int atime;    struct sigaction act;    struct sigaction oldact;    /* set alarm to break out of potentially infinite read *//*    act.sa_handler = SIG_IGN; */    act.sa_handler = empty_alarm_handler;  /* need SOME handler or blocking read never gets interrupted */    sigemptyset(&act.sa_mask);    act.sa_flags = 0;    sigaction(SIGALRM, &act, &oldact);    atime = alarm(5);    while (1)      {      /* wait for server to close connection */      /* NOTE:  if read of 'sock' is blocking, request below may hang forever         -- hence the signal handler above */      if (read_ac_socket(sock, &tmp_buf, sizeof(tmp_buf)) < 1)        break;      }    alarm(atime);    sigaction(SIGALRM, &oldact, NULL);    }  if (chan != NULL)    DIS_tcp_cleanup(chan);  close(sock);  return(0);  }  /* END pbs_disconnect_socket() */

int PBSD_commit(    int   connect,  /* I */    char *jobid)    /* I */{    struct batch_reply *reply;    int                 rc;    int                 sock;    struct tcp_chan *chan = NULL;    pthread_mutex_lock(connection[connect].ch_mutex);    sock = connection[connect].ch_socket;    pthread_mutex_unlock(connection[connect].ch_mutex);    if ((chan = DIS_tcp_setup(sock)) == NULL)    {        return(PBSE_MEM_MALLOC);    }    else if ((rc = encode_DIS_ReqHdr(chan, PBS_BATCH_Commit, pbs_current_user)) ||             (rc = encode_DIS_JobId(chan, jobid)) ||             (rc = encode_DIS_ReqExtend(chan, NULL)))    {        pthread_mutex_lock(connection[connect].ch_mutex);        connection[connect].ch_errtxt = strdup(dis_emsg[rc]);        pthread_mutex_unlock(connection[connect].ch_mutex);        DIS_tcp_cleanup(chan);        return(PBSE_PROTOCOL);    }    if (DIS_tcp_wflush(chan))    {        DIS_tcp_cleanup(chan);        return(PBSE_PROTOCOL);    }    DIS_tcp_cleanup(chan);    /* PBSD_rdrpy sets connection[connect].ch_errno */    reply = PBSD_rdrpy(&rc, connect);    PBSD_FreeReply(reply);    return(rc);}  /* END PBSD_commit() */

int PBSD_async_sig_put(    int   c,    char *jobid,    char *signal,    char *extend){    int sock;    int rc = 0;    struct tcp_chan *chan = NULL;    pthread_mutex_lock(connection[c].ch_mutex);    sock = connection[c].ch_socket;    if ((chan = DIS_tcp_setup(sock)) == NULL)    {        rc = PBSE_PROTOCOL;        return rc;    }    else if ((rc = encode_DIS_ReqHdr(chan,PBS_BATCH_SignalJob,pbs_current_user))             || (rc = encode_DIS_SignalJob(chan,jobid,signal))             || (rc = encode_DIS_ReqExtend(chan,extend)))    {        connection[c].ch_errtxt = strdup(dis_emsg[rc]);        pthread_mutex_unlock(connection[c].ch_mutex);        DIS_tcp_cleanup(chan);        return (PBSE_PROTOCOL);    }    pthread_mutex_unlock(connection[c].ch_mutex);    if (DIS_tcp_wflush(chan))    {        rc = PBSE_PROTOCOL;    }    DIS_tcp_cleanup(chan);    return(rc);} /* END PBSD_async_sig_put() */

static int PBS_resc(      int          c,  int          reqtype,   char       **rescl,  int          ct,  resource_t   rh)  {  int rc;  int sock;  struct tcp_chan *chan = NULL;  sock = connection[c].ch_socket;  /* setup DIS support routines for following DIS calls */  if ((chan = DIS_tcp_setup(sock)) == NULL)    {    pthread_mutex_unlock(connection[c].ch_mutex);    rc = PBSE_PROTOCOL;    return rc;    }  else if ((rc = encode_DIS_ReqHdr(chan, reqtype, pbs_current_user)) ||      (rc = encode_DIS_Resc(chan, rescl, ct, rh)) ||      (rc = encode_DIS_ReqExtend(chan, (char *)0)))    {    connection[c].ch_errtxt = strdup(dis_emsg[rc]);    pthread_mutex_unlock(connection[c].ch_mutex);    DIS_tcp_cleanup(chan);    return (PBSE_PROTOCOL);    }  if (DIS_tcp_wflush(chan))    {    rc = PBSE_PROTOCOL;    }  DIS_tcp_cleanup(chan);  return rc;  } /* END PBS_resc() */

int PBSD_status_put(  int           c,  int           function,  char         *id,  struct attrl *attrib,  char         *extend)  {  int rc = 0;  int sock;  sock = connection[c].ch_socket;  DIS_tcp_setup(sock);  if ((rc = encode_DIS_ReqHdr(sock, function, pbs_current_user)) ||      (rc = encode_DIS_Status(sock, id, attrib)) ||      (rc = encode_DIS_ReqExtend(sock, extend)))    {    connection[c].ch_errtxt = strdup(dis_emsg[rc]);    pbs_errno = PBSE_PROTOCOL;    return(pbs_errno);    }  if (DIS_tcp_wflush(sock))    {    pbs_errno = PBSE_PROTOCOL;    return(pbs_errno);    }  /* success */  return(0);  }  /* END PBSD_status_put() */

int PBSD_gpu_put(      int   c,  char *node,  char *gpuid,  int   gpumode,  int   reset_perm,  int   reset_vol,  char *extend)    {  int sock;  int rc = 0;  struct tcp_chan *chan = NULL;  sock = connection[c].ch_socket;  if ((chan = DIS_tcp_setup(sock)) == NULL)    {    return PBSE_PROTOCOL;    }  else if ((rc = encode_DIS_ReqHdr(chan, PBS_BATCH_GpuCtrl, pbs_current_user)) ||      (rc = encode_DIS_GpuCtrl(chan, node, gpuid, gpumode, reset_perm, reset_vol)) ||      (rc = encode_DIS_ReqExtend(chan, extend)))    {    connection[c].ch_errtxt = strdup(dis_emsg[rc]);    DIS_tcp_cleanup(chan);    return(PBSE_PROTOCOL);    }  if (DIS_tcp_wflush(chan))    {    rc = PBSE_PROTOCOL;    }  DIS_tcp_cleanup(chan);  return(rc);  }

int PBSD_munge_authenticate(  int psock,  /* I */  int handle) /* I */  {  int                 rc;  munge_ctx_t         mctx = NULL;  munge_err_t         mret = EMUNGE_SNAFU;  char               *mcred = NULL;  /* user id and name stuff */  struct passwd      *pwent;  uid_t               myrealuid;  struct batch_reply *reply;  unsigned short      user_port = 0;  struct sockaddr_in  sockname;  socklen_t           socknamelen = sizeof(sockname);  struct tcp_chan    *chan;  if ((mctx = munge_ctx_create()) == NULL)    {    return(-1);    }  if ((mret = munge_encode (&mcred, mctx, NULL, 0)) != EMUNGE_SUCCESS)    {    const char *merrmsg = NULL;    if (!(merrmsg = munge_ctx_strerror(mctx)))      {      merrmsg = munge_strerror(mret);      }    fprintf(stderr, "munge_encode failed: %s (%d)/n", merrmsg, mret);    munge_ctx_destroy(mctx);    return(PBSE_MUNGE_NOT_FOUND); /*TODO more fine-grained error codes? */    }    munge_ctx_destroy(mctx);  /* We got the certificate. Now make the PBS_BATCH_AltAuthenUser request */  myrealuid = getuid();  pwent = getpwuid(myrealuid);    rc = getsockname(psock, (struct sockaddr *)&sockname, &socknamelen);    if (rc == -1)    {    fprintf(stderr, "getsockname failed: %d/n", errno);    return(-1);    }    user_port = ntohs(sockname.sin_port);    if ((chan = DIS_tcp_setup(psock)) == NULL)    {    }  else if ((rc = encode_DIS_ReqHdr(chan, PBS_BATCH_AltAuthenUser, pwent->pw_name)) ||           (rc = diswui(chan, user_port)) ||           (rc = diswst(chan, mcred)) ||           (rc = encode_DIS_ReqExtend(chan, NULL)) ||           (rc = DIS_tcp_wflush(chan)))    {    PBSD_munge_cred_destroy(&mcred);    /* ERROR */    return(rc);    }  else    {    int local_err = PBSE_NONE;    PBSD_munge_cred_destroy(&mcred);    /* read the reply */    if ((reply = PBSD_rdrpy(&local_err, handle)) != NULL)      free(reply);        return(PBSE_NONE);    }  return(-1);  }

//.........这里部分代码省略.........        fprintf(stderr, "ERROR:  cannot get servername (%s) errno=%d (%s)/n",                (server != NULL) ? server : "NULL",                errno,                strerror(errno));        }      return(-1);      }    memcpy((char *)&server_addr.sin_addr, hp->h_addr_list[0], hp->h_length);    server_addr.sin_port = htons(server_port);    if (connect(          connection[out].ch_socket,          (struct sockaddr *)&server_addr,          sizeof(server_addr)) < 0)      {      close(connection[out].ch_socket);      connection[out].ch_inuse = 0;      pbs_errno = errno;      if (getenv("PBSDEBUG"))        {        fprintf(stderr, "ERROR:  cannot connect to server, errno=%d (%s)/n",                errno,                strerror(errno));        }      return(-1);      }  DIS_tcp_setup(connection[out].ch_socket);  if ((ptr = getenv("PBSAPITIMEOUT")) != NULL)    {    pbs_tcp_timeout = strtol(ptr,NULL,0);	    if (pbs_tcp_timeout <= 0)      {      pbs_tcp_timeout = 10800;      /* set for 3 hour time out */      }    }  else    {    pbs_tcp_timeout = 10800;      /* set for 3 hour time out */    }  /* If we have GSSAPI, then try gssapi authentication first.  If that fails, fall back to iff.     If there's no GSSAPI, then just use iff.   */#ifdef GSSAPI  if (!getenv("TORQUE_IGNORE_KERBEROS") &&      !ignore_kerberos_for_connection &&      pbsgss_can_get_creds())    {    if (encode_DIS_ReqHdr(connection[out].ch_socket, PBS_BATCH_GSSAuthenUser, pbs_current_user) ||        encode_DIS_ReqExtend(connection[out].ch_socket,0))      {      if (getenv("PBSDEBUG"))        {        fprintf(stderr,"ERROR:  cannot authenticate connection with gssapi, errno=%d (%s)/n", errno, strerror(errno));        }      neediff = 1;

int pbs_asyrunjob_err(  int   c,  char *jobid,    /* I */  char *location,  char *extend,  int  *local_errno)  {  int                 rc;  struct batch_reply *reply;  unsigned int        resch = 0;  int                 sock;  struct tcp_chan *chan = NULL;  if ((c < 0) || (jobid == NULL) || (*jobid == '/0'))    {    return(PBSE_IVALREQ);    }  if (location == NULL)    location = "";  pthread_mutex_lock(connection[c].ch_mutex);  sock = connection[c].ch_socket;  /* setup DIS support routines for following DIS calls */  if ((chan = DIS_tcp_setup(sock)) == NULL)    {    rc = PBSE_PROTOCOL;    return rc;    }  else if ((rc = encode_DIS_ReqHdr(chan, PBS_BATCH_AsyrunJob, pbs_current_user))      || (rc = encode_DIS_RunJob(chan, jobid, location, resch))      || (rc = encode_DIS_ReqExtend(chan, extend)))    {    connection[c].ch_errtxt = strdup(dis_emsg[rc]);    pthread_mutex_unlock(connection[c].ch_mutex);    DIS_tcp_cleanup(chan);    return(PBSE_PROTOCOL);    }  if (DIS_tcp_wflush(chan))    {    pthread_mutex_unlock(connection[c].ch_mutex);    DIS_tcp_cleanup(chan);    return(PBSE_PROTOCOL);    }  /* get reply */  reply = PBSD_rdrpy(local_errno, c);  rc = connection[c].ch_errno;  pthread_mutex_unlock(connection[c].ch_mutex);  PBSD_FreeReply(reply);  DIS_tcp_cleanup(chan);  return(rc);  }  /* END pbs_asyrunjob_err() */

int pbs_runjob_err(  int   c,  char *jobid,  char *location,  char *extend,  int  *rc)  {  struct batch_reply   *reply;  unsigned int          resch = 0;  int                   sock;  struct tcp_chan *chan = NULL;  /* NOTE:  set_task sets WORK_Deferred_Child : request remains until child terminates */  if ((jobid == NULL) ||      (*jobid == '/0'))    {    *rc = PBSE_IVALREQ;    return (*rc) * -1;    }    if ((c < 0) ||       (c >= PBS_NET_MAX_CONNECTIONS))    {    return(PBSE_IVALREQ * -1);    }  if (location == NULL)    {    location = (char *)"";    }  pthread_mutex_lock(connection[c].ch_mutex);  sock = connection[c].ch_socket;  /* setup DIS support routines for following DIS calls */  if ((chan = DIS_tcp_setup(sock)) == NULL)    {    pthread_mutex_unlock(connection[c].ch_mutex);    *rc = PBSE_PROTOCOL;    return(*rc);    }  /* send run request */  else if ((*rc = encode_DIS_ReqHdr(chan, PBS_BATCH_RunJob, pbs_current_user)) ||           (*rc = encode_DIS_RunJob(chan, jobid, location, resch)) ||           (*rc = encode_DIS_ReqExtend(chan, extend)))    {    connection[c].ch_errtxt = strdup(dis_emsg[*rc]);    pthread_mutex_unlock(connection[c].ch_mutex);    DIS_tcp_cleanup(chan);    return(PBSE_PROTOCOL);    }  if ((*rc = DIS_tcp_wflush(chan)) != PBSE_NONE)    {    pthread_mutex_unlock(connection[c].ch_mutex);        DIS_tcp_cleanup(chan);    return(PBSE_PROTOCOL);    }  /* get reply */  reply = PBSD_rdrpy(rc, c);/*  *rc = connection[c].ch_errno; */  pthread_mutex_unlock(connection[c].ch_mutex);  PBSD_FreeReply(reply);      DIS_tcp_cleanup(chan);  return(*rc);  }  /* END pbs_runjob_err() */

int do_mom(  char *HPtr,  int   MOMPort,  int   CmdIndex)  {  int socket;  int local_errno = 0;  struct tcp_chan *chan = NULL;  int rc;  if ((socket = openrm(HPtr, MOMPort)) < 0)    {    /* FAILURE */    extern char TRMEMsg[];    fprintf(stderr, "cannot connect to MOM on node '%s', errno=%d (%s)/n",      HPtr,      errno,      strerror(errno));    if (TRMEMsg[0] != '/0')      {      fprintf(stderr, " %s/n",              TRMEMsg);      }    return(socket);    }  else if ((chan = DIS_tcp_setup(socket)) == NULL)    {    fprintf(stderr, "%s: can not allocate memory of socket buffers/n", __func__);    return -1;    }  /* send protocol and version, plus how many queries we're sending */  if (QueryI == 0)    QueryI = 1;  if (start_dialogue(chan) != DIS_SUCCESS)    {    fprintf(stderr,"ERROR:    Unable to write the number of queries to %s (errno=%d-%s)/n",        HPtr,        errno,        strerror(errno));        send_command(chan,RM_CMD_CLOSE);    DIS_tcp_cleanup(chan);    return(-1);    }  if (IsVerbose == TRUE)    {    fprintf(stderr, "INFO:     successfully connected to %s/n",            HPtr);    }  switch (CmdIndex)    {    case momClear:      {      char tmpLine[1024];      char *Value;      snprintf(tmpLine, 1024, "clearjob=%s",               (JPtr != NULL) ? JPtr : "all");      if (send_command_str(chan, RM_CMD_REQUEST, tmpLine) != 0)        {        /* FAILURE */        fprintf(stderr,"ERROR:    cannot request job clear on %s (errno=%d-%s)/n",          HPtr,          errno,          strerror(errno));        send_command(chan,RM_CMD_CLOSE);        return(-1);        }      if ((Value = (char *)read_mom_reply(&local_errno, chan)) == NULL)        {        /* FAILURE */        fprintf(stderr,"ERROR:    job clear failed on %s (errno=%d - %s: %d - %s)/n",          HPtr,          errno,          pbs_strerror(errno),          local_errno,          pbs_strerror(local_errno));        send_command(chan,RM_CMD_CLOSE);//.........这里部分代码省略.........

int__pbs_rerunjob(int c, char *jobid, char *extend){	int	rc;	struct batch_reply *reply;	int	sock;	time_t  old_tcp_timeout;	if ((jobid == NULL) || (*jobid == '/0'))		return (pbs_errno = PBSE_IVALREQ);	sock = connection[c].ch_socket;	/* initialize the thread context data, if not already initialized */	if (pbs_client_thread_init_thread_context() != 0)		return pbs_errno;	/* lock pthread mutex here for this connection */	/* blocking call, waits for mutex release */	if (pbs_client_thread_lock_connection(c) != 0)		return pbs_errno;	/* setup DIS support routines for following DIS calls */	DIS_tcp_setup(sock);	if ((rc = encode_DIS_ReqHdr(sock, PBS_BATCH_Rerun, pbs_current_user)) ||		(rc = encode_DIS_JobId(sock, jobid)) ||		(rc = encode_DIS_ReqExtend(sock, extend))) {		connection[c].ch_errtxt = strdup(dis_emsg[rc]);		if (connection[c].ch_errtxt == NULL) {			pbs_errno = PBSE_SYSTEM;		} else {			pbs_errno = PBSE_PROTOCOL;		}		(void)pbs_client_thread_unlock_connection(c);		return pbs_errno;	}	/* write data */	if (DIS_tcp_wflush(sock)) {		pbs_errno = PBSE_PROTOCOL;		(void)pbs_client_thread_unlock_connection(c);		return pbs_errno;	}	/* Set timeout value to very long value as rerun request */	/* goes from Server to Mom and may take a long time      */	old_tcp_timeout = pbs_tcp_timeout;	pbs_tcp_timeout = PBS_DIS_TCP_TIMEOUT_VLONG;	/* read reply from stream into presentation element */	reply = PBSD_rdrpy(c);	/* reset timeout */	pbs_tcp_timeout = old_tcp_timeout;	PBSD_FreeReply(reply);	rc = connection[c].ch_errno;	/* unlock the thread lock and update the thread context data */	if (pbs_client_thread_unlock_connection(c) != 0)		return pbs_errno;	return rc;}