aos/events/shm_event_loop.cc - RealtimeRoboticsGroup/test - Gitiles

 #include "glog/logging.h"

 #include "aos/events/shm_event_loop.h"

 #include <sys/mman.h>
 #include <sys/stat.h>
 #include <sys/timerfd.h>
 #include <sys/types.h>
 #include <unistd.h>
 #include <algorithm>
 #include <atomic>
 #include <chrono>
 #include <stdexcept>

 #include "aos/events/epoll.h"
 #include "aos/ipc_lib/lockless_queue.h"
 #include "aos/realtime.h"
 #include "aos/util/phased_loop.h"

 DEFINE_string(shm_base, "/dev/shm/aos",
               "Directory to place queue backing mmaped files in.");
 DEFINE_uint32(permissions, 0770,
               "Permissions to make shared memory files and folders.");

 namespace aos {

 std::string ShmFolder(const Channel *channel) {
   CHECK(channel->has_name());
   CHECK_EQ(channel->name()->string_view()[0], '/');
   return FLAGS_shm_base + channel->name()->str() + "/";
 }
 std::string ShmPath(const Channel *channel) {
   CHECK(channel->has_type());
   return ShmFolder(channel) + channel->type()->str() + ".v0";
 }

 class MMapedQueue {
  public:
   MMapedQueue(const Channel *channel) {
     std::string path = ShmPath(channel);

     // TODO(austin): Pull these out into the config if there is a need.
     config_.num_watchers = 10;
     config_.num_senders = 10;
     config_.queue_size = 2 * channel->frequency();
     config_.message_data_size = channel->max_size();

     size_ = ipc_lib::LocklessQueueMemorySize(config_);

     MkdirP(path);

     // There are 2 cases.  Either the file already exists, or it does not
     // already exist and we need to create it.  Start by trying to create it. If
     // that fails, the file has already been created and we can open it
     // normally..  Once the file has been created it wil never be deleted.
     fd_ = open(path.c_str(), O_RDWR | O_CREAT | O_EXCL,
                O_CLOEXEC | FLAGS_permissions);
     if (fd_ == -1 && errno == EEXIST) {
       VLOG(1) << path << " already created.";
       // File already exists.
       fd_ = open(path.c_str(), O_RDWR, O_CLOEXEC);
       PCHECK(fd_ != -1) << ": Failed to open " << path;
       while (true) {
         struct stat st;
         PCHECK(fstat(fd_, &st) == 0);
         if (st.st_size != 0) {
           CHECK_EQ(static_cast<size_t>(st.st_size), size_)
               << ": Size of " << path
               << " doesn't match expected size of backing queue file.  Did the "
                  "queue definition change?";
           break;
         } else {
           // The creating process didn't get around to it yet.  Give it a bit.
           std::this_thread::sleep_for(std::chrono::milliseconds(10));
           VLOG(1) << path << " is zero size, waiting";
         }
       }
     } else {
       VLOG(1) << "Created " << path;
       PCHECK(fd_ != -1) << ": Failed to open " << path;
       PCHECK(ftruncate(fd_, size_) == 0);
     }

     data_ = mmap(NULL, size_, PROT_READ | PROT_WRITE, MAP_SHARED, fd_, 0);
     PCHECK(data_ != MAP_FAILED);

     ipc_lib::InitializeLocklessQueueMemory(memory(), config_);
   }

   ~MMapedQueue() {
     PCHECK(munmap(data_, size_) == 0);
     PCHECK(close(fd_) == 0);
   }

   ipc_lib::LocklessQueueMemory *memory() const {
     return reinterpret_cast<ipc_lib::LocklessQueueMemory *>(data_);
   }

   const ipc_lib::LocklessQueueConfiguration &config() const {
     return config_;
   }

  private:
   void MkdirP(absl::string_view path) {
     struct stat st;
     auto last_slash_pos = path.find_last_of("/");

     std::string folder(last_slash_pos == absl::string_view::npos
                            ? absl::string_view("")
                            : path.substr(0, last_slash_pos));
     if (stat(folder.c_str(), &st) == -1) {
       PCHECK(errno == ENOENT);
       CHECK_NE(folder, "") << ": Base path doesn't exist";
       MkdirP(folder);
       VLOG(1) << "Creating " << folder;
       PCHECK(mkdir(folder.c_str(), FLAGS_permissions) == 0);
     }
   }

   ipc_lib::LocklessQueueConfiguration config_;

   int fd_;

   size_t size_;
   void *data_;
 };

 // Returns the portion of the path after the last /.
 absl::string_view Filename(absl::string_view path) {
   auto last_slash_pos = path.find_last_of("/");

   return last_slash_pos == absl::string_view::npos
              ? path
              : path.substr(last_slash_pos + 1, path.size());
 }

 ShmEventLoop::ShmEventLoop(const Configuration *configuration)
     : EventLoop(configuration), name_(Filename(program_invocation_name)) {}

 namespace {

 namespace chrono = ::std::chrono;

 class ShmFetcher : public RawFetcher {
  public:
   explicit ShmFetcher(const Channel *channel)
       : lockless_queue_memory_(channel),
         lockless_queue_(lockless_queue_memory_.memory(),
                         lockless_queue_memory_.config()),
         data_storage_(static_cast<AlignedChar *>(aligned_alloc(
                           alignof(AlignedChar), channel->max_size())),
                       &free) {
     context_.data = nullptr;
     // Point the queue index at the next index to read starting now.  This
     // makes it such that FetchNext will read the next message sent after
     // the fetcher is created.
     PointAtNextQueueIndex();
   }

   ~ShmFetcher() { data_ = nullptr; }

   // Points the next message to fetch at the queue index which will be
   // populated next.
   void PointAtNextQueueIndex() {
     actual_queue_index_ = lockless_queue_.LatestQueueIndex();
     if (!actual_queue_index_.valid()) {
       // Nothing in the queue.  The next element will show up at the 0th
       // index in the queue.
       actual_queue_index_ =
           ipc_lib::QueueIndex::Zero(lockless_queue_.queue_size());
     } else {
       actual_queue_index_ = actual_queue_index_.Increment();
     }
   }

   bool FetchNext() override {
     // TODO(austin): Write a test which starts with nothing in the queue,
     // and then calls FetchNext() after something is sent.
     // TODO(austin): Get behind and make sure it dies both here and with
     // Fetch.
     ipc_lib::LocklessQueue::ReadResult read_result = lockless_queue_.Read(
         actual_queue_index_.index(), &context_.monotonic_sent_time,
         &context_.realtime_sent_time, &context_.size,
         reinterpret_cast<char *>(data_storage_.get()));
     if (read_result == ipc_lib::LocklessQueue::ReadResult::GOOD) {
       context_.queue_index = actual_queue_index_.index();
       data_ = reinterpret_cast<char *>(data_storage_.get()) +
               lockless_queue_.message_data_size() - context_.size;
       context_.data = data_;
       actual_queue_index_ = actual_queue_index_.Increment();
     }

     // Make sure the data wasn't modified while we were reading it.  This
     // can only happen if you are reading the last message *while* it is
     // being written to, which means you are pretty far behind.
     CHECK(read_result != ipc_lib::LocklessQueue::ReadResult::OVERWROTE)
         << ": Got behind while reading and the last message was modified "
            "out "
            "from under us while we were reading it.  Don't get so far "
            "behind.";

     CHECK(read_result != ipc_lib::LocklessQueue::ReadResult::TOO_OLD)
         << ": The next message is no longer available.";
     return read_result == ipc_lib::LocklessQueue::ReadResult::GOOD;
   }

   bool Fetch() override {
     const ipc_lib::QueueIndex queue_index = lockless_queue_.LatestQueueIndex();
     // actual_queue_index_ is only meaningful if it was set by Fetch or
     // FetchNext.  This happens when valid_data_ has been set.  So, only
     // skip checking if valid_data_ is true.
     //
     // Also, if the latest queue index is invalid, we are empty.  So there
     // is nothing to fetch.
     if ((data_ != nullptr &&
          queue_index == actual_queue_index_.DecrementBy(1u)) ||
         !queue_index.valid()) {
       return false;
     }

     ipc_lib::LocklessQueue::ReadResult read_result =
         lockless_queue_.Read(queue_index.index(), &context_.monotonic_sent_time,
                              &context_.realtime_sent_time, &context_.size,
                              reinterpret_cast<char *>(data_storage_.get()));
     if (read_result == ipc_lib::LocklessQueue::ReadResult::GOOD) {
       context_.queue_index = queue_index.index();
       data_ = reinterpret_cast<char *>(data_storage_.get()) +
               lockless_queue_.message_data_size() - context_.size;
       context_.data = data_;
       actual_queue_index_ = queue_index.Increment();
     }

     // Make sure the data wasn't modified while we were reading it.  This
     // can only happen if you are reading the last message *while* it is
     // being written to, which means you are pretty far behind.
     CHECK(read_result != ipc_lib::LocklessQueue::ReadResult::OVERWROTE)
         << ": Got behind while reading and the last message was modified "
            "out "
            "from under us while we were reading it.  Don't get so far "
            "behind.";

     CHECK(read_result != ipc_lib::LocklessQueue::ReadResult::NOTHING_NEW)
         << ": Queue index went backwards.  This should never happen.";

     // We fell behind between when we read the index and read the value.
     // This isn't worth recovering from since this means we went to sleep
     // for a long time in the middle of this function.
     CHECK(read_result != ipc_lib::LocklessQueue::ReadResult::TOO_OLD)
         << ": The next message is no longer available.";
     return read_result == ipc_lib::LocklessQueue::ReadResult::GOOD;
   }

   bool RegisterWakeup(int priority) {
     return lockless_queue_.RegisterWakeup(priority);
   }

   void UnregisterWakeup() { lockless_queue_.UnregisterWakeup(); }

  private:
   MMapedQueue lockless_queue_memory_;
   ipc_lib::LocklessQueue lockless_queue_;

   ipc_lib::QueueIndex actual_queue_index_ =
       ipc_lib::LocklessQueue::empty_queue_index();

   struct AlignedChar {
     alignas(32) char data;
   };

   std::unique_ptr<AlignedChar, decltype(&free)> data_storage_;
 };

 class ShmSender : public RawSender {
  public:
   explicit ShmSender(const Channel *channel, const ShmEventLoop *shm_event_loop)
       : RawSender(),
         shm_event_loop_(shm_event_loop),
         name_(channel->name()->str()),
         lockless_queue_memory_(channel),
         lockless_queue_(lockless_queue_memory_.memory(),
                         lockless_queue_memory_.config()),
         lockless_queue_sender_(lockless_queue_.MakeSender()) {}

   void *data() override { return lockless_queue_sender_.Data(); }
   size_t size() override { return lockless_queue_sender_.size(); }
   bool Send(size_t size) override {
     lockless_queue_sender_.Send(size);
     lockless_queue_.Wakeup(shm_event_loop_->priority());
     return true;
   }

   bool Send(void *msg, size_t length) override {
     lockless_queue_sender_.Send(reinterpret_cast<char *>(msg), length);
     lockless_queue_.Wakeup(shm_event_loop_->priority());
     // TODO(austin): Return an error if we send too fast.
     return true;
   }

   const absl::string_view name() const override { return name_; }

  private:
   const ShmEventLoop *shm_event_loop_;
   std::string name_;
   MMapedQueue lockless_queue_memory_;
   ipc_lib::LocklessQueue lockless_queue_;
   ipc_lib::LocklessQueue::Sender lockless_queue_sender_;
 };

 }  // namespace

 namespace internal {

 // Class to manage the state for a Watcher.
 class WatcherState {
  public:
   WatcherState(
       const Channel *channel,
       std::function<void(const Context &context, const void *message)> watcher)
       : shm_fetcher_(channel), watcher_(watcher) {}

   ~WatcherState() {}

   // Points the next message to fetch at the queue index which will be populated
   // next.
   void PointAtNextQueueIndex() { shm_fetcher_.PointAtNextQueueIndex(); }

   // Returns true if there is new data available.
   bool HasNewData() {
     if (!has_new_data_) {
       has_new_data_ = shm_fetcher_.FetchNext();
     }

     return has_new_data_;
   }

   // Returns the time of the current data sample.
   aos::monotonic_clock::time_point event_time() const {
     return shm_fetcher_.context().monotonic_sent_time;
   }

   // Consumes the data by calling the callback.
   void CallCallback() {
     CHECK(has_new_data_);
     watcher_(shm_fetcher_.context(), shm_fetcher_.most_recent_data());
     has_new_data_ = false;
   }

   // Starts the thread and waits until it is running.
   bool RegisterWakeup(int priority) {
     return shm_fetcher_.RegisterWakeup(priority);
   }

   void UnregisterWakeup() { return shm_fetcher_.UnregisterWakeup(); }

  private:
   bool has_new_data_ = false;

   ShmFetcher shm_fetcher_;

   std::function<void(const Context &context, const void *message)> watcher_;
 };

 // Adapter class to adapt a timerfd to a TimerHandler.
 // The part of the API which is accessed by the TimerHandler interface needs to
 // be threadsafe.  This means Setup and Disable.
 class TimerHandlerState : public TimerHandler {
  public:
   TimerHandlerState(ShmEventLoop *shm_event_loop, ::std::function<void()> fn)
       : shm_event_loop_(shm_event_loop), fn_(::std::move(fn)) {
     shm_event_loop_->epoll_.OnReadable(timerfd_.fd(), [this]() {
       timerfd_.Read();
       fn_();
     });
   }

   ~TimerHandlerState() { shm_event_loop_->epoll_.DeleteFd(timerfd_.fd()); }

   void Setup(monotonic_clock::time_point base,
              monotonic_clock::duration repeat_offset) override {
     // SetTime is threadsafe already.
     timerfd_.SetTime(base, repeat_offset);
   }

   void Disable() override {
     // Disable is also threadsafe already.
     timerfd_.Disable();
   }

  private:
   ShmEventLoop *shm_event_loop_;

   TimerFd timerfd_;

   // Function to be run on the thread
   ::std::function<void()> fn_;
 };

 // Adapter class to the timerfd and PhasedLoop.
 // The part of the API which is accessed by the PhasedLoopHandler interface
 // needs to be threadsafe.  This means set_interval_and_offset
 class PhasedLoopHandler : public ::aos::PhasedLoopHandler {
  public:
   PhasedLoopHandler(ShmEventLoop *shm_event_loop, ::std::function<void(int)> fn,
                     const monotonic_clock::duration interval,
                     const monotonic_clock::duration offset)
       : shm_event_loop_(shm_event_loop),
         phased_loop_(interval, shm_event_loop_->monotonic_now(), offset),
         fn_(::std::move(fn)) {
     shm_event_loop_->epoll_.OnReadable(timerfd_.fd(), [this]() {
       timerfd_.Read();
       // Call the function.  To avoid needing a recursive mutex, drop the lock
       // before running the function.
       fn_(cycles_elapsed_);
       Reschedule();
     });
   }

   ~PhasedLoopHandler() { shm_event_loop_->epoll_.DeleteFd(timerfd_.fd()); }

   void set_interval_and_offset(
       const monotonic_clock::duration interval,
       const monotonic_clock::duration offset) override {
     phased_loop_.set_interval_and_offset(interval, offset);
   }

   void Startup() {
     phased_loop_.Reset(shm_event_loop_->monotonic_now());
     Reschedule();
   }

  private:
   // Reschedules the timer.  Must be called with the mutex held.
   void Reschedule() {
     cycles_elapsed_ = phased_loop_.Iterate(shm_event_loop_->monotonic_now());
     timerfd_.SetTime(phased_loop_.sleep_time(), ::aos::monotonic_clock::zero());
   }

   ShmEventLoop *shm_event_loop_;

   TimerFd timerfd_;
   time::PhasedLoop phased_loop_;

   int cycles_elapsed_ = 1;

   // Function to be run
   const ::std::function<void(int)> fn_;
 };
 }  // namespace internal

 ::std::unique_ptr<RawFetcher> ShmEventLoop::MakeRawFetcher(
     const Channel *channel) {
   return ::std::unique_ptr<RawFetcher>(new ShmFetcher(channel));
 }

 ::std::unique_ptr<RawSender> ShmEventLoop::MakeRawSender(
     const Channel *channel) {
   Take(channel);
   return ::std::unique_ptr<RawSender>(new ShmSender(channel, this));
 }

 void ShmEventLoop::MakeRawWatcher(
     const Channel *channel,
     std::function<void(const Context &context, const void *message)> watcher) {
   Take(channel);

   ::std::unique_ptr<internal::WatcherState> state(
       new internal::WatcherState(
       channel, std::move(watcher)));
   watchers_.push_back(::std::move(state));
 }

 TimerHandler *ShmEventLoop::AddTimer(::std::function<void()> callback) {
   ::std::unique_ptr<internal::TimerHandlerState> timer(
       new internal::TimerHandlerState(this, ::std::move(callback)));

   timers_.push_back(::std::move(timer));

   return timers_.back().get();
 }

 PhasedLoopHandler *ShmEventLoop::AddPhasedLoop(
     ::std::function<void(int)> callback,
     const monotonic_clock::duration interval,
     const monotonic_clock::duration offset) {
   ::std::unique_ptr<internal::PhasedLoopHandler> phased_loop(
       new internal::PhasedLoopHandler(this, ::std::move(callback), interval,
                                       offset));

   phased_loops_.push_back(::std::move(phased_loop));

   return phased_loops_.back().get();
 }

 void ShmEventLoop::OnRun(::std::function<void()> on_run) {
   on_run_.push_back(::std::move(on_run));
 }

 void ShmEventLoop::Run() {
   std::unique_ptr<ipc_lib::SignalFd> signalfd;

   if (watchers_.size() > 0) {
     signalfd.reset(new ipc_lib::SignalFd({ipc_lib::kWakeupSignal}));

     epoll_.OnReadable(signalfd->fd(), [signalfd_ptr = signalfd.get(), this]() {
       signalfd_siginfo result = signalfd_ptr->Read();
       CHECK_EQ(result.ssi_signo, ipc_lib::kWakeupSignal);

       // TODO(austin): We should really be checking *everything*, not just
       // watchers, and calling the oldest thing first.  That will improve
       // determinism a lot.

       while (true) {
         // Call the handlers in time order of their messages.
         aos::monotonic_clock::time_point min_event_time =
             aos::monotonic_clock::max_time;
         size_t min_watcher_index = -1;
         size_t watcher_index = 0;
         for (::std::unique_ptr<internal::WatcherState> &watcher : watchers_) {
           if (watcher->HasNewData()) {
             if (watcher->event_time() < min_event_time) {
               min_watcher_index = watcher_index;
               min_event_time = watcher->event_time();
             }
           }
           ++watcher_index;
         }

         if (min_event_time == aos::monotonic_clock::max_time) {
           break;
         }

         watchers_[min_watcher_index]->CallCallback();
       }
     });
   }

   // Now, all the threads are up.  Lock everything into memory and go RT.
   if (priority_ != 0) {
     ::aos::InitRT();

     LOG(INFO) << "Setting priority to " << priority_;
     ::aos::SetCurrentThreadRealtimePriority(priority_);
   }

   set_is_running(true);

   // Now that we are realtime (but before the OnRun handlers run), snap the
   // queue index.
   for (::std::unique_ptr<internal::WatcherState> &watcher : watchers_) {
     watcher->PointAtNextQueueIndex();
     CHECK(watcher->RegisterWakeup(priority_));
   }

   // Now that we are RT, run all the OnRun handlers.
   for (const auto &run : on_run_) {
     run();
   }

   // Start up all the phased loops.
   for (::std::unique_ptr<internal::PhasedLoopHandler> &phased_loop :
        phased_loops_) {
     phased_loop->Startup();
   }

   // And start our main event loop which runs all the timers and handles Quit.
   epoll_.Run();

   // Once epoll exits, there is no useful nonrt work left to do.
   set_is_running(false);

   // Nothing time or synchronization critical needs to happen after this point.
   // Drop RT priority.
   ::aos::UnsetCurrentThreadRealtimePriority();

   for (::std::unique_ptr<internal::WatcherState> &watcher : watchers_) {
     watcher->UnregisterWakeup();
   }

   if (watchers_.size() > 0) {
     epoll_.DeleteFd(signalfd->fd());
     signalfd.reset();
   }
 }

 void ShmEventLoop::Exit() { epoll_.Quit(); }

 ShmEventLoop::~ShmEventLoop() {
   CHECK(!is_running()) << ": ShmEventLoop destroyed while running";
 }

 void ShmEventLoop::Take(const Channel *channel) {
   CHECK(!is_running()) << ": Cannot add new objects while running.";

   // Cheat aggresively.  Use the shared memory path as a proxy for a unique
   // identifier for the channel.
   const std::string path = ShmPath(channel);

   const auto prior = ::std::find(taken_.begin(), taken_.end(), path);
   CHECK(prior == taken_.end()) << ": " << path << " is already being used.";

   taken_.emplace_back(path);
 }

 void ShmEventLoop::SetRuntimeRealtimePriority(int priority) {
   if (is_running()) {
     LOG(FATAL) << "Cannot set realtime priority while running.";
   }
   priority_ = priority;
 }

 }  // namespace aos
	#include "glog/logging.h"

	#include "aos/events/shm_event_loop.h"

	#include <sys/mman.h>
	#include <sys/stat.h>
	#include <sys/timerfd.h>
	#include <sys/types.h>
	#include <unistd.h>
	#include <algorithm>
	#include <atomic>
	#include <chrono>
	#include <stdexcept>

	#include "aos/events/epoll.h"
	#include "aos/ipc_lib/lockless_queue.h"
	#include "aos/realtime.h"
	#include "aos/util/phased_loop.h"

	DEFINE_string(shm_base, "/dev/shm/aos",
	"Directory to place queue backing mmaped files in.");
	DEFINE_uint32(permissions, 0770,
	"Permissions to make shared memory files and folders.");

	namespace aos {

	std::string ShmFolder(const Channel *channel) {
	CHECK(channel->has_name());
	CHECK_EQ(channel->name()->string_view()[0], '/');
	return FLAGS_shm_base + channel->name()->str() + "/";
	}
	std::string ShmPath(const Channel *channel) {
	CHECK(channel->has_type());
	return ShmFolder(channel) + channel->type()->str() + ".v0";
	}

	class MMapedQueue {
	public:
	MMapedQueue(const Channel *channel) {
	std::string path = ShmPath(channel);

	// TODO(austin): Pull these out into the config if there is a need.
	config_.num_watchers = 10;
	config_.num_senders = 10;
	config_.queue_size = 2 * channel->frequency();
	config_.message_data_size = channel->max_size();

	size_ = ipc_lib::LocklessQueueMemorySize(config_);

	MkdirP(path);

	// There are 2 cases. Either the file already exists, or it does not
	// already exist and we need to create it. Start by trying to create it. If
	// that fails, the file has already been created and we can open it
	// normally.. Once the file has been created it wil never be deleted.
	fd_ = open(path.c_str(), O_RDWR \| O_CREAT \| O_EXCL,
	O_CLOEXEC \| FLAGS_permissions);
	if (fd_ == -1 && errno == EEXIST) {
	VLOG(1) << path << " already created.";
	// File already exists.
	fd_ = open(path.c_str(), O_RDWR, O_CLOEXEC);
	PCHECK(fd_ != -1) << ": Failed to open " << path;
	while (true) {
	struct stat st;
	PCHECK(fstat(fd_, &st) == 0);
	if (st.st_size != 0) {
	CHECK_EQ(static_cast<size_t>(st.st_size), size_)
	<< ": Size of " << path
	<< " doesn't match expected size of backing queue file. Did the "
	"queue definition change?";
	break;
	} else {
	// The creating process didn't get around to it yet. Give it a bit.
	std::this_thread::sleep_for(std::chrono::milliseconds(10));
	VLOG(1) << path << " is zero size, waiting";
	}
	}
	} else {
	VLOG(1) << "Created " << path;
	PCHECK(fd_ != -1) << ": Failed to open " << path;
	PCHECK(ftruncate(fd_, size_) == 0);
	}

	data_ = mmap(NULL, size_, PROT_READ \| PROT_WRITE, MAP_SHARED, fd_, 0);
	PCHECK(data_ != MAP_FAILED);

	ipc_lib::InitializeLocklessQueueMemory(memory(), config_);
	}

	~MMapedQueue() {
	PCHECK(munmap(data_, size_) == 0);
	PCHECK(close(fd_) == 0);
	}

	ipc_lib::LocklessQueueMemory *memory() const {
	return reinterpret_cast<ipc_lib::LocklessQueueMemory *>(data_);
	}

	const ipc_lib::LocklessQueueConfiguration &config() const {
	return config_;
	}

	private:
	void MkdirP(absl::string_view path) {
	struct stat st;
	auto last_slash_pos = path.find_last_of("/");

	std::string folder(last_slash_pos == absl::string_view::npos
	? absl::string_view("")
	: path.substr(0, last_slash_pos));
	if (stat(folder.c_str(), &st) == -1) {
	PCHECK(errno == ENOENT);
	CHECK_NE(folder, "") << ": Base path doesn't exist";
	MkdirP(folder);
	VLOG(1) << "Creating " << folder;
	PCHECK(mkdir(folder.c_str(), FLAGS_permissions) == 0);
	}
	}

	ipc_lib::LocklessQueueConfiguration config_;

	int fd_;

	size_t size_;
	void *data_;
	};

	// Returns the portion of the path after the last /.
	absl::string_view Filename(absl::string_view path) {
	auto last_slash_pos = path.find_last_of("/");

	return last_slash_pos == absl::string_view::npos
	? path
	: path.substr(last_slash_pos + 1, path.size());
	}

	ShmEventLoop::ShmEventLoop(const Configuration *configuration)
	: EventLoop(configuration), name_(Filename(program_invocation_name)) {}

	namespace {

	namespace chrono = ::std::chrono;

	class ShmFetcher : public RawFetcher {
	public:
	explicit ShmFetcher(const Channel *channel)
	: lockless_queue_memory_(channel),
	lockless_queue_(lockless_queue_memory_.memory(),
	lockless_queue_memory_.config()),
	data_storage_(static_cast<AlignedChar *>(aligned_alloc(
	alignof(AlignedChar), channel->max_size())),
	&free) {
	context_.data = nullptr;
	// Point the queue index at the next index to read starting now. This
	// makes it such that FetchNext will read the next message sent after
	// the fetcher is created.
	PointAtNextQueueIndex();
	}

	~ShmFetcher() { data_ = nullptr; }

	// Points the next message to fetch at the queue index which will be
	// populated next.
	void PointAtNextQueueIndex() {
	actual_queue_index_ = lockless_queue_.LatestQueueIndex();
	if (!actual_queue_index_.valid()) {
	// Nothing in the queue. The next element will show up at the 0th
	// index in the queue.
	actual_queue_index_ =
	ipc_lib::QueueIndex::Zero(lockless_queue_.queue_size());
	} else {
	actual_queue_index_ = actual_queue_index_.Increment();
	}
	}

	bool FetchNext() override {
	// TODO(austin): Write a test which starts with nothing in the queue,
	// and then calls FetchNext() after something is sent.
	// TODO(austin): Get behind and make sure it dies both here and with
	// Fetch.
	ipc_lib::LocklessQueue::ReadResult read_result = lockless_queue_.Read(
	actual_queue_index_.index(), &context_.monotonic_sent_time,
	&context_.realtime_sent_time, &context_.size,
	reinterpret_cast<char *>(data_storage_.get()));
	if (read_result == ipc_lib::LocklessQueue::ReadResult::GOOD) {
	context_.queue_index = actual_queue_index_.index();
	data_ = reinterpret_cast<char *>(data_storage_.get()) +
	lockless_queue_.message_data_size() - context_.size;
	context_.data = data_;
	actual_queue_index_ = actual_queue_index_.Increment();
	}

	// Make sure the data wasn't modified while we were reading it. This
	// can only happen if you are reading the last message while it is
	// being written to, which means you are pretty far behind.
	CHECK(read_result != ipc_lib::LocklessQueue::ReadResult::OVERWROTE)
	<< ": Got behind while reading and the last message was modified "
	"out "
	"from under us while we were reading it. Don't get so far "
	"behind.";

	CHECK(read_result != ipc_lib::LocklessQueue::ReadResult::TOO_OLD)
	<< ": The next message is no longer available.";
	return read_result == ipc_lib::LocklessQueue::ReadResult::GOOD;
	}

	bool Fetch() override {
	const ipc_lib::QueueIndex queue_index = lockless_queue_.LatestQueueIndex();
	// actual_queue_index_ is only meaningful if it was set by Fetch or
	// FetchNext. This happens when valid_data_ has been set. So, only
	// skip checking if valid_data_ is true.
	//
	// Also, if the latest queue index is invalid, we are empty. So there
	// is nothing to fetch.
	if ((data_ != nullptr &&
	queue_index == actual_queue_index_.DecrementBy(1u)) \|\|
	!queue_index.valid()) {
	return false;
	}

	ipc_lib::LocklessQueue::ReadResult read_result =
	lockless_queue_.Read(queue_index.index(), &context_.monotonic_sent_time,
	&context_.realtime_sent_time, &context_.size,
	reinterpret_cast<char *>(data_storage_.get()));
	if (read_result == ipc_lib::LocklessQueue::ReadResult::GOOD) {
	context_.queue_index = queue_index.index();
	data_ = reinterpret_cast<char *>(data_storage_.get()) +
	lockless_queue_.message_data_size() - context_.size;
	context_.data = data_;
	actual_queue_index_ = queue_index.Increment();
	}

	// Make sure the data wasn't modified while we were reading it. This
	// can only happen if you are reading the last message while it is
	// being written to, which means you are pretty far behind.
	CHECK(read_result != ipc_lib::LocklessQueue::ReadResult::OVERWROTE)
	<< ": Got behind while reading and the last message was modified "
	"out "
	"from under us while we were reading it. Don't get so far "
	"behind.";

	CHECK(read_result != ipc_lib::LocklessQueue::ReadResult::NOTHING_NEW)
	<< ": Queue index went backwards. This should never happen.";

	// We fell behind between when we read the index and read the value.
	// This isn't worth recovering from since this means we went to sleep
	// for a long time in the middle of this function.
	CHECK(read_result != ipc_lib::LocklessQueue::ReadResult::TOO_OLD)
	<< ": The next message is no longer available.";
	return read_result == ipc_lib::LocklessQueue::ReadResult::GOOD;
	}

	bool RegisterWakeup(int priority) {
	return lockless_queue_.RegisterWakeup(priority);
	}

	void UnregisterWakeup() { lockless_queue_.UnregisterWakeup(); }

	private:
	MMapedQueue lockless_queue_memory_;
	ipc_lib::LocklessQueue lockless_queue_;

	ipc_lib::QueueIndex actual_queue_index_ =
	ipc_lib::LocklessQueue::empty_queue_index();

	struct AlignedChar {
	alignas(32) char data;
	};

	std::unique_ptr<AlignedChar, decltype(&free)> data_storage_;
	};

	class ShmSender : public RawSender {
	public:
	explicit ShmSender(const Channel channel, const ShmEventLoop shm_event_loop)
	: RawSender(),
	shm_event_loop_(shm_event_loop),
	name_(channel->name()->str()),
	lockless_queue_memory_(channel),
	lockless_queue_(lockless_queue_memory_.memory(),
	lockless_queue_memory_.config()),
	lockless_queue_sender_(lockless_queue_.MakeSender()) {}

	void *data() override { return lockless_queue_sender_.Data(); }
	size_t size() override { return lockless_queue_sender_.size(); }
	bool Send(size_t size) override {
	lockless_queue_sender_.Send(size);
	lockless_queue_.Wakeup(shm_event_loop_->priority());
	return true;
	}

	bool Send(void *msg, size_t length) override {
	lockless_queue_sender_.Send(reinterpret_cast<char *>(msg), length);
	lockless_queue_.Wakeup(shm_event_loop_->priority());
	// TODO(austin): Return an error if we send too fast.
	return true;
	}

	const absl::string_view name() const override { return name_; }

	private:
	const ShmEventLoop *shm_event_loop_;
	std::string name_;
	MMapedQueue lockless_queue_memory_;
	ipc_lib::LocklessQueue lockless_queue_;
	ipc_lib::LocklessQueue::Sender lockless_queue_sender_;
	};

	} // namespace

	namespace internal {

	// Class to manage the state for a Watcher.
	class WatcherState {
	public:
	WatcherState(
	const Channel *channel,
	std::function<void(const Context &context, const void *message)> watcher)
	: shm_fetcher_(channel), watcher_(watcher) {}

	~WatcherState() {}

	// Points the next message to fetch at the queue index which will be populated
	// next.
	void PointAtNextQueueIndex() { shm_fetcher_.PointAtNextQueueIndex(); }

	// Returns true if there is new data available.
	bool HasNewData() {
	if (!has_new_data_) {
	has_new_data_ = shm_fetcher_.FetchNext();
	}

	return has_new_data_;
	}

	// Returns the time of the current data sample.
	aos::monotonic_clock::time_point event_time() const {
	return shm_fetcher_.context().monotonic_sent_time;
	}

	// Consumes the data by calling the callback.
	void CallCallback() {
	CHECK(has_new_data_);
	watcher_(shm_fetcher_.context(), shm_fetcher_.most_recent_data());
	has_new_data_ = false;
	}

	// Starts the thread and waits until it is running.
	bool RegisterWakeup(int priority) {
	return shm_fetcher_.RegisterWakeup(priority);
	}

	void UnregisterWakeup() { return shm_fetcher_.UnregisterWakeup(); }

	private:
	bool has_new_data_ = false;

	ShmFetcher shm_fetcher_;

	std::function<void(const Context &context, const void *message)> watcher_;
	};

	// Adapter class to adapt a timerfd to a TimerHandler.
	// The part of the API which is accessed by the TimerHandler interface needs to
	// be threadsafe. This means Setup and Disable.
	class TimerHandlerState : public TimerHandler {
	public:
	TimerHandlerState(ShmEventLoop *shm_event_loop, ::std::function<void()> fn)
	: shm_event_loop_(shm_event_loop), fn_(::std::move(fn)) {
	shm_event_loop_->epoll_.OnReadable(timerfd_.fd(), [this]() {
	timerfd_.Read();
	fn_();
	});
	}

	~TimerHandlerState() { shm_event_loop_->epoll_.DeleteFd(timerfd_.fd()); }

	void Setup(monotonic_clock::time_point base,
	monotonic_clock::duration repeat_offset) override {
	// SetTime is threadsafe already.
	timerfd_.SetTime(base, repeat_offset);
	}

	void Disable() override {
	// Disable is also threadsafe already.
	timerfd_.Disable();
	}

	private:
	ShmEventLoop *shm_event_loop_;

	TimerFd timerfd_;

	// Function to be run on the thread
	::std::function<void()> fn_;
	};

	// Adapter class to the timerfd and PhasedLoop.
	// The part of the API which is accessed by the PhasedLoopHandler interface
	// needs to be threadsafe. This means set_interval_and_offset
	class PhasedLoopHandler : public ::aos::PhasedLoopHandler {
	public:
	PhasedLoopHandler(ShmEventLoop *shm_event_loop, ::std::function<void(int)> fn,
	const monotonic_clock::duration interval,
	const monotonic_clock::duration offset)
	: shm_event_loop_(shm_event_loop),
	phased_loop_(interval, shm_event_loop_->monotonic_now(), offset),
	fn_(::std::move(fn)) {
	shm_event_loop_->epoll_.OnReadable(timerfd_.fd(), [this]() {
	timerfd_.Read();
	// Call the function. To avoid needing a recursive mutex, drop the lock
	// before running the function.
	fn_(cycles_elapsed_);
	Reschedule();
	});
	}

	~PhasedLoopHandler() { shm_event_loop_->epoll_.DeleteFd(timerfd_.fd()); }

	void set_interval_and_offset(
	const monotonic_clock::duration interval,
	const monotonic_clock::duration offset) override {
	phased_loop_.set_interval_and_offset(interval, offset);
	}

	void Startup() {
	phased_loop_.Reset(shm_event_loop_->monotonic_now());
	Reschedule();
	}

	private:
	// Reschedules the timer. Must be called with the mutex held.
	void Reschedule() {
	cycles_elapsed_ = phased_loop_.Iterate(shm_event_loop_->monotonic_now());
	timerfd_.SetTime(phased_loop_.sleep_time(), ::aos::monotonic_clock::zero());
	}

	ShmEventLoop *shm_event_loop_;

	TimerFd timerfd_;
	time::PhasedLoop phased_loop_;

	int cycles_elapsed_ = 1;

	// Function to be run
	const ::std::function<void(int)> fn_;
	};
	} // namespace internal

	::std::unique_ptr<RawFetcher> ShmEventLoop::MakeRawFetcher(
	const Channel *channel) {
	return ::std::unique_ptr<RawFetcher>(new ShmFetcher(channel));
	}

	::std::unique_ptr<RawSender> ShmEventLoop::MakeRawSender(
	const Channel *channel) {
	Take(channel);
	return ::std::unique_ptr<RawSender>(new ShmSender(channel, this));
	}

	void ShmEventLoop::MakeRawWatcher(
	const Channel *channel,
	std::function<void(const Context &context, const void *message)> watcher) {
	Take(channel);

	::std::unique_ptr<internal::WatcherState> state(
	new internal::WatcherState(
	channel, std::move(watcher)));
	watchers_.push_back(::std::move(state));
	}

	TimerHandler *ShmEventLoop::AddTimer(::std::function<void()> callback) {
	::std::unique_ptr<internal::TimerHandlerState> timer(
	new internal::TimerHandlerState(this, ::std::move(callback)));

	timers_.push_back(::std::move(timer));

	return timers_.back().get();
	}

	PhasedLoopHandler *ShmEventLoop::AddPhasedLoop(
	::std::function<void(int)> callback,
	const monotonic_clock::duration interval,
	const monotonic_clock::duration offset) {
	::std::unique_ptr<internal::PhasedLoopHandler> phased_loop(
	new internal::PhasedLoopHandler(this, ::std::move(callback), interval,
	offset));

	phased_loops_.push_back(::std::move(phased_loop));

	return phased_loops_.back().get();
	}

	void ShmEventLoop::OnRun(::std::function<void()> on_run) {
	on_run_.push_back(::std::move(on_run));
	}

	void ShmEventLoop::Run() {
	std::unique_ptr<ipc_lib::SignalFd> signalfd;

	if (watchers_.size() > 0) {
	signalfd.reset(new ipc_lib::SignalFd({ipc_lib::kWakeupSignal}));

	epoll_.OnReadable(signalfd->fd(), [signalfd_ptr = signalfd.get(), this]() {
	signalfd_siginfo result = signalfd_ptr->Read();
	CHECK_EQ(result.ssi_signo, ipc_lib::kWakeupSignal);

	// TODO(austin): We should really be checking everything, not just
	// watchers, and calling the oldest thing first. That will improve
	// determinism a lot.

	while (true) {
	// Call the handlers in time order of their messages.
	aos::monotonic_clock::time_point min_event_time =
	aos::monotonic_clock::max_time;
	size_t min_watcher_index = -1;
	size_t watcher_index = 0;
	for (::std::unique_ptr<internal::WatcherState> &watcher : watchers_) {
	if (watcher->HasNewData()) {
	if (watcher->event_time() < min_event_time) {
	min_watcher_index = watcher_index;
	min_event_time = watcher->event_time();
	}
	}
	++watcher_index;
	}

	if (min_event_time == aos::monotonic_clock::max_time) {
	break;
	}

	watchers_[min_watcher_index]->CallCallback();
	}
	});
	}

	// Now, all the threads are up. Lock everything into memory and go RT.
	if (priority_ != 0) {
	::aos::InitRT();

	LOG(INFO) << "Setting priority to " << priority_;
	::aos::SetCurrentThreadRealtimePriority(priority_);
	}

	set_is_running(true);

	// Now that we are realtime (but before the OnRun handlers run), snap the
	// queue index.
	for (::std::unique_ptr<internal::WatcherState> &watcher : watchers_) {
	watcher->PointAtNextQueueIndex();
	CHECK(watcher->RegisterWakeup(priority_));
	}

	// Now that we are RT, run all the OnRun handlers.
	for (const auto &run : on_run_) {
	run();
	}

	// Start up all the phased loops.
	for (::std::unique_ptr<internal::PhasedLoopHandler> &phased_loop :
	phased_loops_) {
	phased_loop->Startup();
	}

	// And start our main event loop which runs all the timers and handles Quit.
	epoll_.Run();

	// Once epoll exits, there is no useful nonrt work left to do.
	set_is_running(false);

	// Nothing time or synchronization critical needs to happen after this point.
	// Drop RT priority.
	::aos::UnsetCurrentThreadRealtimePriority();

	for (::std::unique_ptr<internal::WatcherState> &watcher : watchers_) {
	watcher->UnregisterWakeup();
	}

	if (watchers_.size() > 0) {
	epoll_.DeleteFd(signalfd->fd());
	signalfd.reset();
	}
	}

	void ShmEventLoop::Exit() { epoll_.Quit(); }

	ShmEventLoop::~ShmEventLoop() {
	CHECK(!is_running()) << ": ShmEventLoop destroyed while running";
	}

	void ShmEventLoop::Take(const Channel *channel) {
	CHECK(!is_running()) << ": Cannot add new objects while running.";

	// Cheat aggresively. Use the shared memory path as a proxy for a unique
	// identifier for the channel.
	const std::string path = ShmPath(channel);

	const auto prior = ::std::find(taken_.begin(), taken_.end(), path);
	CHECK(prior == taken_.end()) << ": " << path << " is already being used.";

	taken_.emplace_back(path);
	}

	void ShmEventLoop::SetRuntimeRealtimePriority(int priority) {
	if (is_running()) {
	LOG(FATAL) << "Cannot set realtime priority while running.";
	}
	priority_ = priority;
	}

	} // namespace aos