| /////////////////////////////////////////////////////////////////////////////// |
| // |
| // @@@ START COPYRIGHT @@@ |
| // |
| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| // |
| // @@@ END COPYRIGHT @@@ |
| // |
| /////////////////////////////////////////////////////////////////////////////// |
| |
| #include <iostream> |
| |
| using namespace std; |
| |
| #include <stddef.h> |
| #include <stdio.h> |
| #include <stdlib.h> |
| #include <string.h> |
| #include <fcntl.h> |
| #include <netdb.h> |
| #include <sys/socket.h> |
| #include <netinet/in.h> |
| #include <netinet/tcp.h> |
| #include <sys/epoll.h> |
| #include <sys/file.h> |
| #include <sys/time.h> |
| #include <sys/resource.h> |
| #include <errno.h> |
| #include <limits.h> |
| #include <unistd.h> |
| |
| #include "lnode.h" |
| #include "pnode.h" |
| #include "ptpclient.h" |
| #include "monitor.h" |
| #include "monlogging.h" |
| #include "montrace.h" |
| #include "meas.h" |
| |
| extern CMonitor *Monitor; |
| extern CNode *MyNode; |
| extern CNodeContainer *Nodes; |
| extern bool IsRealCluster; |
| extern CMeas Meas; |
| extern int MyPNID; |
| |
| #define MON2MON_IO_RETRIES 3 |
| |
| CPtpClient::CPtpClient (void) |
| : ptpCommPort_(0) |
| , ptpClusterSocks_(NULL) |
| , seqNum_(0) |
| { |
| const char method_name[] = "CPtpClient::CPtpClient"; |
| TRACE_ENTRY; |
| |
| ptpHost_[0] = '\0'; |
| ptpPortBase_[0] = '\0'; |
| if ( !IsRealCluster ) |
| { |
| SetLocalHost(); |
| } |
| |
| char * env = getenv( "MON2MON_COMM_PORT" ); |
| if ( env ) |
| { |
| ptpCommPort_ = atoi( env ); |
| } |
| else |
| { |
| char buf[MON_STRING_BUF_SIZE]; |
| snprintf( buf, sizeof(buf) |
| , "[%s@%d] MON2MON_COMM_PORT environment variable is not set!\n" |
| , method_name, __LINE__ ); |
| mon_log_write( PTPCLIENT_PTPCLIENT_1, SQ_LOG_CRIT, buf ); |
| abort(); |
| } |
| |
| ptpClusterSocks_ = new int[MAX_NODES]; |
| for (int i=0; i < MAX_NODES; ++i) |
| { |
| ptpClusterSocks_[i] = -1; |
| } |
| |
| TRACE_EXIT; |
| } |
| |
| CPtpClient::~CPtpClient (void) |
| { |
| const char method_name[] = "CPtpClient::~CPtpClient"; |
| TRACE_ENTRY; |
| |
| delete [] ptpClusterSocks_; |
| |
| TRACE_EXIT; |
| } |
| |
| int CPtpClient::InitializePtpClient( int pnid, char * ptpPort ) |
| { |
| const char method_name[] = "CPtpClient::InitializePtpClient"; |
| TRACE_ENTRY; |
| int err = 0; |
| |
| if (ptpClusterSocks_[pnid] == -1) |
| { |
| int sock = Monitor->MkCltSock( ptpPort ); |
| if (sock < 0) |
| { |
| err = sock; |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - MkCltSock failed with error %d\n" |
| , method_name, __LINE__, err ); |
| } |
| } |
| else |
| { |
| ptpClusterSocks_[pnid] = sock; |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - connected to monitor node=%d(%s), sock=%d, " |
| "ptpClusterSocks_[%d]=%d\n" |
| , method_name, __LINE__ |
| , pnid |
| , ptpPort |
| , sock |
| , pnid |
| , ptpClusterSocks_[pnid] ); |
| } |
| } |
| } |
| |
| TRACE_EXIT; |
| return err; |
| } |
| |
| bool CPtpClient::IsTargetRemote( int targetNid ) |
| { |
| const char method_name[] = "CPtpClient::IsTargetRemote"; |
| TRACE_ENTRY; |
| |
| CLNode *targetLNode = Nodes->GetLNode( targetNid ); |
| CNode *targetNode = targetLNode->GetNode(); |
| bool rs = (targetNode && targetNode->GetPNid() == MyPNID) ? false : true ; |
| |
| TRACE_EXIT; |
| return(rs); |
| } |
| |
| int CPtpClient::ProcessAddUniqStr( int nid |
| , int id |
| , const char *stringValue |
| , int targetNid |
| , const char *targetNodeName ) |
| { |
| const char method_name[] = "CPtpClient::ProcessAddUniqStr"; |
| TRACE_ENTRY; |
| |
| if (!IsTargetRemote( targetNid )) |
| { |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Not Sending InternalType_UniqStr request to " |
| "local nid=%d\n" |
| , method_name, __LINE__ |
| , targetNid ); |
| } |
| return(0); |
| } |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Sending InternalType_UniqStr request to %s, " |
| "targetNid=%d\n" |
| , method_name, __LINE__ |
| , targetNodeName |
| , targetNid ); |
| } |
| |
| struct internal_msg_def msg; |
| memset(&msg, 0, sizeof(msg)); |
| msg.type = InternalType_UniqStr; |
| msg.u.uniqstr.nid = nid; |
| msg.u.uniqstr.id = id; |
| |
| char *stringData = & msg.u.uniqstr.valueData; |
| int stringDataLen = strlen(stringValue) + 1; |
| |
| // Copy the string |
| memcpy( stringData, stringValue, stringDataLen ); |
| |
| ptpMsgInfo_t myInfo; |
| myInfo.pnid = MyPNID; |
| myInfo.size = offsetof(struct internal_msg_def, u); |
| myInfo.size += sizeof(msg.u.uniqstr); |
| myInfo.size += stringDataLen; |
| |
| if (trace_settings & TRACE_PROCESS_DETAIL) |
| { |
| trace_printf( "%s@%d - size_=%d, forwarding unique string [%d, %d] (%s)\n" |
| , method_name, __LINE__ |
| , myInfo.size |
| , msg.u.uniqstr.nid |
| , msg.u.uniqstr.id |
| , &msg.u.uniqstr.valueData ); |
| } |
| |
| int error = SendToMon( "process-add-unique-string" |
| , &msg |
| , myInfo |
| , targetNid |
| , targetNodeName); |
| |
| TRACE_EXIT; |
| return error; |
| } |
| |
| int CPtpClient::ProcessClone( CProcess *process ) |
| { |
| const char method_name[] = "CPtpClient::ProcessClone"; |
| TRACE_ENTRY; |
| |
| CLNode *parentLNode = NULL; |
| if (process->GetParentNid() != -1) |
| { |
| parentLNode = Nodes->GetLNode( process->GetParentNid() ); |
| } |
| |
| if (parentLNode == NULL) |
| { |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Not Sending InternalType_Clone request to parentNid=%d" |
| ", process=%s (%d:%d:%d)\n" |
| , method_name, __LINE__ |
| , process->GetParentNid() |
| , process->GetName() |
| , process->GetNid() |
| , process->GetPid() |
| , process->GetVerifier() ); |
| } |
| return(0); |
| } |
| |
| if (!IsTargetRemote( process->GetParentNid() )) |
| { |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Not Sending InternalType_Clone request to " |
| "local nid=%d\n" |
| , method_name, __LINE__ |
| , process->GetParentNid() ); |
| } |
| return(0); |
| } |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Sending InternalType_Clone request to %s, parentNid=%d" |
| ", process=%s (%d:%d:%d)\n" |
| , method_name, __LINE__ |
| , parentLNode->GetNode()->GetName() |
| , process->GetParentNid() |
| , process->GetName() |
| , process->GetNid() |
| , process->GetPid() |
| , process->GetVerifier() ); |
| } |
| |
| struct internal_msg_def msg; |
| memset(&msg, 0, sizeof(msg)); |
| msg.type = InternalType_Clone; |
| msg.u.clone.backup = process->IsBackup(); |
| msg.u.clone.unhooked = process->IsUnhooked(); |
| msg.u.clone.event_messages = process->IsEventMessages(); |
| msg.u.clone.system_messages = process->IsSystemMessages(); |
| msg.u.clone.nid = process->GetNid(); |
| msg.u.clone.verifier = process->GetVerifier(); |
| msg.u.clone.type = process->GetType(); |
| msg.u.clone.priority = process->GetPriority(); |
| msg.u.clone.parent_nid = process->GetParentNid(); |
| msg.u.clone.parent_pid = process->GetParentPid(); |
| msg.u.clone.parent_verifier = process->GetParentVerifier(); |
| msg.u.clone.os_pid = process->GetPid(); |
| msg.u.clone.persistent = process->IsPersistent(); |
| msg.u.clone.persistent_retries = process->GetPersistentRetries(); |
| msg.u.clone.origPNidNs= -1; |
| msg.u.clone.argc = process->argc(); |
| msg.u.clone.creation_time = process->GetCreationTime(); |
| msg.u.clone.pathStrId = process->pathStrId(); |
| msg.u.clone.ldpathStrId = process->ldPathStrId(); |
| msg.u.clone.programStrId = process->programStrId(); |
| |
| msg.u.clone.prior_pid = process->GetPriorPid (); |
| process->SetPriorPid ( 0 ); |
| msg.u.clone.creation_time = process->GetCreationTime(); |
| |
| char *stringData = & msg.u.clone.stringData; |
| int nameLen = strlen(process->GetName()) + 1; |
| int portLen = strlen(process->GetPort()) + 1; |
| int infileLen = strlen(process->infile()) + 1; |
| int outfileLen = strlen(process->outfile()) + 1; |
| int argvLen = process->userArgvLen(); |
| |
| // Copy the process name |
| msg.u.clone.nameLen = nameLen; |
| memcpy( stringData, process->GetName(), nameLen ); |
| stringData += nameLen; |
| |
| // Copy the port |
| msg.u.clone.portLen = portLen; |
| memcpy(stringData, process->GetPort(), portLen ); |
| stringData += portLen; |
| |
| // Copy the standard in file name |
| msg.u.clone.infileLen = infileLen; |
| memcpy( stringData, process->infile(), infileLen ); |
| stringData += infileLen ; |
| |
| // Copy the standard out file name |
| msg.u.clone.outfileLen = outfileLen; |
| memcpy( stringData, process->outfile(), outfileLen ); |
| stringData += outfileLen ; |
| |
| // Copy the program argument strings |
| msg.u.clone.argvLen = argvLen; |
| memcpy( stringData, process->userArgv(), argvLen ); |
| |
| ptpMsgInfo_t myInfo; |
| myInfo.pnid = MyPNID; |
| myInfo.size = offsetof(struct internal_msg_def, u); |
| myInfo.size += sizeof(msg.u.clone); |
| myInfo.size += nameLen ; |
| myInfo.size += portLen ; |
| myInfo.size += infileLen ; |
| myInfo.size += outfileLen ; |
| myInfo.size += argvLen ; |
| |
| if (trace_settings & TRACE_PROCESS_DETAIL) |
| { |
| trace_printf( "%s@%d - size_=%d, programStrId=(%d,%d), " |
| "pathStrId=(%d,%d), ldPathStrId=(%d,%d), " |
| "name=%s, strlen(name)=%d, " |
| "port=%s, strlen(port)=%d, " |
| "infile=%s, strlen(infile)=%d, " |
| "outfile=%s, strlen(outfile)=%d, " |
| "argc=%d, strlen(total argv)=%d, args=[%.*s]\n" |
| , method_name, __LINE__ |
| , myInfo.size |
| , msg.u.clone.programStrId.nid |
| , msg.u.clone.programStrId.id |
| , msg.u.clone.pathStrId.nid |
| , msg.u.clone.pathStrId.id |
| , msg.u.clone.ldpathStrId.nid |
| , msg.u.clone.ldpathStrId.id |
| , &msg.u.clone.stringData |
| , nameLen |
| , &msg.u.clone.stringData+nameLen |
| , portLen |
| , &msg.u.clone.stringData+nameLen |
| , infileLen |
| , &msg.u.clone.stringData+nameLen+infileLen |
| , outfileLen |
| , msg.u.clone.argc |
| , argvLen |
| , argvLen |
| , &msg.u.clone.stringData+nameLen+infileLen+outfileLen); |
| } |
| |
| int error = SendToMon( "process-clone" |
| , &msg |
| , myInfo |
| , process->GetParentNid() |
| , parentLNode->GetNode()->GetName()); |
| |
| TRACE_EXIT; |
| return error; |
| } |
| |
| int CPtpClient::ProcessExit( CProcess *process |
| , int targetNid |
| , const char *targetNodeName ) |
| { |
| const char method_name[] = "CPtpClient::ProcessExit"; |
| TRACE_ENTRY; |
| |
| if (!IsTargetRemote( targetNid )) |
| { |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Not Sending InternalType_Exit request to " |
| "local nid=%d\n" |
| , method_name, __LINE__ |
| , targetNid ); |
| } |
| return(0); |
| } |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Sending InternalType_Exit request to %s, targetNid=%d" |
| ", process=%s (%d,%d:%d) is exiting\n" |
| , method_name, __LINE__ |
| , targetNodeName |
| , targetNid |
| , process->GetName() |
| , process->GetNid() |
| , process->GetPid() |
| , process->GetVerifier() ); |
| } |
| |
| struct internal_msg_def msg; |
| memset(&msg, 0, sizeof(msg)); |
| msg.type = InternalType_Exit; |
| msg.u.exit.nid = process->GetNid(); |
| msg.u.exit.pid = process->GetPid(); |
| msg.u.exit.verifier = process->GetVerifier(); |
| strcpy(msg.u.exit.name, process->GetName()); |
| msg.u.exit.abended = process->IsAbended(); |
| |
| ptpMsgInfo_t myInfo; |
| myInfo.pnid = MyPNID; |
| myInfo.size = offsetof(struct internal_msg_def, u); |
| myInfo.size += sizeof(msg.u.exit); |
| |
| if (trace_settings & TRACE_PROCESS_DETAIL) |
| { |
| trace_printf( "%s@%d - size_=%d, process %s (%d,%d:%d) " |
| "abended=%d\n" |
| , method_name, __LINE__ |
| , myInfo.size |
| , msg.u.exit.name |
| , msg.u.exit.nid |
| , msg.u.exit.pid |
| , msg.u.exit.verifier |
| , msg.u.exit.abended ); |
| } |
| |
| int error = SendToMon( "process-exit" |
| , &msg |
| , myInfo |
| , targetNid |
| , targetNodeName); |
| |
| TRACE_EXIT; |
| return error; |
| } |
| |
| int CPtpClient::ProcessInit( CProcess *process |
| , void *tag |
| , int result |
| , int parentNid ) |
| { |
| const char method_name[] = "CPtpClient::ProcessInit"; |
| TRACE_ENTRY; |
| |
| CLNode *parentLNode = NULL; |
| if (process->GetParentNid() != -1) |
| { |
| parentLNode = Nodes->GetLNode( process->GetParentNid() ); |
| } |
| |
| if (parentLNode == NULL) |
| { |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Not Sending InternalType_ProcessInit request to parentNid=%d" |
| ", process=%s (%d,%d:%d)\n" |
| , method_name, __LINE__ |
| , process->GetParentNid() |
| , process->GetName() |
| , process->GetNid() |
| , process->GetPid() |
| , process->GetVerifier() ); |
| } |
| return(0); |
| } |
| |
| if (!IsTargetRemote( process->GetParentNid() )) |
| { |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Not Sending InternalType_ProcessInit request to " |
| "local nid=%d\n" |
| , method_name, __LINE__ |
| , process->GetParentNid() ); |
| } |
| return(0); |
| } |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d" " - Sending InternalType_ProcessInit to parent node %s, parentNid=%d" |
| ", for process %s (%d,%d:%d), result=%d, tag=%p\n" |
| , method_name, __LINE__ |
| , parentLNode->GetNode()->GetName() |
| , parentNid |
| , process->GetName() |
| , process->GetNid() |
| , process->GetPid() |
| , process->GetVerifier() |
| , result |
| , tag ); |
| } |
| |
| struct internal_msg_def msg; |
| memset(&msg, 0, sizeof(msg)); |
| msg.type = InternalType_ProcessInit; |
| msg.u.processInit.nid = process->GetNid(); |
| msg.u.processInit.pid = process->GetPid(); |
| msg.u.processInit.verifier = process->GetVerifier(); |
| strcpy (msg.u.processInit.name, process->GetName()); |
| msg.u.processInit.state = process->GetState(); |
| msg.u.processInit.result = result; |
| msg.u.processInit.tag = tag; |
| msg.u.processInit.origNid = process->GetParentNid(); |
| |
| ptpMsgInfo_t myInfo; |
| myInfo.pnid = MyPNID; |
| myInfo.size = offsetof(struct internal_msg_def, u); |
| myInfo.size += sizeof(msg.u.processInit); |
| |
| int error = SendToMon( "process-init" |
| , &msg |
| , myInfo |
| , parentNid |
| , parentLNode->GetNode()->GetName() ); |
| |
| TRACE_EXIT; |
| return error; |
| |
| } |
| |
| int CPtpClient::ProcessKill( CProcess *process |
| , bool abort |
| , int targetNid |
| , const char *targetNodeName ) |
| { |
| const char method_name[] = "CPtpClient::ProcessKill"; |
| TRACE_ENTRY; |
| |
| if (!IsTargetRemote( targetNid )) |
| { |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Not Sending InternalType_Kill request to " |
| "local nid=%d\n" |
| , method_name, __LINE__ |
| , targetNid ); |
| } |
| return(0); |
| } |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Sending InternalType_Kill request to %s, targetNid=%d" |
| ", killing process (%d,%d:%d)\n" |
| , method_name, __LINE__ |
| , targetNodeName |
| , targetNid |
| , process->GetNid() |
| , process->GetPid() |
| , process->GetVerifier() ); |
| } |
| |
| struct internal_msg_def msg; |
| memset(&msg, 0, sizeof(msg)); |
| msg.type = InternalType_Kill; |
| msg.u.kill.nid = process->GetNid(); |
| msg.u.kill.pid = process->GetPid(); |
| msg.u.kill.verifier = process->GetVerifier(); |
| msg.u.kill.persistent_abort = abort; |
| |
| ptpMsgInfo_t myInfo; |
| myInfo.pnid = MyPNID; |
| myInfo.size = offsetof(struct internal_msg_def, u); |
| myInfo.size += sizeof(msg.u.exit); |
| |
| if (trace_settings & TRACE_PROCESS_DETAIL) |
| { |
| trace_printf( "%s@%d - size_=%d, process (%d,%d:%d) " |
| "persistent_abort=%d\n" |
| , method_name, __LINE__ |
| , myInfo.size |
| , msg.u.kill.nid |
| , msg.u.kill.pid |
| , msg.u.kill.verifier |
| , msg.u.kill.persistent_abort ); |
| } |
| |
| int error = SendToMon( "process-kill" |
| , &msg |
| , myInfo |
| , targetNid |
| , targetNodeName); |
| |
| TRACE_EXIT; |
| return error; |
| } |
| |
| int CPtpClient::ProcessNew( CProcess *process |
| , int targetNid |
| , const char *targetNodeName ) |
| { |
| const char method_name[] = "CPtpClient::ProcessNew"; |
| TRACE_ENTRY; |
| |
| if (!IsTargetRemote( targetNid )) |
| { |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Not Sending InternalType_Process request to " |
| "local nid=%d\n" |
| , method_name, __LINE__ |
| , targetNid ); |
| } |
| return(0); |
| } |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Sending InternalType_Process request to %s, targetNid=%d" |
| ", program=%s, parent=(%d,%d:%d)\n" |
| , method_name, __LINE__ |
| , targetNodeName |
| , targetNid |
| , process->program() |
| , process->GetParentNid() |
| , process->GetParentPid() |
| , process->GetParentVerifier() ); |
| } |
| |
| struct internal_msg_def msg; |
| memset(&msg, 0, sizeof(msg)); |
| msg.type = InternalType_Process; |
| msg.u.process.nid = process->GetNid(); |
| msg.u.process.pid = process->GetPid(); |
| msg.u.process.type = process->GetType(); |
| msg.u.process.priority = process->GetPriority(); |
| msg.u.process.backup = process->IsBackup(); |
| msg.u.process.unhooked = process->IsUnhooked(); |
| msg.u.process.tag = process; |
| msg.u.process.parent_nid = process->GetParentNid(); |
| msg.u.process.parent_pid = process->GetParentPid(); |
| msg.u.process.parent_verifier = process->GetParentVerifier(); |
| msg.u.process.pair_parent_nid = process->GetPairParentNid(); |
| msg.u.process.pair_parent_pid = process->GetPairParentPid(); |
| msg.u.process.pair_parent_verifier = process->GetPairParentVerifier(); |
| msg.u.process.pathStrId = process->pathStrId(); |
| msg.u.process.ldpathStrId = process->ldPathStrId(); |
| msg.u.process.programStrId = process->programStrId(); |
| msg.u.process.argc = process->argc(); |
| |
| char *stringData = & msg.u.process.stringData; |
| int nameLen = strlen(process->GetName()) + 1; |
| int infileLen = strlen(process->infile()) + 1; |
| int outfileLen = strlen(process->outfile()) + 1; |
| int argvLen = process->userArgvLen(); |
| |
| // Copy the process name |
| msg.u.process.nameLen = nameLen; |
| memcpy( stringData, process->GetName(), nameLen ); |
| stringData += nameLen; |
| |
| // Copy the standard in file name |
| msg.u.process.infileLen = infileLen; |
| memcpy( stringData, process->infile(), infileLen ); |
| stringData += infileLen; |
| |
| // Copy the standard out file name |
| msg.u.process.outfileLen = outfileLen; |
| memcpy( stringData, process->outfile(), outfileLen ); |
| stringData += outfileLen; |
| |
| // Copy the program argument strings |
| msg.u.process.argvLen = argvLen; |
| memcpy( stringData, process->userArgv(), argvLen ); |
| |
| ptpMsgInfo_t myInfo; |
| myInfo.pnid = MyPNID; |
| myInfo.size = offsetof(struct internal_msg_def, u); |
| myInfo.size += sizeof(msg.u.process); |
| myInfo.size += nameLen ; |
| myInfo.size += infileLen ; |
| myInfo.size += outfileLen ; |
| myInfo.size += argvLen ; |
| |
| if (trace_settings & TRACE_PROCESS_DETAIL) |
| { |
| trace_printf( "%s@%d - size_=%d, programStrId=(%d,%d), " |
| "pathStrId=(%d,%d), ldPathStrId=(%d,%d), " |
| "name=%s, strlen(name)=%d, " |
| "infile=%s, strlen(infile)=%d, " |
| "outfile=%s, strlen(outfile)=%d, " |
| "argc=%d, strlen(total argv)=%d, args=[%.*s]\n" |
| , method_name, __LINE__ |
| , myInfo.size |
| , msg.u.process.programStrId.nid |
| , msg.u.process.programStrId.id |
| , msg.u.process.pathStrId.nid |
| , msg.u.process.pathStrId.id |
| , msg.u.process.ldpathStrId.nid |
| , msg.u.process.ldpathStrId.id |
| , &msg.u.process.stringData |
| , nameLen |
| , &msg.u.process.stringData+nameLen |
| , infileLen |
| , &msg.u.process.stringData+nameLen+infileLen |
| , outfileLen |
| , msg.u.process.argc |
| , argvLen |
| , argvLen |
| , &msg.u.process.stringData+nameLen+infileLen+outfileLen); |
| } |
| |
| int error = SendToMon( "process-new" |
| , &msg |
| , myInfo |
| , targetNid |
| , targetNodeName); |
| |
| TRACE_EXIT; |
| return error; |
| } |
| |
| int CPtpClient::ProcessNotify( int nid |
| , int pid |
| , Verifier_t verifier |
| , _TM_Txid_External transId |
| , bool canceled |
| , CProcess *targetProcess |
| , int targetNid |
| , const char *targetNodeName ) |
| { |
| const char method_name[] = "CPtpClient::ProcessNotify"; |
| TRACE_ENTRY; |
| |
| if (!IsTargetRemote( targetNid )) |
| { |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Not Sending InternalType_Notify request to " |
| "local nid=%d\n" |
| , method_name, __LINE__ |
| , targetNid ); |
| } |
| return(0); |
| } |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Sending InternalType_Notify request to %s" |
| ", nid=%d, canceled=%d\n" |
| , method_name, __LINE__ |
| , targetNodeName |
| , targetNid |
| , canceled ); |
| } |
| |
| struct internal_msg_def msg; |
| memset(&msg, 0, sizeof(msg)); |
| msg.type = InternalType_Notify; |
| msg.u.notify.nid = nid; |
| msg.u.notify.pid = pid; |
| msg.u.notify.verifier = verifier; |
| msg.u.notify.canceled = canceled; |
| msg.u.notify.target_nid = targetProcess->GetNid(); |
| msg.u.notify.target_pid = targetProcess->GetPid(); |
| msg.u.notify.target_verifier = targetProcess->GetVerifier(); |
| msg.u.notify.trans_id = transId; |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| if (canceled) |
| { |
| trace_printf( "%s@%d - Process (%d, %d:%d) deleting death " |
| "notice interest for %s (%d, %d:%d), " |
| "trans_id=%lld.%lld.%lld.%lld\n" |
| , method_name, __LINE__ |
| , msg.u.notify.nid |
| , msg.u.notify.pid |
| , msg.u.notify.verifier |
| , targetProcess->GetName() |
| , msg.u.notify.target_nid |
| , msg.u.notify.target_pid |
| , msg.u.notify.target_verifier |
| , msg.u.notify.trans_id.txid[0] |
| , msg.u.notify.trans_id.txid[1] |
| , msg.u.notify.trans_id.txid[2] |
| , msg.u.notify.trans_id.txid[3] ); |
| } |
| else |
| { |
| trace_printf("%s@%d - Process (%d, %d:%d) registering interest " |
| "in death of process %s (%d, %d:%d), " |
| "trans_id=%lld.%lld.%lld.%lld\n" |
| , method_name, __LINE__ |
| , msg.u.notify.nid |
| , msg.u.notify.pid |
| , msg.u.notify.verifier |
| , targetProcess->GetName() |
| , msg.u.notify.target_nid |
| , msg.u.notify.target_pid |
| , msg.u.notify.target_verifier |
| , msg.u.notify.trans_id.txid[0] |
| , msg.u.notify.trans_id.txid[1] |
| , msg.u.notify.trans_id.txid[2] |
| , msg.u.notify.trans_id.txid[3] ); |
| } |
| } |
| |
| ptpMsgInfo_t myInfo; |
| myInfo.pnid = MyPNID; |
| myInfo.size = offsetof(struct internal_msg_def, u); |
| myInfo.size += sizeof(msg.u.notify); |
| |
| int error = SendToMon( "process-notify" |
| , &msg |
| , myInfo |
| , targetNid |
| , targetNodeName); |
| |
| TRACE_EXIT; |
| return error; |
| } |
| |
| int CPtpClient::ProcessStdInReq( int nid |
| , int pid |
| , StdinReqType type |
| , int supplierNid |
| , int supplierPid ) |
| { |
| const char method_name[] = "CPtpClient::ProcessStdInReq"; |
| TRACE_ENTRY; |
| |
| if (trace_settings & (TRACE_REDIRECTION | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Sending InternalType_StdinReq request type =%d " |
| "from (%d,%d), for supplier (%d,%d)\n" |
| , method_name, __LINE__ |
| , type |
| , nid |
| , pid |
| , supplierNid |
| , supplierPid ); |
| } |
| |
| CLNode *lnode = Nodes->GetLNode( supplierNid ); |
| if (lnode == NULL) |
| { |
| char buf[MON_STRING_BUF_SIZE]; |
| snprintf( buf, sizeof(buf) |
| , "[%s], Can't find supplier node nid=%d " |
| "for stdin data request.\n" |
| , method_name |
| , supplierNid ); |
| mon_log_write(PTPCLIENT_STDINREQ_1, SQ_LOG_ERR, buf); |
| |
| TRACE_EXIT; |
| return -1; |
| } |
| |
| CProcess *process = lnode->GetProcessL( supplierPid ); |
| if (process == NULL) |
| { |
| char buf[MON_STRING_BUF_SIZE]; |
| snprintf( buf, sizeof(buf) |
| , "[%s], Can't find process nid=%d, " |
| "pid=%d for stdin data request.\n" |
| , method_name |
| , supplierNid |
| , supplierPid ); |
| mon_log_write(PTPCLIENT_STDINREQ_2, SQ_LOG_ERR, buf); |
| |
| TRACE_EXIT; |
| return -1; |
| } |
| |
| struct internal_msg_def msg; |
| memset(&msg, 0, sizeof(msg)); |
| msg.type = InternalType_StdinReq; |
| msg.u.stdin_req.nid = nid; |
| msg.u.stdin_req.pid = pid; |
| msg.u.stdin_req.reqType = type; |
| msg.u.stdin_req.supplier_nid = supplierNid; |
| msg.u.stdin_req.supplier_pid = supplierPid; |
| |
| ptpMsgInfo_t myInfo; |
| myInfo.pnid = MyPNID; |
| myInfo.size = offsetof(struct internal_msg_def, u); |
| myInfo.size += sizeof(msg.u.stdin_req); |
| |
| if (trace_settings & (TRACE_REDIRECTION | TRACE_PROCESS_DETAIL)) |
| { |
| trace_printf( "%s@%d - size_=%d, type =%d " |
| "from (%d,%d), for supplier (%d,%d)\n" |
| , method_name, __LINE__ |
| , myInfo.size |
| , msg.u.stdin_req.reqType |
| , msg.u.stdin_req.nid |
| , msg.u.stdin_req.pid |
| , msg.u.stdin_req.supplier_nid |
| , msg.u.stdin_req.supplier_pid ); |
| } |
| |
| int error = SendToMon( "process-stdin" |
| , &msg |
| , myInfo |
| , process->GetNid() |
| , lnode->GetNode()->GetName()); |
| |
| TRACE_EXIT; |
| return error; |
| } |
| |
| int CPtpClient::ProcessStdIoData( int nid |
| , int pid |
| , StdIoType type |
| , ssize_t count |
| , char *data ) |
| { |
| const char method_name[] = "CPtpClient::ProcessStdIoData"; |
| TRACE_ENTRY; |
| |
| if (trace_settings & (TRACE_REDIRECTION | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - Sending InternalType_IoData request type =%d " |
| "to (%d,%d), count=%ld\n" |
| , method_name, __LINE__ |
| , type |
| , nid |
| , pid |
| , count ); |
| } |
| |
| CLNode *lnode = Nodes->GetLNode( nid ); |
| if (lnode == NULL) |
| { |
| char buf[MON_STRING_BUF_SIZE]; |
| snprintf( buf, sizeof(buf) |
| , "[%s], Can't find supplier node nid=%d " |
| "for stdin data request.\n" |
| , method_name |
| , nid ); |
| mon_log_write(PTPCLIENT_STDIODATA_1, SQ_LOG_ERR, buf); |
| |
| TRACE_EXIT; |
| return -1; |
| } |
| |
| CProcess *process = lnode->GetProcessL( pid ); |
| if (process == NULL) |
| { |
| char buf[MON_STRING_BUF_SIZE]; |
| snprintf( buf, sizeof(buf) |
| , "[%s], Can't find process nid=%d, " |
| "pid=%d for stdin data request.\n" |
| , method_name |
| , nid |
| , pid ); |
| mon_log_write(PTPCLIENT_STDIODATA_2, SQ_LOG_ERR, buf); |
| |
| TRACE_EXIT; |
| return -1; |
| } |
| |
| struct internal_msg_def msg; |
| memset(&msg, 0, sizeof(msg)); |
| msg.type = InternalType_IoData; |
| msg.u.iodata.nid = nid ; |
| msg.u.iodata.pid = pid ; |
| msg.u.iodata.ioType = type ; |
| msg.u.iodata.length = count; |
| memcpy(&msg.u.iodata.data, data, count); |
| |
| ptpMsgInfo_t myInfo; |
| myInfo.pnid = MyPNID; |
| myInfo.size = offsetof(struct internal_msg_def, u); |
| myInfo.size += sizeof(msg.u.iodata); |
| |
| if (trace_settings & (TRACE_REDIRECTION | TRACE_PROCESS_DETAIL)) |
| { |
| trace_printf( "%s@%d - size_=%d, type =%d " |
| "to (%d,%d), count=%d\n(%s)" |
| , method_name, __LINE__ |
| , myInfo.size |
| , msg.u.iodata.ioType |
| , msg.u.iodata.nid |
| , msg.u.iodata.pid |
| , msg.u.iodata.length |
| , msg.u.iodata.length?msg.u.iodata.data:"\n" ); |
| } |
| |
| int error = SendToMon( "process-stdio-data" |
| , &msg |
| , myInfo |
| , process->GetNid() |
| , lnode->GetNode()->GetName()); |
| |
| TRACE_EXIT; |
| return error; |
| } |
| |
| int CPtpClient::SendToMon(const char *reqType, internal_msg_def *msg |
| , ptpMsgInfo_t &myInfo |
| , int targetNid, const char *hostName) |
| { |
| const char method_name[] = "CPtpClient::SendToMon"; |
| TRACE_ENTRY; |
| |
| char ptpHost[MAX_PROCESSOR_NAME]; |
| char ptpPort[MAX_PROCESSOR_NAME]; |
| int error = 0; |
| int tempPort = ptpCommPort_; |
| int pnid = 0; |
| int sendSock = -1; |
| int retryCount = 0; |
| CNode *node = NULL; |
| CLNode *lnode = NULL; |
| |
| ptpHost[0] = '\0'; |
| lnode = Nodes->GetLNode( targetNid ); |
| node = lnode->GetNode(); |
| pnid = node->GetPNid(); |
| |
| // For virtual env |
| if (!IsRealCluster) |
| { |
| tempPort += targetNid; |
| strcat( ptpHost, ptpHost_ ); |
| } |
| else |
| { |
| strcat( ptpHost, hostName ); |
| } |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - reqType=%s, hostName=%s, targetNid=%d, " |
| "ptpHost=%s, tempPort=%d, ptpCommPort_=%d\n" |
| , method_name, __LINE__ |
| , reqType |
| , hostName |
| , targetNid |
| , ptpHost |
| , tempPort |
| , ptpCommPort_ ); |
| } |
| |
| memset( &ptpPort, 0, MAX_PROCESSOR_NAME ); |
| memset( &ptpPortBase_, 0, MAX_PROCESSOR_NAME+100 ); |
| sprintf( ptpPortBase_,"%s:", ptpHost ); |
| sprintf( ptpPort,"%s%d", ptpPortBase_, tempPort ); |
| |
| retryIO: |
| |
| if (ptpClusterSocks_[pnid] == -1) |
| { |
| error = InitializePtpClient( pnid, ptpPort ); |
| if (error < 0) |
| { |
| ptpClusterSocks_[pnid] = -1; |
| TRACE_EXIT; |
| return error; |
| } |
| } |
| |
| sendSock = ptpClusterSocks_[pnid]; |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - sending %s REQ to Monitor=%s, sock=%d\n" |
| , method_name, __LINE__ |
| , reqType |
| , ptpPort |
| , sendSock ); |
| } |
| |
| error = SockSend((char *) &myInfo, sizeof(ptpMsgInfo_t), sendSock); |
| if (error) |
| { |
| int err = error; |
| char buf[MON_STRING_BUF_SIZE]; |
| snprintf( buf, sizeof(buf) |
| , "[%s], unable to send %s request size %ld to " |
| "node %s, error: %d(%s)\n" |
| , method_name, reqType, sizeof(ptpMsgInfo_t), ptpHost, err, strerror(err) ); |
| mon_log_write(PTPCLIENT_SENDTOMON_1, SQ_LOG_ERR, buf); |
| } |
| else |
| { |
| error = SockSend((char *) msg, myInfo.size, sendSock); |
| if (error) |
| { |
| int err = error; |
| char buf[MON_STRING_BUF_SIZE]; |
| snprintf( buf, sizeof(buf) |
| , "[%s], unable to send %s request to " |
| "node %s, error: %d(%s)\n" |
| , method_name, reqType, ptpHost, err, strerror(err) ); |
| mon_log_write(PTPCLIENT_SENDTOMON_2, SQ_LOG_ERR, buf); |
| } |
| } |
| |
| if (error) |
| { |
| SockClose( pnid ); |
| if ( retryCount < MON2MON_IO_RETRIES ) |
| { |
| retryCount++; |
| if (trace_settings & (TRACE_REQUEST | TRACE_PROCESS)) |
| { |
| trace_printf( "%s@%d - retrying IO (%d) to node %s\n" |
| , method_name, __LINE__ |
| , retryCount |
| , ptpHost ); |
| } |
| goto retryIO; |
| } |
| } |
| |
| TRACE_EXIT; |
| return error; |
| } |
| |
| void CPtpClient::SockClose( int pnid ) |
| { |
| const char method_name[] = "CPtpClient::SockClose"; |
| TRACE_ENTRY; |
| |
| if (ptpClusterSocks_[pnid] != -1) |
| { |
| close( ptpClusterSocks_[pnid] ); |
| ptpClusterSocks_[pnid] = -1; |
| } |
| |
| TRACE_EXIT; |
| } |
| |
| void CPtpClient::SetLocalHost( void ) |
| { |
| gethostname( ptpHost_, MAX_PROCESSOR_NAME ); |
| } |
| |
| int CPtpClient::SockReceive(char *buf, int size, int sockFd) |
| { |
| const char method_name[] = "CPtpClient::SockReceive"; |
| TRACE_ENTRY; |
| |
| bool readAgain = false; |
| int error = 0; |
| int readCount = 0; |
| int received = 0; |
| int sizeCount = size; |
| |
| do |
| { |
| readCount = (int) recv( sockFd |
| , buf |
| , sizeCount |
| , 0 ); |
| if ( readCount > 0 ) Meas.addSockPtpRcvdBytes( readCount ); |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_INIT | TRACE_RECOVERY)) |
| { |
| trace_printf( "%s@%d - Count read %d = recv(%d)\n" |
| , method_name, __LINE__ |
| , readCount |
| , sizeCount ); |
| } |
| |
| if ( readCount > 0 ) |
| { // Got data |
| received += readCount; |
| buf += readCount; |
| if ( received == size ) |
| { |
| readAgain = false; |
| } |
| else |
| { |
| sizeCount -= readCount; |
| readAgain = true; |
| } |
| } |
| else if ( readCount == 0 ) |
| { // EOF |
| error = ENODATA; |
| readAgain = false; |
| } |
| else |
| { // Got an error |
| if ( errno != EINTR) |
| { |
| error = errno; |
| readAgain = false; |
| } |
| else |
| { |
| readAgain = true; |
| } |
| } |
| } |
| while( readAgain ); |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_INIT | TRACE_RECOVERY)) |
| { |
| trace_printf( "%s@%d - recv(), received=%d, error=%d(%s)\n" |
| , method_name, __LINE__ |
| , received |
| , error, strerror(error) ); |
| } |
| |
| TRACE_EXIT; |
| return error; |
| } |
| |
| int CPtpClient::SockSend(char *buf, int size, int sockFd) |
| { |
| const char method_name[] = "CPtpClient::SockSend"; |
| TRACE_ENTRY; |
| |
| bool sendAgain = false; |
| int error = 0; |
| int sendCount = 0; |
| int sent = 0; |
| |
| do |
| { |
| sendCount = (int) send( sockFd |
| , buf |
| , size |
| , 0 ); |
| if ( sendCount > 0 ) Meas.addSockPtpSentBytes( sendCount ); |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_INIT | TRACE_RECOVERY)) |
| { |
| trace_printf( "%s@%d - send(), sendCount=%d\n" |
| , method_name, __LINE__ |
| , sendCount ); |
| } |
| |
| if ( sendCount > 0 ) |
| { // Sent data |
| sent += sendCount; |
| if ( sendCount == size ) |
| { |
| sendAgain = false; |
| } |
| else |
| { |
| sendAgain = true; |
| } |
| } |
| else |
| { // Got an error |
| if ( errno != EINTR) |
| { |
| error = errno; |
| sendAgain = false; |
| } |
| else |
| { |
| sendAgain = true; |
| } |
| } |
| } |
| while( sendAgain ); |
| |
| if (trace_settings & (TRACE_REQUEST | TRACE_INIT | TRACE_RECOVERY)) |
| { |
| trace_printf( "%s@%d - send(), sent=%d, error=%d(%s)\n" |
| , method_name, __LINE__ |
| , sent |
| , error, strerror(error) ); |
| } |
| |
| TRACE_EXIT; |
| return error; |
| } |
| |
