lp.h

Go to the documentation of this file.

#ifndef _LP_H
#define _LP_H

#include "lp.decl.h"

#include "event.h"
#include "pending_heap.h"
#include "processed_queue.h"
#include "random.h"
#include "typedefs.h"

#include <vector>

extern CProxy_LPChare lps;

class LPBase;
class RemoteEvent;
class Scheduler;

using std::vector;

struct LPToken {
  private:
    LPChare* lp;    
    Time ts;        
    uint32_t index; 

  public:
    LPToken() : lp(NULL), ts(0), index(0) {}

    LPToken(LPChare* lp) : lp(lp), ts(0), index(0) {}

    friend class PEQueue;
    friend class Scheduler;
    friend class DistributedScheduler;
    friend class OptimisticScheduler;
};

class LPChare : public CBase_LPChare {
  private:
    LPToken next_token;
    Time current_time;    
    Event* current_event; 

    uint64_t next_event_id;

    PendingHeap events;

    ProcessedQueue processed_events;

    vector<LPBase*> lp_structs;

    Scheduler* scheduler;

    bool isOptimistic;

    int committed_events;   
    int rolled_back_events; 
    int committed_time;     
//@{
    Event *cancel_q;    
    Time min_cancel_q;  

  public:
    AvlTree all_events;

    LPChare();
    LPChare(CkMigrateMessage* m);

    // TODO: Clean up current/event and time. Probably store a sentinel event
    // in processed queue that can be used when rolling back to a point where
    // we have no history. Also have the initial event be not the abort event,
    // probably by again having a sentinel in the processed queue.
    uint64_t get_next_event_id() { return next_event_id++; }
    Event* get_current_event() const { return current_event; }
    Time get_current_time() const { return current_time; }
    void set_current_event(Event* e) {
      current_event = e;
      if (current_event == NULL) {
        current_time = PE_VALUE(g_last_gvt);
      } else {
        current_time = current_event->ts;
      }
    }

    void stop_scheduler();

    void init();
    void finalize();
    int execute_me();
    void fossil_me(Time gvt);


    void reconstruct_causality(Event*, Event**, Event**);
    void reconstruct_pending_event(Event*);
    void reconstruct_processed_event(Event*, Event**, Event**);
    virtual void pup(PUP::er &p);
    void load_balance();
    void ResumeFromSync();
    void UserSetLBLoad();


    void recv_remote_event(RemoteEvent* event);
    void recv_anti_event(RemoteEvent* event);
    void recv_local_event(Event* e);


    void rollback_me(Time ts);
    void rollback_me(Event* event);
    void cancel_event(Event* event);
    void delete_pending(Event* event);
    void add_to_cancel_q(Event* e);
    void process_cancel_q();
};

class LPBase {
  public:
    LPChare* owner;
    unsigned gid;
    RNG rng;

    virtual void initialize() {}
    virtual void forward(Event* e) {}
    virtual void reverse(Event* e) {}
    virtual void commit(Event* e) {}
    virtual void finalize() {}

    virtual bool compare(const Event* e1, const Event* e2) {
      if (e1->src_lp != e2->src_lp) {
        return e1->src_lp < e2->src_lp;
      } else {
        return e1->event_id < e2->event_id;
      }
    }

    void set_current_event(Event* e) { owner->set_current_event(e); }
    Event* get_current_event() const { return owner->get_current_event(); }
    Time get_current_time() const { return owner->get_current_time(); }

    virtual void pup(PUP::er& p) {
      p | gid;
      p | rng;
    }
};

template <typename Derived, typename M, typename... Ms>
class LP : public LPBase {
  private:
    static bool registered;
    static vector<DispatcherBase<Derived>*> dispatchers;

    template <typename M1>
    static void register_msg_type() {
      dispatchers[get_msg_id<M1>()] = new Dispatcher<Derived, M1>();
    }

    template <typename M1, typename M2, typename... M3>
    static void register_msg_type() {
      register_msg_type<M1>();
      register_msg_type<M2, M3...>();
    }

  public:
    LP() {
      dispatchers.resize(g_num_msg_types);
      if (!registered) {
        register_msg_type<M, Ms...>();
        registered = true;
      }
    }

    void forward(Event* e) {
      dispatchers[e->type_id]->forward(static_cast<Derived*>(this), e);
    }
    void reverse(Event* e) {
      dispatchers[e->type_id]->reverse(static_cast<Derived*>(this), e);
    }
    void commit(Event* e) {
      dispatchers[e->type_id]->commit(static_cast<Derived*>(this), e);
    }
};

template <typename Derived, typename M, typename... Ms>
vector<DispatcherBase<Derived>*> LP<Derived, M, Ms...>::dispatchers;

template <typename Derived, typename M, typename... Ms>
bool LP<Derived, M, Ms...>::registered = false;

void init_lps();

#endif