aos/ipc_lib/queue.cc

#if !AOS_DEBUG
#undef NDEBUG
#define NDEBUG
#endif

#include "aos/ipc_lib/queue.h"

#include <assert.h>
#include <errno.h>
#include <stdio.h>
#include <string.h>

#include <algorithm>
#include <memory>

#include "aos/ipc_lib/core_lib.h"
#include "aos/type_traits/type_traits.h"

namespace aos {
namespace {

namespace chrono = ::std::chrono;

static_assert(shm_ok<RawQueue>::value,
              "RawQueue instances go into shared memory");

const bool kReadDebug = false;
const bool kWriteDebug = false;
const bool kRefDebug = false;
const bool kFetchDebug = false;
const bool kReadIndexDebug = false;

// The number of extra messages the pool associated with each queue will be able
// to hold (for readers who are slow about freeing them or who leak one when
// they get killed).
const int kExtraMessages = 20;

}  // namespace

constexpr Options<RawQueue>::Option RawQueue::kPeek;
constexpr Options<RawQueue>::Option RawQueue::kFromEnd;
constexpr Options<RawQueue>::Option RawQueue::kNonBlock;
constexpr Options<RawQueue>::Option RawQueue::kBlock;
constexpr Options<RawQueue>::Option RawQueue::kOverride;

// This is what gets stuck in before each queue message in memory. It is always
// allocated aligned to 8 bytes and its size has to maintain that alignment for
// the message that follows immediately.
struct RawQueue::MessageHeader {
  MessageHeader *next;

  // Gets the message header immediately preceding msg.
  static MessageHeader *Get(const void *msg) {
    return reinterpret_cast<MessageHeader *>(__builtin_assume_aligned(
        static_cast<uint8_t *>(const_cast<void *>(msg)) - sizeof(MessageHeader),
        alignof(MessageHeader)));
  }

  int32_t ref_count() const {
    return __atomic_load_n(&ref_count_, __ATOMIC_RELAXED);
  }
  void set_ref_count(int32_t val) {
    __atomic_store_n(&ref_count_, val, __ATOMIC_RELAXED);
  }

  void ref_count_sub() { __atomic_sub_fetch(&ref_count_, 1, __ATOMIC_RELAXED); }
  void ref_count_add() { __atomic_add_fetch(&ref_count_, 1, __ATOMIC_RELAXED); }

