scheduler.h

Go to the documentation of this file.

/*
 *  scheduler.h
 *
 *  This file is part of NEST.
 *
 *  Copyright (C) 2004 The NEST Initiative
 *
 *  NEST is free software: you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation, either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  NEST is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with NEST.  If not, see <http://www.gnu.org/licenses/>.
 *
 */

#ifndef SCHEDULER_H
#define SCHEDULER_H
#include <queue>
#include <vector>
#include <iostream>
#include <iomanip>
#include <limits>
#include <fstream>
#include <sys/time.h>

#include "nest.h"
#include "nest_time.h"
#include "nodelist.h"
#include "event.h"
#include "event_priority.h"
#include "randomgen.h"
#include "lockptr.h"
#include "communicator.h"

namespace nest
{

  using std::priority_queue;
  using std::vector;
  typedef Communicator::OffGridSpike OffGridSpike;

  class Network;
  
  class Scheduler {
    
  public: // Public methods

    Scheduler(Network &);
    virtual ~Scheduler();

    void reset();

    void clear_pending_spikes();

    void simulate(Time const&);

    void resume();

    void prepare_simulation();

    void finalize_simulation();

    void terminate();

    void send_remote(thread p, SpikeEvent&, const long_t lag = 0);
    
    void send_offgrid_remote(thread p, SpikeEvent&, const long_t lag = 0);

    Node* thread_lid_to_node(thread t, targetindex thread_local_id) const;

    thread get_num_threads() const;
    
    void set_num_threads(thread n_threads);
    
    thread get_num_processes() const;

    thread get_num_rec_processes() const;
    thread get_num_sim_processes() const;

    void set_num_rec_processes(int nrp, bool called_by_reset=false);

    void increment_n_gsd();

    index get_n_gsd();

    bool is_local_node(Node*) const;
    
    bool is_local_vp(thread) const;
    
    thread suggest_vp(index gid) const;

    thread suggest_rec_vp(index gid) const;

    thread vp_to_thread(thread vp) const;
    
    thread thread_to_vp(thread t) const;

    thread get_global_thread_id(thread lt) const;
        
    thread get_process_id(thread vp) const;

    bool get_simulated() const;

    void set_off_grid_communication(bool off_grid_spiking);

    bool get_off_grid_communication() const;

    Time const& get_slice_origin() const;

    Time get_previous_slice_origin() const;

    Time const get_time() const;

    void update();
 
    void set_network_(Network*);

    void set_status(const DictionaryDatum&);
    void get_status(DictionaryDatum &) const;

    librandom::RngPtr get_rng(const thread) const;
    librandom::RngPtr get_grng() const;

    static
    delay get_modulo(delay d);

    static
    delay get_slice_modulo(delay d);

    static
    delay get_min_delay(); 

    static
    delay get_max_delay();

    size_t get_slice() const;

    void calibrate_clock();

    void ensure_valid_thread_local_ids() { update_nodes_vec_(); }

  private:

    void init_();

    void finalize_();
    
    void advance_time_();

    void print_progress_();

    void prepare_nodes();

    void prepare_node_(Node *);
    
    void finalize_nodes();

    void compute_moduli_(); 
    
    void init_moduli_();

    void create_rngs_(const bool ctor_call = false);
    void create_grng_(const bool ctor_call = false);

    static
    void update_delay_extrema_();

    bool initialized_;
    bool simulating_;  
    bool force_singlethreading_;

    index n_threads_; 

    index n_rec_procs_; 
    index n_sim_procs_; 

    index n_gsd_; 

    volatile index   entry_counter_; 
    volatile index   exit_counter_;  

    vector<vector<Node*> > nodes_vec_;   
    index nodes_vec_network_size_;       
    
    Time     clock_;        
    delay   slice_;        
    delay   to_do_;        
    delay   to_do_total_;  
    delay   from_step_;    
    delay   to_step_;      

    timeval t_slice_begin_; 
    timeval t_slice_end_;   
    long t_real_;           
    
    bool terminate_;        
    bool simulated_;        
    bool off_grid_spiking_; 
    bool print_time_;       

    std::vector<long_t> rng_seeds_;  
    long_t grng_seed_;   
    
    static
    Network*  net_;

    static
    delay min_delay_;

    static
    delay max_delay_;
    
    static 
    vector<delay> moduli_; 

    static 
    vector<delay> slice_moduli_;

    vector<librandom::RngPtr> rng_;
    librandom::RngPtr grng_;

    std::vector<std::vector<std::vector<uint_t> > > spike_register_;

    std::vector<std::vector<std::vector<OffGridSpike> > > 
      offgrid_spike_register_;

    std::vector<uint_t> local_grid_spikes_;

    std::vector<uint_t> global_grid_spikes_;

     std::vector<OffGridSpike> local_offgrid_spikes_;

     std::vector<OffGridSpike> global_offgrid_spikes_;  

     std::vector<int> displacements_;
          

    static
    const delay comm_marker_;

    void configure_spike_buffers_();
    

    void update_nodes_vec_();

    void collocate_buffers_();

    void gather_events_();

    void deliver_events_(thread t);
  };

  inline
  Time const& Scheduler::get_slice_origin() const
  {
    return clock_;
  }
  
  inline
  Time Scheduler::get_previous_slice_origin() const
  {
    return clock_ - Time::step(min_delay_);
  }

