src/surf/ptask_L07.cpp

   1 /* Copyright (c) 2007-2010, 2013-2015. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #include <cstdlib>
   8
   9 #include <algorithm>
  10
  11 #include "ptask_L07.hpp"
  12
  13 #include "cpu_interface.hpp"
  14 #include "surf_routing.hpp"
  15 #include "xbt/lib.h"
  16
  17 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(surf_host);
  18 XBT_LOG_EXTERNAL_CATEGORY(xbt_cfg);
  19
  20 /**************************************/
  21 /*** Resource Creation & Destruction **/
  22 /**************************************/
  23 void surf_host_model_init_ptask_L07(void)
  24 {
  25   XBT_CINFO(xbt_cfg,"Switching to the L07 model to handle parallel tasks.");
  26   xbt_assert(!surf_cpu_model_pm, "CPU model type already defined");
  27   xbt_assert(!surf_network_model, "network model type already defined");
  28
  29   // Define the callbacks to parse the XML
  30   simgrid::surf::on_link.connect(netlink_parse_init);
  31
  32   surf_host_model = new simgrid::surf::HostL07Model();
  33   xbt_dynar_push(all_existing_models, &surf_host_model);
  34 }
  35
  36
  37 namespace simgrid {
  38 namespace surf {
  39
  40 HostL07Model::HostL07Model() : HostModel() {
  41   maxminSystem_ = lmm_system_new(1);
  42   surf_network_model = new NetworkL07Model(this,maxminSystem_);
  43   surf_cpu_model_pm = new CpuL07Model(this,maxminSystem_);
  44
  45   routing_model_create(surf_network_model->createLink("__loopback__", 498000000, 0.000015, SURF_LINK_FATPIPE, NULL));
  46 }
  47
  48 HostL07Model::~HostL07Model() {
  49   delete surf_cpu_model_pm;
  50   delete surf_network_model;
  51 }
  52
  53 CpuL07Model::CpuL07Model(HostL07Model *hmodel,lmm_system_t sys)
  54   : CpuModel()
  55   , p_hostModel(hmodel)
  56   {
  57     maxminSystem_ = sys;
  58   }
  59 CpuL07Model::~CpuL07Model() {
  60   surf_cpu_model_pm = NULL;
  61   lmm_system_free(maxminSystem_);
  62   maxminSystem_ = NULL;
  63 }
  64 NetworkL07Model::NetworkL07Model(HostL07Model *hmodel, lmm_system_t sys)
  65   : NetworkModel()
  66   , p_hostModel(hmodel)
  67   {
  68     maxminSystem_ = sys;
  69   }
  70 NetworkL07Model::~NetworkL07Model()
  71 {
  72   surf_network_model = NULL;
  73   maxminSystem_ = NULL; // Avoid multi-free
  74 }
  75
  76
  77 double HostL07Model::next_occuring_event(double /*now*/)
  78 {
  79   L07Action *action;
  80
  81   ActionList *running_actions = getRunningActionSet();
  82   double min = this->shareResourcesMaxMin(running_actions,
  83                                               maxminSystem_,
  84                                               bottleneck_solve);
  85
  86   for(ActionList::iterator it(running_actions->begin()), itend(running_actions->end())
  87    ; it != itend ; ++it) {
  88   action = static_cast<L07Action*>(&*it);
  89     if (action->m_latency > 0) {
  90       if (min < 0) {
  91         min = action->m_latency;
  92         XBT_DEBUG("Updating min (value) with %p (start %f): %f", action,
  93                action->getStartTime(), min);
  94       } else if (action->m_latency < min) {
  95         min = action->m_latency;
  96         XBT_DEBUG("Updating min (latency) with %p (start %f): %f", action,
  97                action->getStartTime(), min);
  98       }
  99     }
 100   }
 101
 102   XBT_DEBUG("min value : %f", min);
 103
 104   return min;
 105 }
 106
 107 void HostL07Model::updateActionsState(double /*now*/, double delta) {
 108
 109   L07Action *action;
 110   ActionList *actionSet = getRunningActionSet();
 111
 112   for(ActionList::iterator it = actionSet->begin(), itNext = it
 113    ; it != actionSet->end()
 114    ; it =  itNext) {
 115   ++itNext;
 116     action = static_cast<L07Action*>(&*it);
 117     if (action->m_latency > 0) {
 118       if (action->m_latency > delta) {
 119         double_update(&(action->m_latency), delta, sg_surf_precision);
 120       } else {
 121         action->m_latency = 0.0;
 122       }
 123       if ((action->m_latency == 0.0) && (action->isSuspended() == 0)) {
 124         action->updateBound();
 125         lmm_update_variable_weight(maxminSystem_, action->getVariable(), 1.0);
 126       }
 127     }
 128     XBT_DEBUG("Action (%p) : remains (%g) updated by %g.",
 129            action, action->getRemains(), lmm_variable_getvalue(action->getVariable()) * delta);
 130     action->updateRemains(lmm_variable_getvalue(action->getVariable()) * delta);
 131
 132     if (action->getMaxDuration() != NO_MAX_DURATION)
 133       action->updateMaxDuration(delta);
 134
 135     XBT_DEBUG("Action (%p) : remains (%g).", action, action->getRemains());
 136
 137     /* In the next if cascade, the action can be finished either because:
 138      *  - The amount of remaining work reached 0
 139      *  - The max duration was reached
 140      * If it's not done, it may have failed.
 141      */
 142
 143     if ((action->getRemains() <= 0) &&
 144         (lmm_get_variable_weight(action->getVariable()) > 0)) {
 145       action->finish();
 146       action->setState(Action::State::done);
 147     } else if ((action->getMaxDuration() != NO_MAX_DURATION) &&
 148                (action->getMaxDuration() <= 0)) {
 149       action->finish();
 150       action->setState(Action::State::done);
 151     } else {
 152       /* Need to check that none of the model has failed */
 153       lmm_constraint_t cnst = NULL;
 154       int i = 0;
 155
 156       while ((cnst = lmm_get_cnst_from_var(maxminSystem_, action->getVariable(), i++))) {
 157         void *constraint_id = lmm_constraint_id(cnst);
 158
 159         if (static_cast<HostImpl*>(constraint_id)->isOff()) {
 160           XBT_DEBUG("Action (%p) Failed!!", action);
 161           action->finish();
 162           action->setState(Action::State::failed);
 163           break;
 164         }
 165       }
 166     }
 167   }
 168   return;
 169 }
 170
 171 Action *HostL07Model::executeParallelTask(int host_nb, sg_host_t *host_list,
 172       double *flops_amount, double *bytes_amount,
 173       double rate) {
 174   return new L07Action(this, host_nb, host_list, flops_amount, bytes_amount, rate);
 175 }
 176
 177
 178 L07Action::L07Action(Model *model, int host_nb, sg_host_t*host_list,
 179     double *flops_amount, double *bytes_amount, double rate)
 180   : CpuAction(model, 1, 0)
 181 {
 182   int nb_link = 0;
 183   int nb_used_host = 0; /* Only the hosts with something to compute (>0 flops) are counted) */
 184   double latency = 0.0;
 185
 186   this->p_netcardList->reserve(host_nb);
 187   for (int i = 0; i<host_nb; i++)
 188     this->p_netcardList->push_back(host_list[i]->pimpl_netcard);
 189
 190   /* Compute the number of affected resources... */
 191   if(bytes_amount != NULL) {
 192     xbt_dict_t ptask_parallel_task_link_set = xbt_dict_new_homogeneous(NULL);
 193
 194     for (int i = 0; i < host_nb; i++) {
 195       for (int j = 0; j < host_nb; j++) {
 196
 197         if (bytes_amount[i * host_nb + j] > 0) {
 198           double lat=0.0;
 199           std::vector<Link*> *route = new std::vector<Link*>();
 200
 201           routing_platf->getRouteAndLatency((*p_netcardList)[i], (*p_netcardList)[j], route, &lat);
 202           latency = MAX(latency, lat);
 203
 204           for (auto link : *route)
 205             xbt_dict_set(ptask_parallel_task_link_set, link->getName(), link, NULL);
 206           delete route;
 207         }
 208       }
 209     }
 210
 211     nb_link = xbt_dict_length(ptask_parallel_task_link_set);
 212     xbt_dict_free(&ptask_parallel_task_link_set);
 213   }
 214
 215   for (int i = 0; i < host_nb; i++)
 216     if (flops_amount[i] > 0)
 217       nb_used_host++;
 218
 219   XBT_DEBUG("Creating a parallel task (%p) with %d hosts and %d unique links.", this, host_nb, nb_link);
 220   this->p_computationAmount = flops_amount;
 221   this->p_communicationAmount = bytes_amount;
 222   this->m_latency = latency;
 223   this->m_rate = rate;
 224
 225   this->p_variable = lmm_variable_new(model->getMaxminSystem(), this, 1.0,
 226       (rate > 0 ? rate : -1.0),
 227       host_nb + nb_link);
 228
 229   if (this->m_latency > 0)
 230     lmm_update_variable_weight(model->getMaxminSystem(), this->getVariable(), 0.0);
 231
 232   for (int i = 0; i < host_nb; i++)
 233     lmm_expand(model->getMaxminSystem(), host_list[i]->pimpl_cpu->getConstraint(),
 234         this->getVariable(), flops_amount[i]);
 235
 236   if(bytes_amount != NULL) {
 237     for (int i = 0; i < host_nb; i++) {
 238       for (int j = 0; j < host_nb; j++) {
 239
 240         if (bytes_amount[i * host_nb + j] == 0.0)
 241           continue;
 242         std::vector<Link*> *route = new std::vector<Link*>();
 243
 244         routing_platf->getRouteAndLatency((*p_netcardList)[i], (*p_netcardList)[j], route, NULL);
 245
 246         for (auto link : *route)
 247           lmm_expand_add(model->getMaxminSystem(), link->getConstraint(), this->getVariable(), bytes_amount[i * host_nb + j]);
 248
 249         delete route;
 250       }
 251     }
 252   }
 253
 254   if (nb_link + nb_used_host == 0) {
 255     this->setCost(1.0);
 256     this->setRemains(0.0);
 257   }
 258   xbt_free(host_list);
 259 }
 260
 261 Action *NetworkL07Model::communicate(NetCard *src, NetCard *dst, double size, double rate)
 262 {
 263   sg_host_t*host_list = xbt_new0(sg_host_t, 2);
 264   double *flops_amount = xbt_new0(double, 2);
 265   double *bytes_amount = xbt_new0(double, 4);
 266
 267   host_list[0] = sg_host_by_name(src->name());
 268   host_list[1] = sg_host_by_name(dst->name());
 269   bytes_amount[1] = size;
 270
 271   return p_hostModel->executeParallelTask(2, host_list, flops_amount, bytes_amount, rate);
 272 }
 273
 274 Cpu *CpuL07Model::createCpu(simgrid::s4u::Host *host,  xbt_dynar_t speedPerPstate, int core)
 275 {
 276   return new CpuL07(this, host, speedPerPstate, core);
 277 }
 278
 279 Link* NetworkL07Model::createLink(const char *name, double bandwidth, double latency,
 280     e_surf_link_sharing_policy_t policy, xbt_dict_t properties)
 281 {
 282   return new LinkL07(this, name, properties, bandwidth, latency, policy);
 283 }
 284
 285 /************
 286  * Resource *
 287  ************/
 288
 289 CpuL07::CpuL07(CpuL07Model *model, simgrid::s4u::Host *host, xbt_dynar_t speedPerPstate, int core)
 290  : Cpu(model, host, speedPerPstate, core)
 291 {
 292   constraint_ = lmm_constraint_new(model->getMaxminSystem(), this, xbt_dynar_get_as(speedPerPstate,0,double));
 293 }
 294
 295 CpuL07::~CpuL07()
 296 {
 297 }
 298
 299 LinkL07::LinkL07(NetworkL07Model *model, const char* name, xbt_dict_t props, double bandwidth, double latency,
 300              e_surf_link_sharing_policy_t policy)
 301  : Link(model, name, props, lmm_constraint_new(model->getMaxminSystem(), this, bandwidth))
 302 {
 303   m_bandwidth.peak = bandwidth;
 304   m_latency.peak = latency;
 305
 306   if (policy == SURF_LINK_FATPIPE)
 307     lmm_constraint_shared(getConstraint());
 308
 309   Link::onCreation(this);
 310 }
 311
 312 Action *CpuL07::execution_start(double size)
 313 {
 314   sg_host_t*host_list = xbt_new0(sg_host_t, 1);
 315   double *flops_amount = xbt_new0(double, 1);
 316
 317   host_list[0] = getHost();
 318   flops_amount[0] = size;
 319
 320   return static_cast<CpuL07Model*>(getModel())->p_hostModel
 321     ->executeParallelTask( 1, host_list, flops_amount, NULL, -1);
 322 }
 323
 324 Action *CpuL07::sleep(double duration)
 325 {
 326   L07Action *action = static_cast<L07Action*>(execution_start(1.0));
 327   action->m_maxDuration = duration;
 328   action->m_suspended = 2;
 329   lmm_update_variable_weight(getModel()->getMaxminSystem(), action->getVariable(), 0.0);
 330
 331   return action;
 332 }
 333
 334 bool CpuL07::isUsed(){
 335   return lmm_constraint_used(getModel()->getMaxminSystem(), getConstraint());
 336 }
 337
 338 /** @brief take into account changes of speed (either load or max) */
 339 void CpuL07::onSpeedChange() {
 340   lmm_variable_t var = NULL;
 341   lmm_element_t elem = NULL;
 342
 343     lmm_update_constraint_bound(getModel()->getMaxminSystem(), getConstraint(), speed_.peak * speed_.scale);
 344     while ((var = lmm_get_var_from_cnst
 345             (getModel()->getMaxminSystem(), getConstraint(), &elem))) {
 346       Action *action = static_cast<Action*>(lmm_variable_id(var));
 347
 348       lmm_update_variable_bound(getModel()->getMaxminSystem(),
 349                                 action->getVariable(),
 350                                 speed_.scale * speed_.peak);
 351     }
 352
 353   Cpu::onSpeedChange();
 354 }
 355
 356
 357 bool LinkL07::isUsed(){
 358   return lmm_constraint_used(getModel()->getMaxminSystem(), getConstraint());
 359 }
 360
 361 void CpuL07::apply_event(tmgr_trace_iterator_t triggered, double value){
 362   XBT_DEBUG("Updating cpu %s (%p) with value %g", getName(), this, value);
 363   if (triggered == speed_.event) {
 364     speed_.scale = value;
 365     onSpeedChange();
 366     tmgr_trace_event_unref(&speed_.event);
 367
 368   } else if (triggered == stateEvent_) {
 369     if (value > 0)
 370       turnOn();
 371     else
 372       turnOff();
 373     tmgr_trace_event_unref(&stateEvent_);
 374
 375   } else {
 376     xbt_die("Unknown event!\n");
 377   }
 378 }
 379
 380 void LinkL07::apply_event(tmgr_trace_iterator_t triggered, double value) {
 381   XBT_DEBUG("Updating link %s (%p) with value=%f", getName(), this, value);
 382   if (triggered == m_bandwidth.event) {
 383     updateBandwidth(value);
 384     tmgr_trace_event_unref(&m_bandwidth.event);
 385
 386   } else if (triggered == m_latency.event) {
 387     updateLatency(value);
 388     tmgr_trace_event_unref(&m_latency.event);
 389
 390   } else if (triggered == m_stateEvent) {
 391     if (value > 0)
 392       turnOn();
 393     else
 394       turnOff();
 395     tmgr_trace_event_unref(&m_stateEvent);
 396
 397   } else {
 398     xbt_die("Unknown event ! \n");
 399   }
 400 }
 401
 402 void LinkL07::updateBandwidth(double value)
 403 {
 404   m_bandwidth.peak = value;
 405   lmm_update_constraint_bound(getModel()->getMaxminSystem(), getConstraint(), m_bandwidth.peak * m_bandwidth.scale);
 406 }
 407
 408 void LinkL07::updateLatency(double value)
 409 {
 410   lmm_variable_t var = NULL;
 411   L07Action *action;
 412   lmm_element_t elem = NULL;
 413
 414   m_latency.peak = value;
 415   while ((var = lmm_get_var_from_cnst(getModel()->getMaxminSystem(), getConstraint(), &elem))) {
 416     action = static_cast<L07Action*>(lmm_variable_id(var));
 417     action->updateBound();
 418   }
 419 }
 420
 421 /**********
 422  * Action *
 423  **********/
 424
 425 L07Action::~L07Action(){
 426   delete p_netcardList;
 427   free(p_communicationAmount);
 428   free(p_computationAmount);
 429 }
 430
 431 void L07Action::updateBound()
 432 {
 433   double lat_current = 0.0;
 434   double lat_bound = -1.0;
 435   int i, j;
 436
 437   int hostNb = p_netcardList->size();
 438
 439   if (p_communicationAmount != NULL) {
 440     for (i = 0; i < hostNb; i++) {
 441       for (j = 0; j < hostNb; j++) {
 442
 443         if (p_communicationAmount[i * hostNb + j] > 0) {
 444           double lat = 0.0;
 445           std::vector<Link*> *route = new std::vector<Link*>();
 446           routing_platf->getRouteAndLatency((*p_netcardList)[i], (*p_netcardList)[j], route, &lat);
 447
 448           lat_current = MAX(lat_current, lat * p_communicationAmount[i * hostNb + j]);
 449           delete route;
 450         }
 451       }
 452     }
 453   }
 454   lat_bound = sg_tcp_gamma / (2.0 * lat_current);
 455   XBT_DEBUG("action (%p) : lat_bound = %g", this, lat_bound);
 456   if ((m_latency == 0.0) && (m_suspended == 0)) {
 457     if (m_rate < 0)
 458       lmm_update_variable_bound(getModel()->getMaxminSystem(), getVariable(), lat_bound);
 459     else
 460       lmm_update_variable_bound(getModel()->getMaxminSystem(), getVariable(),
 461         std::min(m_rate, lat_bound));
 462   }
 463 }
 464
 465 int L07Action::unref()
 466 {
 467   m_refcount--;
 468   if (!m_refcount) {
 469     if (action_hook.is_linked())
 470       p_stateSet->erase(p_stateSet->iterator_to(*this));
 471     if (getVariable())
 472       lmm_variable_free(getModel()->getMaxminSystem(), getVariable());
 473     delete this;
 474     return 1;
 475   }
 476   return 0;
 477 }
 478
 479 }
 480 }