runtime.hpp

#ifndef HMLP_RUNTIME_HPP
#define HMLP_RUNTIME_HPP

#include <time.h>
#include <tuple>
#include <algorithm>
#include <vector>
#include <deque>
#include <map>
#include <unordered_map>
#include <limits>
#include <cstdint>
#include <cassert>
#include <stdio.h>
#include <stdlib.h>
#include <iostream>
#include <cstddef>
#include <omp.h>

//#ifdef USE_INTEL
//#include <mkl.h>
//#endif

#ifdef USE_PTHREAD_RUNTIME
#include <pthread.h>
#endif

#include <base/thread.hpp>
#include <base/tci.hpp>
#include <hmlp_mpi.hpp>

#define MAX_WORKER 68


using namespace std;






namespace hmlp
{

//typedef enum 
//{
//  HMLP_SCHEDULE_DEFAULT,
//  HMLP_SCHEDULE_ROUND_ROBIN,
//  HMLP_SCHEDULE_UNIFORM,
//  HMLP_SCHEDULE_HEFT
//} SchedulePolicy;

typedef enum { ALLOCATED, NOTREADY, QUEUED, RUNNING, EXECUTED, DONE, CANCELLED } TaskStatus;

typedef enum { R, W, RW } ReadWriteType;


// *  class Lock
// */ 
//
//class Lock
//{
//  public:
//
//    Lock();
//
//    ~Lock();
//
//    void Acquire();
//
//    void Release();
//
//  private:
//#ifdef USE_PTHREAD_RUNTIME
//    pthread_mutex_t lock;
//#else
//    omp_lock_t lock;
//#endif
//}; /** end class Lock */
//
//

class Event
{
  public:

    Event(); 

    void Set( string, double, double );

    void AddFlopsMops( double, double );

    void Begin( size_t );

    double GetBegin();

    double GetEnd();

    double GetDuration();

    double GetFlops();

    double GetMops();

    double GflopsPerSecond();

    void Normalize( double shift );

    void Terminate();

    void Print();

    void Timeline( bool isbeg, size_t tag );

    void MatlabTimeline( FILE *pFile );

  private:

    size_t tid = 0;

     string label;

    double flops = 0.0;

    double mops = 0.0;

    double beg = 0.0;

    double end = 0.0;

    double sec = 0.0;

}; 
class Task
{
  public:

    Task();

    ~Task();

    class Worker *worker = NULL;

    string name;

    string label;

    int taskid;

    float cost = 0;

    bool priority = false;

    Event event;

    TaskStatus GetStatus();

    void SetStatus( TaskStatus status );
    void SetBatchStatus( TaskStatus status );

    void Submit();


    virtual void Set( string user_name, void (*user_function)(Task*), void *user_arg );

    virtual void Prefetch( Worker* );

    void Enqueue();

    void Enqueue( size_t tid );

    bool TryEnqueue();

    void ForceEnqueue( size_t tid );

    void CallBackWhileWaiting();

    virtual void Execute( Worker* ) = 0;

    virtual void GetEventRecord();

    virtual void DependencyAnalysis();

    /* function ptr */
    void (*function)(Task*);

    /* function context */
    void *arg;

    /* Dependencies related members */
    volatile int n_dependencies_remaining = 0;
    void DependenciesUpdate();

    deque<Task*> in;
    deque<Task*> out;

    void Acquire();
    void Release();
    
    Lock* task_lock = NULL;

    Task *next = NULL;

    //bool ContextSwitchToNextTask( Worker* );

    volatile bool stealable = true;

    volatile int created_by = 0;

    bool IsNested();

  private:

    volatile TaskStatus status;

    volatile bool is_created_in_epoch_session = false;

}; 
void hmlp_msg_dependency_analysis( 
    int key, int p, hmlp::ReadWriteType type, hmlp::Task *task );




template<typename ARGUMENT>
class NULLTask : public Task
{
  public:

    void Set( ARGUMENT *arg ) {};

    void Execute( Worker* ) {};

}; 
class MessageTask : public Task
{
  public:

    mpi::Comm comm;

    int tar = 0;
    int src = 0;
    int key = 0;

    MessageTask( int src, int tar, int key );

    void Submit();
}; 
class ListenerTask : public MessageTask
{
  public:

