Skip to content

Commit 84c478d

Browse files
henricasanovahenricasanova
committed
Test fixes
More mailbox removal
1 parent bb25963 commit 84c478d

File tree

34 files changed

+186
-205
lines changed

34 files changed

+186
-205
lines changed

‎include/wrench/failure_causes/NetworkError.h

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,12 +54,12 @@ namespace wrench {
5454
bool whileReceiving();
5555
bool whileSending();
5656
bool isTimeout();
57-
std::string getMailbox();
57+
std::string getCommPortName();
5858

5959
private:
6060
NetworkError::OperationType operation_type;
6161
NetworkError::ErrorType error_type;
62-
std::string commport_name = "";
62+
std::string commport_name;
6363
};
6464

6565

‎include/wrench/managers/job_manager/JobManager.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -154,7 +154,7 @@ namespace wrench {
154154

155155
void processPilotJobFailure(const std::shared_ptr<PilotJob> &job, std::shared_ptr<ComputeService> compute_service, std::shared_ptr<FailureCause> cause);
156156

157-
// Mailbox of the creator of this job manager
157+
// CommPort of the creator of this job manager
158158
S4U_CommPort *creator_commport;
159159

160160
std::vector<std::shared_ptr<CompoundJob>> jobs_to_dispatch;

‎include/wrench/services/compute/batch/BatschedNetworkListener.h

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ namespace wrench {
3030

3131
public:
3232
BatschedNetworkListener(std::string hostname, std::shared_ptr<BatchComputeService> batch_service,
33-
S4U_Mailbox *batch_service_commport, std::string sched_port,
33+
S4U_CommPort *batch_service_commport, std::string sched_port,
3434
std::string data_to_send, WRENCH_PROPERTY_COLLECTION_TYPE property_list = {});
3535

3636
private:
@@ -50,7 +50,7 @@ namespace wrench {
5050
std::shared_ptr<BatchComputeService> batch_service;
5151
S4U_Commport *batch_service_commport;
5252

53-
void sendExecuteMessageToBatchComputeService(simgrid::s4u::Mailbox *answer_commport, std::string execute_job_reply_data);
53+
void sendExecuteMessageToBatchComputeService(S4U_CommPort *answer_commport, std::string execute_job_reply_data);
5454
void sendQueryAnswerMessageToBatchComputeService(double estimated_waiting_time);
5555
void send_receive();
5656
};

‎include/wrench/services/storage/StorageServiceMessage.h

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ namespace wrench {
4343
public:
4444
StorageServiceFreeSpaceRequestMessage(S4U_CommPort *answer_commport, const std::string &path, double payload);
4545

46-
/** @brief Mailbox to which the answer message should be sent */
46+
/** @brief CommPort to which the answer message should be sent */
4747
S4U_CommPort *answer_commport;
4848
/** @brief The path */
4949
std::string path;
@@ -69,7 +69,7 @@ namespace wrench {
6969
const std::shared_ptr<FileLocation> &location,
7070
double payload);
7171

72-
/** @brief Mailbox to which the answer message should be sent */
72+
/** @brief CommPort to which the answer message should be sent */
7373
S4U_CommPort *answer_commport;
7474
/** @brief The location to lookup */
7575
std::shared_ptr<FileLocation> location;
@@ -98,7 +98,7 @@ namespace wrench {
9898
const std::shared_ptr<FileLocation> &location,
9999
double payload);
100100

101-
/** @brief Mailbox to which the answer message should be sent */
101+
/** @brief CommPort to which the answer message should be sent */
102102
S4U_CommPort *answer_commport;
103103
/** @brief The location to delete */
104104
std::shared_ptr<FileLocation> location;
@@ -135,7 +135,7 @@ namespace wrench {
135135
std::shared_ptr<FileLocation> dst,
136136
double payload);
137137

138-
/** @brief Mailbox to which the answer message should be sent */
138+
/** @brief CommPort to which the answer message should be sent */
139139
S4U_CommPort *answer_commport;
140140
/** @brief The source location */
141141
std::shared_ptr<FileLocation> src;
@@ -175,7 +175,7 @@ namespace wrench {
175175
double num_bytes_to_write,
176176
double payload);
177177

178-
/** @brief Mailbox to which the answer message should be sent */
178+
/** @brief CommPort to which the answer message should be sent */
179179
S4U_CommPort *answer_commport;
180180
/** @brief The requesting host */
181181
simgrid::s4u::Host *requesting_host;

‎include/wrench/services/storage/compound/CompoundStorageServiceMessage.h

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ namespace wrench {
3737
public:
3838
CompoundStorageAllocationRequestMessage(S4U_CommPort *answer_commport, std::shared_ptr<DataFile> file, double payload);
3939

40-
/** @brief Mailbox to which the answer message should be sent */
40+
/** @brief CommPort to which the answer message should be sent */
4141
S4U_CommPort *answer_commport;
4242
/** @brief The path */
4343
std::shared_ptr<DataFile> file;
@@ -61,7 +61,7 @@ namespace wrench {
6161
public:
6262
CompoundStorageLookupRequestMessage(S4U_CommPort *answer_commport, std::shared_ptr<DataFile> file, double payload);
6363

64-
/** @brief Mailbox to which the answer message should be sent */
64+
/** @brief CommPort to which the answer message should be sent */
6565
S4U_CommPort *answer_commport;
6666
/** @brief The path */
6767
std::shared_ptr<DataFile> file;

‎include/wrench/services/storage/xrootd/XRootDMessage.h

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ namespace wrench {
5151
std::shared_ptr<bool> answered,
5252
int timeToLive);
5353
ContinueSearchMessage(ContinueSearchMessage *toCopy);
54-
/** @brief Mailbox to which the FINAL answer message should be sent */
54+
/** @brief CommPort to which the FINAL answer message should be sent */
5555
S4U_CommPort *answer_commport;
5656

5757
/** @brief The original file read request that kicked off the search (if null this was a lookup request)*/
@@ -78,7 +78,7 @@ namespace wrench {
7878
double payload, std::shared_ptr<bool> answered);
7979
UpdateCacheMessage(UpdateCacheMessage &other);
8080
UpdateCacheMessage(UpdateCacheMessage *other);
81-
/** @brief Mailbox to which the FINAL answer message should be sent */
81+
/** @brief CommPort to which the FINAL answer message should be sent */
8282
S4U_CommPort *answer_commport;
8383
/** @brief The original file read request that kicked off the search (if null this was a lookup request)*/
8484
std::shared_ptr<StorageServiceFileReadRequestMessage> original;
@@ -102,7 +102,7 @@ namespace wrench {
102102
bool fileReadRequest,
103103
std::shared_ptr<bool> answered);
104104

105-
/** @brief Mailbox to which the FINAL answer message should be sent */
105+
/** @brief CommPort to which the FINAL answer message should be sent */
106106
S4U_CommPort *answer_commport;
107107
/** @brief The file being searched for */
108108
std::shared_ptr<DataFile> file;

‎include/wrench/simgrid_S4U_util/S4U_PendingCommunication.h

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -39,10 +39,10 @@ namespace wrench {
3939
/**
4040
* @brief Constructor
4141
*
42-
* @param mailbox: the S4U mailbox
42+
* @param commport: the CommPort
4343
* @param operation_type: the operation type
4444
*/
45-
S4U_PendingCommunication(simgrid::s4u::Mailbox *mailbox, OperationType operation_type) : s4u_mb(mailbox), operation_type(operation_type) {}
45+
S4U_PendingCommunication(S4U_CommPort *commport, OperationType operation_type) : commport(commport), operation_type(operation_type) {}
4646

4747
std::unique_ptr<SimulationMessage> wait();
4848
std::unique_ptr<SimulationMessage> wait(double timeout);
@@ -62,8 +62,8 @@ namespace wrench {
6262

6363
/** @brief The message */
6464
std::unique_ptr<SimulationMessage> simulation_message;
65-
/** @brief The S4U mailbox */
66-
simgrid::s4u::Mailbox *s4u_mb;
65+
/** @brief The CommPort */
66+
S4U_CommPort *commport;
6767
/** @brief The operation type */
6868
OperationType operation_type;
6969
};

‎include/wrench/util/MessageManager.h

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -29,17 +29,17 @@ namespace wrench {
2929
* a message was sent but never received). That is, if messages are in flight when the receiver daemon fails,
3030
* than, because the receiver is the one freeing memory, we have memory leaks.
3131
* This takes extra time however, and many simulations never simulate failures anyway, so it's use is only optional
32-
* at compile time. Perhaps woudl be a good idea to make its usage optional at runtime?
32+
* at compile time. Perhaps would be a good idea to make its usage optional at runtime?
3333
*/
3434

3535
class MessageManager {
3636

37-
static std::unordered_map<std::string, std::unordered_set<SimulationMessage *>> mailbox_messages;
37+
static std::unordered_map<const S4U_CommPort *, std::unordered_set<SimulationMessage *>> messages;
3838

3939
public:
40-
static void manageMessage(const std::string &commport_name, SimulationMessage *msg);
41-
static void cleanUpMessages(const std::string &commport_name);
42-
static void removeReceivedMessage(const std::string &commport_name, SimulationMessage *msg);
40+
static void manageMessage(const S4U_CommPort *commport, SimulationMessage *msg);
41+
static void cleanUpMessages(const S4U_CommPort *commport);
42+
static void removeReceivedMessage(const S4U_CommPort *commport, SimulationMessage *msg);
4343
static void cleanUpAllMessages();
4444
static void print();
4545
};

‎src/wrench/failure_causes/NetworkError.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -63,10 +63,10 @@ namespace wrench {
6363
}
6464

6565
/**
66-
* @brief Returns the commport_name name on which the error occurred
66+
* @brief Returns the name of the CommPort on which the error occurred
6767
* @return the commport_name name
6868
*/
69-
std::string NetworkError::getMailbox() {
69+
std::string NetworkError::getCommPortName() {
7070
return this->commport_name;
7171
}
7272

‎src/wrench/services/compute/batch/BatschedNetworkListener.cpp

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ namespace wrench {
4646
* @param property_list: property list ({} means "use all defaults")
4747
*/
4848
BatschedNetworkListener::BatschedNetworkListener(std::string hostname, std::shared_ptr<BatchComputeService> batch_service,
49-
simgrid::s4u::Mailbox *batch_service_commport,
49+
S4U_CommPort *batch_service_commport,
5050
std::string sched_port,
5151
std::string data_to_send,
5252
WRENCH_PROPERTY_COLLECTION_TYPE property_list) : BatschedNetworkListener(hostname, batch_service, batch_service_commport,
@@ -65,7 +65,7 @@ namespace wrench {
6565
* @param suffix the suffix to append
6666
*/
6767
BatschedNetworkListener::BatschedNetworkListener(
68-
std::string hostname, std::shared_ptr<BatchComputeService> batch_service, simgrid::s4u::Mailbox *batch_service_commport,
68+
std::string hostname, std::shared_ptr<BatchComputeService> batch_service, S4U_CommPort *batch_service_commport,
6969
std::string sched_port,
7070
std::string data_to_send, WRENCH_PROPERTY_COLLECTION_TYPE property_list,
7171
std::string suffix = "") : Service(hostname, "batch_network_listener" + suffix) {
@@ -102,7 +102,7 @@ namespace wrench {
102102
* @param answer_commport: commport_name on which ack will be received
103103
* @param execute_job_reply_data: message to send
104104
*/
105-
void BatschedNetworkListener::sendExecuteMessageToBatchComputeService(simgrid::s4u::Mailbox *answer_commport,
105+
void BatschedNetworkListener::sendExecuteMessageToBatchComputeService(S4U_CommPort *answer_commport,
106106
std::string execute_job_reply_data) {
107107
S4U_CommPort::putMessage(this->batch_service_commport,
108108
new BatchExecuteJobFromBatSchedMessage(answer_commport, execute_job_reply_data, 0));
@@ -157,7 +157,7 @@ namespace wrench {
157157
reply_decisions = nlohmann::json::parse(reply_data);
158158
decision_events = reply_decisions["events"];
159159

160-
auto answer_commport = S4U_Daemon::getRunningActorRecvMailbox();
160+
auto answer_commport = S4U_Daemon::getRunningActorRecvCommPort();
161161
for (auto decisions: decision_events) {
162162
std::string decision_type = decisions["type"];
163163
double decision_timestamp = decisions["timestamp"];

‎src/wrench/services/storage/compound/CompoundStorageService.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -309,7 +309,7 @@ namespace wrench {
309309
* @return A vector of shared_ptr on a FileLocation if the DataFile is known to the CompoundStorageService or empty vector if it's not.
310310
*/
311311
std::vector<std::shared_ptr<FileLocation>> CompoundStorageService::lookupFileLocation(const std::shared_ptr<DataFile> &file, S4U_CommPort *answer_commport) {
312-
WRENCH_INFO("CSS::lookupFileLocation() - DataFile + Mailbox");
312+
WRENCH_INFO("CSS::lookupFileLocation() - DataFile + CommPort");
313313

314314
this->commport->putMessage(
315315
new CompoundStorageLookupRequestMessage(

‎src/wrench/services/storage/simple/SimpleStorageServiceBufferized.cpp

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -221,7 +221,6 @@ namespace wrench {
221221

222222
// Generate a commport_name name on which to receive the file
223223
auto file_reception_commport = S4U_CommPort::getTemporaryCommPort();
224-
// auto file_reception_commport = S4U_CommPort::generateUniqueMailbox("faa_does_not_work");
225224

226225
// Reply with a "go ahead, send me the file" message
227226
answer_commport->dputMessage(

‎src/wrench/services/storage/storage_helper_classes/FileTransferThread.cpp

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -519,14 +519,6 @@ namespace wrench {
519519

520520
// Send a message to the source
521521
auto request_answer_commport = S4U_Daemon::getRunningActorRecvCommPort();
522-
// auto commport_that_should_receive_file_content = S4U_CommPort::generateUniqueMailbox("works_by_itself");
523-
524-
// S4U_CommPort *commport_that_should_receive_file_content;
525-
// if (src_loc->getStorageService()->buffer_size > DBL_EPSILON) {
526-
// commport_that_should_receive_file_content = S4U_CommPort::getTemporaryCommPort();
527-
// } else {
528-
// commport_that_should_receive_file_content = nullptr;
529-
// }
530522

531523
src_loc->getStorageService()->commport->putMessage(
532524
new StorageServiceFileReadRequestMessage(

‎src/wrench/simgrid_S4U_util/S4U_Daemon.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -231,7 +231,7 @@ namespace wrench {
231231
if ((S4U_Daemon::num_non_daemonized_actors_running == 0) or (not this->isSetToAutoRestart())) {
232232
// Service::increaseNumCompletedServicesCount();
233233
#ifdef MESSAGE_MANAGER
234-
MessageManager::cleanUpMessages(this->commport_name);
234+
MessageManager::cleanUpMessages(this->commport);
235235
#endif
236236
this->deleteLifeSaver();
237237
}

‎src/wrench/simgrid_S4U_util/S4U_PendingCommunication.cpp

Lines changed: 12 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -43,12 +43,15 @@ namespace wrench {
4343
} catch (simgrid::NetworkFailureException &e) {
4444
if (this->operation_type == S4U_PendingCommunication::OperationType::SENDING) {
4545
throw ExecutionException(std::make_shared<NetworkError>(
46-
NetworkError::OperationType::SENDING, NetworkError::FAILURE, s4u_mb->get_name()));
46+
NetworkError::OperationType::SENDING, NetworkError::FAILURE, this->commport->s4u_mb->get_name()));
4747
} else {
4848
throw ExecutionException(std::make_shared<NetworkError>(
49-
NetworkError::OperationType::RECEIVING, NetworkError::FAILURE, s4u_mb->get_name()));
49+
NetworkError::OperationType::RECEIVING, NetworkError::FAILURE, this->commport->s4u_mb->get_name()));
5050
}
5151
}
52+
#ifdef MESSAGE_MANAGER
53+
MessageManager::removeReceivedMessage(this->commport, this->simulation_message.get());
54+
#endif
5255
return std::move(this->simulation_message);
5356
}
5457

@@ -68,20 +71,23 @@ namespace wrench {
6871
} catch (simgrid::NetworkFailureException &e) {
6972
if (this->operation_type == S4U_PendingCommunication::OperationType::SENDING) {
7073
throw ExecutionException(std::make_shared<NetworkError>(
71-
NetworkError::OperationType::SENDING, NetworkError::FAILURE, s4u_mb->get_name()));
74+
NetworkError::OperationType::SENDING, NetworkError::FAILURE, this->commport->s4u_mb->get_name()));
7275
} else {
7376
throw ExecutionException(std::make_shared<NetworkError>(
74-
NetworkError::OperationType::RECEIVING, NetworkError::FAILURE, s4u_mb->get_name()));
77+
NetworkError::OperationType::RECEIVING, NetworkError::FAILURE, this->commport->s4u_mb->get_name()));
7578
}
7679
} catch (simgrid::TimeoutException &e) {
7780
if (this->operation_type == S4U_PendingCommunication::OperationType::SENDING) {
7881
throw ExecutionException(std::make_shared<NetworkError>(
79-
NetworkError::OperationType::SENDING, NetworkError::TIMEOUT, s4u_mb->get_name()));
82+
NetworkError::OperationType::SENDING, NetworkError::TIMEOUT, this->commport->s4u_mb->get_name()));
8083
} else {
8184
throw ExecutionException(std::make_shared<NetworkError>(
82-
NetworkError::OperationType::RECEIVING, NetworkError::TIMEOUT, s4u_mb->get_name()));
85+
NetworkError::OperationType::RECEIVING, NetworkError::TIMEOUT, this->commport->s4u_mb->get_name()));
8386
}
8487
}
88+
#ifdef MESSAGE_MANAGER
89+
MessageManager::removeReceivedMessage(this->commport, this->simulation_message.get());
90+
#endif
8591
return std::move(this->simulation_message);
8692
}
8793

@@ -139,9 +145,6 @@ namespace wrench {
139145
simgrid::s4u::ActivityPtr finished_activity = nullptr;
140146
try {
141147
finished_activity = pending_activities.wait_any_for(timeout);
142-
#ifdef MESSAGE_MANAGER
143-
MessageManager::removeReceivedMessage(pending_comms[index]->mailbox_name, pending_comms[index]->simulation_message.get());
144-
#endif
145148
} catch (simgrid::Exception &e) {
146149
auto failed_activity = pending_activities.get_failed_activity();
147150
for (unsigned long idx = 0; idx < pending_comms.size(); idx++ ) {

0 commit comments

Comments
 (0)