  inline
  Time const Scheduler::get_time() const
  {
    assert(!simulating_);
    return clock_ + Time::step(from_step_);
  }

  inline 
  thread Scheduler::get_num_threads() const
  {
    return n_threads_;
  }

  inline 
  thread Scheduler::get_num_processes() const
  {
    return Communicator::get_num_processes();
  }

  inline 
  thread Scheduler::get_num_rec_processes() const
  {
    return n_rec_procs_;
  }

  inline 
  thread Scheduler::get_num_sim_processes() const
  {
    return n_sim_procs_;
  }

  inline 
  void Scheduler::increment_n_gsd()
  {
    ++n_gsd_;
  }
  
  inline
  index Scheduler::get_n_gsd()
  {
    return n_gsd_;
  }

  inline
  thread Scheduler::get_process_id(thread vp) const
  {
    if (vp >= static_cast<thread>(n_sim_procs_ * n_threads_)) // vp belongs to recording VPs
    {
      return (vp-n_sim_procs_*n_threads_) % n_rec_procs_ + n_sim_procs_;
    }
    else  // vp belongs to simulating VPs
    {
      return vp % n_sim_procs_;
    }
  }

  inline
  bool Scheduler::is_local_node(Node* n) const
  {
    return is_local_vp(n->get_vp());
  }

  inline
  bool Scheduler::is_local_vp(thread vp) const
  {
    return get_process_id(vp) == Communicator::get_rank();
  }

  inline
  thread Scheduler::suggest_vp(index gid) const
  {
    return gid % (n_sim_procs_*n_threads_);
  }

  inline
  thread Scheduler::suggest_rec_vp(index gid) const
  {
    return gid % (n_rec_procs_*n_threads_)+n_sim_procs_*n_threads_;
  }

  inline
  thread Scheduler::vp_to_thread(thread vp) const
  {
    if (vp >= static_cast<thread>(n_sim_procs_ * n_threads_))
    {
      return (vp + n_sim_procs_*(1-n_threads_) - Communicator::get_rank() ) / n_rec_procs_;
    }
    else
    {
      return vp / n_sim_procs_;
    }
  }

  inline
  thread Scheduler::thread_to_vp(thread t) const
  {
    if (Communicator::get_rank() >= static_cast<int>(n_sim_procs_)) // Rank is a recording process
    {
      return t * n_rec_procs_ + Communicator::get_rank() - n_sim_procs_ + n_sim_procs_*n_threads_;
    }
    else // Rank is a simulating process
    {
      return t * n_sim_procs_ + Communicator::get_rank();
    }
  }

  inline
  bool Scheduler::get_simulated() const
  {
    return simulated_;
  }

  inline
  void Scheduler::set_off_grid_communication(bool off_grid_spiking)
  {
    off_grid_spiking_ = off_grid_spiking;
  }

  inline
  bool Scheduler::get_off_grid_communication() const
  {
    return off_grid_spiking_;
  }
  
  inline
  librandom::RngPtr Scheduler::get_rng(const thread thrd) const
  {
    assert(thrd < static_cast<thread>(rng_.size()));
    return rng_[thrd];
  }

  inline
  librandom::RngPtr Scheduler::get_grng() const
  {
    return grng_;
  }

  inline 
  void Scheduler::calibrate_clock()
  {
    clock_.calibrate();
  }

  inline
  void Scheduler::prepare_node_(Node *n)
  {
    // Frozen nodes are initialized and calibrated, so that they
    // have ring buffers and can accept incoming spikes.
    n->init_buffers();
    n->calibrate();
  }

  inline
  Node* Scheduler::thread_lid_to_node(thread t, targetindex thread_local_id) const
  {
    return nodes_vec_[t][thread_local_id];
  }

  inline
  void Scheduler::send_remote(thread t, SpikeEvent& e, const delay lag)
  {
    // Put the spike in a buffer for the remote machines
    for (int_t i = 0; i < e.get_multiplicity(); ++i)
      spike_register_[t][lag].push_back(e.get_sender().get_gid());
  }

  inline
  void Scheduler::send_offgrid_remote(thread t, SpikeEvent& e, const delay lag)
  {
    // Put the spike in a buffer for the remote machines
    OffGridSpike ogs(e.get_sender().get_gid(), e.get_offset());
    for (int_t i = 0; i < e.get_multiplicity(); ++i)
      offgrid_spike_register_[t][lag].push_back(ogs);
  }

  inline
  delay Scheduler::get_modulo(delay d)
  {
    // Note, here d may be 0, since bin 0 represents the "current" time
    // when all evens due are read out.
    assert(static_cast<vector<delay>::size_type>(d) < moduli_.size());

    return moduli_[d];
  }

  inline
  delay Scheduler::get_slice_modulo(delay d)
  {
    // Note, here d may be 0, since bin 0 represents the "current" time
    // when all evens due are read out.
    assert(static_cast<vector<delay>::size_type>(d) < slice_moduli_.size());

    return slice_moduli_[d];
  }

  inline
  delay Scheduler::get_min_delay()
  {
    return min_delay_;
  }

  inline
  delay Scheduler::get_max_delay()
  {
    return max_delay_;
  }

  inline
  size_t Scheduler::get_slice() const
  {
    return slice_;
  }
  
  inline 
  void Scheduler::terminate()
  {
    terminate_=true;
  }

}

#endif //SCHEDULER_H