    ListenerTask( int src, int tar, int key );

    void Submit();

    virtual void Listen() = 0;
}; 
template<typename T, typename ARG>
class SendTask : public MessageTask
{
  public:
    
    ARG *arg = NULL;

    vector<size_t> send_sizes;
    vector<size_t> send_skels;
    vector<T>      send_buffs;

    SendTask( ARG *user_arg, int src, int tar, int key ) : MessageTask( src, tar, key )
    {
      this->arg = user_arg;
    };

    void Set( ARG *user_arg, int src, int tar, int key )
    {
      name = string( "Send" );
      label = to_string( tar );
      this->arg = user_arg;
      this->src = src;
      this->tar = tar;
      this->key = key;
      double flops = 0, mops = 0;
      event.Set( label + name, flops, mops );
      priority = true;
    };

    virtual void Pack() = 0;

    void Execute( Worker *user_worker ) 
    {
      Pack();
      mpi::Request req1, req2, req3;
      mpi::Isend( this->send_sizes.data(), this->send_sizes.size(),
          this->tar, this->key + 0, this->comm, &req1 ); 
      //mpi::Isend( this->send_skels.data(), this->send_skels.size(),
      //    this->tar, this->key + 1, this->comm, &req2 ); 
      mpi::Isend( this->send_buffs.data(), this->send_buffs.size(),
          this->tar, this->key + 2, this->comm, &req3 ); 
    };
}; 
template<typename T, typename ARG>
class RecvTask : public ListenerTask
{
   public:

    ARG *arg = NULL;

    vector<size_t> recv_sizes;
    vector<size_t> recv_skels;
    vector<T>      recv_buffs;

    RecvTask( ARG *user_arg, int src, int tar, int key ) 
      : ListenerTask( src, tar, key ) { this->arg = user_arg; };

    void Set( ARG *user_arg, int src, int tar, int key )
    {
      name = string( "Listener" );
      label = to_string( src );
      this->arg = user_arg;
      this->src = src;
      this->tar = tar;
      this->key = key;
      double flops = 0, mops = 0;
      event.Set( label + name, flops, mops );
    };

    void DependencyAnalysis()
    {
      hmlp_msg_dependency_analysis( this->key, this->src, RW, this );
    };

    void Listen()
    {
      int src = this->src;
      int tar = this->tar;
      int key = this->key;
      mpi::Comm comm = this->comm;
      int cnt = 0;
      mpi::Status status;
      mpi::Probe( src, key + 0, comm, &status );
      mpi::Get_count( &status, HMLP_MPI_SIZE_T, &cnt );
      recv_sizes.resize( cnt );
      mpi::Recv( recv_sizes.data(), cnt, src, key + 0, comm, &status );
      cnt = 0;
      for ( auto c : recv_sizes ) cnt += c;
      recv_buffs.resize( cnt );
      mpi::Recv( recv_buffs.data(), cnt, src, key + 2, comm, &status );
    };

    virtual void Unpack() = 0;

    void Execute( Worker *user_worker ) { Unpack(); };

}; 
template<typename ARG>
void RecuTaskSubmit( ARG *arg ) {  }; 


template<typename ARG, typename TASK, typename... Args>
void RecuTaskSubmit( ARG *arg, TASK& dummy, Args&... dummyargs )
{
  using NULLTASK = NULLTask<ARG>;
  if ( !std::is_same<NULLTASK, TASK>::value )
  {
    auto task = new TASK();
    task->Submit();
    task->Set( arg );
    task->DependencyAnalysis();
  }
  RecuTaskSubmit( arg, dummyargs... );
}; 
template<typename ARG>
void RecuTaskExecute( ARG *arg ) {  }; 


template<typename ARG, typename TASK, typename... Args>
void RecuTaskExecute( ARG *arg, TASK& dummy, Args&... dummyargs )
{
  using NULLTASK = NULLTask<ARG>;
  if ( !std::is_same<NULLTASK, TASK>::value )
  {
    auto *task = new TASK();
    task->Set( arg );
    task->Execute( NULL );
    delete task;
  }
  RecuTaskExecute( arg, dummyargs... );
}; 
class ReadWrite
{
  public:

    ReadWrite();

    deque<Task*> read;

    deque<Task*> write;

