srsLTE/lib/include/srsran/upper/byte_buffer_queue.h

/**
 *
 * \section COPYRIGHT
 *
 * Copyright 2013-2021 Software Radio Systems Limited
 *
 * By using this file, you agree to the terms and conditions set
 * forth in the LICENSE file which can be found at the top level of
 * the distribution.
 *
 */

/*
 * @file byte_buffer_queue.h
 *
 * @brief Queue of unique pointers to byte buffers used in PDCP and RLC TX queues.
 *        Uses a blocking queue with bounded capacity to block higher layers
 *        when pushing uplink traffic
 */

#ifndef SRSRAN_BYTE_BUFFERQUEUE_H
#define SRSRAN_BYTE_BUFFERQUEUE_H

#include "srsran/adt/circular_buffer.h"
#include "srsran/common/block_queue.h"
#include "srsran/common/byte_buffer.h"
#include "srsran/common/common.h"
#include <functional>
#include <pthread.h>

namespace srsran {

class byte_buffer_queue
{
public:
  byte_buffer_queue(int capacity = 128) :
    queue(capacity, push_callback(unread_bytes, n_sdus), pop_callback(unread_bytes, n_sdus))
  {}

  void write(unique_byte_buffer_t msg) { queue.push_blocking(std::move(msg)); }

  srsran::error_type<unique_byte_buffer_t> try_write(unique_byte_buffer_t&& msg)
  {
    return queue.try_push(std::move(msg));
  }

  unique_byte_buffer_t read() { return queue.pop_blocking(); }

  bool try_read(unique_byte_buffer_t* msg) { return queue.try_pop(*msg); }

  void     resize(uint32_t capacity) { queue.set_size(capacity); }
  uint32_t size() { return (uint32_t)queue.size(); }
  uint32_t get_n_sdus() { return n_sdus; }

  uint32_t size_bytes() { return unread_bytes; }

  uint32_t size_tail_bytes()
  {
    uint32_t size_next = 0;
    queue.try_call_on_front([&size_next](const unique_byte_buffer_t& front_val) {
      if (front_val != nullptr) {
        size_next += front_val->N_bytes;
      }
    });
    return size_next;
  }

  // This is a hack to reset N_bytes counter when queue is corrupted (see line 89)
  void reset() { unread_bytes = 0; }

  bool is_empty() { return queue.empty(); }

  bool is_full() { return queue.full(); }

  template <typename F>
  bool apply_first(const F& func)
  {
    return queue.apply_first(func);
  }

private:
  struct push_callback {
    explicit push_callback(uint32_t& unread_bytes_, uint32_t& n_sdus_) : unread_bytes(&unread_bytes_), n_sdus(&n_sdus_)
    {}
    void operator()(const unique_byte_buffer_t& msg)
    {
      *unread_bytes += msg->N_bytes;
      (*n_sdus)++;
    }
    uint32_t* unread_bytes;
    uint32_t* n_sdus;
  };
  struct pop_callback {
    explicit pop_callback(uint32_t& unread_bytes_, uint32_t& n_sdus_) : unread_bytes(&unread_bytes_), n_sdus(&n_sdus_)
    {}
    void operator()(const unique_byte_buffer_t& msg)
    {
      if (msg == nullptr) {
        return;
      }
      *unread_bytes -= std::min(msg->N_bytes, *unread_bytes);
      *n_sdus = std::max(0, (int32_t)(*n_sdus) - 1);
    }
    uint32_t* unread_bytes;
    uint32_t* n_sdus;
  };

  uint32_t unread_bytes = 0;
  uint32_t n_sdus       = 0;

public:
  dyn_blocking_queue<unique_byte_buffer_t, push_callback, pop_callback> queue;
};

} // namespace srsran

#endif // SRSRAN_BYTE_BUFFERQUEUE_H