PsySpace.h

Go to the documentation of this file.

#if !defined(_PSYSPACE_H_)
#define _PSYSPACE_H_
 
#include "MemoryManager.h"
#include "PsyTime.h"
#include "xml_parser.h"
#include "DataMessage.h"
#include "Subscriptions.h"
 
namespace cmlabs {
 
class PsyAPI;
typedef int8 (* CrankFunction)(PsyAPI* api);

#define LOG_SPACE 7

class TimeQueueSchedule {
public:
    TimeQueueSchedule(uint32 interval, uint64 start = 0, uint64 end = 0);
    ~TimeQueueSchedule();
 
    uint32 id;          
    uint64 nextTrigger; 
 
    uint32 interval;    
    uint64 start;       
    uint64 end;         
 
    PsyType msgType;    
    uint32 msgTag;      
};

class TimeQueue {
public:
    TimeQueue();
    ~TimeQueue();

    bool addSchedule(TimeQueueSchedule* schedule);
    bool removeSchedule(uint32 id);

    DataMessage* waitForNextEvent(uint32 ms);
 
private:
    utils::Timer* timer;
    std::map<uint32, TimeQueueSchedule*> schedules;
 
    utils::Mutex mutex;
    utils::Semaphore semaphore;
};

struct SignalStruct {
    DataMessage* lastSignalMsg; 
    utils::Event event;         
    utils::Mutex mutex;         
};

#define PROCMESSAGE     0 
#define PROCSIGNAL      1 
#define PROCREQUEST     2 

class PsySpace : public Runnable, public LogReceiver {
    friend class PsyAPI;
public:
    friend THREAD_RET THREAD_FUNCTION_CALL PsySpaceRun(THREAD_ARG arg);
    friend THREAD_RET THREAD_FUNCTION_CALL PsySpacePoolRun(THREAD_ARG arg);
    friend THREAD_RET THREAD_FUNCTION_CALL PsySpaceContinuousRun(THREAD_ARG arg);
 
    // ***************** Constructors *****************
    // Create and destroy resources

    PsySpace(const char* name = NULL, bool isAdHoc = true, uint16 procID = 0, bool isLocal = false);
    ~PsySpace();

    bool connect(uint16 systemID, bool isMaster = false, const char* cmdline = NULL);
    bool isConnected(uint32 timeoutMS = 5000);
    bool reset();
    bool start(uint16 threadCount = 5);
    bool shutdown();
    bool hasShutdown();
    uint16 getID();

    bool postMessage(DataMessage* msg);
 
    // ***************** Signals *****************
    bool emitSignal(const PsyType& type, DataMessage* msg);
    DataMessage* waitForSignal(const PsyType& type, uint32 timeout, uint64 lastReceivedTime = 0);

    uint8 query(DataMessage* msg, DataMessage** result, uint32 timeout);
    bool queryReply(uint32 id, uint8 status, DataMessage* result);

    uint32 getComponentID(const char* name);
//  PsyAPI* getComponentAPI(const char* name);
//  PsyAPI* getComponentAPI(uint32 compID);
//  bool registerComponentCallback(const char* name, CrankFunction func);
//  bool registerComponentCallback(uint32 compID, CrankFunction func);

    PsyAPI* getCrankAPI(const char* name);
    bool registerCrankCallback(const char* name, CrankFunction func);

    bool registerInternalCrank(const char* name, CrankFunction func);

    bool addPsyProbeCustomView(uint32 compID, const char* name, const char* templateURL);

    bool logEntry(LogEntry* entry);

    MemoryManager* manager;
private:
 
    std::map<std::string, CrankFunction> internalCranks;
 
    bool threadPoolDispatch();
    bool runContinuousComponent();
    bool startContinuousComponent(uint32 compID);
    bool setThreadPoolSize(uint16 threadCount);
    CrankFunction loadCrankFromLibrary(const char* crankName, const char* libraryFilename);
 
    bool pullRemoteComponentData(uint32 compID, uint16 fromNodeID);
 
    bool run();
 
    std::string name;
    bool isLocal;
    bool isMaster;
    bool isAdHoc;
    bool finishedShuttingDown;
    uint64 currentInstID;
 
    uint16 procID;
    uint64 masterCreatedTime;
 
    uint32 waitCounters[3];
    uint32 procCounters[3];
 
    uint32 threadPoolWaitLowerThreshold;
    uint32 threadPoolWaitUpperThreshold;
    uint32 threadPoolWaitIncrement;
    uint64 lastThreadPoolCheck;
    uint64 threadPoolCheckInterval;
    utils::Mutex threadPoolMutex;
 
    TimeQueue* timeQ;
 
    uint16 threadTarget;
 
    std::map<uint16, CrankFunction> cranks;
    std::map<uint32, uint16> compCrankIDs;
    std::map<std::string, utils::Library*> libraries;
    std::map<uint32, PsyAPI*> psyAPIs;
    std::map<PsyType, SignalStruct*> signalList;
    utils::Mutex signalsMutex;
 
    std::map<uint32, uint8> threadPool;
    std::map<uint32, uint32> continuousComponentThreads;
};

THREAD_RET THREAD_FUNCTION_CALL PsySpaceRun(THREAD_ARG arg);
THREAD_RET THREAD_FUNCTION_CALL PsySpacePoolRun(THREAD_ARG arg);
THREAD_RET THREAD_FUNCTION_CALL PsySpaceContinuousRun(THREAD_ARG arg);
 
} // namespace cmlabs
 
#include "PsyAPI.h"
#include "PsyInternal.h"
 
#endif //_PSYSPACE_H_