Skip to content
This repository has been archived by the owner on Apr 1, 2023. It is now read-only.

Cleanup idle transactions after timeout #294

Closed
wants to merge 5 commits into from
Closed

Cleanup idle transactions after timeout #294

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ee52773
Cleanup idle transactions after timeout, associate queries with trans…
May 4, 2022
d6ab5d0
Implement using single periodic timer
May 9, 2022
a139c81
Return iterator from find
May 10, 2022
55544f3
Add more check
May 10, 2022
1cc17e8
Add check before erase
May 10, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/k2/cmd/httpproxy/http_proxy_main.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ int main(int argc, char** argv) {
("partition_request_timeout", bpo::value<k2::ParseableDuration>(), "Timeout of K23SI operations, as chrono literals")
("cpo", bpo::value<k2::String>(), "URL of Control Plane Oracle (CPO), e.g. 'tcp+k2rpc://192.168.1.2:12345'")
("cpo_request_timeout", bpo::value<k2::ParseableDuration>(), "CPO request timeout")
("cpo_request_backoff", bpo::value<k2::ParseableDuration>(), "CPO request backoff");
("cpo_request_backoff", bpo::value<k2::ParseableDuration>(), "CPO request backoff")
("httpproxy_txn_timeout", bpo::value<k2::ParseableDuration>(), "HTTP Proxy Txn idle timeout");
return app.start(argc, argv);
}
160 changes: 160 additions & 0 deletions src/k2/common/MapWithExpiry.h
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,160 @@
/*
MIT License

Copyright(c) 2022 Futurewei Cloud

Permission is hereby granted,
free of charge, to any person obtaining a copy of this software and associated documentation files(the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and / or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions :

The above copyright notice and this permission notice shall be included in all copies
or
substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS",
WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
DAMAGES OR OTHER
LIABILITY,
WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/

#pragma once
#include "Timer.h"

namespace k2 {
namespace nsbi = boost::intrusive;

// Utility map class that also keeps a timestamp ordered elements list
template <class KeyT, class ValT> class TimestampOrderedMap {
public:
class MapElem {
public:
ValT& getValue() {return entry;}
TimePoint timestamp() {return ts;}
KeyT key;
ValT entry;
TimePoint ts;
nsbi::list_member_hook<> tsLink;
};

typedef typename std::unordered_map<KeyT, MapElem>::iterator iterator;
typedef nsbi::list<MapElem, nsbi::member_hook<MapElem, nsbi::list_member_hook<>, &MapElem::tsLink>> TsOrderedList;


auto insert(const KeyT &key, ValT&& val, TimePoint ts) {
auto pair = elems.emplace(key, MapElem {
.key = key,
.entry = std::move(val),
.ts = ts,
.tsLink = {},
});
MapElem& elem = pair.first->second;
tsOrderedList.push_back(elem);
}

// TODO: Implement using boost iterator helpers to reduce boilerplate code.
iterator end() {return elems.end();}
iterator begin() { return elems.begin();}
iterator find(const KeyT &key) {return elems.find(key);}
ValT& at(const KeyT &key) {return elems.at(key).getValue();}
auto size() {return elems.size();}
void unlink(iterator iter) {
tsOrderedList.erase(tsOrderedList.iterator_to(iter->second));
}
auto extract(iterator iter) {
unlink(iter);
// Remove from map
return elems.extract(iter);
}
auto erase(const KeyT &key) {
iterator iter = elems.find(key);
unlink(iter);
return elems.erase(iter);
}

void resetTs(iterator iter, TimePoint ts) {
// new TS needs to be >= current maximum TS
if (ts < tsOrderedList.back().ts) {
assert(false);
throw std::invalid_argument("too small ts");
}
unlink(iter);
iter->second.ts = ts;
tsOrderedList.push_back(iter->second);
}

iterator getFirst() {
if (tsOrderedList.empty())
return elems.end();
return find(tsOrderedList.front().key);
}

TsOrderedList tsOrderedList;
std::unordered_map<KeyT, MapElem> elems;
};

template <class KeyT, class ValT, class ClockT=Clock>
class MapWithExpiry {
public:
template <typename Func>
void start(Duration timeout, Func&& func) {
_timeout = timeout;
_expiryTimer.setCallback([this, func = std::move(func)] {
return seastar::do_until(
[this] {
auto iter = _elems.getFirst();
return (iter == _elems.end() || iter->second.timestamp() > ClockT::now());
},
[this, func] {
auto iter = _elems.getFirst();
// First Remove element from map before calling callback
auto node = _elems.extract(iter);
if (!node) return seastar::make_ready_future();
return func(node.key(), node.mapped().getValue());
});

});
_expiryTimer.armPeriodic(_timeout/2);
}

seastar::future<> stop() {
return _expiryTimer.stop()
.then([this] {
std::vector<seastar::future<>> bgFuts;
for(auto iter = _elems.begin(); iter != _elems.end(); iter++) {
_elems.unlink(iter);
bgFuts.push_back(iter->second.getValue().cleanup().discard_result());
}
return seastar::when_all_succeed(bgFuts.begin(), bgFuts.end());
});
}

typedef typename TimestampOrderedMap<KeyT, ValT>::iterator iterator;

// Check if key is present, with optionally push back expiry if present
iterator find(const KeyT &key, bool reorder) {
ahsank marked this conversation as resolved.
Show resolved Hide resolved
auto iter = _elems.find(key);
if (iter == _elems.end()) return iter;
if (reorder) _elems.resetTs(iter, getExpiry());
return iter;
}

TimePoint getExpiry() {return ClockT::now() + _timeout;}
void insert(const KeyT &key, ValT&& val) {_elems.insert(key, std::move(val), getExpiry());}
void erase(const KeyT &key) {_elems.erase(key);}
auto size() {return _elems.size(); }
ValT& at(const KeyT &key) {return _elems.at(key);}
iterator end() {return _elems.end();}
ValT& getValue(iterator it) {return it->second.getValue();}

private:
TimestampOrderedMap<KeyT, ValT> _elems;
// heartbeats checks are driven off single timer.
PeriodicTimer _expiryTimer;
Duration _timeout = 10s;

};
}
83 changes: 51 additions & 32 deletions src/k2/httpproxy/HTTPProxy.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,31 +168,40 @@ HTTPProxy::HTTPProxy():

seastar::future<> HTTPProxy::gracefulStop() {
_stopped = true;

auto it = _txns.begin();
for(; it != _txns.end(); ++it) {
_endFuts.push_back(it->second.end(false).discard_result());
}

return seastar::when_all_succeed(_endFuts.begin(), _endFuts.end());
return _txns.stop()
.then_wrapped([] (auto&& fut) {
if (fut.failed()) {
K2LOG_W_EXC(log::httpproxy, fut.get_exception(), "txn failed background task");
}
K2LOG_I(log::httpproxy, "stopped");
return seastar::make_ready_future();
});
}

seastar::future<> HTTPProxy::start() {
_stopped = false;
_registerMetrics();
_registerAPI();
auto _startFut = seastar::make_ready_future<>();
_txns.start(httpproxy_txn_timeout(), [this] (uint64_t txnID, TxnInfo& txnInfo){
K2LOG_I(log::httpproxy, "Erasing txnid={}", txnID);
_numQueries -= txnInfo.queries.size();
_timedoutTxns++;
return txnInfo.txn.end(false).discard_result();
});
_startFut = _startFut.then([this] {return _client.start();});
return _startFut;
}


seastar::future<nlohmann::json> HTTPProxy::_handleBegin(nlohmann::json&& request) {
(void) request;
return _client.beginTxn(k2::K2TxnOptions())
.then([this] (auto&& txn) {
K2LOG_D(k2::log::httpproxy, "begin txn: {}", txn.mtr());
_txns[_txnID++] = std::move(txn);
return JsonResponse(Statuses::S201_Created("Begin txn success"), nlohmann::json{{"txnID", _txnID - 1}});
auto txnid = _txnID++;
_txns.insert(txnid, TxnInfo{std::move(txn), {}});
return JsonResponse(Statuses::S201_Created("Begin txn success"), nlohmann::json{{"txnID", txnid}});
});
}

Expand All @@ -209,12 +218,12 @@ seastar::future<nlohmann::json> HTTPProxy::_handleEnd(nlohmann::json&& request)
return JsonResponse(Statuses::S400_Bad_Request("Bad json for end request"));
}

std::unordered_map<uint64_t, k2::K2TxnHandle>::iterator it = _txns.find(id);
if (it == _txns.end()) {
auto iter = _txns.find(id, false);
if (iter == _txns.end()) {
return JsonResponse(Statuses::S400_Bad_Request("Could not find txnID for end request"));
}

return it->second.end(commit)
return _txns.getValue(iter).txn.end(commit)
.then([this, id] (k2::EndResult&& result) {
ahsank marked this conversation as resolved.
Show resolved Hide resolved
_txns.erase(id);
ahsank marked this conversation as resolved.
Show resolved Hide resolved
return JsonResponse(std::move(result.status));
Expand Down Expand Up @@ -242,8 +251,7 @@ seastar::future<nlohmann::json> HTTPProxy::_handleRead(nlohmann::json&& request)
return JsonResponse(Statuses::S400_Bad_Request("Invalid json for read request"));
}

std::unordered_map<uint64_t, k2::K2TxnHandle>::iterator it = _txns.find(id);
if (it == _txns.end()) {
if (_txns.find(id, true) == _txns.end()) {
return JsonResponse(Statuses::S400_Bad_Request("Could not find txnID for read request"));
}

Expand All @@ -261,7 +269,8 @@ seastar::future<nlohmann::json> HTTPProxy::_handleRead(nlohmann::json&& request)
_deserializationErrors++;
return JsonResponse(Statuses::S400_Bad_Request(e.what()));
}
return _txns[id].read(std::move(record))
auto& txnInfo = _txns.at(id);
return txnInfo.txn.read(std::move(record))
.then([this] (k2::ReadResult<k2::dto::SKVRecord>&& result) {
if(!result.status.is2xxOK()) {
return JsonResponse(std::move(result.status));
Expand Down Expand Up @@ -296,8 +305,7 @@ seastar::future<nlohmann::json> HTTPProxy::_handleWrite(nlohmann::json&& request
return JsonResponse(Statuses::S400_Bad_Request("Bad json for write request"));
}

std::unordered_map<uint64_t, k2::K2TxnHandle>::iterator it = _txns.find(id);
if (it == _txns.end()) {
if (_txns.find(id, true) == _txns.end()) {
return JsonResponse(Statuses::S400_Bad_Request("Could not find txnID for write request"));
}

Expand All @@ -315,8 +323,7 @@ seastar::future<nlohmann::json> HTTPProxy::_handleWrite(nlohmann::json&& request
_deserializationErrors++;
return JsonResponse(Statuses::S400_Bad_Request(e.what()));
}

return _txns[id].write(record)
return _txns.at(id).txn.write(record)
.then([] (k2::WriteResult&& result) {
return JsonResponse(std::move(result.status));
});
Expand Down Expand Up @@ -384,17 +391,23 @@ seastar::future<nlohmann::json> HTTPProxy::_handleGetKeyString(nlohmann::json&&
seastar::future<nlohmann::json> HTTPProxy::_handleCreateQuery(nlohmann::json&& jsonReq) {
std::string collectionName;
std::string schemaName;
uint64_t txnID;

try {
jsonReq.at("collectionName").get_to(collectionName);
jsonReq.at("schemaName").get_to(schemaName);
jsonReq.at("txnID").get_to(txnID);
} catch(...) {
_deserializationErrors++;
return JsonResponse(Statuses::S400_Bad_Request("Bad json for query request"));
}

if (_txns.find(txnID, true) == _txns.end()) {
return JsonResponse(Statuses::S400_Bad_Request("Could not find txnID for query request"));
}

return _client.createQuery(collectionName, schemaName)
.then([this, req=std::move(jsonReq)] (auto&& result) mutable {
.then([this, txnID, req=std::move(jsonReq)] (auto&& result) mutable {
if(!result.status.is2xxOK()) {
return JsonResponse(std::move(result.status));
}
Expand All @@ -411,8 +424,10 @@ seastar::future<nlohmann::json> HTTPProxy::_handleCreateQuery(nlohmann::json&& j
if (req.contains("reverse")) {
result.query.setReverseDirection(req["reverse"]);
}
_queries[_queryID++] = std::move(result.query);
nlohmann::json resp{{"queryID", _queryID - 1}};
auto queryid = _queryID++;
_txns.at(txnID).queries[queryid] = std::move(result.query);
_numQueries++;
nlohmann::json resp{{"queryID", queryid}};
return JsonResponse(std::move(result.status), std::move(resp));
});
}
Expand All @@ -428,18 +443,19 @@ seastar::future<nlohmann::json> HTTPProxy::_handleQuery(nlohmann::json&& jsonReq
_deserializationErrors++;
return JsonResponse(Statuses::S400_Bad_Request("Bad json for query request"));
}
auto txnIter = _txns.find(txnID);
if (txnIter == _txns.end()) {
auto iter = _txns.find(txnID, true);
if (iter == _txns.end()) {
return JsonResponse(Statuses::S400_Bad_Request("Could not find txnID for query request"));
}
auto queryIter = _queries.find(queryID);

if (queryIter == _queries.end()) {
auto& txnInfo = _txns.getValue(iter);
auto queryIter = txnInfo.queries.find(queryID);
if (queryIter == txnInfo.queries.end()) {
return JsonResponse(Statuses::S400_Bad_Request("Could not find queryID for query request"));
}

return txnIter->second.query(queryIter->second)
.then([this, queryID](QueryResult&& result) {
return txnInfo.txn.query(queryIter->second)
.then([this, txnID, queryID](QueryResult&& result) {
if(!result.status.is2xxOK()) {
return JsonResponse(std::move(result.status));
}
Expand All @@ -449,10 +465,12 @@ seastar::future<nlohmann::json> HTTPProxy::_handleQuery(nlohmann::json&& jsonReq
for (auto& record: result.records) {
records.push_back(serializeJSONFromRecord(record));
}

bool isDone = _queries[queryID].isDone();
auto& txnInfo = _txns.at(txnID);
auto& query = txnInfo.queries.at(queryID);
bool isDone = query.isDone();
if (isDone) {
_queries.erase(queryID);
txnInfo.queries.erase(queryID);
_numQueries--;
}
nlohmann::json resp;
resp["records"] = std::move(records);
Expand Down Expand Up @@ -509,9 +527,10 @@ void HTTPProxy::_registerMetrics() {
_metric_groups.add_group("session",
{
sm::make_counter("deserialization_errors", _deserializationErrors, sm::description("Total number of deserialization errors"), labels),
sm::make_counter("timed_out_txns", _timedoutTxns, sm::description("Total number of txn timed out"), labels),

sm::make_gauge("open_txns", [this]{ return _txns.size();}, sm::description("Total number of open txn handles"), labels),
sm::make_gauge("open_queries", [this]{ return _queries.size();}, sm::description("Total number of open queries"), labels),
sm::make_gauge("open_queries", [this]{ return _numQueries;}, sm::description("Total number of open queries"), labels),
});
}
}
Loading