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

Don't use shared_ptr for Dcenters.

This commit is contained in:
John Preston 2019-11-15 10:28:33 +03:00
parent f37ab6e38e
commit 055b99f5b0
9 changed files with 123 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Telegram/SourceFiles/core/utils.h
View File

@ -25,12 +25,6 @@ https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL
namespace base {
template <typename T>
using set_of_unique_ptr = std::set<std::unique_ptr<T>, base::pointer_comparator<T>>;
template <typename T>
using set_of_shared_ptr = std::set<std::shared_ptr<T>, base::pointer_comparator<T>>;
template <typename Value, typename From, typename Till>
inline bool in_range(Value &&value, From &&from, Till &&till) {
return (value >= from) && (value < till);

9
Telegram/SourceFiles/mtproto/connection.cpp
View File

@ -249,7 +249,6 @@ ConnectionPrivate::ConnectionPrivate(
connect(thread, &QThread::started, this, [=] { connectToServer(); });
connect(thread, &QThread::finished, this, [=] { finishAndDestroy(); });
connect(this, SIGNAL(finished(internal::Connection*)), _instance, SLOT(connectionFinished(internal::Connection*)), Qt::QueuedConnection);
connect(_sessionData->owner(), SIGNAL(authKeyCreated()), this, SLOT(updateAuthKey()), Qt::QueuedConnection);
connect(_sessionData->owner(), SIGNAL(needToRestart()), this, SLOT(restartNow()), Qt::QueuedConnection);
@ -1227,7 +1226,11 @@ void ConnectionPrivate::doDisconnect() {
void ConnectionPrivate::finishAndDestroy() {
doDisconnect();
_finished = true;
emit finished(_owner);
const auto connection = _owner;
const auto instance = _instance;
InvokeQueued(instance, [=] {
instance->connectionFinished(connection);
});
deleteLater();
}
@ -1249,7 +1252,7 @@ void ConnectionPrivate::handleReceived() {
onReceivedSome();
auto restartOnError = [this, &lockFinished] {
const auto restartOnError = [&] {
lockFinished.unlock();
restart();
};

2
Telegram/SourceFiles/mtproto/connection.h
View File

@ -94,8 +94,6 @@ signals:
void resendManyAsync(QVector<quint64> msgIds, qint64 msCanWait, bool forceContainer, bool sendMsgStateInfo);
void resendAllAsync();
void finished(internal::Connection *connection);
public slots:
void restartNow();

15
Telegram/SourceFiles/mtproto/dcenter.cpp
View File

@ -57,5 +57,20 @@ void Dcenter::destroyKey() {
setKey(AuthKeyPtr());
}
bool Dcenter::connectionInited() const {
const auto lock = QMutexLocker(&_initLock);
return _connectionInited;
}
void Dcenter::setConnectionInited(bool connectionInited) {
auto lock = QMutexLocker(&_initLock);
_connectionInited = connectionInited;
lock.unlock();
if (connectionInited) {
emit connectionWasInited();
}
}
} // namespace internal
} // namespace MTP

17
Telegram/SourceFiles/mtproto/dcenter.h
View File

@ -26,19 +26,8 @@ public:
void setKey(AuthKeyPtr &&key);
void destroyKey();
bool connectionInited() const {
QMutexLocker lock(&initLock);
return _connectionInited;
}
void setConnectionInited(bool connectionInited = true) {
QMutexLocker lock(&initLock);
_connectionInited = connectionInited;
lock.unlock();
if (connectionInited) {
emit connectionWasInited();
}
}
[[nodiscard]] bool connectionInited() const;
void setConnectionInited(bool connectionInited = true);
signals:
void authKeyCreated();
@ -49,7 +38,7 @@ private slots:
private:
mutable QReadWriteLock keyLock;
mutable QMutex initLock;
mutable QMutex _initLock;
not_null<Instance*> _instance;
DcId _id = 0;
AuthKeyPtr _key;

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

@ -81,9 +81,9 @@ public:
void reInitConnection(DcId dcId);
void logout(RPCDoneHandlerPtr onDone, RPCFailHandlerPtr onFail);
std::shared_ptr<Dcenter> getDcById(ShiftedDcId shiftedDcId);
std::shared_ptr<Dcenter> findDc(ShiftedDcId shiftedDcId);
std::shared_ptr<Dcenter> addDc(
not_null<Dcenter*> getDcById(ShiftedDcId shiftedDcId);
Dcenter *findDc(ShiftedDcId shiftedDcId);
not_null<Dcenter*> addDc(
ShiftedDcId shiftedDcId,
AuthKeyPtr &&key = nullptr);
void removeDc(ShiftedDcId shiftedDcId);
@ -91,7 +91,7 @@ public:
void queueQuittingConnection(
std::unique_ptr<Connection> &&connection);
void connectionFinished(Connection *connection);
void connectionFinished(not_null<Connection*> connection);
void sendRequest(
mtpRequestId requestId,
@ -138,12 +138,12 @@ public:
}
void scheduleKeyDestroy(ShiftedDcId shiftedDcId);
void checkIfKeyWasDestroyed(ShiftedDcId shiftedDcId);
void performKeyDestroy(ShiftedDcId shiftedDcId);
void completedKeyDestroy(ShiftedDcId shiftedDcId);
void checkMainDcKey();
void keyDestroyedOnServer(DcId dcId, uint64 keyId);
void clearKilledSessions();
void prepareToDestroy();
private:
@ -156,7 +156,7 @@ private:
Session *findSession(ShiftedDcId shiftedDcId);
not_null<Session*> startSession(ShiftedDcId shiftedDcId);
Session *removeSessionToKilled(ShiftedDcId shiftedDcId);
Session *removeSession(ShiftedDcId shiftedDcId);
void applyDomainIps(
const QString &host,
@ -187,18 +187,19 @@ private:
const not_null<DcOptions*> _dcOptions;
const Instance::Mode _mode = Instance::Mode::Normal;
DcId _mainDcId = Config::kDefaultMainDc;
bool _mainDcIdForced = false;
base::flat_map<DcId, std::shared_ptr<Dcenter>> _dcenters;
QString _deviceModel;
QString _systemVersion;
DcId _mainDcId = Config::kDefaultMainDc;
bool _mainDcIdForced = false;
base::flat_map<DcId, std::unique_ptr<Dcenter>> _dcenters;
std::vector<std::unique_ptr<Dcenter>> _dcentersToDestroy;
Session *_mainSession = nullptr;
base::flat_map<ShiftedDcId, std::unique_ptr<Session>> _sessions;
std::vector<std::unique_ptr<Session>> _killedSessions; // delayed delete
std::vector<std::unique_ptr<Session>> _sessionsToDestroy;
base::set_of_unique_ptr<Connection> _quittingConnections;
std::vector<std::unique_ptr<Connection>> _connectionsToDestroy;
std::unique_ptr<ConfigLoader> _configLoader;
std::unique_ptr<DomainResolver> _domainResolver;
@ -563,22 +564,20 @@ int32 Instance::Private::state(mtpRequestId requestId) {
void Instance::Private::killSession(ShiftedDcId shiftedDcId) {
const auto checkIfMainAndKill = [&](ShiftedDcId shiftedDcId) {
const auto killed = removeSessionToKilled(shiftedDcId);
return killed && (killed == _mainSession);
if (const auto removed = removeSession(shiftedDcId)) {
return (removed == _mainSession);
}
return false;
};
if (checkIfMainAndKill(shiftedDcId)) {
checkIfMainAndKill(_mainDcId);
_mainSession = startSession(_mainDcId);
}
InvokeQueued(_instance, [=] {
clearKilledSessions();
_sessionsToDestroy.clear();
});
}
void Instance::Private::clearKilledSessions() {
_killedSessions.clear();
}
void Instance::Private::stopSession(ShiftedDcId shiftedDcId) {
if (const auto session = findSession(shiftedDcId)) {
if (session != _mainSession) { // don't stop main session
@ -635,38 +634,47 @@ bool Instance::Private::logoutGuestDone(mtpRequestId requestId) {
return false;
}
std::shared_ptr<Dcenter> Instance::Private::findDc(ShiftedDcId shiftedDcId) {
Dcenter *Instance::Private::findDc(ShiftedDcId shiftedDcId) {
const auto i = _dcenters.find(shiftedDcId);
return (i != _dcenters.end()) ? i->second : nullptr;
return (i != _dcenters.end()) ? i->second.get() : nullptr;
}
std::shared_ptr<Dcenter> Instance::Private::addDc(
not_null<Dcenter*> Instance::Private::addDc(
ShiftedDcId shiftedDcId,
AuthKeyPtr &&key) {
const auto dcId = BareDcId(shiftedDcId);
return _dcenters.emplace(
shiftedDcId,
std::make_shared<Dcenter>(_instance, dcId, std::move(key))
).first->second;
std::make_unique<Dcenter>(_instance, dcId, std::move(key))
).first->second.get();
}
void Instance::Private::removeDc(ShiftedDcId shiftedDcId) {
_dcenters.erase(shiftedDcId);
const auto i = _dcenters.find(shiftedDcId);
if (i != _dcenters.end()) {
_dcentersToDestroy.push_back(std::move(i->second));
_dcenters.erase(i);
}
}
std::shared_ptr<Dcenter> Instance::Private::getDcById(
not_null<Dcenter*> Instance::Private::getDcById(
ShiftedDcId shiftedDcId) {
if (auto result = findDc(shiftedDcId)) {
if (const auto result = findDc(shiftedDcId)) {
return result;
}
auto dcId = BareDcId(shiftedDcId);
if (isTemporaryDcId(dcId)) {
if (const auto realDcId = getRealIdFromTemporaryDcId(dcId)) {
dcId = realDcId;
const auto dcId = [&] {
const auto result = BareDcId(shiftedDcId);
if (isTemporaryDcId(result)) {
if (const auto realDcId = getRealIdFromTemporaryDcId(result)) {
return realDcId;
}
}
}
if (auto result = findDc(dcId)) {
return result;
}();
if (dcId != shiftedDcId) {
if (const auto result = findDc(dcId)) {
return result;
}
}
return addDc(dcId);
}
@ -737,13 +745,17 @@ void Instance::Private::unpaused() {
void Instance::Private::queueQuittingConnection(
std::unique_ptr<Connection> &&connection) {
_quittingConnections.insert(std::move(connection));
_connectionsToDestroy.push_back(std::move(connection));
}
void Instance::Private::connectionFinished(Connection *connection) {
const auto it = _quittingConnections.find(connection);
if (it != _quittingConnections.end()) {
_quittingConnections.erase(it);
void Instance::Private::connectionFinished(
not_null<Connection*> connection) {
const auto i = ranges::find(
_connectionsToDestroy,
connection.get(),
&std::unique_ptr<Connection>::get);
if (i != _connectionsToDestroy.end()) {
_connectionsToDestroy.erase(i);
}
}
@ -1005,7 +1017,7 @@ void Instance::Private::clearCallbacksDelayed(
).arg(idsString.join(", ")));
}
crl::on_main(_instance, [this, list = std::move(ids)] {
InvokeQueued(_instance, [=, list = std::move(ids)] {
clearCallbacks(list);
});
}
@ -1456,23 +1468,26 @@ Session *Instance::Private::findSession(ShiftedDcId shiftedDcId) {
not_null<Session*> Instance::Private::startSession(ShiftedDcId shiftedDcId) {
Expects(BareDcId(shiftedDcId) != 0);
const auto dc = (GetDcIdShift(shiftedDcId) != kCheckKeyDcShift)
? getDcById(shiftedDcId).get()
: nullptr;
const auto result = _sessions.emplace(
shiftedDcId,
std::make_unique<Session>(_instance, shiftedDcId)
std::make_unique<Session>(_instance, shiftedDcId, dc)
).first->second.get();
result->start();
return result;
}
Session *Instance::Private::removeSessionToKilled(ShiftedDcId shiftedDcId) {
Session *Instance::Private::removeSession(ShiftedDcId shiftedDcId) {
const auto i = _sessions.find(shiftedDcId);
if (i == _sessions.cend()) {
return nullptr;
}
i->second->kill();
_killedSessions.push_back(std::move(i->second));
_sessionsToDestroy.push_back(std::move(i->second));
_sessions.erase(i);
return _killedSessions.back().get();
return _sessionsToDestroy.back().get();
}
void Instance::Private::scheduleKeyDestroy(ShiftedDcId shiftedDcId) {
@ -1484,13 +1499,29 @@ void Instance::Private::scheduleKeyDestroy(ShiftedDcId shiftedDcId) {
_instance->send(MTPauth_LogOut(), rpcDone([=](const MTPBool &) {
performKeyDestroy(shiftedDcId);
}), rpcFail([=](const RPCError &error) {
if (isDefaultHandledError(error)) return false;
if (isDefaultHandledError(error)) {
return false;
}
performKeyDestroy(shiftedDcId);
return true;
}), shiftedDcId);
}
}
void Instance::Private::checkIfKeyWasDestroyed(ShiftedDcId shiftedDcId) {
InvokeQueued(_instance, [=] {
if (isKeysDestroyer()) {
LOG(("MTP Info: checkIfKeyWasDestroyed on destroying key %1, "
"assuming it is destroyed.").arg(shiftedDcId));
completedKeyDestroy(shiftedDcId);
} else if (BareDcId(shiftedDcId) == mainDcId()) {
LOG(("MTP Info: checkIfKeyWasDestroyed for main dc %1, "
"checking.").arg(shiftedDcId));
checkMainDcKey();
}
});
}
void Instance::Private::performKeyDestroy(ShiftedDcId shiftedDcId) {
Expects(isKeysDestroyer());
@ -1657,7 +1688,7 @@ void Instance::requestCDNConfig() {
_private->requestCDNConfig();
}
void Instance::connectionFinished(Connection *connection) {
void Instance::connectionFinished(not_null<Connection*> connection) {
_private->connectionFinished(connection);
}
@ -1705,10 +1736,6 @@ void Instance::logout(RPCDoneHandlerPtr onDone, RPCFailHandlerPtr onFail) {
_private->logout(onDone, onFail);
}
std::shared_ptr<Dcenter> Instance::getDcById(ShiftedDcId shiftedDcId) {
return _private->getDcById(shiftedDcId);
}
void Instance::setKeyForWrite(DcId dcId, const AuthKeyPtr &key) {
_private->setKeyForWrite(dcId, key);
}
@ -1799,17 +1826,7 @@ void Instance::scheduleKeyDestroy(ShiftedDcId shiftedDcId) {
}
void Instance::checkIfKeyWasDestroyed(ShiftedDcId shiftedDcId) {
crl::on_main(this, [=] {
if (isKeysDestroyer()) {
LOG(("MTP Info: checkIfKeyWasDestroyed on destroying key %1, "
"assuming it is destroyed.").arg(shiftedDcId));
_private->completedKeyDestroy(shiftedDcId);
} else if (BareDcId(shiftedDcId) == mainDcId()) {
LOG(("MTP Info: checkIfKeyWasDestroyed for main dc %1, "
"checking.").arg(shiftedDcId));
_private->checkMainDcKey();
}
});
_private->checkIfKeyWasDestroyed(shiftedDcId);
}
void Instance::keyDestroyedOnServer(DcId dcId, uint64 keyId) {

6
Telegram/SourceFiles/mtproto/mtp_instance.h
View File

@ -149,7 +149,6 @@ public:
void reInitConnection(DcId dcId);
void logout(RPCDoneHandlerPtr onDone, RPCFailHandlerPtr onFail);
std::shared_ptr<internal::Dcenter> getDcById(ShiftedDcId shiftedDcId);
void unpaused();
void queueQuittingConnection(std::unique_ptr<internal::Connection> &&connection);
@ -185,10 +184,9 @@ public:
void syncHttpUnixtime();
~Instance();
void connectionFinished(not_null<internal::Connection*> connection);
public slots:
void connectionFinished(internal::Connection *connection);
~Instance();
signals:
void configLoaded();

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

@ -140,19 +140,32 @@ void SessionData::clear(Instance *instance) {
instance->clearCallbacksDelayed(std::move(clearCallbacks));
}
Session::Session(not_null<Instance*> instance, ShiftedDcId shiftedDcId)
Session::Session(
not_null<Instance*> instance,
ShiftedDcId shiftedDcId,
Dcenter *dc)
: QObject()
, _instance(instance)
, _shiftedDcId(shiftedDcId)
, _dc(dc)
, _data(this)
, _timeouter([=] { checkRequestsByTimer(); })
, _sender([=] { needToResumeAndSend(); }) {
_timeouter.callEach(1000);
refreshOptions();
if (_dc) {
if (const auto lock = ReadLockerAttempt(keyMutex())) {
_data.setKey(_dc->getKey());
if (_dc->connectionInited()) {
_data.setConnectionInited();
}
}
connect(_dc, SIGNAL(authKeyCreated()), this, SLOT(authKeyCreatedForDC()), Qt::QueuedConnection);
connect(_dc, SIGNAL(connectionWasInited()), this, SLOT(connectionWasInitedForDC()), Qt::QueuedConnection);
}
}
void Session::start() {
createDcData();
_connection = std::make_unique<Connection>(_instance);
_connection->start(&_data, _shiftedDcId);
if (_instance->isKeysDestroyer()) {
@ -160,22 +173,6 @@ void Session::start() {
}
}
void Session::createDcData() {
if (_dc || GetDcIdShift(_shiftedDcId) == kCheckKeyDcShift) {
return;
}
_dc = _instance->getDcById(_shiftedDcId);
if (auto lock = ReadLockerAttempt(keyMutex())) {
_data.setKey(_dc->getKey());
if (_dc->connectionInited()) {
_data.setConnectionInited();
}
}
connect(_dc.get(), SIGNAL(authKeyCreated()), this, SLOT(authKeyCreatedForDC()), Qt::QueuedConnection);
connect(_dc.get(), SIGNAL(connectionWasInited()), this, SLOT(connectionWasInitedForDC()), Qt::QueuedConnection);
}
bool Session::rpcErrorOccured(
mtpRequestId requestId,
const RPCFailHandlerPtr &onFail,

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

@ -317,7 +317,10 @@ class Session : public QObject {
Q_OBJECT
public:
Session(not_null<Instance*> instance, ShiftedDcId shiftedDcId);
Session(
not_null<Instance*> instance,
ShiftedDcId shiftedDcId,
Dcenter *dc);
void start();
void restart();
@ -376,13 +379,13 @@ public slots:
void sendMsgsStateInfo(quint64 msgId, QByteArray data);
private:
void createDcData();
void checkRequestsByTimer();
bool rpcErrorOccured(mtpRequestId requestId, const RPCFailHandlerPtr &onFail, const RPCError &err);
const not_null<Instance*> _instance;
const ShiftedDcId _shiftedDcId = 0;
Dcenter *_dc = nullptr;
std::unique_ptr<Connection> _connection;
@ -391,7 +394,6 @@ private:
SessionData _data;
std::shared_ptr<Dcenter> _dc;
AuthKeyPtr _dcKeyForCheck;
crl::time _msSendCall = 0;