tqueue.cpp

//
// Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2020
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
#include "td/db/binlog/Binlog.h"
#include "td/db/binlog/BinlogEvent.h"
#include "td/db/binlog/BinlogHelper.h"
#include "td/db/TQueue.h"

#include "td/utils/buffer.h"
#include "td/utils/int_types.h"
#include "td/utils/misc.h"
#include "td/utils/port/path.h"
#include "td/utils/Random.h"
#include "td/utils/Slice.h"
#include "td/utils/Span.h"
#include "td/utils/Status.h"
#include "td/utils/tests.h"
#include "td/utils/VectorQueue.h"

#include <map>
#include <set>

TEST(TQueue, hands) {
  td::TQueue::Event events[100];
  auto events_span = td::MutableSpan<td::TQueue::Event>(events, 100);

  auto tqueue = td::TQueue::create();
  auto qid = 12;
  ASSERT_EQ(true, tqueue->get_head(qid).empty());
  ASSERT_EQ(true, tqueue->get_tail(qid).empty());
  tqueue->push(qid, "hello", 1, 0, td::TQueue::EventId());
  auto head = tqueue->get_head(qid);
  auto tail = tqueue->get_tail(qid);
  ASSERT_EQ(head.next().ok(), tail);
  ASSERT_EQ(1u, tqueue->get(qid, head, true, 0, events_span).move_as_ok());
  ASSERT_EQ(1u, tqueue->get(qid, head, true, 0, events_span).move_as_ok());
  ASSERT_EQ(1u, tqueue->get(qid, tail, false, 0, events_span).move_as_ok());
  ASSERT_EQ(1u, tqueue->get(qid, head, true, 0, events_span).move_as_ok());
  ASSERT_EQ(0u, tqueue->get(qid, tail, true, 0, events_span).move_as_ok());
  ASSERT_EQ(0u, tqueue->get(qid, head, true, 0, events_span).move_as_ok());
}

class TestTQueue {
 public:
  using EventId = td::TQueue::EventId;

  static td::CSlice binlog_path() {
    return td::CSlice("tqueue_binlog");
  }

  TestTQueue() {
    baseline_ = td::TQueue::create();

    memory_ = td::TQueue::create();
    auto memory_storage = td::make_unique<td::TQueueMemoryStorage>();
    memory_storage_ = memory_storage.get();
    memory_->set_callback(std::move(memory_storage));

    binlog_ = td::TQueue::create();
    auto tqueue_binlog = td::make_unique<td::TQueueBinlog<td::Binlog>>();
    td::Binlog::destroy(binlog_path()).ensure();
    auto binlog = std::make_shared<td::Binlog>();
    binlog->init(binlog_path().str(), [&](const td::BinlogEvent &event) { UNREACHABLE(); }).ensure();
    tqueue_binlog->set_binlog(std::move(binlog));
    binlog_->set_callback(std::move(tqueue_binlog));
  }

  void restart(td::Random::Xorshift128plus &rnd, td::int32 now) {
    if (rnd.fast(0, 10) == 0) {
      baseline_->run_gc(now);
    }

    memory_->extract_callback().release();
    auto memory_storage = td::unique_ptr<td::TQueueMemoryStorage>(memory_storage_);
    memory_ = td::TQueue::create();
    memory_storage->replay(*memory_);
    memory_->set_callback(std::move(memory_storage));
    if (rnd.fast(0, 10) == 0) {
      memory_->run_gc(now);
    }

    if (rnd.fast(0, 30) != 0) {
      return;
    }

    LOG(INFO) << "Restart binlog";
    binlog_ = td::TQueue::create();
    auto tqueue_binlog = td::make_unique<td::TQueueBinlog<td::Binlog>>();
    auto binlog = std::make_shared<td::Binlog>();
    binlog
        ->init(binlog_path().str(),
               [&](const td::BinlogEvent &event) { tqueue_binlog->replay(event, *binlog_).ignore(); })
        .ensure();
    tqueue_binlog->set_binlog(std::move(binlog));
    binlog_->set_callback(std::move(tqueue_binlog));
    if (rnd.fast(0, 2) == 0) {
      binlog_->run_gc(now);
    }
  }

  EventId push(td::TQueue::QueueId queue_id, td::string data, td::int32 expires_at, EventId new_id = EventId()) {
    auto a_id = baseline_->push(queue_id, data, expires_at, 0, new_id).move_as_ok();
    auto b_id = memory_->push(queue_id, data, expires_at, 0, new_id).move_as_ok();
    auto c_id = binlog_->push(queue_id, data, expires_at, 0, new_id).move_as_ok();
    ASSERT_EQ(a_id, b_id);
    ASSERT_EQ(a_id, c_id);
    return a_id;
  }

