procxx
/
kepka
mirror of https://github.com/procxx/kepka.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Improve working with sessions.

This commit is contained in:
John Preston 2019-11-14 20:39:15 +03:00
parent 1e252a6505
commit fe1e627cac
3 changed files with 139 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

235
Telegram/SourceFiles/mtproto/mtp_instance.cpp
View File

@ -35,6 +35,8 @@ constexpr auto kConfigBecomesOldIn = 2 * 60 * crl::time(1000);
constexpr auto kConfigBecomesOldForBlockedIn = 8 * crl::time(1000); constexpr auto kConfigBecomesOldForBlockedIn = 8 * crl::time(1000);
constexpr auto kCheckKeyEach = 60 * crl::time(1000); constexpr auto kCheckKeyEach = 60 * crl::time(1000);
using namespace internal;
} // namespace } // namespace
class Instance::Private : private Sender { class Instance::Private : private Sender {
@ -79,12 +81,17 @@ public:
void reInitConnection(DcId dcId); void reInitConnection(DcId dcId);
void logout(RPCDoneHandlerPtr onDone, RPCFailHandlerPtr onFail); void logout(RPCDoneHandlerPtr onDone, RPCFailHandlerPtr onFail);
std::shared_ptr<internal::Dcenter> getDcById(ShiftedDcId shiftedDcId); std::shared_ptr<Dcenter> getDcById(ShiftedDcId shiftedDcId);
std::shared_ptr<Dcenter> findDc(ShiftedDcId shiftedDcId);
std::shared_ptr<Dcenter> addDc(
ShiftedDcId shiftedDcId,
AuthKeyPtr &&key = nullptr);
void removeDc(ShiftedDcId shiftedDcId);
void unpaused(); void unpaused();
void queueQuittingConnection( void queueQuittingConnection(
std::unique_ptr<internal::Connection> &&connection); std::unique_ptr<Connection> &&connection);
void connectionFinished(internal::Connection *connection); void connectionFinished(Connection *connection);
void sendRequest( void sendRequest(
mtpRequestId requestId, mtpRequestId requestId,
@ -121,7 +128,7 @@ public:
void setSessionResetHandler(Fn<void(ShiftedDcId shiftedDcId)> handler); void setSessionResetHandler(Fn<void(ShiftedDcId shiftedDcId)> handler);
void clearGlobalHandlers(); void clearGlobalHandlers();
not_null<internal::Session*> getSession(ShiftedDcId shiftedDcId); [[nodiscard]] not_null<Session*> getSession(ShiftedDcId shiftedDcId);
bool isNormal() const { bool isNormal() const {
return (_mode == Instance::Mode::Normal); return (_mode == Instance::Mode::Normal);
@ -147,6 +154,10 @@ private:
bool exportFail(const RPCError &error, mtpRequestId requestId); bool exportFail(const RPCError &error, mtpRequestId requestId);
bool onErrorDefault(mtpRequestId requestId, const RPCError &error); bool onErrorDefault(mtpRequestId requestId, const RPCError &error);
Session *findSession(ShiftedDcId shiftedDcId);
not_null<Session*> startSession(ShiftedDcId shiftedDcId);
Session *removeSessionToKilled(ShiftedDcId shiftedDcId);
void applyDomainIps( void applyDomainIps(
const QString &host, const QString &host,
const QStringList &ips, const QStringList &ips,
@ -178,18 +189,18 @@ private:
DcId _mainDcId = Config::kDefaultMainDc; DcId _mainDcId = Config::kDefaultMainDc;
bool _mainDcIdForced = false; bool _mainDcIdForced = false;
std::map<DcId, std::shared_ptr<internal::Dcenter>> _dcenters; base::flat_map<DcId, std::shared_ptr<Dcenter>> _dcenters;
QString _deviceModel; QString _deviceModel;
QString _systemVersion; QString _systemVersion;
internal::Session *_mainSession = nullptr; Session *_mainSession = nullptr;
std::map<ShiftedDcId, std::unique_ptr<internal::Session>> _sessions; base::flat_map<ShiftedDcId, std::unique_ptr<Session>> _sessions;
std::vector<std::unique_ptr<internal::Session>> _killedSessions; // delayed delete std::vector<std::unique_ptr<Session>> _killedSessions; // delayed delete
base::set_of_unique_ptr<internal::Connection> _quittingConnections; base::set_of_unique_ptr<Connection> _quittingConnections;
std::unique_ptr<internal::ConfigLoader> _configLoader; std::unique_ptr<ConfigLoader> _configLoader;
std::unique_ptr<DomainResolver> _domainResolver; std::unique_ptr<DomainResolver> _domainResolver;
std::unique_ptr<SpecialConfigRequest> _httpUnixtimeLoader; std::unique_ptr<SpecialConfigRequest> _httpUnixtimeLoader;
QString _userPhone; QString _userPhone;
@ -258,9 +269,7 @@ void Instance::Private::start(Config &&config) {
} }
} }
_keysForWrite[shiftedDcId] = key; _keysForWrite[shiftedDcId] = key;
addDc(shiftedDcId, std::move(key));
auto dc = std::make_shared<internal::Dcenter>(_instance, dcId, std::move(key));
_dcenters.emplace(shiftedDcId, std::move(dc));
} }
if (config.mainDcId != Config::kNotSetMainDc) { if (config.mainDcId != Config::kNotSetMainDc) {
@ -269,17 +278,11 @@ void Instance::Private::start(Config &&config) {
} }
if (isKeysDestroyer()) { if (isKeysDestroyer()) {
for (auto &dc : _dcenters) { for (const auto &[shiftedDcId, dc] : _dcenters) {
auto shiftedDcId = dc.first; startSession(shiftedDcId);
auto session = std::make_unique<internal::Session>(_instance, shiftedDcId);
auto it = _sessions.emplace(shiftedDcId, std::move(session)).first;
it->second->start();
} }
} else if (_mainDcId != Config::kNoneMainDc) { } else if (_mainDcId != Config::kNoneMainDc) {
auto main = std::make_unique<internal::Session>(_instance, _mainDcId); _mainSession = startSession(_mainDcId);
_mainSession = main.get();
_sessions.emplace(_mainDcId, std::move(main));
_mainSession->start();
} }
_checkDelayedTimer.setCallback([this] { checkDelayedRequests(); }); _checkDelayedTimer.setCallback([this] { checkDelayedRequests(); });
@ -335,8 +338,8 @@ void Instance::Private::applyDomainIps(
} }
if (applyToProxy(Global::RefSelectedProxy()) if (applyToProxy(Global::RefSelectedProxy())
&& (Global::ProxySettings() == ProxyData::Settings::Enabled)) { && (Global::ProxySettings() == ProxyData::Settings::Enabled)) {
for (auto &session : _sessions) { for (const auto &[shiftedDcId, session] : _sessions) {
session.second->refreshOptions(); session->refreshOptions();
} }
} }
emit _instance->proxyDomainResolved(host, ips, expireAt); emit _instance->proxyDomainResolved(host, ips, expireAt);
@ -398,7 +401,7 @@ void Instance::Private::requestConfig() {
if (_configLoader || isKeysDestroyer()) { if (_configLoader || isKeysDestroyer()) {
return; return;
} }
_configLoader = std::make_unique<internal::ConfigLoader>( _configLoader = std::make_unique<ConfigLoader>(
_instance, _instance,
_userPhone, _userPhone,
rpcDone([=](const MTPConfig &result) { configLoadDone(result); }), rpcDone([=](const MTPConfig &result) { configLoadDone(result); }),
@ -472,16 +475,16 @@ void Instance::Private::requestCDNConfig() {
} }
void Instance::Private::restart() { void Instance::Private::restart() {
for (auto &session : _sessions) { for (const auto &[shiftedDcId, session] : _sessions) {
session.second->restart(); session->restart();
} }
} }
void Instance::Private::restart(ShiftedDcId shiftedDcId) { void Instance::Private::restart(ShiftedDcId shiftedDcId) {
auto dcId = BareDcId(shiftedDcId); const auto dcId = BareDcId(shiftedDcId);
for (auto &session : _sessions) { for (const auto &[shiftedDcId, session] : _sessions) {
if (BareDcId(session.second->getDcWithShift()) == dcId) { if (BareDcId(shiftedDcId) == dcId) {
session.second->restart(); session->restart();
} }
} }
} }
@ -497,11 +500,9 @@ int32 Instance::Private::dcstate(ShiftedDcId shiftedDcId) {
shiftedDcId += BareDcId(_mainSession->getDcWithShift()); shiftedDcId += BareDcId(_mainSession->getDcWithShift());
} }
auto it = _sessions.find(shiftedDcId); if (const auto session = findSession(shiftedDcId)) {
if (it != _sessions.cend()) { return session->getState();
return it->second->getState();
} }
return DisconnectedState; return DisconnectedState;
} }
@ -515,11 +516,9 @@ QString Instance::Private::dctransport(ShiftedDcId shiftedDcId) {
shiftedDcId += BareDcId(_mainSession->getDcWithShift()); shiftedDcId += BareDcId(_mainSession->getDcWithShift());
} }
auto it = _sessions.find(shiftedDcId); if (const auto session = findSession(shiftedDcId)) {
if (it != _sessions.cend()) { return session->transport();
return it->second->transport();
} }
return QString(); return QString();
} }
@ -563,25 +562,15 @@ int32 Instance::Private::state(mtpRequestId requestId) {
} }
void Instance::Private::killSession(ShiftedDcId shiftedDcId) { void Instance::Private::killSession(ShiftedDcId shiftedDcId) {
auto checkIfMainAndKill = [this](ShiftedDcId shiftedDcId) { const auto checkIfMainAndKill = [&](ShiftedDcId shiftedDcId) {
auto it = _sessions.find(shiftedDcId); const auto killed = removeSessionToKilled(shiftedDcId);
if (it != _sessions.cend()) { return killed && (killed == _mainSession);
_killedSessions.push_back(std::move(it->second));
_sessions.erase(it);
_killedSessions.back()->kill();
return (_killedSessions.back().get() == _mainSession);
}
return false;
}; };
if (checkIfMainAndKill(shiftedDcId)) { if (checkIfMainAndKill(shiftedDcId)) {
checkIfMainAndKill(_mainDcId); checkIfMainAndKill(_mainDcId);
_mainSession = startSession(_mainDcId);
auto main = std::make_unique<internal::Session>(_instance, _mainDcId);
_mainSession = main.get();
_sessions.emplace(_mainDcId, std::move(main));
_mainSession->start();
} }
InvokeQueued(_instance, [this] { InvokeQueued(_instance, [=] {
clearKilledSessions(); clearKilledSessions();
}); });
} }
@ -591,18 +580,17 @@ void Instance::Private::clearKilledSessions() {
} }
void Instance::Private::stopSession(ShiftedDcId shiftedDcId) { void Instance::Private::stopSession(ShiftedDcId shiftedDcId) {
auto it = _sessions.find(shiftedDcId); if (const auto session = findSession(shiftedDcId)) {
if (it != _sessions.end()) { if (session != _mainSession) { // don't stop main session
if (it->second.get() != _mainSession) { // don't stop main session session->stop();
it->second->stop();
} }
} }
} }
void Instance::Private::reInitConnection(DcId dcId) { void Instance::Private::reInitConnection(DcId dcId) {
for (auto &session : _sessions) { for (const auto &[shiftedDcId, session] : _sessions) {
if (BareDcId(session.second->getDcWithShift()) == dcId) { if (BareDcId(shiftedDcId) == dcId) {
session.second->reInitConnection(); session->reInitConnection();
} }
} }
} }
@ -647,22 +635,40 @@ bool Instance::Private::logoutGuestDone(mtpRequestId requestId) {
return false; return false;
} }
std::shared_ptr<internal::Dcenter> Instance::Private::getDcById(ShiftedDcId shiftedDcId) { std::shared_ptr<Dcenter> Instance::Private::findDc(ShiftedDcId shiftedDcId) {
auto it = _dcenters.find(shiftedDcId); const auto i = _dcenters.find(shiftedDcId);
if (it == _dcenters.cend()) { return (i != _dcenters.end()) ? i->second : nullptr;
auto dcId = BareDcId(shiftedDcId); }
if (isTemporaryDcId(dcId)) {
if (auto realDcId = getRealIdFromTemporaryDcId(dcId)) { std::shared_ptr<Dcenter> Instance::Private::addDc(
dcId = realDcId; ShiftedDcId shiftedDcId,
} AuthKeyPtr &&key) {
} const auto dcId = BareDcId(shiftedDcId);
it = _dcenters.find(dcId); return _dcenters.emplace(
if (it == _dcenters.cend()) { shiftedDcId,
auto result = std::make_shared<internal::Dcenter>(_instance, dcId, AuthKeyPtr()); std::make_shared<Dcenter>(_instance, dcId, std::move(key))
return _dcenters.emplace(dcId, std::move(result)).first->second; ).first->second;
}
void Instance::Private::removeDc(ShiftedDcId shiftedDcId) {
_dcenters.erase(shiftedDcId);
}
std::shared_ptr<Dcenter> Instance::Private::getDcById(
ShiftedDcId shiftedDcId) {
if (auto result = findDc(shiftedDcId)) {
return result;
}
auto dcId = BareDcId(shiftedDcId);
if (isTemporaryDcId(dcId)) {
if (const auto realDcId = getRealIdFromTemporaryDcId(dcId)) {
dcId = realDcId;
} }
} }
return it->second; if (auto result = findDc(dcId)) {
return result;
}
return addDc(dcId);
} }
void Instance::Private::setKeyForWrite(DcId dcId, const AuthKeyPtr &key) { void Instance::Private::setKeyForWrite(DcId dcId, const AuthKeyPtr &key) {
@ -693,7 +699,7 @@ void Instance::Private::addKeysForDestroy(AuthKeysList &&keys) {
Expects(isKeysDestroyer()); Expects(isKeysDestroyer());
for (auto &key : keys) { for (auto &key : keys) {
auto dcId = key->dcId(); const auto dcId = key->dcId();
auto shiftedDcId = MTP::destroyKeyNextDcId(dcId); auto shiftedDcId = MTP::destroyKeyNextDcId(dcId);
{ {
@ -706,12 +712,8 @@ void Instance::Private::addKeysForDestroy(AuthKeysList &&keys) {
_keysForWrite[shiftedDcId] = key; _keysForWrite[shiftedDcId] = key;
} }
auto dc = std::make_shared<internal::Dcenter>(_instance, dcId, std::move(key)); addDc(shiftedDcId, std::move(key));
_dcenters.emplace(shiftedDcId, std::move(dc)); startSession(shiftedDcId);
auto session = std::make_unique<internal::Session>(_instance, shiftedDcId);
auto it = _sessions.emplace(shiftedDcId, std::move(session)).first;
it->second->start();
} }
} }
@ -728,18 +730,18 @@ QString Instance::Private::systemVersion() const {
} }
void Instance::Private::unpaused() { void Instance::Private::unpaused() {
for (auto &session : _sessions) { for (const auto &[shiftedDcId, session] : _sessions) {
session.second->unpaused(); session->unpaused();
} }
} }
void Instance::Private::queueQuittingConnection( void Instance::Private::queueQuittingConnection(
std::unique_ptr<internal::Connection> &&connection) { std::unique_ptr<Connection> &&connection) {
_quittingConnections.insert(std::move(connection)); _quittingConnections.insert(std::move(connection));
} }
void Instance::Private::connectionFinished(internal::Connection *connection) { void Instance::Private::connectionFinished(Connection *connection) {
auto it = _quittingConnections.find(connection); const auto it = _quittingConnections.find(connection);
if (it != _quittingConnections.end()) { if (it != _quittingConnections.end()) {
_quittingConnections.erase(it); _quittingConnections.erase(it);
} }
@ -1430,23 +1432,47 @@ bool Instance::Private::onErrorDefault(mtpRequestId requestId, const RPCError &e
return false; return false;
} }
not_null<internal::Session*> Instance::Private::getSession( not_null<Session*> Instance::Private::getSession(
ShiftedDcId shiftedDcId) { ShiftedDcId shiftedDcId) {
if (!shiftedDcId) { if (!shiftedDcId) {
Assert(_mainSession != nullptr); Assert(_mainSession != nullptr);
return _mainSession; return _mainSession;
} } else if (!BareDcId(shiftedDcId)) {
if (!BareDcId(shiftedDcId)) {
Assert(_mainSession != nullptr); Assert(_mainSession != nullptr);
shiftedDcId += BareDcId(_mainSession->getDcWithShift()); shiftedDcId += BareDcId(_mainSession->getDcWithShift());
} }
auto it = _sessions.find(shiftedDcId); if (const auto session = findSession(shiftedDcId)) {
if (it == _sessions.cend()) { return session;
it = _sessions.emplace(shiftedDcId, std::make_unique<internal::Session>(_instance, shiftedDcId)).first;
it->second->start();
} }
return it->second.get(); return startSession(shiftedDcId);
}
Session *Instance::Private::findSession(ShiftedDcId shiftedDcId) {
const auto i = _sessions.find(shiftedDcId);
return (i != _sessions.end()) ? i->second.get() : nullptr;
}
not_null<Session*> Instance::Private::startSession(ShiftedDcId shiftedDcId) {
Expects(BareDcId(shiftedDcId) != 0);
const auto result = _sessions.emplace(
shiftedDcId,
std::make_unique<Session>(_instance, shiftedDcId)
).first->second.get();
result->start();
return result;
}
Session *Instance::Private::removeSessionToKilled(ShiftedDcId shiftedDcId) {
const auto i = _sessions.find(shiftedDcId);
if (i == _sessions.cend()) {
return nullptr;
}
i->second->kill();
_killedSessions.push_back(std::move(i->second));
_sessions.erase(i);
return _killedSessions.back().get();
} }
void Instance::Private::scheduleKeyDestroy(ShiftedDcId shiftedDcId) { void Instance::Private::scheduleKeyDestroy(ShiftedDcId shiftedDcId) {
@ -1488,7 +1514,7 @@ void Instance::Private::performKeyDestroy(ShiftedDcId shiftedDcId) {
void Instance::Private::completedKeyDestroy(ShiftedDcId shiftedDcId) { void Instance::Private::completedKeyDestroy(ShiftedDcId shiftedDcId) {
Expects(isKeysDestroyer()); Expects(isKeysDestroyer());
_dcenters.erase(shiftedDcId); removeDc(shiftedDcId);
{ {
QWriteLocker lock(&_keysForWriteLock); QWriteLocker lock(&_keysForWriteLock);
_keysForWrite.erase(shiftedDcId); _keysForWrite.erase(shiftedDcId);
@ -1502,7 +1528,7 @@ void Instance::Private::completedKeyDestroy(ShiftedDcId shiftedDcId) {
void Instance::Private::checkMainDcKey() { void Instance::Private::checkMainDcKey() {
const auto id = mainDcId(); const auto id = mainDcId();
const auto shiftedDcId = ShiftDcId(id, kCheckKeyDcShift); const auto shiftedDcId = ShiftDcId(id, kCheckKeyDcShift);
if (_sessions.find(shiftedDcId) != _sessions.end()) { if (findSession(shiftedDcId)) {
return; return;
} }
const auto key = [&] { const auto key = [&] {
@ -1527,9 +1553,8 @@ void Instance::Private::keyDestroyedOnServer(DcId dcId, uint64 keyId) {
} }
restart(); restart();
} else { } else {
const auto i = _dcenters.find(dcId); if (const auto dc = findDc(dcId)) {
if (i != end(_dcenters)) { return dc->destroyKey();
i->second->destroyKey();
} }
restart(dcId); restart(dcId);
} }
@ -1564,8 +1589,8 @@ void Instance::Private::prepareToDestroy() {
requestCancellingDiscard(); requestCancellingDiscard();
for (auto &session : base::take(_sessions)) { for (const auto &[shiftedDcId, session] : base::take(_sessions)) {
session.second->kill(); session->kill();
} }
_mainSession = nullptr; _mainSession = nullptr;
} }
@ -1632,7 +1657,7 @@ void Instance::requestCDNConfig() {
_private->requestCDNConfig(); _private->requestCDNConfig();
} }
void Instance::connectionFinished(internal::Connection *connection) { void Instance::connectionFinished(Connection *connection) {
_private->connectionFinished(connection); _private->connectionFinished(connection);
} }
@ -1680,7 +1705,7 @@ void Instance::logout(RPCDoneHandlerPtr onDone, RPCFailHandlerPtr onFail) {
_private->logout(onDone, onFail); _private->logout(onDone, onFail);
} }
std::shared_ptr<internal::Dcenter> Instance::getDcById(ShiftedDcId shiftedDcId) { std::shared_ptr<Dcenter> Instance::getDcById(ShiftedDcId shiftedDcId) {
return _private->getDcById(shiftedDcId); return _private->getDcById(shiftedDcId);
} }
@ -1713,7 +1738,7 @@ void Instance::unpaused() {
} }
void Instance::queueQuittingConnection( void Instance::queueQuittingConnection(
std::unique_ptr<internal::Connection> &&connection) { std::unique_ptr<Connection> &&connection) {
_private->queueQuittingConnection(std::move(connection)); _private->queueQuittingConnection(std::move(connection));
} }

11
Telegram/SourceFiles/mtproto/session.cpp
View File

@ -143,12 +143,11 @@ void SessionData::clear(Instance *instance) {
Session::Session(not_null<Instance*> instance, ShiftedDcId shiftedDcId) Session::Session(not_null<Instance*> instance, ShiftedDcId shiftedDcId)
: QObject() : QObject()
, _instance(instance) , _instance(instance)
, _data(this)
, _shiftedDcId(shiftedDcId) , _shiftedDcId(shiftedDcId)
, _data(this)
, _timeouter([=] { checkRequestsByTimer(); })
, _sender([=] { needToResumeAndSend(); }) { , _sender([=] { needToResumeAndSend(); }) {
connect(&_timeouter, SIGNAL(timeout()), this, SLOT(checkRequestsByTimer())); _timeouter.callEach(1000);
_timeouter.start(1000);
refreshOptions(); refreshOptions();
} }
@ -291,9 +290,7 @@ void Session::needToResumeAndSend() {
} }
if (!_connection) { if (!_connection) {
DEBUG_LOG(("Session Info: resuming session dcWithShift %1").arg(_shiftedDcId)); DEBUG_LOG(("Session Info: resuming session dcWithShift %1").arg(_shiftedDcId));
createDcData(); start();
_connection = std::make_unique<Connection>(_instance);
_connection->start(&_data, _shiftedDcId);
} }
if (_ping) { if (_ping) {
_ping = false; _ping = false;

9
Telegram/SourceFiles/mtproto/session.h
View File

@ -378,7 +378,6 @@ public slots:
void connectionWasInitedForDC(); void connectionWasInitedForDC();
void tryToReceive(); void tryToReceive();
void checkRequestsByTimer();
void onConnectionStateChange(qint32 newState); void onConnectionStateChange(qint32 newState);
void onResetDone(); void onResetDone();
@ -388,10 +387,13 @@ public slots:
private: private:
void createDcData(); void createDcData();
void checkRequestsByTimer();
bool rpcErrorOccured(mtpRequestId requestId, const RPCFailHandlerPtr &onFail, const RPCError &err); bool rpcErrorOccured(mtpRequestId requestId, const RPCFailHandlerPtr &onFail, const RPCError &err);
not_null<Instance*> _instance; const not_null<Instance*> _instance;
const ShiftedDcId _shiftedDcId = 0;
std::unique_ptr<Connection> _connection; std::unique_ptr<Connection> _connection;
bool _killed = false; bool _killed = false;
@ -399,7 +401,6 @@ private:
SessionData _data; SessionData _data;
ShiftedDcId _shiftedDcId = 0;
std::shared_ptr<Dcenter> _dc; std::shared_ptr<Dcenter> _dc;
AuthKeyPtr _dcKeyForCheck; AuthKeyPtr _dcKeyForCheck;
@ -408,7 +409,7 @@ private:
bool _ping = false; bool _ping = false;
QTimer _timeouter; base::Timer _timeouter;
base::Timer _sender; base::Timer _sender;
}; };