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 "simgrid/s4u/Barrier.hpp"
   9 #include "simgrid/s4u/ConditionVariable.hpp"
  10 #include "smpi/smpi.h"
  11 #include "xbt/asserts.h"
  12 #include "xbt/ex.h"
  13 #include "xbt/log.h"
  14 #include "xbt/string.hpp"
  15 #include <simgrid/actor.h>
  16 #include <simgrid/s4u/Actor.hpp>
  17 #include <simgrid/s4u/Engine.hpp>
  18 #include <simgrid/s4u/Mutex.hpp>
  19 #include <simgrid/s4u/NetZone.hpp>
  20 #include <simgrid/s4u/Semaphore.hpp>
  21 #include <xbt/base.h>
  22 #include <xbt/sysdep.h>
  23
  24 #include "src/internal_config.h"
  25 #include "src/sthread/sthread.h"
  26
  27 #include <cmath>
  28 #include <dlfcn.h>
  29 #include <pthread.h>
  30 #include <semaphore.h>
  31 #include <sstream>
  32 #include <stdio.h>
  33 #include <stdlib.h>
  34 #include <string_view>
  35 #include <thread>
  36
  37 XBT_LOG_NEW_DEFAULT_CATEGORY(sthread, "pthread intercepter");
  38 namespace sg4 = simgrid::s4u;
  39
  40 static sg4::Host* lilibeth = nullptr;
  41
  42 int sthread_main(int argc, char** argv, char** envp, int (*raw_main)(int, char**, char**))
  43 {
  44   /* Do not intercept the main when run from SMPI: it will initialize the simulation properly */
  45   for (int i = 0; envp[i] != nullptr; i++)
  46     if (std::string_view(envp[i]).rfind("SMPI_GLOBAL_SIZE", 0) == 0) {
  47       printf("sthread refuses to intercept the SMPI application %s directly, as its interception is done otherwise.\n",
  48              argv[0]);
  49       return raw_main(argc, argv, envp);
  50     }
  51
  52   /* Do not intercept valgrind step 1 */
  53   if (not strcmp(argv[0], "/usr/bin/valgrind.bin") || not strcmp(argv[0], "/bin/sh")|| not strcmp(argv[0], "/bin/bash")|| not strcmp(argv[0], "gdb")) {
  54     printf("sthread refuses to intercept the execution of %s. Running the application unmodified.\n", argv[0]);
  55     fflush(stdout);
  56     return raw_main(argc, argv, envp);
  57   }
  58
  59   /* If not in SMPI, the old main becomes an actor in a newly created simulation */
  60   printf("sthread is intercepting the execution of %s\n", argv[0]);
  61   fflush(stdout);
  62
  63   sg4::Engine e(&argc, argv);
  64   auto* zone = sg4::create_full_zone("world");
  65   lilibeth   = zone->create_host("Lilibeth", 1e15);
  66   zone->seal();
  67
  68   /* Launch the user's main() on an actor */
  69   sthread_enable();
  70   sg4::ActorPtr main_actor = sg4::Actor::create("main thread", lilibeth, raw_main, argc, argv, envp);
  71
  72   sg4::Engine::get_instance()->run();
  73   sthread_disable();
  74   XBT_INFO("All threads exited. Terminating the simulation.");
  75
  76   return 0;
  77 }
  78
  79 struct sthread_mutex {
  80   s4u_Mutex* mutex;
  81 };
  82
  83 int sthread_create(unsigned long int* thread, const void* /*pthread_attr_t* attr*/, void* (*start_routine)(void*),
  84                    void* arg)
  85 {
  86   static int TID = 0;
  87   TID++;
  88   XBT_VERB("Create thread %d", TID);
  89   std::string name = std::string("thread ") + std::to_string(TID);
  90 #if HAVE_SMPI
  91   if (SMPI_is_inited()) {
  92     int rank = 0;
  93     MPI_Comm_rank(MPI_COMM_WORLD, &rank);
  94     name = simgrid::xbt::string_printf("%d:%d", rank, TID);
  95   }
  96 #endif
  97   sg4::ActorPtr actor = sg4::Actor::create(
  98       name, lilibeth,
  99       [](auto* user_function, auto* param) {
 100 #if HAVE_SMPI
 101         if (SMPI_is_inited())
 102           SMPI_thread_create();
 103 #endif
 104         sthread_enable();
 105         user_function(param);
 106         sthread_disable();
 107       },
 108       start_routine, arg);
 109
 110   intrusive_ptr_add_ref(actor.get());
 111   *thread = reinterpret_cast<unsigned long>(actor.get());
 112   return 0;
 113 }
 114 int sthread_join(sthread_t thread, void** /*retval*/)
 115 {
 116   sg4::ActorPtr actor(reinterpret_cast<sg4::Actor*>(thread));
 117   actor->join();
 118   intrusive_ptr_release(actor.get());
 119
 120   return 0;
 121 }
 122
 123 int sthread_mutexattr_init(sthread_mutexattr_t* attr)
 124 {
 125   memset(attr, 0, sizeof(*attr));
 126   return 0;
 127 }
 128 int sthread_mutexattr_settype(sthread_mutexattr_t* attr, int type)
 129 {
 130   switch (type) {
 131     case PTHREAD_MUTEX_NORMAL:
 132       xbt_assert(not attr->recursive, "S4U does not allow to remove the recursivness of a mutex.");
 133       attr->recursive = 0;
 134       break;
 135     case PTHREAD_MUTEX_RECURSIVE:
 136       attr->recursive = 1;
 137       attr->errorcheck = 0; // reset
 138       break;
 139     case PTHREAD_MUTEX_ERRORCHECK:
 140       attr->errorcheck = 1;
 141       THROW_UNIMPLEMENTED;
 142       break;
 143     default:
 144       THROW_IMPOSSIBLE;
 145   }
 146   return 0;
 147 }
 148 int sthread_mutexattr_gettype(const sthread_mutexattr_t* attr, int* type)
 149 {
 150   if (attr->recursive)
 151     *type = PTHREAD_MUTEX_RECURSIVE;
 152   else if (attr->errorcheck)
 153     *type = PTHREAD_MUTEX_ERRORCHECK;
 154   else
 155     *type = PTHREAD_MUTEX_NORMAL;
 156   return 0;
 157 }
 158 int sthread_mutexattr_getrobust(const sthread_mutexattr_t* attr, int* robustness)
 159 {
 160   *robustness = attr->robust;
 161   return 0;
 162 }
 163 int sthread_mutexattr_setrobust(sthread_mutexattr_t* attr, int robustness)
 164 {
 165   attr->robust = robustness;
 166   if (robustness)
 167     THROW_UNIMPLEMENTED;
 168   return 0;
 169 }
 170
 171 int sthread_mutex_init(sthread_mutex_t* mutex, const sthread_mutexattr_t* attr)
 172 {
 173   auto m = sg4::Mutex::create(attr != nullptr && attr->recursive);
 174   intrusive_ptr_add_ref(m.get());
 175
 176   mutex->mutex = m.get();
 177   return 0;
 178 }
 179
 180 int sthread_mutex_lock(sthread_mutex_t* mutex)
 181 {
 182   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
 183   if (mutex->mutex == nullptr)
 184     sthread_mutex_init(mutex, nullptr);
 185
 186   XBT_DEBUG("%s(%p)", __func__, mutex);
 187   static_cast<sg4::Mutex*>(mutex->mutex)->lock();
 188   return 0;
 189 }
 190
 191 int sthread_mutex_trylock(sthread_mutex_t* mutex)
 192 {
 193   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
 194   if (mutex->mutex == nullptr)
 195     sthread_mutex_init(mutex, nullptr);
 196
 197   XBT_DEBUG("%s(%p)", __func__, mutex);
 198   if (static_cast<sg4::Mutex*>(mutex->mutex)->try_lock())
 199     return 0;
 200   return EBUSY;
 201 }
 202
 203 int sthread_mutex_unlock(sthread_mutex_t* mutex)
 204 {
 205   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
 206   if (mutex->mutex == nullptr)
 207     sthread_mutex_init(mutex, nullptr);
 208
 209   XBT_DEBUG("%s(%p)", __func__, mutex);
 210   static_cast<sg4::Mutex*>(mutex->mutex)->unlock();
 211   return 0;
 212 }
 213 int sthread_mutex_destroy(sthread_mutex_t* mutex)
 214 {
 215   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
 216   if (mutex->mutex == nullptr)
 217     sthread_mutex_init(mutex, nullptr);
 218
 219   XBT_DEBUG("%s(%p)", __func__, mutex);
 220   intrusive_ptr_release(static_cast<sg4::Mutex*>(mutex->mutex));
 221   return 0;
 222 }
 223
 224 int sthread_barrier_init(sthread_barrier_t* barrier, const sthread_barrierattr_t* attr, unsigned count){
 225   auto b = sg4::Barrier::create(count);
 226   intrusive_ptr_add_ref(b.get());
 227
 228   barrier->barrier = b.get();
 229   return 0;
 230 }
 231 int sthread_barrier_wait(sthread_barrier_t* barrier){
 232   XBT_DEBUG("%s(%p)", __func__, barrier);
 233   static_cast<sg4::Barrier*>(barrier->barrier)->wait();
 234   return 0;
 235 }
 236 int sthread_barrier_destroy(sthread_barrier_t* barrier){
 237   XBT_DEBUG("%s(%p)", __func__, barrier);
 238   intrusive_ptr_release(static_cast<sg4::Barrier*>(barrier->barrier));
 239   return 0;
 240 }
 241
 242 int sthread_cond_init(sthread_cond_t* cond, sthread_condattr_t* attr)
 243 {
 244   auto cv = sg4::ConditionVariable::create();
 245   intrusive_ptr_add_ref(cv.get());
 246
 247   cond->cond = cv.get();
 248   return 0;
 249 }
 250 int sthread_cond_signal(sthread_cond_t* cond)
 251 {
 252   XBT_DEBUG("%s(%p)", __func__, cond);
 253   static_cast<sg4::ConditionVariable*>(cond->cond)->notify_one();
 254   return 0;
 255 }
 256 int sthread_cond_broadcast(sthread_cond_t* cond)
 257 {
 258   XBT_DEBUG("%s(%p)", __func__, cond);
 259   static_cast<sg4::ConditionVariable*>(cond->cond)->notify_all();
 260   return 0;
 261 }
 262 int sthread_cond_wait(sthread_cond_t* cond, sthread_mutex_t* mutex)
 263 {
 264   XBT_DEBUG("%s(%p)", __func__, cond);
 265   static_cast<sg4::ConditionVariable*>(cond->cond)->wait(static_cast<sg4::Mutex*>(mutex->mutex));
 266   return 0;
 267 }
 268 int sthread_cond_destroy(sthread_cond_t* cond)
 269 {
 270   XBT_DEBUG("%s(%p)", __func__, cond);
 271   intrusive_ptr_release(static_cast<sg4::ConditionVariable*>(cond->cond));
 272   return 0;
 273 }
 274
 275 int sthread_sem_init(sthread_sem_t* sem, int /*pshared*/, unsigned int value)
 276 {
 277   auto s = sg4::Semaphore::create(value);
 278   intrusive_ptr_add_ref(s.get());
 279
 280   sem->sem = s.get();
 281   return 0;
 282 }
 283 int sthread_sem_destroy(sthread_sem_t* sem)
 284 {
 285   intrusive_ptr_release(static_cast<sg4::Semaphore*>(sem->sem));
 286   return 0;
 287 }
 288 int sthread_sem_post(sthread_sem_t* sem)
 289 {
 290   static_cast<sg4::Semaphore*>(sem->sem)->release();
 291   return 0;
 292 }
 293 int sthread_sem_wait(sthread_sem_t* sem)
 294 {
 295   static_cast<sg4::Semaphore*>(sem->sem)->acquire();
 296   return 0;
 297 }
 298 int sthread_sem_trywait(sthread_sem_t* sem)
 299 {
 300   auto* s = static_cast<sg4::Semaphore*>(sem->sem);
 301   if (s->would_block()) {
 302     errno = EAGAIN;
 303     return -1;
 304   }
 305   s->acquire();
 306   return 0;
 307 }
 308 int sthread_sem_timedwait(sthread_sem_t* sem, const struct timespec* abs_timeout)
 309 {
 310   if (static_cast<sg4::Semaphore*>(sem->sem)->acquire_timeout(static_cast<double>(abs_timeout->tv_sec) +
 311                                                               static_cast<double>(abs_timeout->tv_nsec) / 1E9)) {
 312     errno = ETIMEDOUT;
 313     return -1;
 314   }
 315   return 0;
 316 }
 317
 318 int sthread_gettimeofday(struct timeval* tv)
 319 {
 320   if (tv) {
 321     double now   = simgrid::s4u::Engine::get_clock();
 322     double secs  = trunc(now);
 323     double usecs = (now - secs) * 1e6;
 324     tv->tv_sec   = static_cast<time_t>(secs);
 325     tv->tv_usec  = static_cast<decltype(tv->tv_usec)>(usecs); // suseconds_t
 326   }
 327   return 0;
 328 }
 329
 330 unsigned int sthread_sleep(double seconds)
 331 {
 332   XBT_DEBUG("sleep(%lf)", seconds);
 333   simgrid::s4u::this_actor::sleep_for(seconds);
 334   return 0;
 335 }
 336 int sthread_usleep(double seconds)
 337 {
 338   XBT_DEBUG("sleep(%lf)", seconds);
 339   simgrid::s4u::this_actor::sleep_for(seconds);
 340   return 0;
 341 }
 342
 343 #if 0
 344 int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr) {
 345     *cond = sg_cond_init();
 346     return 0;
 347 }
 348
 349 int pthread_cond_signal(pthread_cond_t *cond) {
 350         sg_cond_notify_one(*cond);
 351     return 0;
 352 }
 353
 354 int pthread_cond_broadcast(pthread_cond_t *cond) {
 355         sg_cond_notify_all(*cond);
 356     return 0;
 357 }
 358
 359 int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex) {
 360         sg_cond_wait(*cond, *mutex);
 361     return 0;
 362 }
 363
 364 int pthread_cond_destroy(pthread_cond_t *cond) {
 365         sg_cond_destroy(*cond);
 366     return 0;
 367 }
 368 #endif