  void check_head_tail(td::TQueue::QueueId qid) {
    //ASSERT_EQ(baseline_->get_head(qid), memory_->get_head(qid));
    //ASSERT_EQ(baseline_->get_head(qid), binlog_->get_head(qid));
    ASSERT_EQ(baseline_->get_tail(qid), memory_->get_tail(qid));
    ASSERT_EQ(baseline_->get_tail(qid), binlog_->get_tail(qid));
  }

  void check_get(td::TQueue::QueueId qid, td::Random::Xorshift128plus &rnd, td::int32 now) {
    td::TQueue::Event a[10];
    td::MutableSpan<td::TQueue::Event> a_span(a, 10);
    td::TQueue::Event b[10];
    td::MutableSpan<td::TQueue::Event> b_span(b, 10);
    td::TQueue::Event c[10];
    td::MutableSpan<td::TQueue::Event> c_span(c, 10);

    auto a_from = baseline_->get_head(qid);
    //auto b_from = memory_->get_head(qid);
    //auto c_from = binlog_->get_head(qid);
    //ASSERT_EQ(a_from, b_from);
    //ASSERT_EQ(a_from, c_from);

    auto tmp = a_from.advance(rnd.fast(-10, 10));
    if (tmp.is_ok()) {
      a_from = tmp.move_as_ok();
    }
    baseline_->get(qid, a_from, true, now, a_span).move_as_ok();
    memory_->get(qid, a_from, true, now, b_span).move_as_ok();
    binlog_->get(qid, a_from, true, now, c_span).move_as_ok();
    ASSERT_EQ(a_span.size(), b_span.size());
    ASSERT_EQ(a_span.size(), c_span.size());
    for (size_t i = 0; i < a_span.size(); i++) {
      ASSERT_EQ(a_span[i].id, b_span[i].id);
      ASSERT_EQ(a_span[i].id, c_span[i].id);
      ASSERT_EQ(a_span[i].data, b_span[i].data);
      ASSERT_EQ(a_span[i].data, c_span[i].data);
    }
  }

 private:
  td::unique_ptr<td::TQueue> baseline_;
  td::unique_ptr<td::TQueue> memory_;
  td::unique_ptr<td::TQueue> binlog_;
  td::TQueueMemoryStorage *memory_storage_{nullptr};
};

TEST(TQueue, random) {
  using EventId = td::TQueue::EventId;
  td::Random::Xorshift128plus rnd(123);
  auto next_queue_id = [&rnd] {
    return rnd.fast(1, 10);
  };
  auto next_first_id = [&rnd] {
    if (rnd.fast(0, 3) == 0) {
      return EventId::from_int32(EventId::MAX_ID - 20).move_as_ok();
    }
    return EventId::from_int32(rnd.fast(1000000000, 1500000000)).move_as_ok();
  };

  TestTQueue q;
  td::int32 now = 1000;
  auto push_event = [&] {
    auto data = PSTRING() << rnd();
    if (rnd.fast(0, 10000) == 0) {
      data = td::string(1 << 19, '\0');
    }
    q.push(next_queue_id(), data, now + rnd.fast(-10, 10) * 10 + 5, next_first_id());
  };
  auto inc_now = [&] {
    now += 10;
  };
  auto check_head_tail = [&] {
    q.check_head_tail(next_queue_id());
  };
  auto restart = [&] {
    q.restart(rnd, now);
  };
  auto get = [&] {
    q.check_get(next_queue_id(), rnd, now);
  };
  td::RandomSteps steps({{push_event, 100}, {check_head_tail, 10}, {get, 40}, {inc_now, 5}, {restart, 1}});
  for (int i = 0; i < 100000; i++) {
    steps.step(rnd);
  }
}

TEST(TQueue, memory_leak) {
  return;
  auto tqueue = td::TQueue::create();
  auto tqueue_binlog = td::make_unique<td::TQueueBinlog<td::Binlog>>();
  std::string binlog_path = "test_tqueue.binlog";
  td::Binlog::destroy(binlog_path).ensure();
  auto binlog = std::make_shared<td::Binlog>();
  binlog->init(binlog_path, [&](const td::BinlogEvent &event) { UNREACHABLE(); }).ensure();
  tqueue_binlog->set_binlog(std::move(binlog));
  tqueue->set_callback(std::move(tqueue_binlog));

  td::int32 now = 0;
  std::vector<td::TQueue::EventId> ids;
  td::Random::Xorshift128plus rnd(123);
  int i = 0;
  while (true) {
    auto id = tqueue->push(1, "a", now + 600000, 0, {}).move_as_ok();
    ids.push_back(id);
    if (ids.size() > static_cast<std::size_t>(rnd()) % 100000) {
      auto it = static_cast<std::size_t>(rnd()) % ids.size();
      std::swap(ids.back(), ids[it]);
      tqueue->forget(1, ids.back());
      ids.pop_back();
    }
    now++;
    if (i++ % 100000 == 0) {
      LOG(ERROR) << td::BufferAllocator::get_buffer_mem() << " " << tqueue->get_size(1) << " "
                 << td::BufferAllocator::get_buffer_slice_size();
    }
  }
}