christoph/fail
1
Fork 0
Code Releases Activity

Bugfix for server-client communication

Browse Source 
git-svn-id: https://www4.informatik.uni-erlangen.de/i4svn/danceos/trunk/devel/fail@1965 8c4709b5-6ec9-48aa-a5cd-a96041d1645a
This commit is contained in:
hellwig
2012-11-30 18:13:13 +00:00
parent c689c159fb
commit fc1d21fe53
2 changed files with 45 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
src/core/cpn/JobServer.cc
View File

@ -247,40 +247,45 @@ void CommThread::sendPendingExperimentData(Minion& minion)
ctrlmsg.set_build_id(42); ctrlmsg.set_build_id(42);
ctrlmsg.set_run_id(m_js.m_runid); ctrlmsg.set_run_id(m_js.m_runid);
ctrlmsg.set_command(FailControlMessage::WORK_FOLLOWS); ctrlmsg.set_command(FailControlMessage::WORK_FOLLOWS);
for(i = 0; i < m_job_size ; i++) { for(i = 0; i < m_job_size ; i++) {
if (m_js.m_undoneJobs.Dequeue_nb(temp_exp) == true) { if (m_js.m_undoneJobs.Dequeue_nb(temp_exp) == true) {
// Got an element from queue, assign ID to workload and send to minion // Got an element from queue, assign ID to workload and send to minion
workloadID = m_js.m_counter.increment(); // increment workload counter workloadID = m_js.m_counter.increment(); // increment workload counter
temp_exp->setWorkloadID(workloadID); // store ID for identification when receiving result temp_exp->setWorkloadID(workloadID); // store ID for identification when receiving result
ctrlmsg.add_workloadid(workloadID); ctrlmsg.add_workloadid(workloadID);
exp.push_back(temp_exp); exp.push_back(temp_exp);
}
if (!m_js.m_runningJobs.insert(workloadID, temp_exp)) {
cout << "!![Server]could not insert workload id: [" << workloadID << "] double entry?" << endl;
}
}
ctrlmsg.set_job_size(exp.size());
cout << " >>[";
for ( i = 0; i < exp.size() ; i++) {
cout << " "<< ctrlmsg.workloadid(i) <<" ";
}
cout << "] " << flush;
if (SocketComm::sendMsg(minion.getSocketDescriptor(), ctrlmsg)) {
for (i = 0; i < ctrlmsg.job_size() ; i++) {
if(SocketComm::sendMsg(minion.getSocketDescriptor(), exp.front()->getMessage())) {
exp.erase(exp.begin());
} else { } else {
break; break;
} }
if (!m_js.m_runningJobs.insert(workloadID, temp_exp)) {
cout << "!![Server]could not insert workload id: [" << workloadID << "] double entry?" << endl;
sleep(10);
}
}
if (exp.size() != 0) {
ctrlmsg.set_job_size(exp.size());
cout << " >>[";
for ( i = 0; i < exp.size() ; i++) {
cout << " "<< ctrlmsg.workloadid(i) <<" ";
}
cout << "] " << flush;
if (SocketComm::sendMsg(minion.getSocketDescriptor(), ctrlmsg)) {
for (i = 0; i < ctrlmsg.job_size() ; i++) {
if(SocketComm::sendMsg(minion.getSocketDescriptor(), exp.front()->getMessage())) {
exp.erase(exp.begin());
} else {
break;
}
}
return;
}
} }
return;
}
#ifndef __puma #ifndef __puma
// Prevent receiveExperimentResults from modifying (or indirectly, via // Prevent receiveExperimentResults from modifying (or indirectly, via

20
src/core/efw/JobClient.cc
View File

@ -148,15 +148,17 @@ FailControlMessage_Command JobClient::tryToGetExperimentData(ExperimentData& exp
} }
close(m_sockfd); close(m_sockfd);
//Take front from m_parameters and copy to exp. if (m_parameters.size() != 0) {
exp.getMessage().CopyFrom(m_parameters.front()->getMessage()); //Take front from m_parameters and copy to exp.
exp.setWorkloadID(m_parameters.front()->getWorkloadID()); exp.getMessage().CopyFrom(m_parameters.front()->getMessage());
//Delete front element of m_parameters exp.setWorkloadID(m_parameters.front()->getWorkloadID());
delete &m_parameters.front()->getMessage(); //Delete front element of m_parameters
delete m_parameters.front(); delete &m_parameters.front()->getMessage();
m_parameters.erase(m_parameters.begin()); delete m_parameters.front();
//start time measurement for throughput calculation m_parameters.erase(m_parameters.begin());
m_job_runtime.startTimer(); //start time measurement for throughput calculation
m_job_runtime.startTimer();
}
return ctrlmsg.command(); return ctrlmsg.command();
} }