cbComputeGraph.hpp

Go to the documentation of this file.

 
#ifndef __SERVER_CB_COMPUTE_GRAPH_HPP_
#define __SERVER_CB_COMPUTE_GRAPH_HPP_
 
#ifdef _WIN32
#if _MSC_VER > 1000
#pragma once
#endif
#endif  
 
#if defined(__unix__) && defined(__clang__)
#pragma once
#endif  
 
#include <workflow/WFGraphTask.h>
#include <workflow/WFFacilities.h>
#include <workflow/WFTaskFactory.h>
 
#include "luaEngine.hpp"
#include "cbOperator.hpp"
#include "trivial/cbVirtualDevice.hpp"
 
typedef std::function<void(WFGraphTask*)> graph_callback;
 
namespace cb {
namespace graph {
 
class cbComputeGraph;
 
enum class nodeType : uint32_t {
  Leaf = 1,
  Output = 2,
  Operator = 3,
};
 
struct cbGraphSharedMem {
 public:
  cbGraphSharedMem() = default;
 
  cbGraphSharedMem(cbGraphSharedMem& rhs) = delete;
 
  cbGraphSharedMem(const cbGraphSharedMem& rhs) = delete;
 
  cbGraphSharedMem operator=(cbGraphSharedMem& rhs) = delete;
 
  cbGraphSharedMem operator=(const cbGraphSharedMem& rhs) = delete;
 
  ~cbGraphSharedMem();
 
  void push(cbVirtualSharedTable* v);
 
  void push(cbMySQLCell* v);
 
  void push(cbMySQLField* v);
 
  void setOutStruct(const cbShape<2>& shape, cbMySQLField** info);
 
  size_t getMemUsed();
 
  int32_t getCellNum();
 
  cbOutputTableStruct* getOutStruct();
 
  void clear();
 
 private:
  cbOutputTableStruct* m_outStruct = nullptr;
  std::vector<cbVirtualSharedTable*> m_dataFromDevice;
  std::vector<cbMySQLCell*> m_dataPool;
  std::vector<cbMySQLField*> m_fields;
};
 
struct cbGraphSharedLuaStack {
  cbGraphSharedLuaStack() = default;
 
  cbGraphSharedLuaStack(const cbGraphSharedLuaStack&) = delete;
 
  cbGraphSharedLuaStack operator=(const cbGraphSharedLuaStack&) = delete;
 
  void execScriptFile(const std::string& filePath);
 
  void execScript(const std::string& script);
 
  luaJitThread& get();
 
 private:
  luaJitThread m_lua;
};
 
struct cbNode {
  virtual ~cbNode(){};
  cbNode(const nodeType& nt);
  void PointTo(cbNode* ptr);
 
  virtual void* generateTask() = 0;
 
  static WFMySQLTask* asSQLTask(void* metaTask);
  static WFGoTask* asGoTask(void* metaTask);
 
  // The output.
  cbOpIO io;
 
  // others
  bool isFinalOutput = false;
  bool visited = false;
  nodeType nodeT;
  void* task = nullptr;
  cbNode* nextNode = nullptr;
  cbComputeGraph* graph = nullptr;
};
 
struct cbVirtualDeviceNode final : public cbNode {
  friend cbComputeGraph;
 
  ~cbVirtualDeviceNode() override final;
  cbVirtualDeviceNode();
 
  void* generateTask() override final;
 
  void addQuery(const std::string& q);
 
  trivial::cbMySqlDevice* getDevice();
 
 private:
  void setMySQLDevice(trivial::cbMySqlDevice* device = nullptr);
  trivial::cbMySqlDevice* m_device;
  std::vector<std::string> m_queries;
};
 
struct cbRedisCachingNode final : public cbNode {
  friend cbComputeGraph;
 
  ~cbRedisCachingNode() override final;
  cbRedisCachingNode() = delete;
  cbRedisCachingNode(int32_t idx);
  cbRedisCachingNode(const cbRedisCachingNode& v) = delete;
 
  cbRedisCachingNode operator=(const cbRedisCachingNode& v) = delete;
 
  void* generateTask() override final;
 
  WFRedisTask* _generateSetTask(const std::vector<std::string>& params,
                                const redis_callback& callback_func = nullptr,
                                void* usrData = nullptr, int32_t retryTimes = 3);
 
 private:
  void setRedisDevice(trivial::cbRedisDevice* device = nullptr);
  int32_t m_idx;
  trivial::cbRedisDevice* m_device = nullptr;
};
 
struct cbOperatorNode : public cbNode {
  friend cbComputeGraph;
 
  ~cbOperatorNode() override;
  cbOperatorNode(baseOp* op);
 
  void* generateTask() override;
 
  void overload(sol::function& funcPtr);
 
  baseOp* Op;
};
 
// operator Node;
 
class cbComputeGraph {
 public:
  friend cbVirtualDeviceNode;
  friend cbRedisCachingNode;
  friend cbOperatorNode;
 
  cbComputeGraph(int32_t idx);
  ~cbComputeGraph();
 
  bool isDAG();
 
  bool isSingleOutput();
 
  // create cells belongs to this graph.
  cbMySQLCell* createCell();
 
  cbMySQLCell* createCell(int value);
 
  cbMySQLCell* createCell(float value);
 
  cbMySQLCell* createCell(double value);
 
  cbMySQLCell* createCell(unsigned long long value);
 
  cbMySQLCell* createCell(const std::string& value);
 
  cbMySQLCell* createCell(const std::string& value, const cbMySQLType& t);
 
  cbVirtualDeviceNode* createVirtualDeviceNode(int32_t idx);  
 
  cbRedisCachingNode* createRedisCachingNode(int32_t idx);
 
  cbOperatorNode* createCombineNode(const std::vector<std::string>& keys, const std::string& name);
 
  cbOperatorNode* createFilterNode(const sol::function& boolF, const sol::function& exF);
 
  void setVirtualDeviceManager(trivial::cbVirtualDeviceManager* virtualDeviceM);
 
  void registerNode(cbNode* node);
 
  int32_t getId();
 
  WFGraphTask* generateGraphTask(const graph_callback& func = nullptr);
 
  static void execMain(WFGraphTask* task, cbComputeGraph* graph);
 
  void execScriptFile(const std::string& filePath);
 
  void execScript(const std::string& script);
 
  void addCacheServer(cbRedisCachingNode* v);
 
  cbOutputTableStruct* getOutput();
 
  std::vector<cbNode*> getNodes();
 
 private:
  int32_t m_idx = 0;
  cbOpIO io;
  std::vector<cbNode*> m_nodes;
  cbRedisCachingNode* m_cacheNode = nullptr;
  cbGraphSharedMem* m_sharedMem = nullptr;
  cbGraphSharedLuaStack* m_sharedLuaStack = nullptr;
  // Have the state of all Virtual device
  trivial::cbVirtualDeviceManager* m_virtualDevice = nullptr;
};
 
};  // namespace graph
}  // namespace cb
 
#endif  //!__SERVER_CB_COMPUTE_GRAPH_HPP_