logplus/Workflow/WFEngine/Module/include/WorkflowContext-.h

/**
 * @file   WorkflowContext.h
 * @brief  文件定义了与WorkflowContext相关的类
 * @author Ben
 * @date   2012-10-25
 */

#ifndef PAI_FRAME_MODULEAPI_WORKFLOW_CONTEXT_H
#define PAI_FRAME_MODULEAPI_WORKFLOW_CONTEXT_H

// #include "error.h"
#include "Turtle.h"
#include <string>
using std::string;

#include <map>
using std::map;

namespace pai
{
    namespace workflow
    {
        class ModuleContainer;
    }
}

// for test
class WorkflowContextTest;
namespace pai
{
namespace module
{
/**
 * WORKFLOW CONTEXT KEY- DATASETINFO
 */
const std::string WORKLFOWCONTEXT_KEY_DATASETINFO = "DataSetInfo";                         /**< WorkflowContext中文件头的key常量 */
const std::string KEY_DATABASE_ID = "database_id";                                         /**< WorkflowContext中database_id的key常量*/
const std::string KEY_CHAIN_ID = "submitJob.info.chainId";                                 /**< WorkflowContext中作业链id的key常量*/
const std::string KEY_HADOOP_JOB_ID = "pai.workflow.job.hadoop.id";                        /**< WorkflowContext中hadoopId的key常量*/
const std::string KEY_JOB_DB_ID = "pai.workflow.job.db.id";                                /**< WorkflowContext中工作流id的key常量*/
const std::string INPUT_GATHER_ORDER = "INPUT_GATHER_ORDER";                               /**< WorkflowContext中输入道集顺序的key常量*/
const std::string STACK_FLAG="STACK_FLAG";                                                 /**< WorkflowContext中Stack_flag的key常量*/
const std::string KEY_SURCON_MODULE_ISAPPLY = "pai.workflow.surcon.isapply";               /**< WorkflowContext中isapply的key常量*/
const std::string HADOOP_CONF_KEY_PRE_HADOOP_JOB_ID = "pai.workflow.pre.job.hadoop.id";    /**< WorkflowContext中上一个作业的hadoopId的key常量*/

const std::string RESETDATASETID = "resetDataSetID";                                       /**< WorkflowContext中resetDataSetID的key常量*/
const std::string CONTEXT_IDXSORTINGCODE="CONTEXT_IDXSORTINGCODE";                         /**< WorkflowContext中idxSortingCode的key常量*/
const std::string KEY_MAPRED_SPLIT="split";                                                /**< WorkflowContext中切片的key常量*/
/**
 * 用于多波模块中
 */
const std::string MC_FILEPATH_X="mc_filepath_x";											/**< WorkflowContext中三分量中x分量的文件*/
const std::string MC_FILEPATH_Y="mc_filepath_y";											/**< WorkflowContext中三分量中y分量的文件*/
const std::string MC_FILEPATH_Z="mc_filepath_z";											/**< WorkflowContext中三分量中z分量的文件*/
/**
 * @class Placeholder
 * @brief 占位符基类
 */
class Placeholder
{
public:
	/**
	 * @brief 虚析构函数
	 */
    virtual ~Placeholder()
    {
    }
    /**
     * @brief 克隆函数
     */
    virtual Placeholder * Clone() const = 0;
};
/**
 * @class Holder
 * @brief 占位符实现类，使用模板方式实现，具体存储类型由T决定
 */
template<typename T> class Holder: public Placeholder
{
public:
	/**
	 * @brief 构造函数
	 */
    Holder(const T & t) :
        m_held(t)
    {
    }
    /**
     * @brief 克隆函数
     */
    virtual Placeholder * Clone() const
    {
        return new Holder(m_held);
    }
    /**
     * @brief 获取<code>Holder</code>中的值
     * @return <code>Holder</code>中的值
     */
    T Get()
    {
        return m_held;
    }
private:
    Holder & operator=(const Holder&);//unimplemented
    T m_held;
};
/**
 * @class Any
 * @brief 代表Key-Value中的Value类型，在创建和获取value时，使用模板方法获得具体类型
 */
class Any
{
public:
	/**
	 * @brief 构造函数
	 */
    Any() :
        m_content(NULL)
    {
    }
    /**
     * @brief 构造函数
     */
    template<typename T> Any(const T & t) :
        m_content(new Holder<T> (t))
    {
    }
    /**
     * @brief 拷贝构造函数
     */
    Any(const Any& other) :
        m_content(other.m_content ? other.m_content->Clone() : NULL)
    {
    }
    /**
     * @brief 析构函数
     */
    ~Any()
    {
    	if (m_content != NULL)
    	{
    		delete m_content;
    		m_content = NULL;
    	}
    }
    /**
     * @brief 用参数中的<code>Any</code>的值交换
     * @return 交换后的对象
     */
    Any & swap(Any & rhs)
    {
        std::swap(m_content, rhs.m_content);
        return *this;
    }
    /**
     * @brief 重载=运算符
     */
    template<typename T> Any& operator=(const T & rhs)
    {
        anything(rhs).swap(*this);
        return *this;
    }
    /**
     * @brief 重载=运算符
     */
    Any & operator=(Any rhs)
    {
        rhs.swap(*this);
        return *this;
    }
    /**
     * @brief 判断值是否为空
     * @return 为空返回true，否则返回false
     */
    bool empty() const
    {
        return !m_content;
    }
    /**
     * @brief 获取<code>Any</code>中的值
     */
    template<typename T> T Get()
    {
        return (dynamic_cast<Holder<T>*> (m_content))->Get();
    }
private:
    Placeholder * m_content;
};

/**
 * @class Context
 * @brief 基于Key-Value的上线文容器
 * <p>注意为了实现简单，易于维护，没有释放内部的资源,所以不要在一个进程中频繁申请此Context,否则会存在大量的内存泄漏
 */
class PAI_MODULE_EXPORT Context
{

public:
	/**
	 * @brief 构造函数
	 */
	Context() : m_container(new map<string, Any>())
    {
		init ();
    }
	/**
	 * 拷贝构造函数
	 */
	Context(const Context& another) : m_container(another.m_container)
    {
		init ();
    }
	/**
	 * @brief 重载=运算符
	 */
	Context& operator=(const Context& rhs)
    {
		if (this == &rhs)
		   return *this;
		pthread_mutex_lock(&m_Locker);
		if (m_container != NULL)
		{
			delete m_container;
			m_container = NULL;
		}
        m_container = rhs.m_container;
        pthread_mutex_unlock(&m_Locker);
        return *this;
    }
    /**
     * TODO 	   存在内存泄漏
     * @brief 析构方法
     * @remark 本方法中没有释放内部的资源，这样做的原因是
     * - 需要释放的时机通常为工作流结束后，此时操作系统会自动释放资源
     * - 实现简单，易于维护
     */
	virtual ~Context(){
		pthread_mutex_destroy(&m_Locker);
	}
	/**
	 * @brief 向<code>Context</code>中放入键值对
	 * @param[in] key 键
	 * @param[in] anything 值
	 */
    void Put(const string& key, Any anything)
    {
    	pthread_mutex_lock(&m_Locker);
        (*m_container)[key] = anything;
        pthread_mutex_unlock(&m_Locker);
    }
    /**
     * @brief 从<code>Context</code>中获取指定key的值
     * @param[in] key 键
     * @return 键值为key的值
     */
     template<typename T>
    T  Get(const string& key)
    {
    	pthread_mutex_lock(&m_Locker);
        if (m_container->count(key) == 0)
        {
        	pthread_mutex_unlock(&m_Locker);
            // throw pai::error::key_not_found_error("Key [" + key + "] not found.");
        }
        T rest =(*m_container)[key].Get<T> ();
        pthread_mutex_unlock(&m_Locker);
        return rest;
    }
     /**
      * @brief 判断<code>Context</code>中是否包含指定key的键值对
      * @param[in] key 键
      * @return 包含则返回true，否则返回false
      */
     bool ContainsKey(const string& key)
	{
		pthread_mutex_lock(&m_Locker);
		bool rest = false;
	   if (m_container->count(key) == 0)
		   rest = false;
	   else
		   rest = true;
	   pthread_mutex_unlock(&m_Locker);
	   return rest;
	}

private:
	void init ()
	{
		pthread_mutex_init(&m_Locker,NULL);
	}
    map<string, Any> * m_container;
    pthread_mutex_t m_Locker;
};

/**
 * @class BufferContext
 * @brief 模块之间的Buffer上下文信息
 */
class BufferContext:public Context
{
public:
	/**
	 * @brief 构造函数
	 */
	BufferContext() : Context()
    {
    }
	/**
	 * @brief 拷贝构造函数
	 */
	BufferContext(const BufferContext& another) : Context(another)
    {
    }
	/**
	 * @brief 重载=运算符
	 */
	BufferContext& operator=(const BufferContext& rhs)
    {
        if (this == &rhs)
           return *this;
        Context::operator=(rhs);
        return *this;
    }
    /**
     * @brief 析构函数
     * @remark 本方法中没有释放内部的资源，这样做的原因是
     * - 需要释放的时机通常为工作流结束后，此时操作系统会自动释放资源
     * - 实现简单，易于维护
     */
	virtual ~BufferContext(){}
};

/**
 * @class WorkflowContext
 * @brief 工作流全局上下文，整个工作流只有一份
 * <p>工作流中的模块通过WorkflowContext获取全局性的信息以及共享信息.
 */
class WorkflowContext:public Context
{
private:
	friend class ::WorkflowContextTest;//for test
	friend class pai::workflow::ModuleContainer;
	//module 所在的分支
	std::vector<std::string> branchs;
	/**
	 * need_branch == true，模块所在的分支都存储一份，模块中使用put操作期望这样的结果
	 * 在内存中以(branch-id)-key 》 anything保存多分
	 *
	 * need_branch == false，整个工作流中保存一份，非模块中使用期望这样的操作
	 * 在内存中以key 》 anything保存多分
	 */
	void Put(const string& key, Any anything, bool need_branch)
	{
		if(need_branch)
		{
			if(branchs.size() == 0){
                throw /*pai::error::runtime_error*/(" Illegal use of context or program error!");
			}
			for(std::vector<std::string>::iterator it = branchs.begin(); it != branchs.end(); it++)
			{
				std::string key_tmp = (*it) + "-" + key;
				Context::Put(key_tmp, anything);
			}
		}
		else
		{
			Context::Put(key, anything);
		}
	}
public:
	/**
	 * @brief 构造函数
	 */
	WorkflowContext() : Context(), branchs()
    {
    }
	/**
	 * @brief 拷贝构造函数
	 */
	WorkflowContext(const WorkflowContext& another) : Context(another), branchs(another.branchs)
    {
    }
	/**
	 * @brief 向<code>WorkflowContext</code>中存放键值对
	 * @param[in] key 键
	 * @param[in] anything 值
	 */
	void Put(const string& key, Any anything)
	{
		Put(key, anything, true);
	}
	/**
	 * @brief 获取<code>WorkflowContext</code>中指定键的值
	 * <p>做get操作时，首先通过常规方式获取，如果没有则从分支中获取，如果没有，抛出异常
	 * 如果一个模块存在在多个分支中，选取第一个找到的
	 * @param[in] key 键
	 * @return 值
	 */
    // template<typename T>
    // T Get(const string& key)
    // {
    // 	try{
    // 		return Context::Get<T>(key);
    // 	}catch(pai::error::key_not_found_error& e){
    // 		for(std::vector<std::string>::iterator it = branchs.begin(); it != branchs.end(); it++)
    // 		{
    // 			std::string key_tmp = (*it) + "-" + key;
    // 			try{
    // 				return Context::Get<T>(key_tmp);
    // 			}catch(pai::error::key_not_found_error& e){
    // 				continue;
    // 			}
    // 		}
    // 	}
    // 	throw pai::error::key_not_found_error("Key [" + key + "] not found.");
    // }
	/**
	 * @brief 判断<code>WorkflowContext</code>中是否包含指定key的键值对
	 * @param[in] key 键
	 * @return 包含则返回true，否则返回false
	 */
	bool ContainsKey(const string& key)
	{
		bool res = Context::ContainsKey(key);
		for(std::vector<std::string>::iterator it = branchs.begin(); it != branchs.end(); it++)
		{
			const std::string key_tmp = (*it) + "-" + key;
			res = (res || Context::ContainsKey(key_tmp));
		}
	   return res;
	}
	/**
	 * @brief 重载=运算符
	 */
	WorkflowContext& operator=(const WorkflowContext& rhs)
    {
        if (this == &rhs)
           return *this;
        branchs = rhs.branchs;
        Context::operator=(rhs);
        return *this;
    }
    /**
     * @brief 析构方法
     * @remark 本方法中没有释放内部的资源，这样做的原因是
     * - 需要释放的时机通常为工作流结束后，此时操作系统会自动释放资源
     * - 实现简单，易于维护
     */
	virtual ~WorkflowContext(){}
};
}
}

#endif