hiro/tdesktop
1
0
Fork 0
mirror of https://github.com/vale981/tdesktop synced 2025-03-06 02:01:40 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

Clear callbacks async in MTP::Instance.

Browse source 
Also fix previous build.
This commit is contained in:
John Preston 2018-01-02 16:44:12 +03:00
parent 3325106837
commit 06c724df01
6 changed files with 103 additions and 76 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Telegram/SourceFiles/history/history_item.cpp
View file

@ -41,6 +41,7 @@ Copyright (c) 2014-2017 John Preston, https://desktop.telegram.org
#include "messenger.h"
#include "mainwindow.h"
#include "window/window_controller.h"
#include "core/crash_reports.h"
namespace {

60
Telegram/SourceFiles/mtproto/connection.cpp
View file

@ -1580,10 +1580,17 @@ ConnectionPrivate::HandleResult ConnectionPrivate::handleOneReceived(const mtpPr
_pingId = 0;
}
int32 errorCode = data.verror_code.v;
if (errorCode == 16 || errorCode == 17 || errorCode == 32 || errorCode == 33 || errorCode == 64) { // can handle
bool needResend = (errorCode == 16 || errorCode == 17); // bad msg_id
if (false
|| errorCode == 16
|| errorCode == 17
|| errorCode == 32
|| errorCode == 33
|| errorCode == 64) { // can handle
const auto needResend = false
|| (errorCode == 16) // bad msg_id
|| (errorCode == 17) // bad msg_id
|| (errorCode == 64); // bad container
if (errorCode == 64) { // bad container!
needResend = true;
if (cDebug()) {
mtpRequest request;
{
@ -1600,7 +1607,7 @@ ConnectionPrivate::HandleResult ConnectionPrivate::handleOneReceived(const mtpPr
if (request) {
if (mtpRequestData::isSentContainer(request)) {
QStringList lst;
const mtpMsgId *ids = (const mtpMsgId *)(request->constData() + 8);
const auto ids = (const mtpMsgId *)(request->constData() + 8);
for (uint32 i = 0, l = (request->size() - 8) >> 1; i < l; ++i) {
lst.push_back(QString::number(ids[i]));
}
@ -1613,11 +1620,14 @@ ConnectionPrivate::HandleResult ConnectionPrivate::handleOneReceived(const mtpPr
}
if (!wasSent(resendId)) {
DEBUG_LOG(("Message Error: such message was not sent recently %1").arg(resendId));
return (badTime ? HandleResult::Ignored : HandleResult::Success);
DEBUG_LOG(("Message Error: "
"such message was not sent recently %1").arg(resendId));
return badTime
? HandleResult::Ignored
: HandleResult::Success;
}
if (needResend) { // bad msg_id
if (needResend) { // bad msg_id or bad container
if (serverSalt) sessionData->setSalt(serverSalt);
unixtimeSet(serverTime, true);
@ -1634,17 +1644,25 @@ ConnectionPrivate::HandleResult ConnectionPrivate::handleOneReceived(const mtpPr
return HandleResult::ResetSession;
}
} else { // fatal (except 48, but it must not get here)
mtpMsgId resendId = data.vbad_msg_id.v;
mtpRequestId requestId = wasSent(resendId);
const auto badMsgId = mtpMsgId(data.vbad_msg_id.v);
const auto requestId = wasSent(resendId);
if (requestId) {
LOG(("Message Error: bad message notification received, msgId %1, error_code %2, fatal: clearing callbacks").arg(data.vbad_msg_id.v).arg(errorCode));
_instance->clearCallbacksDelayed(RPCCallbackClears(
1,
RPCCallbackClear(requestId, -errorCode)));
LOG(("Message Error: "
"bad message notification received, "
"msgId %1, error_code %2, fatal: clearing callbacks"
).arg(badMsgId
).arg(errorCode
));
_instance->clearCallbacksDelayed({ 1, RPCCallbackClear(
requestId,
-errorCode) });
} else {
DEBUG_LOG(("Message Error: such message was not sent recently %1").arg(resendId));
DEBUG_LOG(("Message Error: "
"such message was not sent recently %1").arg(badMsgId));
}
return (badTime ? HandleResult::Ignored : HandleResult::Success);
return badTime
? HandleResult::Ignored
: HandleResult::Success;
}
} return HandleResult::Success;
@ -2101,7 +2119,7 @@ void ConnectionPrivate::requestsAcked(const QVector<MTPlong> &ids, bool byRespon
DEBUG_LOG(("Message Info: requests acked, ids %1").arg(LogIdsVector(ids)));
RPCCallbackClears clearedAcked;
auto clearedBecauseTooOld = std::vector<RPCCallbackClear>();
QVector<MTPlong> toAckMore;
{
QWriteLocker locker1(sessionData->wereAckedMutex());
@ -2177,10 +2195,10 @@ void ConnectionPrivate::requestsAcked(const QVector<MTPlong> &ids, bool byRespon
uint32 ackedCount = wereAcked.size();
if (ackedCount > MTPIdsBufferSize) {
DEBUG_LOG(("Message Info: removing some old acked sent msgIds %1").arg(ackedCount - MTPIdsBufferSize));
clearedAcked.reserve(ackedCount - MTPIdsBufferSize);
clearedBecauseTooOld.reserve(ackedCount - MTPIdsBufferSize);
while (ackedCount-- > MTPIdsBufferSize) {
mtpRequestIdsMap::iterator i(wereAcked.begin());
clearedAcked.push_back(RPCCallbackClear(
auto i = wereAcked.begin();
clearedBecauseTooOld.push_back(RPCCallbackClear(
i.key(),
RPCError::TimeoutError));
wereAcked.erase(i);
@ -2188,8 +2206,8 @@ void ConnectionPrivate::requestsAcked(const QVector<MTPlong> &ids, bool byRespon
}
}
if (clearedAcked.size()) {
_instance->clearCallbacksDelayed(clearedAcked);
if (!clearedBecauseTooOld.empty()) {
_instance->clearCallbacksDelayed(std::move(clearedBecauseTooOld));
}
if (toAckMore.size()) {

98
Telegram/SourceFiles/mtproto/mtp_instance.cpp
View file

@ -76,15 +76,13 @@ public:
std::unique_ptr<internal::Connection> &&connection);
void connectionFinished(internal::Connection *connection);
void registerRequest(mtpRequestId requestId, int32 dcWithShift);
void registerRequest(mtpRequestId requestId, ShiftedDcId dcWithShift);
void unregisterRequest(mtpRequestId requestId);
mtpRequestId storeRequest(
mtpRequest &request,
RPCResponseHandler &&callbacks);
mtpRequest getRequest(mtpRequestId requestId);
void clearCallbacks(mtpRequestId requestId, int32 errorCode = RPCError::NoError); // 0 - do not toggle onError callback
void clearCallbacksDelayed(const RPCCallbackClears &requestIds);
void performDelayedClear();
void clearCallbacksDelayed(std::vector<RPCCallbackClear> &&ids);
void execCallback(mtpRequestId requestId, const mtpPrime *from, const mtpPrime *end);
bool hasCallbacks(mtpRequestId requestId);
void globalCallback(const mtpPrime *from, const mtpPrime *end);
@ -140,6 +138,12 @@ private:
void cdnConfigLoadDone(const MTPCdnConfig &result);
bool cdnConfigLoadFail(const RPCError &error);
// RPCError::NoError means do not toggle onError callback.
void clearCallbacks(
mtpRequestId requestId,
int32 errorCode = RPCError::NoError);
void clearCallbacks(const std::vector<RPCCallbackClear> &ids);
void checkDelayedRequests();
not_null<Instance*> _instance;
@ -185,9 +189,6 @@ private:
std::map<DcId, std::vector<mtpRequestId>> _authWaiters;
QMutex _toClearLock;
RPCCallbackClears _toClear;
RPCResponseHandler _globalHandler;
base::lambda<void(ShiftedDcId shiftedDcId, int32 state)> _stateChangedHandler;
base::lambda<void(ShiftedDcId shiftedDcId)> _sessionResetHandler;
@ -681,12 +682,11 @@ void Instance::Private::checkDelayedRequests() {
}
}
void Instance::Private::registerRequest(mtpRequestId requestId, int32 dcWithShift) {
{
QMutexLocker locker(&_requestByDcLock);
_requestsByDc.emplace(requestId, dcWithShift);
}
performDelayedClear(); // need to do it somewhere...
void Instance::Private::registerRequest(
mtpRequestId requestId,
ShiftedDcId dcWithShift) {
QMutexLocker locker(&_requestByDcLock);
_requestsByDc.emplace(requestId, dcWithShift);
}
void Instance::Private::unregisterRequest(mtpRequestId requestId) {
@ -748,46 +748,50 @@ void Instance::Private::clearCallbacks(mtpRequestId requestId, int32 errorCode)
}
}
void Instance::Private::clearCallbacksDelayed(const RPCCallbackClears &requestIds) {
uint32 idsCount = requestIds.size();
if (!idsCount) return;
void Instance::Private::clearCallbacksDelayed(
std::vector<RPCCallbackClear> &&ids) {
if (ids.empty()) {
return;
}
if (cDebug()) {
QString idsStr = QString("%1").arg(requestIds[0].requestId);
for (uint32 i = 1; i < idsCount; ++i) {
idsStr += QString(", %1").arg(requestIds[i].requestId);
auto idsString = QStringList();
idsString.reserve(ids.size());
for (auto &value : ids) {
idsString.push_back(QString::number(value.requestId));
}
DEBUG_LOG(("RPC Info: clear callbacks delayed, msgIds: %1").arg(idsStr));
DEBUG_LOG(("RPC Info: clear callbacks delayed, msgIds: %1"
).arg(idsString.join(", ")));
}
QMutexLocker lock(&_toClearLock);
uint32 toClearNow = _toClear.size();
if (toClearNow) {
_toClear.resize(toClearNow + idsCount);
memcpy(_toClear.data() + toClearNow, requestIds.constData(), idsCount * sizeof(RPCCallbackClear));
} else {
_toClear = requestIds;
}
crl::on_main(_instance, [this, list = std::move(ids)] {
clearCallbacks(list);
});
}
void Instance::Private::performDelayedClear() {
QMutexLocker lock(&_toClearLock);
if (!_toClear.isEmpty()) {
for (auto &clearRequest : _toClear) {
if (cDebug()) {
QMutexLocker locker(&_parserMapLock);
if (_parserMap.find(clearRequest.requestId) != _parserMap.end()) {
DEBUG_LOG(("RPC Info: clearing delayed callback %1, error code %2").arg(clearRequest.requestId).arg(clearRequest.errorCode));
}
void Instance::Private::clearCallbacks(
const std::vector<RPCCallbackClear> &ids) {
Expects(!ids.empty());
for (const auto &clearRequest : ids) {
if (cDebug()) {
QMutexLocker locker(&_parserMapLock);
if (_parserMap.find(clearRequest.requestId) != _parserMap.end()) {
DEBUG_LOG(("RPC Info: "
"clearing delayed callback %1, error code %2"
).arg(clearRequest.requestId
).arg(clearRequest.errorCode));
}
clearCallbacks(clearRequest.requestId, clearRequest.errorCode);
unregisterRequest(clearRequest.requestId);
}
_toClear.clear();
clearCallbacks(clearRequest.requestId, clearRequest.errorCode);
unregisterRequest(clearRequest.requestId);
}
}
void Instance::Private::execCallback(mtpRequestId requestId, const mtpPrime *from, const mtpPrime *end) {
void Instance::Private::execCallback(
mtpRequestId requestId,
const mtpPrime *from,
const mtpPrime *end) {
RPCResponseHandler h;
{
QMutexLocker locker(&_parserMapLock);
@ -1023,7 +1027,9 @@ bool Instance::Private::onErrorDefault(mtpRequestId requestId, const RPCError &e
request = it->second;
}
if (auto session = getSession(newdcWithShift)) {
registerRequest(requestId, (dcWithShift < 0) ? -newdcWithShift : newdcWithShift);
registerRequest(
requestId,
(dcWithShift < 0) ? -newdcWithShift : newdcWithShift);
session->sendPrepared(request);
}
return true;
@ -1424,7 +1430,9 @@ void Instance::onSessionReset(ShiftedDcId dcWithShift) {
_private->onSessionReset(dcWithShift);
}
void Instance::registerRequest(mtpRequestId requestId, ShiftedDcId dcWithShift) {
void Instance::registerRequest(
mtpRequestId requestId,
ShiftedDcId dcWithShift) {
_private->registerRequest(requestId, dcWithShift);
}
@ -1438,8 +1446,8 @@ mtpRequest Instance::getRequest(mtpRequestId requestId) {
return _private->getRequest(requestId);
}
void Instance::clearCallbacksDelayed(const RPCCallbackClears &requestIds) {
_private->clearCallbacksDelayed(requestIds);
void Instance::clearCallbacksDelayed(std::vector<RPCCallbackClear> &&ids) {
_private->clearCallbacksDelayed(std::move(ids));
}
void Instance::execCallback(mtpRequestId requestId, const mtpPrime *from, const mtpPrime *end) {

2
Telegram/SourceFiles/mtproto/mtp_instance.h
View file

@ -135,7 +135,7 @@ public:
mtpRequest &request,
RPCResponseHandler &&callbacks);
mtpRequest getRequest(mtpRequestId requestId);
void clearCallbacksDelayed(const RPCCallbackClears &requestIds);
void clearCallbacksDelayed(std::vector<RPCCallbackClear> &&ids);
void execCallback(mtpRequestId requestId, const mtpPrime *from, const mtpPrime *end);
bool hasCallbacks(mtpRequestId requestId);

6
Telegram/SourceFiles/mtproto/rpc_sender.h
View file

@ -270,7 +270,9 @@ private:
};
struct RPCCallbackClear {
RPCCallbackClear(mtpRequestId id = 0, int32 code = RPCError::NoError) : requestId(id), errorCode(code) {
RPCCallbackClear(mtpRequestId id , int32 code = RPCError::NoError)
: requestId(id)
, errorCode(code) {
}
mtpRequestId requestId;
@ -278,8 +280,6 @@ struct RPCCallbackClear {
};
using RPCCallbackClears = QVector<RPCCallbackClear> ;
template <typename TReturn>
inline RPCDoneHandlerPtr rpcDone(TReturn (*onDone)(const mtpPrime *, const mtpPrime *)) { // done(from, end)
return RPCDoneHandlerPtr(new RPCDoneHandlerBare<TReturn>(onDone));

12
Telegram/SourceFiles/mtproto/session.cpp
View file

@ -56,7 +56,7 @@ void SessionData::setKey(const AuthKeyPtr &key) {
}
void SessionData::clear(Instance *instance) {
RPCCallbackClears clearCallbacks;
auto clearCallbacks = std::vector<RPCCallbackClear>();
{
QReadLocker locker1(haveSentMutex()), locker2(toResendMutex()), locker3(haveReceivedMutex()), locker4(wereAckedMutex());
auto receivedResponsesEnd = _receivedResponses.cend();
@ -96,7 +96,7 @@ void SessionData::clear(Instance *instance) {
QWriteLocker locker(receivedIdsMutex());
_receivedIds.clear();
}
instance->clearCallbacksDelayed(clearCallbacks);
instance->clearCallbacksDelayed(std::move(clearCallbacks));
}
Session::Session(not_null<Instance*> instance, ShiftedDcId shiftedDcId) : QObject()
@ -305,12 +305,12 @@ void Session::checkRequestsByTimer() {
}
}
if (!removingIds.isEmpty()) {
RPCCallbackClears clearCallbacks;
auto clearCallbacks = std::vector<RPCCallbackClear>();
{
QWriteLocker locker(data.haveSentMutex());
mtpRequestMap &haveSent(data.haveSentMap());
auto &haveSent = data.haveSentMap();
for (uint32 i = 0, l = removingIds.size(); i < l; ++i) {
mtpRequestMap::iterator j = haveSent.find(removingIds[i]);
auto j = haveSent.find(removingIds[i]);
if (j != haveSent.cend()) {
if (j.value()->requestId) {
clearCallbacks.push_back(j.value()->requestId);
@ -319,7 +319,7 @@ void Session::checkRequestsByTimer() {
}
}
}
_instance->clearCallbacksDelayed(clearCallbacks);
_instance->clearCallbacksDelayed(std::move(clearCallbacks));
}
}