/** * @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 using std::string; #include 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 class Holder: public Placeholder { public: /** * @brief 构造函数 */ Holder(const T & t) : m_held(t) { } /** * @brief 克隆函数 */ virtual Placeholder * Clone() const { return new Holder(m_held); } /** * @brief 获取Holder中的值 * @return Holder中的值 */ 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 Any(const T & t) : m_content(new Holder (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 用参数中的Any的值交换 * @return 交换后的对象 */ Any & swap(Any & rhs) { std::swap(m_content, rhs.m_content); return *this; } /** * @brief 重载=运算符 */ template 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 获取Any中的值 */ template T Get() { return (dynamic_cast*> (m_content))->Get(); } private: Placeholder * m_content; }; /** * @class Context * @brief 基于Key-Value的上线文容器 *

注意为了实现简单,易于维护,没有释放内部的资源,所以不要在一个进程中频繁申请此Context,否则会存在大量的内存泄漏 */ class PAI_MODULE_EXPORT Context { public: /** * @brief 构造函数 */ Context() : m_container(new map()) { 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 向Context中放入键值对 * @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 从Context中获取指定key的值 * @param[in] key 键 * @return 键值为key的值 */ template 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 (); pthread_mutex_unlock(&m_Locker); return rest; } /** * @brief 判断Context中是否包含指定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 * 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 工作流全局上下文,整个工作流只有一份 *

工作流中的模块通过WorkflowContext获取全局性的信息以及共享信息. */ class WorkflowContext:public Context { private: friend class ::WorkflowContextTest;//for test friend class pai::workflow::ModuleContainer; //module 所在的分支 std::vector 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::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 向WorkflowContext中存放键值对 * @param[in] key 键 * @param[in] anything 值 */ void Put(const string& key, Any anything) { Put(key, anything, true); } /** * @brief 获取WorkflowContext中指定键的值 *

做get操作时,首先通过常规方式获取,如果没有则从分支中获取,如果没有,抛出异常 * 如果一个模块存在在多个分支中,选取第一个找到的 * @param[in] key 键 * @return 值 */ // template // T Get(const string& key) // { // try{ // return Context::Get(key); // }catch(pai::error::key_not_found_error& e){ // for(std::vector::iterator it = branchs.begin(); it != branchs.end(); it++) // { // std::string key_tmp = (*it) + "-" + key; // try{ // return Context::Get(key_tmp); // }catch(pai::error::key_not_found_error& e){ // continue; // } // } // } // throw pai::error::key_not_found_error("Key [" + key + "] not found."); // } /** * @brief 判断WorkflowContext中是否包含指定key的键值对 * @param[in] key 键 * @return 包含则返回true,否则返回false */ bool ContainsKey(const string& key) { bool res = Context::ContainsKey(key); for(std::vector::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