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 system binaries such as valgrind step 1 */
  53   std::vector<std::string> binaries = {"/usr/bin/valgrind.bin", "/bin/sh", "/bin/bash", "gdb", "addr2line"};
  54   for (int i = 0; envp[i] != nullptr; i++) {
  55     auto view = std::string_view(envp[i]);
  56     /* If you want to ignore more than one binary, export STHREAD_IGNORE_BINARY1=toto STHREAD_IGNORE_BINARY2=tutu */
  57     /* Note that this cannot be configured with --cfg because we are before the main() */
  58     if (view.rfind("STHREAD_IGNORE_BINARY", 0) == 0) {
  59       view.remove_prefix(std::min(view.rfind("=") + 1, view.size()));
  60       binaries.push_back(std::string(view));
  61     }
  62   }
  63   auto binary_view = std::string_view(argv[0]);
  64   for (auto binary : binaries) {
  65     if (binary_view.rfind(binary) != std::string_view::npos) {
  66       printf("sthread refuses to intercept the execution of %s. Running the application unmodified.\n", argv[0]);
  67       fflush(stdout);
  68       return raw_main(argc, argv, envp);
  69     }
  70   }
  71
  72   /* If not in SMPI, the old main becomes an actor in a newly created simulation */
  73   printf("sthread is intercepting the execution of %s. If it's not what you want, export STHREAD_IGNORE_BINARY=%s\n",
  74          argv[0], argv[0]);
  75   fflush(stdout);
  76
  77   sg4::Engine e(&argc, argv);
  78   auto* zone = sg4::create_full_zone("world");
  79   lilibeth   = zone->create_host("Lilibeth", 1e15);
  80   zone->seal();
  81
  82   /* Launch the user's main() on an actor */
  83   sthread_enable();
  84   sg4::ActorPtr main_actor = sg4::Actor::create("main thread", lilibeth, raw_main, argc, argv, envp);
  85
  86   sg4::Engine::get_instance()->run();
  87   sthread_disable();
  88   XBT_INFO("All threads exited. Terminating the simulation.");
  89
  90   return 0;
  91 }
  92
  93 struct sthread_mutex {
  94   s4u_Mutex* mutex;
  95 };
  96
  97 int sthread_create(unsigned long int* thread, const void* /*pthread_attr_t* attr*/, void* (*start_routine)(void*),
  98                    void* arg)
  99 {
 100   static int TID = 0;
 101   TID++;
 102   XBT_VERB("Create thread %d", TID);
 103   std::string name = std::string("thread ") + std::to_string(TID);
 104 #if HAVE_SMPI
 105   if (SMPI_is_inited()) {
 106     int rank = 0;
 107     MPI_Comm_rank(MPI_COMM_WORLD, &rank);
 108     name = simgrid::xbt::string_printf("%d:%d", rank, TID);
 109   }
 110 #endif
 111   sg4::ActorPtr actor = sg4::Actor::create(
 112       name, lilibeth,
 113       [](auto* user_function, auto* param) {
 114 #if HAVE_SMPI
 115         if (SMPI_is_inited())
 116           SMPI_thread_create();
 117 #endif
 118         sthread_enable();
 119         user_function(param);
 120         sthread_disable();
 121       },
 122       start_routine, arg);
 123
 124   intrusive_ptr_add_ref(actor.get());
 125   *thread = reinterpret_cast<unsigned long>(actor.get());
 126   return 0;
 127 }
 128 int sthread_join(sthread_t thread, void** /*retval*/)
 129 {
 130   sg4::ActorPtr actor(reinterpret_cast<sg4::Actor*>(thread));
 131   actor->join();
 132   intrusive_ptr_release(actor.get());
 133
 134   return 0;
 135 }
 136
 137 int sthread_mutexattr_init(sthread_mutexattr_t* attr)
 138 {
 139   memset(attr, 0, sizeof(*attr));
 140   return 0;
 141 }
 142 int sthread_mutexattr_settype(sthread_mutexattr_t* attr, int type)
 143 {
 144   switch (type) {
 145     case PTHREAD_MUTEX_NORMAL:
 146       xbt_assert(not attr->recursive, "S4U does not allow to remove the recursivness of a mutex.");
 147       attr->recursive = 0;
 148       break;
 149     case PTHREAD_MUTEX_RECURSIVE:
 150       attr->recursive = 1;
 151       attr->errorcheck = 0; // reset
 152       break;
 153     case PTHREAD_MUTEX_ERRORCHECK:
 154       attr->errorcheck = 1;
 155       THROW_UNIMPLEMENTED;
 156       break;
 157     default:
 158       THROW_IMPOSSIBLE;
 159   }
 160   return 0;
 161 }
 162 int sthread_mutexattr_gettype(const sthread_mutexattr_t* attr, int* type)
 163 {
 164   if (attr->recursive)
 165     *type = PTHREAD_MUTEX_RECURSIVE;
 166   else if (attr->errorcheck)
 167     *type = PTHREAD_MUTEX_ERRORCHECK;
 168   else
 169     *type = PTHREAD_MUTEX_NORMAL;
 170   return 0;
 171 }
 172 int sthread_mutexattr_getrobust(const sthread_mutexattr_t* attr, int* robustness)
 173 {
 174   *robustness = attr->robust;
 175   return 0;
 176 }
 177 int sthread_mutexattr_setrobust(sthread_mutexattr_t* attr, int robustness)
 178 {
 179   attr->robust = robustness;
 180   if (robustness)
 181     THROW_UNIMPLEMENTED;
 182   return 0;
 183 }
 184
 185 int sthread_mutex_init(sthread_mutex_t* mutex, const sthread_mutexattr_t* attr)
 186 {
 187   auto m = sg4::Mutex::create(attr != nullptr && attr->recursive);
 188   intrusive_ptr_add_ref(m.get());
 189
 190   mutex->mutex = m.get();
 191   return 0;
 192 }
 193
 194 int sthread_mutex_lock(sthread_mutex_t* mutex)
 195 {
 196   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
 197   if (mutex->mutex == nullptr)
 198     sthread_mutex_init(mutex, nullptr);
 199
 200   XBT_DEBUG("%s(%p)", __func__, mutex);
 201   static_cast<sg4::Mutex*>(mutex->mutex)->lock();
 202   return 0;
 203 }
 204
 205 int sthread_mutex_trylock(sthread_mutex_t* mutex)
 206 {
 207   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
 208   if (mutex->mutex == nullptr)
 209     sthread_mutex_init(mutex, nullptr);
 210
 211   XBT_DEBUG("%s(%p)", __func__, mutex);
 212   if (static_cast<sg4::Mutex*>(mutex->mutex)->try_lock())
 213     return 0;
 214   return EBUSY;
 215 }
 216
 217 int sthread_mutex_unlock(sthread_mutex_t* mutex)
 218 {
 219   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
 220   if (mutex->mutex == nullptr)
 221     sthread_mutex_init(mutex, nullptr);
 222
 223   XBT_DEBUG("%s(%p)", __func__, mutex);
 224   static_cast<sg4::Mutex*>(mutex->mutex)->unlock();
 225   return 0;
 226 }
 227 int sthread_mutex_destroy(sthread_mutex_t* mutex)
 228 {
 229   /* At least in glibc, PTHREAD_STATIC_INITIALIZER sets every fields to 0 */
 230   if (mutex->mutex == nullptr)
 231     sthread_mutex_init(mutex, nullptr);
 232
 233   XBT_DEBUG("%s(%p)", __func__, mutex);
 234   intrusive_ptr_release(static_cast<sg4::Mutex*>(mutex->mutex));
 235   return 0;
 236 }
 237
 238 int sthread_barrier_init(sthread_barrier_t* barrier, const sthread_barrierattr_t* attr, unsigned count){
 239   auto b = sg4::Barrier::create(count);
 240   intrusive_ptr_add_ref(b.get());
 241
 242   barrier->barrier = b.get();
 243   return 0;
 244 }
 245 int sthread_barrier_wait(sthread_barrier_t* barrier){
 246   XBT_DEBUG("%s(%p)", __func__, barrier);
 247   static_cast<sg4::Barrier*>(barrier->barrier)->wait();
 248   return 0;
 249 }
 250 int sthread_barrier_destroy(sthread_barrier_t* barrier){
 251   XBT_DEBUG("%s(%p)", __func__, barrier);
 252   intrusive_ptr_release(static_cast<sg4::Barrier*>(barrier->barrier));
 253   return 0;
 254 }
 255
 256 int sthread_cond_init(sthread_cond_t* cond, sthread_condattr_t* attr)
 257 {
 258   auto cv = sg4::ConditionVariable::create();
 259   intrusive_ptr_add_ref(cv.get());
 260
 261   cond->cond = cv.get();
 262   cond->mutex = nullptr;
 263   return 0;
 264 }
 265 int sthread_cond_signal(sthread_cond_t* cond)
 266 {
 267   XBT_DEBUG("%s(%p)", __func__, cond);
 268
 269   if (cond->mutex == nullptr)
 270     XBT_WARN("No mutex was associated so far with condition variable %p. Safety checks skipped.", cond);
 271   else {
 272     auto* owner = static_cast<sg4::Mutex*>(cond->mutex)->get_owner();
 273     if (owner == nullptr)
 274       XBT_WARN("The mutex associated to condition %p is not currently owned by anyone when calling "
 275                "pthread_cond_signal(). The signal could get lost.",
 276                cond);
 277     else if (owner != simgrid::s4u::Actor::self())
 278       XBT_WARN("The mutex associated to condition %p is currently owned by %s, not by the thread currently calling "
 279                "calling pthread_cond_signal(). The signal could get lost.",
 280                cond, owner->get_cname());
 281   }
 282
 283   static_cast<sg4::ConditionVariable*>(cond->cond)->notify_one();
 284   return 0;
 285 }
 286 int sthread_cond_broadcast(sthread_cond_t* cond)
 287 {
 288   XBT_DEBUG("%s(%p)", __func__, cond);
 289
 290   if (cond->mutex == nullptr)
 291     XBT_WARN("No mutex was associated so far with condition variable %p. Safety checks skipped.", cond);
 292   else {
 293     auto* owner = static_cast<sg4::Mutex*>(cond->mutex)->get_owner();
 294     if (owner == nullptr)
 295       XBT_WARN("The mutex associated to condition %p is not currently owned by anyone when calling "
 296                "pthread_cond_broadcast(). The signal could get lost.",
 297                cond);
 298     else if (owner != simgrid::s4u::Actor::self())
 299       XBT_WARN("The mutex associated to condition %p is currently owned by %s, not by the thread currently calling "
 300                "calling pthread_cond_broadcast(). The signal could get lost.",
 301                cond, owner->get_cname());
 302   }
 303
 304   static_cast<sg4::ConditionVariable*>(cond->cond)->notify_all();
 305   return 0;
 306 }
 307 int sthread_cond_wait(sthread_cond_t* cond, sthread_mutex_t* mutex)
 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   static_cast<sg4::ConditionVariable*>(cond->cond)->wait(static_cast<sg4::Mutex*>(mutex->mutex));
 319   return 0;
 320 }
 321 int sthread_cond_timedwait(sthread_cond_t* cond, sthread_mutex_t* mutex, const struct timespec* abs_timeout)
 322 {
 323   XBT_DEBUG("%s(%p)", __func__, cond);
 324
 325   if (cond->mutex == nullptr)
 326     cond->mutex = mutex->mutex;
 327   else if (cond->mutex != mutex->mutex)
 328     XBT_WARN("The condition %p is now waited with mutex %p while it was previoulsy waited with mutex %p. sthread may "
 329              "not work with such a dangerous code.",
 330              cond, cond->mutex, mutex->mutex);
 331
 332   THROW_UNIMPLEMENTED;
 333 }
 334 int sthread_cond_destroy(sthread_cond_t* cond)
 335 {
 336   XBT_DEBUG("%s(%p)", __func__, cond);
 337   intrusive_ptr_release(static_cast<sg4::ConditionVariable*>(cond->cond));
 338   return 0;
 339 }
 340
 341 int sthread_sem_init(sthread_sem_t* sem, int /*pshared*/, unsigned int value)
 342 {
 343   auto s = sg4::Semaphore::create(value);
 344   intrusive_ptr_add_ref(s.get());
 345
 346   sem->sem = s.get();
 347   return 0;
 348 }
 349 int sthread_sem_destroy(sthread_sem_t* sem)
 350 {
 351   intrusive_ptr_release(static_cast<sg4::Semaphore*>(sem->sem));
 352   return 0;
 353 }
 354 int sthread_sem_post(sthread_sem_t* sem)
 355 {
 356   static_cast<sg4::Semaphore*>(sem->sem)->release();
 357   return 0;
 358 }
 359 int sthread_sem_wait(sthread_sem_t* sem)
 360 {
 361   static_cast<sg4::Semaphore*>(sem->sem)->acquire();
 362   return 0;
 363 }
 364 int sthread_sem_trywait(sthread_sem_t* sem)
 365 {
 366   auto* s = static_cast<sg4::Semaphore*>(sem->sem);
 367   if (s->would_block()) {
 368     errno = EAGAIN;
 369     return -1;
 370   }
 371   s->acquire();
 372   return 0;
 373 }
 374 int sthread_sem_timedwait(sthread_sem_t* sem, const struct timespec* abs_timeout)
 375 {
 376   if (static_cast<sg4::Semaphore*>(sem->sem)->acquire_timeout(static_cast<double>(abs_timeout->tv_sec) +
 377                                                               static_cast<double>(abs_timeout->tv_nsec) / 1E9)) {
 378     errno = ETIMEDOUT;
 379     return -1;
 380   }
 381   return 0;
 382 }
 383
 384 int sthread_gettimeofday(struct timeval* tv)
 385 {
 386   if (tv) {
 387     double now   = simgrid::s4u::Engine::get_clock();
 388     double secs  = trunc(now);
 389     double usecs = (now - secs) * 1e6;
 390     tv->tv_sec   = static_cast<time_t>(secs);
 391     tv->tv_usec  = static_cast<decltype(tv->tv_usec)>(usecs); // suseconds_t
 392   }
 393   return 0;
 394 }
 395
 396 unsigned int sthread_sleep(double seconds)
 397 {
 398   XBT_DEBUG("sleep(%lf)", seconds);
 399   simgrid::s4u::this_actor::sleep_for(seconds);
 400   return 0;
 401 }
 402 int sthread_usleep(double seconds)
 403 {
 404   XBT_DEBUG("sleep(%lf)", seconds);
 405   simgrid::s4u::this_actor::sleep_for(seconds);
 406   return 0;
 407 }