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