Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
ae5a74ec34905e32fd12615ef931015cf1565416
[simgrid.git] / src / smpi / smpi_sender.c
1 #include "private.h"
2
3 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_sender, smpi,
4                                 "Logging specific to SMPI (sender)");
5
6 int smpi_sender(int argc,char*argv[]) {
7         smpi_host_data_t mydata = SIMIX_process_get_data(SIMIX_process_self());
8   smx_process_t self;
9   smx_host_t shost;
10
11   int index;
12
13   xbt_fifo_t request_queue;
14
15   smpi_mpi_request_t request;
16
17   smx_host_t dhost;
18
19   smx_action_t action;
20
21   e_surf_action_state_t state;
22
23   smpi_received_message_t message;
24
25   int dindex;
26
27   self = SIMIX_process_self();
28   shost = SIMIX_host_self();
29
30   index = mydata->index;
31
32   request_queue = mydata->pending_send_request_queue;
33
34   while (1) {
35     request = xbt_fifo_shift(request_queue);
36
37     if (NULL != request) {
38       message = xbt_mallocator_get(smpi_global->message_mallocator);
39
40       SIMIX_mutex_lock(request->mutex);
41
42       message->comm = request->comm;
43       message->src = request->comm->index_to_rank_map[index];
44       message->tag = request->tag;
45       message->data = request->data;
46       message->buf = xbt_malloc(request->datatype->size * request->count);
47       memcpy(message->buf, request->buf,
48              request->datatype->size * request->count);
49
50       dindex = request->comm->rank_to_index_map[request->dst];
51       dhost = smpi_global->hosts[dindex];
52
53       message->forward = (request->forward - 1) / 2;
54       request->forward = request->forward / 2;
55
56       if (0 < request->forward) {
57         request->dst =
58           (request->dst + message->forward + 1) % request->comm->size;
59         xbt_fifo_push(request_queue, request);
60       } else {
61         request->completed = 1;
62       }
63
64       action =
65         SIMIX_action_communicate(shost, dhost, "communication",
66                                  request->datatype->size * request->count,
67                                  -1.0);
68
69       SIMIX_register_action_to_condition(action, request->cond);
70
71       for (state = SIMIX_action_get_state(action);
72            state == SURF_ACTION_READY ||
73            state == SURF_ACTION_RUNNING;
74            state = SIMIX_action_get_state(action)
75         ) {
76         SIMIX_cond_wait(request->cond, request->mutex);
77       }
78
79       xbt_fifo_push(smpi_global->received_message_queues[dindex], message);
80
81       SIMIX_unregister_action_to_condition(action, request->cond);
82       SIMIX_action_destroy(action);
83
84       SIMIX_mutex_unlock(request->mutex);
85
86       // wake up receiver if necessary
87       smpi_host_data_t remote_host = SIMIX_host_get_data(SIMIX_process_get_host(smpi_global->main_processes[dindex]));
88       SIMIX_process_resume(remote_host->receiver);
89
90     } else if (mydata->finalize>0) { /* main wants me to die and nothing to do */
91         mydata->finalize--;
92         SIMIX_cond_signal(mydata->cond);
93         return 0;
94     } else {
95         SIMIX_process_suspend(self);
96     }
97   }
98   return 0;
99 }