-
Notifications
You must be signed in to change notification settings - Fork 0
/
KvServer.cpp
138 lines (99 loc) · 3.39 KB
/
KvServer.cpp
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
126
127
128
129
130
131
132
133
134
135
136
137
138
#include "include/HashTable.hpp"
#include "include/IpcEndpoint.hpp"
#include "include/ComProtocol.hpp"
#include "third_party/cxxopts.hpp"
#include <thread>
#include <list>
#include <iostream>
#include <cstring>
#include <cassert>
int main(int argc, const char* argv[]) {
cxxopts::Options options(argv[0], "Key Value server");
options
.allow_unrecognised_options()
.add_options()
("s, size", "size of HashTable", cxxopts::value<size_t>())
("h, help", "Help");
auto args = options.parse(argc, argv);
if (args.count("help")) {
std::cout << options.help() << std::endl;
return 0;
}
if (!args.count("size"))
{
std::cout << "-s arg is required" << std::endl;
return 1;
}
size_t hashSize = args["size"].as<size_t>();
cskv::HashTable table(hashSize);
auto workerFunc = [&table](std::string_view clientEndpoint) {
auto endpoint = cskv::IpcEndpoint(clientEndpoint);
bool exit = false;
while(not exit) {
cskv::protocol::KvRequest req;
cskv::protocol::KvResponse res;
auto readLen = endpoint.readReq(&req, sizeof(req));
if (not readLen or readLen != sizeof(req)) {
std::cerr << "error reading req" << std::endl;
continue;
}
switch (req.cmd)
{
case cskv::protocol::KvCommand::FINISH:
res.success = true;
exit = true;
break;
case cskv::protocol::KvCommand::INSERT:
res.success = table.insert(req.key, std::string(req.value, req.len));
break;
case cskv::protocol::KvCommand::DEL:
res.success = true;
table.del(req.key);
break;
case cskv::protocol::KvCommand::GET:
{
auto value = table.get(req.key);
if (not value) {
res.success = false;
break;
}
res.len = value.value().length();
assert(res.len < cskv::protocol::VALUE_SIZE);
std::memcpy(res.value, value.value().c_str(), res.len);
res.success = true;
}
break;
default:
std::cerr << "invalid cmd" << std::endl;
break;
}
auto writtenLen = endpoint.writeRes(&res, sizeof(res));
writtenLen = 0;
}
};
std::list<std::thread> workerThreads;
auto listenServer = cskv::IpcEndpoint(cskv::protocol::LISTEN_SERVER, true);
bool exit = false;
while (not exit) {
cskv::protocol::ConnectionRequest req;
auto readLen = listenServer.readReq(&req, sizeof(req));
if (not readLen or readLen != sizeof(req)) {
std::cerr << "error reading req" << std::endl;
continue;
}
std::string endpointName = std::string(req.name, req.len);
if (endpointName.compare(cskv::protocol::SPECIAL_EXIT_SEQUENCE) == 0) {
exit = true;
}
else {
workerThreads.emplace_back(workerFunc, endpointName);
}
cskv::protocol::ConnectionResponse res;
res.success = true;
auto writtenLen = listenServer.writeRes(&res, sizeof(res));
}
for (auto& t : workerThreads) {
t.join();
}
return 0;
}