Ton

download-archive-slice.cpp
206 строк · 7.5 Кб
Перенос по словам
1
/*
2
    This file is part of TON Blockchain Library.
3

4
    TON Blockchain Library is free software: you can redistribute it and/or modify
5
    it under the terms of the GNU Lesser General Public License as published by
6
    the Free Software Foundation, either version 2 of the License, or
7
    (at your option) any later version.
8

9
    TON Blockchain Library is distributed in the hope that it will be useful,
10
    but WITHOUT ANY WARRANTY; without even the implied warranty of
11
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12
    GNU Lesser General Public License for more details.
13

14
    You should have received a copy of the GNU Lesser General Public License
15
    along with TON Blockchain Library.  If not, see <http://www.gnu.org/licenses/>.
16

17
    Copyright 2017-2020 Telegram Systems LLP
18
*/
19
#include "download-archive-slice.hpp"
20
#include "td/utils/port/path.h"
21
#include "td/utils/overloaded.h"
22

23
namespace ton {
24

25
namespace validator {
26

27
namespace fullnode {
28

29
DownloadArchiveSlice::DownloadArchiveSlice(
30
    BlockSeqno masterchain_seqno, std::string tmp_dir, adnl::AdnlNodeIdShort local_id,
31
    overlay::OverlayIdShort overlay_id, adnl::AdnlNodeIdShort download_from, td::Timestamp timeout,
32
    td::actor::ActorId<ValidatorManagerInterface> validator_manager, td::actor::ActorId<adnl::AdnlSenderInterface> rldp,
33
    td::actor::ActorId<overlay::Overlays> overlays, td::actor::ActorId<adnl::Adnl> adnl,
34
    td::actor::ActorId<adnl::AdnlExtClient> client, td::Promise<std::string> promise)
35
    : masterchain_seqno_(masterchain_seqno)
36
    , tmp_dir_(std::move(tmp_dir))
37
    , local_id_(local_id)
38
    , overlay_id_(overlay_id)
39
    , download_from_(download_from)
40
    , timeout_(timeout)
41
    , validator_manager_(validator_manager)
42
    , rldp_(rldp)
43
    , overlays_(overlays)
44
    , adnl_(adnl)
45
    , client_(client)
46
    , promise_(std::move(promise)) {
47
}
48

49
void DownloadArchiveSlice::abort_query(td::Status reason) {
50
  if (promise_) {
51
    promise_.set_error(std::move(reason));
52
    if (!fd_.empty()) {
53
      td::unlink(tmp_name_).ensure();
54
      fd_.close();
55
    }
56
  }
57
  stop();
58
}
59

60
void DownloadArchiveSlice::alarm() {
61
  abort_query(td::Status::Error(ErrorCode::timeout, "timeout"));
62
}
63

64
void DownloadArchiveSlice::finish_query() {
65
  if (promise_) {
66
    promise_.set_value(std::move(tmp_name_));
67
    fd_.close();
68
  }
69
  stop();
70
}
71

72
void DownloadArchiveSlice::start_up() {
73
  alarm_timestamp() = timeout_;
74

75
  auto R = td::mkstemp(tmp_dir_);
76
  if (R.is_error()) {
77
    abort_query(R.move_as_error_prefix("failed to open temp file: "));
78
    return;
79
  }
80
  auto r = R.move_as_ok();
81
  fd_ = std::move(r.first);
82
  tmp_name_ = std::move(r.second);
83

84
  if (download_from_.is_zero() && client_.empty()) {
85
    auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<std::vector<adnl::AdnlNodeIdShort>> R) {
86
      if (R.is_error()) {
87
        td::actor::send_closure(SelfId, &DownloadArchiveSlice::abort_query, R.move_as_error());
88
      } else {
89
        auto vec = R.move_as_ok();
90
        if (vec.size() == 0) {
91
          td::actor::send_closure(SelfId, &DownloadArchiveSlice::abort_query,
92
                                  td::Status::Error(ErrorCode::notready, "no nodes"));
93
        } else {
94
          td::actor::send_closure(SelfId, &DownloadArchiveSlice::got_node_to_download, vec[0]);
95
        }
96
      }
97
    });
98

99
    td::actor::send_closure(overlays_, &overlay::Overlays::get_overlay_random_peers, local_id_, overlay_id_, 1,
100
                            std::move(P));
101
  } else {
102
    got_node_to_download(download_from_);
103
  }
104
}
105

106
void DownloadArchiveSlice::got_node_to_download(adnl::AdnlNodeIdShort download_from) {
107
  download_from_ = download_from;
108

109
  auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<td::BufferSlice> R) {
110
    if (R.is_error()) {
111
      td::actor::send_closure(SelfId, &DownloadArchiveSlice::abort_query, R.move_as_error());
112
    } else {
113
      td::actor::send_closure(SelfId, &DownloadArchiveSlice::got_archive_info, R.move_as_ok());
114
    }
115
  });
