iOS/Pods/Realm/Realm/ObjectStore/src/impl/apple/external_commit_helper.cpp

   1 ////////////////////////////////////////////////////////////////////////////
   2 //
   3 // Copyright 2015 Realm Inc.
   4 //
   5 // Licensed under the Apache License, Version 2.0 (the "License");
   6 // you may not use this file except in compliance with the License.
   7 // You may obtain a copy of the License at
   8 //
   9 // http://www.apache.org/licenses/LICENSE-2.0
  10 //
  11 // Unless required by applicable law or agreed to in writing, software
  12 // distributed under the License is distributed on an "AS IS" BASIS,
  13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 // See the License for the specific language governing permissions and
  15 // limitations under the License.
  16 //
  17 ////////////////////////////////////////////////////////////////////////////
  18
  19 #include "impl/external_commit_helper.hpp"
  20
  21 #include "impl/realm_coordinator.hpp"
  22
  23 #include <asl.h>
  24 #include <assert.h>
  25 #include <fcntl.h>
  26 #include <sstream>
  27 #include <sys/event.h>
  28 #include <sys/stat.h>
  29 #include <sys/time.h>
  30 #include <system_error>
  31 #include <unistd.h>
  32
  33 using namespace realm;
  34 using namespace realm::_impl;
  35
  36 namespace {
  37 // Write a byte to a pipe to notify anyone waiting for data on the pipe
  38 void notify_fd(int fd, int read_fd)
  39 {
  40     while (true) {
  41         char c = 0;
  42         ssize_t ret = write(fd, &c, 1);
  43         if (ret == 1) {
  44             break;
  45         }
  46
  47         // If the pipe's buffer is full, we need to read some of the old data in
  48         // it to make space. We don't just read in the code waiting for
  49         // notifications so that we can notify multiple waiters with a single
  50         // write.
  51         assert(ret == -1 && errno == EAGAIN);
  52         char buff[1024];
  53         read(read_fd, buff, sizeof buff);
  54     }
  55 }
  56 } // anonymous namespace
  57
  58 void ExternalCommitHelper::FdHolder::close()
  59 {
  60     if (m_fd != -1) {
  61         ::close(m_fd);
  62     }
  63     m_fd = -1;
  64 }
  65
  66 // Inter-thread and inter-process notifications of changes are done using a
  67 // named pipe in the filesystem next to the Realm file. Everyone who wants to be
  68 // notified of commits waits for data to become available on the pipe, and anyone
  69 // who commits a write transaction writes data to the pipe after releasing the
  70 // write lock. Note that no one ever actually *reads* from the pipe: the data
  71 // actually written is meaningless, and trying to read from a pipe from multiple
  72 // processes at once is fraught with race conditions.
  73
  74 // When a RLMRealm instance is created, we add a CFRunLoopSource to the current
  75 // thread's runloop. On each cycle of the run loop, the run loop checks each of
  76 // its sources for work to do, which in the case of CFRunLoopSource is just
  77 // checking if CFRunLoopSourceSignal has been called since the last time it ran,
  78 // and if so invokes the function pointer supplied when the source is created,
  79 // which in our case just invokes `[realm handleExternalChange]`.
  80
  81 // Listening for external changes is done using kqueue() on a background thread.
  82 // kqueue() lets us efficiently wait until the amount of data which can be read
  83 // from one or more file descriptors has changed, and tells us which of the file
  84 // descriptors it was that changed. We use this to wait on both the shared named
  85 // pipe, and a local anonymous pipe. When data is written to the named pipe, we
  86 // signal the runloop source and wake up the target runloop, and when data is
  87 // written to the anonymous pipe the background thread removes the runloop
  88 // source from the runloop and and shuts down.
  89 ExternalCommitHelper::ExternalCommitHelper(RealmCoordinator& parent)
  90 : m_parent(parent)
  91 {
  92     m_kq = kqueue();
  93     if (m_kq == -1) {
  94         throw std::system_error(errno, std::system_category());
  95     }
  96
  97 #if !TARGET_OS_TV
  98     auto path = parent.get_path() + ".note";
  99
 100     // Create and open the named pipe
 101     int ret = mkfifo(path.c_str(), 0600);
 102     if (ret == -1) {
 103         int err = errno;
 104         if (err == ENOTSUP) {
 105             // Filesystem doesn't support named pipes, so try putting it in tmp instead
 106             // Hash collisions are okay here because they just result in doing
 107             // extra work, as opposed to correctness problems
 108             std::ostringstream ss;
 109             ss << getenv("TMPDIR");
 110             ss << "realm_" << std::hash<std::string>()(path) << ".note";
 111             path = ss.str();
 112             ret = mkfifo(path.c_str(), 0600);
 113             err = errno;
 114         }
 115         // the fifo already existing isn't an error
 116         if (ret == -1 && err != EEXIST) {
 117             throw std::system_error(err, std::system_category());
 118         }
 119     }
 120
 121     m_notify_fd = open(path.c_str(), O_RDWR);
 122     if (m_notify_fd == -1) {
 123         throw std::system_error(errno, std::system_category());
 124     }
 125
 126     // Make writing to the pipe return -1 when the pipe's buffer is full
 127     // rather than blocking until there's space available
 128     ret = fcntl(m_notify_fd, F_SETFL, O_NONBLOCK);
 129     if (ret == -1) {
 130         throw std::system_error(errno, std::system_category());
 131     }
 132
 133 #else // !TARGET_OS_TV
 134
 135     // tvOS does not support named pipes, so use an anonymous pipe instead
 136     int notification_pipe[2];
 137     int ret = pipe(notification_pipe);
 138     if (ret == -1) {
 139         throw std::system_error(errno, std::system_category());
 140     }
 141
 142     m_notify_fd = notification_pipe[0];
 143     m_notify_fd_write = notification_pipe[1];
 144
 145 #endif // TARGET_OS_TV
 146
 147     // Create the anonymous pipe for shutdown notifications
 148     int shutdown_pipe[2];
 149     ret = pipe(shutdown_pipe);
 150     if (ret == -1) {
 151         throw std::system_error(errno, std::system_category());
 152     }
 153
 154     m_shutdown_read_fd = shutdown_pipe[0];
 155     m_shutdown_write_fd = shutdown_pipe[1];
 156
 157     m_thread = std::async(std::launch::async, [=] {
 158         try {
 159             listen();
 160         }
 161 #pragma clang diagnostic push
 162 #pragma clang diagnostic ignored "-Wdeprecated-declarations"
 163         catch (std::exception const& e) {
 164             fprintf(stderr, "uncaught exception in notifier thread: %s: %s\n", typeid(e).name(), e.what());
 165             asl_log(nullptr, nullptr, ASL_LEVEL_ERR, "uncaught exception in notifier thread: %s: %s", typeid(e).name(), e.what());
 166             throw;
 167         }
 168         catch (...) {
 169             fprintf(stderr,  "uncaught exception in notifier thread\n");
 170             asl_log(nullptr, nullptr, ASL_LEVEL_ERR, "uncaught exception in notifier thread");
 171             throw;
 172         }
 173 #pragma clang diagnostic pop
 174     });
 175 }
 176
 177 ExternalCommitHelper::~ExternalCommitHelper()
 178 {
 179     notify_fd(m_shutdown_write_fd, m_shutdown_read_fd);
 180     m_thread.wait(); // Wait for the thread to exit
 181 }
 182
 183 void ExternalCommitHelper::listen()
 184 {
 185     pthread_setname_np("RLMRealm notification listener");
 186
 187     // Set up the kqueue
 188     // EVFILT_READ indicates that we care about data being available to read
 189     // on the given file descriptor.
 190     // EV_CLEAR makes it wait for the amount of data available to be read to
 191     // change rather than just returning when there is any data to read.
 192     struct kevent ke[2];
 193     EV_SET(&ke[0], m_notify_fd, EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, 0);
 194     EV_SET(&ke[1], m_shutdown_read_fd, EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, 0);
 195     int ret = kevent(m_kq, ke, 2, nullptr, 0, nullptr);
 196     assert(ret == 0);
 197
 198     while (true) {
 199         struct kevent event;
 200         // Wait for data to become on either fd
 201         // Return code is number of bytes available or -1 on error
 202         ret = kevent(m_kq, nullptr, 0, &event, 1, nullptr);
 203         assert(ret >= 0);
 204         if (ret == 0) {
 205             // Spurious wakeup; just wait again
 206             continue;
 207         }
 208
 209         // Check which file descriptor had activity: if it's the shutdown
 210         // pipe, then someone called -stop; otherwise it's the named pipe
 211         // and someone committed a write transaction
 212         if (event.ident == (uint32_t)m_shutdown_read_fd) {
 213             return;
 214         }
 215         assert(event.ident == (uint32_t)m_notify_fd);
 216
 217         m_parent.on_change();
 218     }
 219 }
 220
 221 void ExternalCommitHelper::notify_others()
 222 {
 223     if (m_notify_fd_write != -1) {
 224         notify_fd(m_notify_fd_write, m_notify_fd);
 225     }
 226     else {
 227         notify_fd(m_notify_fd, m_notify_fd);
 228     }
 229 }