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   cond->mutex = nullptr;
 249   return 0;
 250 }
 251 int sthread_cond_signal(sthread_cond_t* cond)
 252 {
 253   XBT_DEBUG("%s(%p)", __func__, cond);
 254
 255   if (cond->mutex == nullptr)
 256     XBT_WARN("No mutex was associated so far with condition variable %p. Safety checks skipped.", cond);
 257   else {
 258     auto* owner = static_cast<sg4::Mutex*>(cond->mutex)->get_owner();
 259     if (owner == nullptr)
 260       XBT_WARN("The mutex associated to condition %p is not currently owned by anyone when calling "
 261                "pthread_cond_signal(). The signal could get lost.",
 262                cond);
 263     else if (owner != simgrid::s4u::Actor::self())
 264       XBT_WARN("The mutex associated to condition %p is currently owned by %s, not by the thread currently calling "
 265                "calling pthread_cond_signal(). The signal could get lost.",
 266                cond, owner->get_cname());
 267   }
 268
 269   static_cast<sg4::ConditionVariable*>(cond->cond)->notify_one();
 270   return 0;
 271 }
 272 int sthread_cond_broadcast(sthread_cond_t* cond)
 273 {
 274   XBT_DEBUG("%s(%p)", __func__, cond);
 275
 276   if (cond->mutex == nullptr)
 277     XBT_WARN("No mutex was associated so far with condition variable %p. Safety checks skipped.", cond);
 278   else {
 279     auto* owner = static_cast<sg4::Mutex*>(cond->mutex)->get_owner();
 280     if (owner == nullptr)
 281       XBT_WARN("The mutex associated to condition %p is not currently owned by anyone when calling "
 282                "pthread_cond_broadcast(). The signal could get lost.",
 283                cond);
 284     else if (owner != simgrid::s4u::Actor::self())
 285       XBT_WARN("The mutex associated to condition %p is currently owned by %s, not by the thread currently calling "
 286                "calling pthread_cond_broadcast(). The signal could get lost.",
 287                cond, owner->get_cname());
 288   }
 289
 290   static_cast<sg4::ConditionVariable*>(cond->cond)->notify_all();
 291   return 0;
 292 }
 293 int sthread_cond_wait(sthread_cond_t* cond, sthread_mutex_t* mutex)
 294 {
 295   XBT_DEBUG("%s(%p)", __func__, cond);
 296
 297   if (cond->mutex == nullptr)
 298     cond->mutex = mutex->mutex;
 299   else if (cond->mutex != mutex->mutex)
 300     XBT_WARN("The condition %p is now waited with mutex %p while it was previoulsy waited with mutex %p. sthread may "
 301              "not work with such a dangerous code.",
 302              cond, cond->mutex, mutex->mutex);
 303
 304   static_cast<sg4::ConditionVariable*>(cond->cond)->wait(static_cast<sg4::Mutex*>(mutex->mutex));
 305   return 0;
 306 }
 307 int sthread_cond_timedwait(sthread_cond_t* cond, sthread_mutex_t* mutex, const struct timespec* abs_timeout)
 308 {
 309   XBT_DEBUG("%s(%p)", __func__, cond);
 310
 311   if (cond->mutex == nullptr)
 312     cond->mutex = mutex->mutex;
 313   else if (cond->mutex != mutex->mutex)
 314     XBT_WARN("The condition %p is now waited with mutex %p while it was previoulsy waited with mutex %p. sthread may "
 315              "not work with such a dangerous code.",
 316              cond, cond->mutex, mutex->mutex);
 317
 318   THROW_UNIMPLEMENTED;
 319 }
 320 int sthread_cond_destroy(sthread_cond_t* cond)
 321 {
 322   XBT_DEBUG("%s(%p)", __func__, cond);
 323   intrusive_ptr_release(static_cast<sg4::ConditionVariable*>(cond->cond));
 324   return 0;
 325 }
 326
 327 int sthread_sem_init(sthread_sem_t* sem, int /*pshared*/, unsigned int value)
 328 {
 329   auto s = sg4::Semaphore::create(value);
 330   intrusive_ptr_add_ref(s.get());
 331
 332   sem->sem = s.get();
 333   return 0;
 334 }
 335 int sthread_sem_destroy(sthread_sem_t* sem)
 336 {
 337   intrusive_ptr_release(static_cast<sg4::Semaphore*>(sem->sem));
 338   return 0;
 339 }
 340 int sthread_sem_post(sthread_sem_t* sem)
 341 {
 342   static_cast<sg4::Semaphore*>(sem->sem)->release();
 343   return 0;
 344 }
 345 int sthread_sem_wait(sthread_sem_t* sem)
 346 {
 347   static_cast<sg4::Semaphore*>(sem->sem)->acquire();
 348   return 0;
 349 }
 350 int sthread_sem_trywait(sthread_sem_t* sem)
 351 {
 352   auto* s = static_cast<sg4::Semaphore*>(sem->sem);
 353   if (s->would_block()) {
 354     errno = EAGAIN;
 355     return -1;
 356   }
 357   s->acquire();
 358   return 0;
 359 }
 360 int sthread_sem_timedwait(sthread_sem_t* sem, const struct timespec* abs_timeout)
 361 {
 362   if (static_cast<sg4::Semaphore*>(sem->sem)->acquire_timeout(static_cast<double>(abs_timeout->tv_sec) +
 363                                                               static_cast<double>(abs_timeout->tv_nsec) / 1E9)) {
 364     errno = ETIMEDOUT;
 365     return -1;
 366   }
 367   return 0;
 368 }
 369
 370 int sthread_gettimeofday(struct timeval* tv)
 371 {
 372   if (tv) {
 373     double now   = simgrid::s4u::Engine::get_clock();
 374     double secs  = trunc(now);
 375     double usecs = (now - secs) * 1e6;
 376     tv->tv_sec   = static_cast<time_t>(secs);
 377     tv->tv_usec  = static_cast<decltype(tv->tv_usec)>(usecs); // suseconds_t
 378   }
 379   return 0;
 380 }
 381
 382 unsigned int sthread_sleep(double seconds)
 383 {
 384   XBT_DEBUG("sleep(%lf)", seconds);
 385   simgrid::s4u::this_actor::sleep_for(seconds);
 386   return 0;
 387 }
 388 int sthread_usleep(double seconds)
 389 {
 390   XBT_DEBUG("sleep(%lf)", seconds);
 391   simgrid::s4u::this_actor::sleep_for(seconds);
 392   return 0;
 393 }