#105: more on htcondor pilot support

wrench-project · Jun 17, 2019 · eabe556 · eabe556
1 parent be0ec50
commit eabe556
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 8 deletions.
diff --git a/src/wrench/services/compute/htcondor/HTCondorComputeService.cpp b/src/wrench/services/compute/htcondor/HTCondorComputeService.cpp
@@ -83,7 +83,7 @@ namespace wrench {
                 new ComputeServiceSubmitStandardJobRequestMessage(
                         answer_mailbox, job, service_specific_args,
                         this->getMessagePayloadValue(
-                                HTCondorServiceMessagePayload::SUBMIT_STANDARD_JOB_REQUEST_MESSAGE_PAYLOAD)));
+                                HTCondorComputeServiceMessagePayload::SUBMIT_STANDARD_JOB_REQUEST_MESSAGE_PAYLOAD)));
       } catch (std::shared_ptr<NetworkError> &cause) {
         throw WorkflowExecutionException(cause);
       }
@@ -128,7 +128,7 @@ namespace wrench {
                 new ComputeServiceSubmitPilotJobRequestMessage(
                         answer_mailbox, job, service_specific_args,
                         this->getMessagePayloadValue(
-                                HTCondorServiceMessagePayload::SUBMIT_PILOT_JOB_REQUEST_MESSAGE_PAYLOAD)));
+                                HTCondorComputeServiceMessagePayload::SUBMIT_PILOT_JOB_REQUEST_MESSAGE_PAYLOAD)));
       } catch (std::shared_ptr<NetworkError> &cause) {
         throw WorkflowExecutionException(cause);
       }
@@ -243,7 +243,7 @@ namespace wrench {
         try {
           S4U_Mailbox::putMessage(msg->ack_mailbox,
                                   new ServiceDaemonStoppedMessage(this->getMessagePayloadValue(
-                                          HTCondorServiceMessagePayload::DAEMON_STOPPED_MESSAGE_PAYLOAD)));
+                                          HTCondorComputeServiceMessagePayload::DAEMON_STOPPED_MESSAGE_PAYLOAD)));
         } catch (std::shared_ptr<NetworkError> &cause) {
           return false;
         }
@@ -283,7 +283,7 @@ namespace wrench {
                           job, this->getSharedPtr<HTCondorComputeService>(), false, std::shared_ptr<FailureCause>(
                                   new JobTypeNotSupported(job, this->getSharedPtr<HTCondorComputeService>())),
                           this->getMessagePayloadValue(
-                                  HTCondorServiceMessagePayload::SUBMIT_STANDARD_JOB_ANSWER_MESSAGE_PAYLOAD)));
+                                  HTCondorComputeServiceMessagePayload::SUBMIT_STANDARD_JOB_ANSWER_MESSAGE_PAYLOAD)));
         } catch (std::shared_ptr<NetworkError> &cause) {
           return;
         }
@@ -298,7 +298,7 @@ namespace wrench {
                 answer_mailbox,
                 new ComputeServiceSubmitStandardJobAnswerMessage(
                         job, this->getSharedPtr<HTCondorComputeService>(), true, nullptr, this->getMessagePayloadValue(
-                                HTCondorServiceMessagePayload::SUBMIT_STANDARD_JOB_ANSWER_MESSAGE_PAYLOAD)));
+                                HTCondorComputeServiceMessagePayload::SUBMIT_STANDARD_JOB_ANSWER_MESSAGE_PAYLOAD)));
         return;
       } catch (std::shared_ptr<NetworkError> &cause) {
         return;
@@ -326,7 +326,7 @@ namespace wrench {
                           job, this->getSharedPtr<HTCondorComputeService>(), false, std::shared_ptr<FailureCause>(
                                   new JobTypeNotSupported(job, this->getSharedPtr<HTCondorComputeService>())),
                           this->getMessagePayloadValue(
-                                  HTCondorServiceMessagePayload::SUBMIT_PILOT_JOB_ANSWER_MESSAGE_PAYLOAD)));
+                                  HTCondorComputeServiceMessagePayload::SUBMIT_PILOT_JOB_ANSWER_MESSAGE_PAYLOAD)));
         } catch (std::shared_ptr<NetworkError> &cause) {
           return;
         }
@@ -341,7 +341,7 @@ namespace wrench {
                 answer_mailbox,
                 new ComputeServiceSubmitPilotJobAnswerMessage(
                         job, this->getSharedPtr<HTCondorComputeService>(), true, nullptr, this->getMessagePayloadValue(
-                                HTCondorServiceMessagePayload::SUBMIT_PILOT_JOB_ANSWER_MESSAGE_PAYLOAD)));
+                                HTCondorComputeServiceMessagePayload::SUBMIT_PILOT_JOB_ANSWER_MESSAGE_PAYLOAD)));
         return;
       } catch (std::shared_ptr<NetworkError> &cause) {
         return;

diff --git a/src/wrench/services/compute/htcondor/HTCondorNegotiatorService.cpp b/src/wrench/services/compute/htcondor/HTCondorNegotiatorService.cpp
@@ -103,7 +103,12 @@ namespace wrench {
         } else if (auto pilot_job = dynamic_cast<PilotJob *>(job)) {
           // TODO: handle pilot job
           for (auto &item : *this->compute_resources) {
-
+            pilot_job->pushCallbackMailbox(this->reply_mailbox);
+            item.first->submitPilotJob(pilot_job, specific_args);
+            this->running_jobs->insert(std::make_pair(job, item.first));
+            scheduled_jobs.push_back(job);
+            WRENCH_INFO("Dispatched pilot job %s", pilot_job->getName().c_str());
+            break;
           }
         }
       }