Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Implement the semaphore functions in sthread
[simgrid.git] / src / sthread / sthread_impl.cpp
1 /* Copyright (c) 2002-2023. The SimGrid Team. All rights reserved.          */
2
3 /* This program is free software; you can redistribute it and/or modify it
4  * under the terms of the license (GNU LGPL) which comes with this package. */
5
6 /* SimGrid's pthread interposer. Actual implementation of the symbols (see the comment in sthread.h) */
7
8 #include "smpi/smpi.h"
9 #include "xbt/string.hpp"
10 #include <simgrid/actor.h>
11 #include <simgrid/s4u/Actor.hpp>
12 #include <simgrid/s4u/Engine.hpp>
13 #include <simgrid/s4u/Mutex.hpp>
14 #include <simgrid/s4u/NetZone.hpp>
15 #include <simgrid/s4u/Semaphore.hpp>
16 #include <xbt/base.h>
17 #include <xbt/sysdep.h>
18
19 #include "src/internal_config.h"
20 #include "src/sthread/sthread.h"
21
22 #include <cmath>
23 #include <dlfcn.h>
24 #include <pthread.h>
25 #include <semaphore.h>
26 #include <sstream>
27 #include <stdio.h>
28 #include <stdlib.h>
29 #include <string_view>
30 #include <thread>
31
32 XBT_LOG_NEW_DEFAULT_CATEGORY(sthread, "pthread intercepter");
33 namespace sg4 = simgrid::s4u;
34
35 static sg4::Host* lilibeth = nullptr;
36
37 int sthread_main(int argc, char** argv, char** envp, int (*raw_main)(int, char**, char**))
38 {
39   /* Do not intercept the main when run from SMPI: it will initialize the simulation properly */
40   for (int i = 0; envp[i] != nullptr; i++)
41     if (std::string_view(envp[i]).rfind("SMPI_GLOBAL_SIZE", 0) == 0)
42       return raw_main(argc, argv, envp);
43
44   /* If not in SMPI, the old main becomes an actor in a newly created simulation */
45   std::ostringstream id;
46   id << std::this_thread::get_id();
47
48   XBT_DEBUG("sthread main() is starting in thread %s", id.str().c_str());
49
50   sg4::Engine e(&argc, argv);
51   auto* zone = sg4::create_full_zone("world");
52   lilibeth   = zone->create_host("Lilibeth", 1e15);
53   zone->seal();
54
55   /* Launch the user's main() on an actor */
56   sthread_enable();
57   sg4::ActorPtr main_actor = sg4::Actor::create("main thread", lilibeth, raw_main, argc, argv, envp);
58
59   XBT_INFO("Starting the simulation.");
60   sg4::Engine::get_instance()->run();
61   sthread_disable();
62   XBT_INFO("All threads exited. Terminating the simulation.");
63
64   return 0;
65 }
66
67 struct sthread_mutex {
68   s4u_Mutex* mutex;
69 };
70
71 int sthread_create(unsigned long int* thread, const void* /*pthread_attr_t* attr*/, void* (*start_routine)(void*),
72                    void* arg)
73 {
74   static int TID = 0;
75   TID++;
76   XBT_VERB("Create thread %d", TID);
77   int rank = 0;
78 #if HAVE_SMPI
79   if (SMPI_is_inited())
80     MPI_Comm_rank(MPI_COMM_WORLD, &rank);
81 #endif
82   std::string name    = simgrid::xbt::string_printf("%d:%d", rank, TID);
83   sg4::ActorPtr actor = sg4::Actor::create(
84       name, lilibeth,
85       [](auto* user_function, auto* param) {
86 #if HAVE_SMPI
87         if (SMPI_is_inited())
88           SMPI_thread_create();
89 #endif
90         sthread_enable();
91         user_function(param);
92         sthread_disable();
93       },
94       start_routine, arg);
95
96   intrusive_ptr_add_ref(actor.get());
97   *thread = reinterpret_cast<unsigned long>(actor.get());
98   return 0;
99 }
100 int sthread_join(sthread_t thread, void** /*retval*/)
101 {
102   sg4::ActorPtr actor(reinterpret_cast<sg4::Actor*>(thread));
103   actor->join();
104   intrusive_ptr_release(actor.get());
105
106   return 0;
107 }
108
109 int sthread_mutex_init(sthread_mutex_t* mutex, const void* /*pthread_mutexattr_t* attr*/)
110 {
111   auto m = sg4::Mutex::create();
112   intrusive_ptr_add_ref(m.get());
113
114   mutex->mutex = m.get();
115   return 0;
116 }
117
118 int sthread_mutex_lock(sthread_mutex_t* mutex)
119 {
120   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
121   if (mutex->mutex == nullptr)
122     sthread_mutex_init(mutex, nullptr);
123
124   static_cast<sg4::Mutex*>(mutex->mutex)->lock();
125   return 0;
126 }
127
128 int sthread_mutex_trylock(sthread_mutex_t* mutex)
129 {
130   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
131   if (mutex->mutex == nullptr)
132     sthread_mutex_init(mutex, nullptr);
133
134   return static_cast<sg4::Mutex*>(mutex->mutex)->try_lock();
135 }
136
137 int sthread_mutex_unlock(sthread_mutex_t* mutex)
138 {
139   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
140   if (mutex->mutex == nullptr)
141     sthread_mutex_init(mutex, nullptr);
142
143   static_cast<sg4::Mutex*>(mutex->mutex)->unlock();
144   return 0;
145 }
146 int sthread_mutex_destroy(sthread_mutex_t* mutex)
147 {
148   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
149   if (mutex->mutex == nullptr)
150     sthread_mutex_init(mutex, nullptr);
151
152   intrusive_ptr_release(static_cast<sg4::Mutex*>(mutex->mutex));
153   return 0;
154 }
155 int sthread_sem_init(sthread_sem_t* sem, int pshared, unsigned int value)
156 {
157   auto s = sg4::Semaphore::create(value);
158   intrusive_ptr_add_ref(s.get());
159
160   sem->sem = s.get();
161   return 0;
162 }
163 int sthread_sem_destroy(sthread_sem_t* sem)
164 {
165   intrusive_ptr_release(static_cast<sg4::Semaphore*>(sem->sem));
166   return 0;
167 }
168 int sthread_sem_post(sthread_sem_t* sem)
169 {
170   static_cast<sg4::Semaphore*>(sem->sem)->release();
171   return 0;
172 }
173 int sthread_sem_wait(sthread_sem_t* sem)
174 {
175   static_cast<sg4::Semaphore*>(sem->sem)->acquire();
176   return 0;
177 }
178 int sthread_sem_trywait(sthread_sem_t* sem)
179 {
180   auto* s = static_cast<sg4::Semaphore*>(sem->sem);
181   if (s->would_block()) {
182     errno = EAGAIN;
183     return -1;
184   }
185   s->acquire();
186   return 0;
187 }
188 int sthread_sem_timedwait(sthread_sem_t* sem, const struct timespec* abs_timeout)
189 {
190   if (static_cast<sg4::Semaphore*>(sem->sem)->acquire_timeout(abs_timeout->tv_sec +
191                                                               static_cast<double>(abs_timeout->tv_nsec) / 1E9)) {
192     errno = ETIMEDOUT;
193     return -1;
194   }
195   return 0;
196 }
197
198 int sthread_gettimeofday(struct timeval* tv)
199 {
200   if (tv) {
201     double now   = simgrid::s4u::Engine::get_clock();
202     double secs  = trunc(now);
203     double usecs = (now - secs) * 1e6;
204     tv->tv_sec   = static_cast<time_t>(secs);
205     tv->tv_usec  = static_cast<decltype(tv->tv_usec)>(usecs); // suseconds_t
206   }
207   return 0;
208 }
209
210 void sthread_sleep(double seconds)
211 {
212   simgrid::s4u::this_actor::sleep_for(seconds);
213 }
214
215 #if 0
216 int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr) {
217     *cond = sg_cond_init();
218     return 0;
219 }
220
221 int pthread_cond_signal(pthread_cond_t *cond) {
222         sg_cond_notify_one(*cond);
223     return 0;
224 }
225
226 int pthread_cond_broadcast(pthread_cond_t *cond) {
227         sg_cond_notify_all(*cond);
228     return 0;
229 }
230
231 int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex) {
232         sg_cond_wait(*cond, *mutex);
233     return 0;
234 }
235
236 int pthread_cond_destroy(pthread_cond_t *cond) {
237         sg_cond_destroy(*cond);
238     return 0;
239 }
240 #endif