tableCache/TableCacheProcessor.h at master · highras/tableCache · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
#ifndef Table_Cache_Processor_H
#define Table_Cache_Processor_H

#include <atomic>
#include <unordered_map>
#include "jenkins.h"
#include "hashint.h"
#include "TableRow.h"
#include "LruHashMap.h"
#include "RWLocker.hpp"
#include "IQuestProcessor.h"
#include "ClusterNotifier.h"

using namespace fpnn;

struct TableKey
{
	int64_t hintId;
	std::string tableName;

	bool operator == (const struct TableKey& key)
	{
		return this->hintId == key.hintId && this->tableName == key.tableName;
	}

	bool operator < (const struct TableKey& right) const
	{
		if(this->hintId != right.hintId)
			return this->hintId < right.hintId;
		return this->tableName < right.tableName;
	}

	unsigned int hash() const
	{
		uint32_t val = hash32_uint64(hintId);
		return jenkins_hash(tableName.data(), tableName.length(), val);
	}
};

class WriteCallback;
template<typename TYPE>
class FetchRowCallback;

struct FetchStatistics
{
	std::atomic<uint64_t> fetchCount;
	std::atomic<uint64_t> partHitCount;
	std::atomic<uint64_t> fullHitCount;

	std::atomic<uint64_t> itemFetchCount;
	std::atomic<uint64_t> itemHitCount;

	FetchStatistics(): fetchCount(0), partHitCount(0), fullHitCount(0), itemFetchCount(0), itemHitCount(0) {}
};

class TableCacheProcessor: virtual public IQuestProcessor, virtual public std::enable_shared_from_this<TableCacheProcessor>
{
	QuestProcessorClassPrivateFields(TableCacheProcessor)

	ClusterNotifierPtr _clusterNotifier;
	TCPClientPtr _dbproxy;

	RWLocker _rwlocker;
	std::unordered_map<std::string, TABLEPtr> _tableInfo;

	typedef LruHashMap<TableKey, ROWPtr> CacheMap;
	typedef std::shared_ptr<CacheMap> CacheMapPtr;
	CacheMapPtr _cachaMap;

	std::unordered_map<std::string, std::set<CacheMap::node_type*>> _tableDataIndexes;

	FetchStatistics _statistics;

	void configure();
	bool loadTableScheme(const std::string& tableName, std::vector<std::vector<std::string>>& scheme);
	std::string loadSplitColumn(const std::string& tableName);
	TABLEPtr loadTableInfo(const std::string& tableName);
	TABLEPtr getTableScheme(const std::string& tableName);
	void cleanCache(const std::string& tableName, int64_t hintId);

	FPAnswerPtr real_fetch(const FPQuestPtr quest, const std::string& tableName, TABLEPtr scheme,
		const std::vector<std::string>& fields, const std::set<int64_t>& hintIds);
	FPAnswerPtr real_fetch(const FPQuestPtr quest, const std::string& tableName, TABLEPtr scheme,
		const std::vector<std::string>& fields, const std::set<std::string>& hintStrings);

	FPAnswerPtr real_fetch_from_database(const FPQuestPtr quest,
		const std::string& tableName, TABLEPtr scheme, std::vector<uint16_t>& fieldIndexes,
		const std::set<int64_t>& lackedHintIds, std::map<int64_t, std::vector<std::string>>& result, bool jsonCompatible);
	FPAnswerPtr real_fetch_from_database(const FPQuestPtr quest,
		const std::string& tableName, TABLEPtr scheme, std::vector<uint16_t>& fieldIndexes,
		const std::set<std::string>& lackedHintStrings, std::map<std::string, std::vector<std::string>>& result);

	friend class WriteCallback;
	friend class FetchRowCallback<int64_t>;
	friend class FetchRowCallback<std::string>;

	void addRows(TABLEPtr orginalScheme, const std::vector<std::vector<std::string>>& data);

public:
	FPAnswerPtr modify(const FPReaderPtr args, const FPQuestPtr quest, const ConnectionInfo& ci);
	FPAnswerPtr fetch(const FPReaderPtr args, const FPQuestPtr quest, const ConnectionInfo& ci);
	FPAnswerPtr deleteData(const FPReaderPtr args, const FPQuestPtr quest, const ConnectionInfo& ci);
	FPAnswerPtr invalidateTable(const FPReaderPtr args, const FPQuestPtr quest, const ConnectionInfo& ci);
	FPAnswerPtr refreshCluster(const FPReaderPtr args, const FPQuestPtr quest, const ConnectionInfo& ci);
	FPAnswerPtr invalidate(const FPReaderPtr args, const FPQuestPtr quest, const ConnectionInfo& ci);

	virtual std::string infos();

	TableCacheProcessor()
	{
		registerMethod("modify", &TableCacheProcessor::modify);
		registerMethod("fetch", &TableCacheProcessor::fetch);
		registerMethod("delete", &TableCacheProcessor::deleteData);
		registerMethod("invalidateTable", &TableCacheProcessor::invalidateTable);
		registerMethod("refreshCluster", &TableCacheProcessor::refreshCluster);
		registerMethod("invalidate", &TableCacheProcessor::invalidate);

		configure();
	}

	QuestProcessorClassBasicPublicFuncs
};
typedef std::shared_ptr<TableCacheProcessor> TableCacheProcessorPtr;

#endif