Ton

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 "output-queue-merger.h"
20

21
namespace block {
22

23
/*
24
 * 
25
 *  OUTPUT QUEUE MERGER 
26
 * 
27
 */
28

29
bool OutputQueueMerger::MsgKeyValue::operator<(const MsgKeyValue& other) const {
30
  return lt < other.lt ||
31
         (lt == other.lt && td::bitstring::bits_memcmp(key.cbits() + 96, other.key.cbits() + 96, 256) < 0);
32
}
33

34
bool OutputQueueMerger::MsgKeyValue::less(const std::unique_ptr<MsgKeyValue>& he1,
35
                                          const std::unique_ptr<MsgKeyValue>& he2) {
36
  return *he1 < *he2;
37
}
38

39
bool OutputQueueMerger::MsgKeyValue::greater(const std::unique_ptr<MsgKeyValue>& he1,
40
                                             const std::unique_ptr<MsgKeyValue>& he2) {
41
  return *he2 < *he1;
42
}
43

44
OutputQueueMerger::MsgKeyValue::MsgKeyValue(td::ConstBitPtr key_pfx, int key_pfx_len, int _src, Ref<vm::Cell> node)
45
    : source(_src) {
46
  unpack_node(key_pfx, key_pfx_len, std::move(node));
47
}
48

49
OutputQueueMerger::MsgKeyValue::MsgKeyValue(int _src, Ref<vm::Cell> node) : source(_src) {
50
  unpack_node(td::ConstBitPtr{nullptr}, 0, std::move(node));
51
}
52

53
bool OutputQueueMerger::MsgKeyValue::invalidate() {
54
  msg.clear();
55
  lt = 0;
56
  source = -1;
57
  return false;
58
}
59

60
ton::LogicalTime OutputQueueMerger::MsgKeyValue::get_node_lt(Ref<vm::Cell> node, int key_pfx_len) {
61
  if (node.is_null() || (unsigned)key_pfx_len > (unsigned)max_key_len) {
62
    return std::numeric_limits<td::uint64>::max();
63
  }
64
  vm::dict::LabelParser label{std::move(node), max_key_len - key_pfx_len, vm::dict::LabelParser::chk_size};
65
  if (!label.is_valid()) {
66
    return std::numeric_limits<td::uint64>::max();
67
  }
68
  label.skip_label();
69
  return label.remainder->prefetch_ulong(64);
70
}
71

72
bool OutputQueueMerger::MsgKeyValue::unpack_node(td::ConstBitPtr key_pfx, int key_pfx_len, Ref<vm::Cell> node) {
73
  if (node.is_null() || (unsigned)key_pfx_len >= (unsigned)max_key_len) {
74
    return invalidate();
75
  }
76
  if (!key_pfx.is_null()) {
77
    td::bitstring::bits_memcpy(key.bits(), key_pfx, key_pfx_len);
78
  }
79
  vm::dict::LabelParser label{std::move(node), max_key_len - key_pfx_len, vm::dict::LabelParser::chk_size};
80
  if (!label.is_valid()) {
81
    return invalidate();
82
  }
83
  label.extract_label_to(key.bits() + key_pfx_len);
84
  key_len = key_pfx_len + label.l_bits;
85
  msg = std::move(label.remainder);
86
  if (!msg.write().fetch_uint_to(64, lt)) {
87
    return invalidate();
88
  }
89
  if (is_fork() && msg->size_ext() != 0x20000) {
90
    return invalidate();
91
  }
92
  return true;
93
}
94

95
bool OutputQueueMerger::MsgKeyValue::replace_with_child(bool child_idx) {
96
  if (!is_fork() || msg.is_null() || msg->size_ext() != 0x20000) {
97
    return false;
98
  }
99
  key[key_len] = child_idx;
100
  return unpack_node(td::ConstBitPtr{nullptr}, key_len + 1, msg->prefetch_ref(child_idx));
101
}
102

103
bool OutputQueueMerger::MsgKeyValue::replace_by_prefix(td::ConstBitPtr req_pfx, int req_pfx_len) {
104
  do {
105
    if (td::bitstring::bits_memcmp(req_pfx, key.cbits(), std::min(req_pfx_len, key_len))) {
106
      return false;
107
    }
108
    if (key_len >= req_pfx_len) {
109
      return true;
110
    }
111
  } while (replace_with_child(req_pfx[key_len]));
112
  return false;
113
}
114

115
bool OutputQueueMerger::MsgKeyValue::split(MsgKeyValue& second) {
116
  if (!is_fork() || msg.is_null()) {
117
    return false;
118
  }
119
  unsigned long long keep_lt = lt;
120
  unsigned long long left_lt = get_node_lt(msg->prefetch_ref(0), key_len + 1);
121
  bool sw = (left_lt == lt);
122
  second.source = source;
123
  key[key_len] = sw;
124
  if (!second.unpack_node(key.cbits(), key_len + 1, msg->prefetch_ref(sw))) {
125
    return false;
126
  }
127
  key[key_len] = 1 - sw;
128
  if (!unpack_node(td::ConstBitPtr{nullptr}, key_len + 1, msg->prefetch_ref(1 - sw))) {
129
    return false;
130
  }
131
  if (lt != keep_lt || second.lt < keep_lt) {
132
    return false;
133
  }
134
  return true;
135
}
136

137
bool OutputQueueMerger::add_root(int src, Ref<vm::Cell> outmsg_root) {
138
  if (outmsg_root.is_null()) {
139
    return true;
140
  }
141
  //block::gen::HashmapAug{352, block::gen::t_EnqueuedMsg, block::gen::t_uint64}.print_ref(std::cerr, outmsg_root);
142
  auto kv = std::make_unique<MsgKeyValue>(src, std::move(outmsg_root));
143
  if (kv->replace_by_prefix(common_pfx.cbits(), common_pfx_len)) {
144
    heap.push_back(std::move(kv));
145
  }
146
  return true;
147
}
148

149
OutputQueueMerger::OutputQueueMerger(ton::ShardIdFull _queue_for, std::vector<Neighbor> _neighbors)
150
    : queue_for(_queue_for), neighbors(std::move(_neighbors)), eof(false), failed(false) {
151
  init();
152
}
153

154
OutputQueueMerger::OutputQueueMerger(ton::ShardIdFull _queue_for, std::vector<block::McShardDescr> _neighbors)
155
    : queue_for(_queue_for), eof(false), failed(false) {
156
  for (auto& nb : _neighbors) {
157
    neighbors.emplace_back(nb.top_block_id(), nb.outmsg_root, nb.is_disabled());
158
  }
159
  init();
160
}
161

162
void OutputQueueMerger::init() {
163
  common_pfx.bits().store_int(queue_for.workchain, 32);
164
  int l = queue_for.pfx_len();
165
  td::bitstring::bits_store_long_top(common_pfx.bits() + 32, queue_for.shard, l);
166
  common_pfx_len = 32 + l;
167
  int i = 0;
168
  for (Neighbor& neighbor : neighbors) {
169
    if (!neighbor.disabled_) {
170
      LOG(DEBUG) << "adding " << (neighbor.outmsg_root_.is_null() ? "" : "non-") << "empty output queue for neighbor #"
171
                 << i << " (" << neighbor.block_id_.to_str() << ")";
172
      add_root(i++, neighbor.outmsg_root_);
173
    } else {
174
      LOG(DEBUG) << "skipping output queue for disabled neighbor #" << i;
175
      i++;
176
    }
177
  }
178
  std::make_heap(heap.begin(), heap.end(), MsgKeyValue::greater);
179
  eof = heap.empty();
180
  if (!eof) {
181
    load();
182
  }
183
}
184

185
OutputQueueMerger::MsgKeyValue* OutputQueueMerger::cur() {
186
  return eof ? nullptr : msg_list.at(pos).get();
187
}
188

189
std::unique_ptr<OutputQueueMerger::MsgKeyValue> OutputQueueMerger::extract_cur() {
190
  return eof ? std::unique_ptr<MsgKeyValue>{} : std::move(msg_list.at(pos));
191
}
192

193
bool OutputQueueMerger::next() {
194
  if (eof) {
195
    return false;
196
  } else if (++pos < msg_list.size() || load()) {
197
    return true;
198
  } else {
199
    eof = true;
200
    return false;
201
  }
202
}
203

204
bool OutputQueueMerger::load() {
205
  if (heap.empty() || failed) {
206
    return false;
207
  }
208
  unsigned long long lt = heap[0]->lt;
209
  std::size_t orig_size = msg_list.size();
210
  do {
211
    while (heap[0]->is_fork()) {
212
      auto other = std::make_unique<MsgKeyValue>();
213
      if (!heap[0]->split(*other)) {
214
        failed = true;
215
        return false;
216
      }
217
      heap.push_back(std::move(other));
218
      std::push_heap(heap.begin(), heap.end(), MsgKeyValue::greater);
219
    }
220
    assert(heap[0]->lt == lt);
221
    std::pop_heap(heap.begin(), heap.end(), MsgKeyValue::greater);
222
    msg_list.push_back(std::move(heap.back()));
223
    heap.pop_back();
224
  } while (!heap.empty() && heap[0]->lt <= lt);
225
  std::sort(msg_list.begin() + orig_size, msg_list.end(), MsgKeyValue::less);
226
  return true;
227
}
228

229
}  // namespace block
230