116

117
  auto q = create_serialize_tl_object<ton_api::tonNode_getArchiveInfo>(masterchain_seqno_);
118
  if (client_.empty()) {
119
    td::actor::send_closure(overlays_, &overlay::Overlays::send_query, download_from_, local_id_, overlay_id_,
120
                            "get_archive_info", std::move(P), td::Timestamp::in(3.0), std::move(q));
121
  } else {
122
    td::actor::send_closure(client_, &adnl::AdnlExtClient::send_query, "get_archive_info",
123
                            create_serialize_tl_object_suffix<ton_api::tonNode_query>(std::move(q)),
124
                            td::Timestamp::in(1.0), std::move(P));
125
  }
126
}
127

128
void DownloadArchiveSlice::got_archive_info(td::BufferSlice data) {
129
  auto F = fetch_tl_object<ton_api::tonNode_ArchiveInfo>(std::move(data), true);
130
  if (F.is_error()) {
131
    abort_query(F.move_as_error_prefix("failed to parse ArchiveInfo answer"));
132
    return;
133
  }
134
  auto f = F.move_as_ok();
135

136
  bool fail = false;
137
  ton_api::downcast_call(*f.get(), td::overloaded(
138
                                       [&](const ton_api::tonNode_archiveNotFound &obj) {
139
                                         abort_query(td::Status::Error(ErrorCode::notready, "remote db not found"));
140
                                         fail = true;
141
                                       },
142
                                       [&](const ton_api::tonNode_archiveInfo &obj) { archive_id_ = obj.id_; }));
143
  if (fail) {
144
    return;
145
  }
146

147
  prev_logged_timer_ = td::Timer();
148
  LOG(INFO) << "downloading archive slice #" << masterchain_seqno_ << " from " << download_from_;
149
  get_archive_slice();
150
}
151

152
void DownloadArchiveSlice::get_archive_slice() {
153
  auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<td::BufferSlice> R) {
154
    if (R.is_error()) {
155
      td::actor::send_closure(SelfId, &DownloadArchiveSlice::abort_query, R.move_as_error());
156
    } else {
157
      td::actor::send_closure(SelfId, &DownloadArchiveSlice::got_archive_slice, R.move_as_ok());
158
    }
159
  });
160

161
  auto q = create_serialize_tl_object<ton_api::tonNode_getArchiveSlice>(archive_id_, offset_, slice_size());
162
  if (client_.empty()) {
163
    td::actor::send_closure(overlays_, &overlay::Overlays::send_query_via, download_from_, local_id_, overlay_id_,
164
                            "get_archive_slice", std::move(P), td::Timestamp::in(15.0), std::move(q),
165
                            slice_size() + 1024, rldp_);
166
  } else {
167
    td::actor::send_closure(client_, &adnl::AdnlExtClient::send_query, "get_archive_slice",
168
                            create_serialize_tl_object_suffix<ton_api::tonNode_query>(std::move(q)),
169
                            td::Timestamp::in(15.0), std::move(P));
170
  }
171
}
172

173
void DownloadArchiveSlice::got_archive_slice(td::BufferSlice data) {
174
  auto R = fd_.write(data.as_slice());
175
  if (R.is_error()) {
176
    abort_query(R.move_as_error_prefix("failed to write temp file: "));
177
    return;
178
  }
179
  if (R.move_as_ok() != data.size()) {
180
    abort_query(td::Status::Error(ErrorCode::error, "short write to temp file"));
181
    return;
182
  }
183

184
  offset_ += data.size();
185

186
  double elapsed = prev_logged_timer_.elapsed();
187
  if (elapsed > 10.0) {
188
    prev_logged_timer_ = td::Timer();
189
    LOG(INFO) << "downloading archive slice #" << masterchain_seqno_ << ": total=" << offset_ << " ("
190
              << td::format::as_size((td::uint64)(double(offset_ - prev_logged_sum_) / elapsed)) << "/s)";
191
    prev_logged_sum_ = offset_;
192
  }
193

194
  if (data.size() < slice_size()) {
195
    LOG(INFO) << "finished downloading arcrive slice #" << masterchain_seqno_ << ": total=" << offset_;
196
    finish_query();
197
  } else {
198
    get_archive_slice();
199
  }
200
}
201

202
}  // namespace fullnode
203

204
}  // namespace validator
205

206
}  // namespace ton
207
Ton

Использование cookies