 private:
  // This gets accessed with atomic instructions without any
  // locks held by various member functions.
  int32_t ref_count_;

// Padding to make the total size 8 bytes if we have 4-byte pointers or bump
// it to 16 if a pointer is 8 bytes by itself.
#if __SIZEOF_POINTER__ == 8
#ifdef __clang__
  // Clang is smart enough to realize this is unused, but GCC doesn't like the
  // attribute here...
  __attribute__((unused))
#endif
  char padding[4];
#elif __SIZEOF_POINTER__ == 4
// No padding needed to get 8 byte total size.
#else
#error Unknown pointer size.
#endif
};

inline int RawQueue::index_add1(int index) {
  // Doing it this way instead of with % is more efficient on ARM.
  int r = index + 1;
  assert(index <= data_length_);
  if (r == data_length_) {
    return 0;
  } else {
    return r;
  }
}

void RawQueue::DecrementMessageReferenceCount(const void *msg) {
  MessageHeader *header = MessageHeader::Get(msg);
  header->ref_count_sub();
  if (kRefDebug) {
    printf("%p ref dec count: %p count=%d\n", this, msg, header->ref_count());
  }

  // The only way it should ever be 0 is if we were the last one to decrement,
  // in which case nobody else should have it around to re-increment it or
  // anything in the middle, so this is safe to do not atomically with the
  // decrement.
  if (header->ref_count() == 0) {
    DoFreeMessage(msg);
  } else {
    assert(header->ref_count() > 0);
  }
}

inline void RawQueue::IncrementMessageReferenceCount(const void *msg) const {
  MessageHeader *const header = MessageHeader::Get(msg);
  header->ref_count_add();
  if (kRefDebug) {
    printf("%p ref inc count: %p\n", this, msg);
  }
}

inline void RawQueue::DoFreeMessage(const void *msg) {
  MessageHeader *header = MessageHeader::Get(msg);
  if (kRefDebug) {
    printf("%p ref free to %p: %p\n", this, recycle_, msg);
  }

  if (__builtin_expect(recycle_ != nullptr, 0)) {
    void *const new_msg = recycle_->GetMessage();
    if (new_msg == nullptr) {
      fprintf(stderr,
              "queue: couldn't get a message"
              " for recycle queue %p\n",
              recycle_);
    } else {
      header->ref_count_add();
      if (!recycle_->WriteMessage(const_cast<void *>(msg), kOverride)) {
        fprintf(stderr,
                "queue: %p->WriteMessage(%p, kOverride) failed."
                " aborting\n",
                recycle_, msg);
        printf("see stderr\n");
        abort();
      }
      msg = new_msg;
      header = MessageHeader::Get(new_msg);
    }
  }

  // This works around GCC bug 60272 (fixed in 4.8.3).
  // new_next should just get replaced with header->next (and the body of the
  // loop should become empty).
  // The bug is that the store to new_next after the compare/exchange is
  // unconditional but it should only be if it fails, which could mean
  // overwriting what somebody else who preempted us right then changed it to.
  // TODO(brians): Get rid of this workaround once we get a new enough GCC.
  MessageHeader *new_next = __atomic_load_n(&free_messages_, __ATOMIC_RELAXED);
  do {
    header->next = new_next;
  } while (__builtin_expect(
      !__atomic_compare_exchange_n(&free_messages_, &new_next, header, true,
                                   __ATOMIC_RELEASE, __ATOMIC_RELAXED),
      0));
}

void *RawQueue::GetMessage() {
  MessageHeader *header = __atomic_load_n(&free_messages_, __ATOMIC_RELAXED);
  do {
    if (__builtin_expect(header == nullptr, 0)) {
      AOS_LOG(FATAL, "overused pool of queue %p (%s)\n", this, name_);
    }
  } while (__builtin_expect(
      !__atomic_compare_exchange_n(&free_messages_, &header, header->next, true,
                                   __ATOMIC_ACQ_REL, __ATOMIC_RELAXED),
      0));
  void *msg = reinterpret_cast<uint8_t *>(header + 1);
  // It might be uninitialized, 0 from a previous use, or 1 from previously
  // being recycled.
  header->set_ref_count(1);
  if (kRefDebug) {
    printf("%p ref alloc: %p\n", this, msg);
  }
  return msg;
}

RawQueue::RawQueue(const char *name, size_t length, int hash, int queue_length)
    : readable_(&data_lock_), writable_(&data_lock_) {
  static_assert(shm_ok<RawQueue::MessageHeader>::value,
                "the whole point is to stick it in shared memory");
  static_assert((sizeof(RawQueue::MessageHeader) % 8) == 0,
                "need to revalidate size/alignent assumptions");

  if (queue_length < 1) {
    AOS_LOG(FATAL, "queue length %d of %s needs to be at least 1\n",
            queue_length, name);
  }

  const size_t name_size = strlen(name) + 1;
  char *temp = static_cast<char *>(shm_malloc(name_size));
  memcpy(temp, name, name_size);
  name_ = temp;
  length_ = length;
  hash_ = hash;
  queue_length_ = queue_length;

  next_ = NULL;
  recycle_ = NULL;

  if (kFetchDebug) {
    printf("initializing name=%s, length=%zd, hash=%d, queue_length=%d\n", name,
           length, hash, queue_length);
  }

  data_length_ = queue_length + 1;
  data_ = static_cast<void **>(shm_malloc(sizeof(void *) * data_length_));
  data_start_ = 0;
  data_end_ = 0;
  messages_ = 0;

  msg_length_ = length + sizeof(MessageHeader);

  // Create all of the messages for the free list and stick them on.
  {
    MessageHeader *previous = nullptr;
    for (int i = 0; i < queue_length + kExtraMessages; ++i) {
      MessageHeader *const message =
          static_cast<MessageHeader *>(shm_malloc(msg_length_));
      free_messages_ = message;
      message->next = previous;
      previous = message;
    }
  }

  readable_waiting_ = false;

  if (kFetchDebug) {
    printf("made queue %s\n", name);
  }
}

RawQueue *RawQueue::Fetch(const char *name, size_t length, int hash,
                          int queue_length) {
  if (kFetchDebug) {
    printf("fetching queue %s\n", name);
  }
  if (mutex_lock(&global_core->mem_struct->queues.lock) != 0) {
    AOS_LOG(FATAL, "mutex_lock(%p) failed\n",
            &global_core->mem_struct->queues.lock);
  }
  RawQueue *current =
      static_cast<RawQueue *>(global_core->mem_struct->queues.pointer);
  if (current != NULL) {
    while (true) {
      // If we found a matching queue.
      if (strcmp(current->name_, name) == 0 && current->length_ == length &&
          current->hash_ == hash && current->queue_length_ == queue_length) {
        mutex_unlock(&global_core->mem_struct->queues.lock);
        return current;
      } else {
        if (kFetchDebug) {
          printf("rejected queue %s strcmp=%d target=%s\n", current->name_,
                 strcmp(current->name_, name), name);
        }
      }
      // If this is the last one.
      if (current->next_ == NULL) break;
      current = current->next_;
    }
  }

  RawQueue *r = new (shm_malloc(sizeof(RawQueue)))
      RawQueue(name, length, hash, queue_length);
  if (current == NULL) {  // if we don't already have one
    global_core->mem_struct->queues.pointer = r;
  } else {
    current->next_ = r;
  }

  mutex_unlock(&global_core->mem_struct->queues.lock);
  return r;
}

RawQueue *RawQueue::Fetch(const char *name, size_t length, int hash,
                          int queue_length, int recycle_hash,
                          int recycle_length, RawQueue **recycle) {
  RawQueue *r = Fetch(name, length, hash, queue_length);
  r->recycle_ = Fetch(name, length, recycle_hash, recycle_length);
  if (r == r->recycle_) {
    fprintf(stderr, "queue: r->recycle_(=%p) == r(=%p)\n", r->recycle_, r);
    printf("see stderr\n");
    r->recycle_ = NULL;
    abort();
  }
  *recycle = r->recycle_;
  return r;
}

bool RawQueue::DoWriteMessage(void *msg, Options<RawQueue> options) {
  if (kWriteDebug) {
    printf("queue: %p->WriteMessage(%p, %x)\n", this, msg, options.printable());
  }

  bool signal_readable;

  {
    IPCMutexLocker locker(&data_lock_);
    AOS_CHECK(!locker.owner_died());

    int new_end;
    while (true) {
      new_end = index_add1(data_end_);
      // If there is room in the queue right now.
      if (new_end != data_start_) break;
      if (options & kNonBlock) {
        if (kWriteDebug) {
          printf("queue: not blocking on %p. returning false\n", this);
        }
        DecrementMessageReferenceCount(msg);
        return false;
      } else if (options & kOverride) {
        if (kWriteDebug) {
          printf("queue: overriding on %p\n", this);
        }
        // Avoid leaking the message that we're going to overwrite.
        DecrementMessageReferenceCount(data_[data_start_]);
        data_start_ = index_add1(data_start_);
      } else {  // kBlock
        assert(options & kBlock);
        if (kWriteDebug) {
          printf("queue: going to wait for writable_ of %p\n", this);
        }
        AOS_CHECK(!writable_.Wait());
      }
    }
    data_[data_end_] = msg;
    ++messages_;
    data_end_ = new_end;

    signal_readable = readable_waiting_;
    readable_waiting_ = false;
  }

  if (signal_readable) {
    if (kWriteDebug) {
      printf("queue: broadcasting to readable_ of %p\n", this);
    }
    readable_.Broadcast();
  } else if (kWriteDebug) {
    printf("queue: skipping broadcast to readable_ of %p\n", this);
  }

  if (kWriteDebug) {
    printf("queue: write returning true on queue %p\n", this);
  }
  return true;
}

inline void RawQueue::ReadCommonEnd() {
  if (is_writable()) {
    if (kReadDebug) {
      printf("queue: %ssignalling writable_ of %p\n",
             writable_start_ ? "not " : "", this);
    }
    if (!writable_start_) writable_.Broadcast();
  }
}

bool RawQueue::ReadCommonStart(Options<RawQueue> options, int *index,
                               chrono::nanoseconds timeout) {
  while (data_start_ == data_end_ || ((index != NULL) && messages_ <= *index)) {
    if (options & kNonBlock) {
      if (kReadDebug) {
        printf("queue: not going to block waiting on %p\n", this);
      }
      return false;
    } else {  // kBlock
      assert(options & kBlock);
      if (kReadDebug) {
        printf("queue: going to wait for readable_ of %p\n", this);
      }
      readable_waiting_ = true;
      // Wait for a message to become readable.
      while (true) {
        Condition::WaitResult wait_result = readable_.WaitTimed(timeout);
        if (wait_result == Condition::WaitResult::kOk) {
          break;
        }
        AOS_CHECK(wait_result != Condition::WaitResult::kOwnerDied);
        if (wait_result == Condition::WaitResult::kTimeout) {
          return false;
        }
      }

      if (kReadDebug) {
        printf("queue: done waiting for readable_ of %p\n", this);
      }
    }
  }
  // We have to check down here because we might have unlocked the mutex while
  // Wait()ing above so this value might have changed.
  writable_start_ = is_writable();
  if (kReadDebug) {
    printf("queue: %p->read(%p) start=%d end=%d writable_start=%d\n", this,
           index, data_start_, data_end_, writable_start_);
  }
  return true;
}

inline int RawQueue::LastMessageIndex() const {
  int pos = data_end_ - 1;
  if (pos < 0) {  // If it wrapped around.
    pos = data_length_ - 1;
  }
  return pos;
}

const void *RawQueue::DoReadMessage(Options<RawQueue> options) {
  // TODO(brians): Test this function.
  if (kReadDebug) {
    printf("queue: %p->ReadMessage(%x)\n", this, options.printable());
  }
  void *msg = NULL;

  IPCMutexLocker locker(&data_lock_);
  AOS_CHECK(!locker.owner_died());

  if (!ReadCommonStart(options, nullptr, chrono::nanoseconds(0))) {
    if (kReadDebug) {
      printf("queue: %p common returned false\n", this);
    }
    return NULL;
  }

  if (options & kFromEnd) {
    if (options & kPeek) {
      if (kReadDebug) {
        printf("queue: %p shortcutting c2: %d\n", this, LastMessageIndex());
      }
      msg = data_[LastMessageIndex()];
      IncrementMessageReferenceCount(msg);
    } else {
      while (true) {
        if (kReadDebug) {
          printf("queue: %p start of c2\n", this);
        }
        // This loop pulls each message out of the buffer.
        const int pos = data_start_;
        data_start_ = index_add1(data_start_);
        // If this is the last one.
        if (data_start_ == data_end_) {
          if (kReadDebug) {
            printf("queue: %p reading from c2: %d\n", this, pos);
          }
          msg = data_[pos];
          break;
        }
        // This message is not going to be in the queue any more.
        DecrementMessageReferenceCount(data_[pos]);
      }
    }
  } else {
    if (kReadDebug) {
      printf("queue: %p reading from d2: %d\n", this, data_start_);
    }
    msg = data_[data_start_];
    if (options & kPeek) {
      IncrementMessageReferenceCount(msg);
    } else {
      data_start_ = index_add1(data_start_);
    }
  }
  ReadCommonEnd();
  if (kReadDebug) {
    printf("queue: %p read returning %p\n", this, msg);
  }
  return msg;
}

const void *RawQueue::DoReadMessageIndex(Options<RawQueue> options, int *index,
                                         chrono::nanoseconds timeout) {
  if (kReadDebug) {
    printf("queue: %p->ReadMessageIndex(%x, %p(*=%d))\n", this,
           options.printable(), index, *index);
  }
  void *msg = NULL;

  IPCMutexLocker locker(&data_lock_);
  AOS_CHECK(!locker.owner_died());

  if (!ReadCommonStart(options, index, timeout)) {
    if (kReadDebug) {
      printf("queue: %p common returned false\n", this);
    }
    return NULL;
  }

  // TODO(parker): Handle integer wrap on the index.

  if (options & kFromEnd) {
    if (kReadDebug) {
      printf("queue: %p reading from c1: %d\n", this, LastMessageIndex());
    }
    msg = data_[LastMessageIndex()];

    // We'd skip this if we had kPeek, but kPeek | kFromEnd isn't valid for
    // reading with an index.
    *index = messages_;
  } else {
    // Where we're going to start reading.
    int my_start;

    const int unread_messages = messages_ - *index;
    assert(unread_messages > 0);
    int current_messages = data_end_ - data_start_;
    if (current_messages < 0) current_messages += data_length_;
    if (kReadIndexDebug) {
      printf("queue: %p start=%d end=%d current=%d\n", this, data_start_,
             data_end_, current_messages);
    }
    assert(current_messages > 0);
    // If we're behind the available messages.
    if (unread_messages > current_messages) {
      // Catch index up to the last available message.
      *index = messages_ - current_messages;
      // And that's the one we're going to read.
      my_start = data_start_;
      if (kReadIndexDebug) {
        printf("queue: %p jumping ahead to message %d (have %d) (at %d)\n",
               this, *index, messages_, data_start_);
      }
    } else {
      // Just start reading at the first available message that we haven't yet
      // read.
      my_start = data_end_ - unread_messages;
      if (kReadIndexDebug) {
        printf("queue: %p original read from %d\n", this, my_start);
      }
      if (data_start_ < data_end_) {
        assert(my_start >= 0);
      }
      if (my_start < 0) my_start += data_length_;
    }

    if (kReadDebug) {
      printf("queue: %p reading from d1: %d\n", this, my_start);
    }
    // We have to be either after the start or before the end, even if the queue
    // is wrapped around (should be both if it's not).
    assert((my_start >= data_start_) || (my_start < data_end_));
    // More sanity checking.
    assert((my_start >= 0) && (my_start < data_length_));
    msg = data_[my_start];
    if (!(options & kPeek)) ++(*index);
  }
  IncrementMessageReferenceCount(msg);

  ReadCommonEnd();
  return msg;
}

int RawQueue::FreeMessages() const {
  int r = 0;
  MessageHeader *header = free_messages_;
  while (header != nullptr) {
    ++r;
    header = header->next;
  }
  return r;
}

}  // namespace aos