    void DependencyAnalysis( ReadWriteType type, Task *task );

    void DependencyCleanUp();

  private:

}; 
class MatrixReadWrite
{
  public:

    MatrixReadWrite();

    void Setup( size_t m, size_t n );

    bool HasBeenSetup();

    void DependencyAnalysis( size_t i, size_t j, ReadWriteType type, Task *task );

    void DependencyCleanUp();

  private:

    bool has_been_setup = false;

    size_t m = 0;

    size_t n = 0;

    vector<vector<ReadWrite> > Submatrices;

}; 
class Scheduler : public mpi::MPIObject
{
  public:

    Scheduler( mpi::Comm comm );
    
    ~Scheduler();

    void Init( int n_worker );

    void Finalize();

    int n_worker = 0;

    size_t timeline_tag;

    double timeline_beg;

    deque<Task*> ready_queue[ MAX_WORKER ];
    //deque<Task*> tasklist;
    Lock ready_queue_lock[ MAX_WORKER ];

    deque<Task*> nested_queue[ MAX_WORKER ];
    //deque<Task*> nested_tasklist;
    Lock nested_queue_lock[ MAX_WORKER ];


    vector<unordered_map<int, ListenerTask*>> listener_tasklist;
    Lock listener_queue_lock;

    float time_remaining[ MAX_WORKER ];

    void ReportRemainingTime();

    static void DependencyAdd( Task *source, Task *target );

    void MessageDependencyAnalysis( int key, int p, ReadWriteType type, Task *task );

    void NewTask( Task *task );

    void NewMessageTask( MessageTask *task );

    void NewListenerTask( ListenerTask *task );

    void ExecuteNestedTasksWhileWaiting( Worker *me, Task *waiting_task );

    void Summary();


  private:

    static void* EntryPoint( void* );

    deque<Task*> tasklist;

    deque<Task*> nested_tasklist;

    Lock tasklist_lock;

    int n_task_completed = 0;
    int n_nested_task_completed = 0;

    Lock n_task_lock;

    vector<Task*> DispatchFromNormalQueue( int tid );

    vector<Task*> DispatchFromNestedQueue( int tid );

    vector<Task*> StealFromOther();

    Task *StealFromQueue( size_t target );

    bool ConsumeTasks( Worker *me, vector<Task*> &batch );

    bool ConsumeTasks( Worker *me, Task *batch, bool is_nested );

    bool IsTimeToExit( int tid );

    void Listen( Worker* );

    Lock task_lock[ 2 * MAX_WORKER ];

    Lock run_lock[ MAX_WORKER ];

    Lock pci_lock;

    Lock gpu_lock;

    unordered_map<int, vector<ReadWrite>> msg_dependencies;

      bool do_terminate = false;

    bool has_ibarrier = false;

    mpi::Request ibarrier_request;

    int ibarrier_consensus = 0;
}; 
class RunTime
{
  public:

    RunTime();

    ~RunTime();

    hmlpError_t init( mpi::Comm comm );

    hmlpError_t init( int *argc, char ***argv, mpi::Comm comm );

    hmlpError_t run();

    hmlpError_t finalize();

    bool isInit() const noexcept;

    bool isInEpochSession() const noexcept;

    void ExecuteNestedTasksWhileWaiting( Task *waiting_task );

    hmlpError_t setNumberOfWorkers( int num_of_workers ) noexcept;

    int getNumberOfWorkers() const noexcept;

    thread_communicator *mycomm;

    class Worker workers[ MAX_WORKER ];

    class Device host;

    class Device* device[ 1 ];

    Scheduler *scheduler;

  private:

    int num_of_workers_ = 1;
    int num_of_max_workers_ = 1;
    int* argc = NULL;
    char*** argv = NULL;
    bool is_init_ = false;
    bool is_mpi_init_by_hmlp_ = false;
    bool is_in_epoch_session_ = false;
    void Print( string msg );
    void ExitWithError( string msg );

}; 
}; 
hmlp::RunTime *hmlp_get_runtime_handle();

hmlp::Device *hmlp_get_device( int i );

int hmlp_get_mpi_rank();

int hmlp_get_mpi_size();

//void hmlp_msg_dependency_analysis( 
//    int key, int p, hmlp::ReadWriteType type, hmlp::Task *task );


#endif