[mojo-edk] Add MojoWatch and MojoCancelWatch APIs

mojo/edk/system/core.cc

 // Copyright 2013 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "mojo/edk/system/core.h"
 
 #include <string.h>
 
 #include <utility>
 
@@ skipping 29 matching lines @@
 
 namespace {
 
 // This is an unnecessarily large limit that is relatively easy to enforce.
 const uint32_t kMaxHandlesPerMessage = 1024 * 1024;
 
 // TODO: Maybe we could negotiate a debugging pipe ID for cross-process pipes
 // too; for now we just use a constant. This only affects bootstrap pipes.
 const uint64_t kUnknownPipeIdForDebug = 0x7f7f7f7f7f7f7f7fUL;
 
+void CallWatchCallback(MojoWatchCallback callback,
+                       uintptr_t context,
+                       MojoResult result,
+                       MojoHandleSignalsState signals_state) {
+  callback(context, result, signals_state);
+}
+
 }  // namespace
 
 Core::Core() {}
 
 Core::~Core() {
   if (node_controller_ && node_controller_->io_task_runner()) {
     // If this races with IO thread shutdown the callback will be dropped and
     // the NodeController will be shutdown on this thread anyway, which is also
     // just fine.
     scoped_refptr<base::TaskRunner> io_task_runner =
@@ skipping 38 matching lines @@
 bool Core::AddDispatchersFromTransit(
     const std::vector<Dispatcher::DispatcherInTransit>& dispatchers,
     MojoHandle* handles) {
   bool failed = false;
   {
     base::AutoLock lock(handles_lock_);
     if (!handles_.AddDispatchersFromTransit(dispatchers, handles))
       failed = true;
   }
   if (failed) {
+    Dispatcher::RequestContext request_context;
     for (auto d : dispatchers)
-      d.dispatcher->Close();
+      d.dispatcher->Close(&request_context);
     return false;
   }
   return true;
 }
 
 MojoResult Core::CreatePlatformHandleWrapper(
     ScopedPlatformHandle platform_handle,
     MojoHandle* wrapper_handle) {
   MojoHandle h = AddDispatcher(
       PlatformHandleDispatcher::Create(std::move(platform_handle)));
   if (h == MOJO_HANDLE_INVALID)
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   *wrapper_handle = h;
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::PassWrappedPlatformHandle(
     MojoHandle wrapper_handle,
     ScopedPlatformHandle* platform_handle) {
   base::AutoLock lock(handles_lock_);
   scoped_refptr<Dispatcher> d;
   MojoResult result = handles_.GetAndRemoveDispatcher(wrapper_handle, &d);
   if (result != MOJO_RESULT_OK)
     return result;
   PlatformHandleDispatcher* phd =
       static_cast<PlatformHandleDispatcher*>(d.get());
   *platform_handle = phd->PassPlatformHandle();
-  phd->Close();
+
+  Dispatcher::RequestContext request_context;
+  phd->Close(&request_context);
+
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::CreateSharedBufferWrapper(
     base::SharedMemoryHandle shared_memory_handle,
     size_t num_bytes,
     bool read_only,
     MojoHandle* mojo_wrapper_handle) {
   DCHECK(num_bytes);
   CHECK(!read_only);
@@ skipping 46 matching lines @@
 
   if (!platform_shared_buffer)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (num_bytes)
     *num_bytes = platform_shared_buffer->GetNumBytes();
   if (read_only)
     *read_only = false;
   *shared_memory_handle = platform_shared_buffer->DuplicateSharedMemoryHandle();
 
-  shm_dispatcher->Close();
+  Dispatcher::RequestContext request_context;
+  shm_dispatcher->Close(&request_context);
+
   return result;
 }
 
 void Core::RequestShutdown(const base::Closure& callback) {
   base::Closure on_shutdown;
   if (base::ThreadTaskRunnerHandle::IsSet()) {
     on_shutdown = base::Bind(base::IgnoreResult(&base::TaskRunner::PostTask),
                              base::ThreadTaskRunnerHandle::Get(),
                              FROM_HERE, callback);
   } else {
@@ skipping 53 matching lines @@
 }
 
 MojoResult Core::Close(MojoHandle handle) {
   scoped_refptr<Dispatcher> dispatcher;
   {
     base::AutoLock lock(handles_lock_);
     MojoResult rv = handles_.GetAndRemoveDispatcher(handle, &dispatcher);
     if (rv != MOJO_RESULT_OK)
       return rv;
   }
-  dispatcher->Close();
+
+  Dispatcher::RequestContext request_context;
+  dispatcher->Close(&request_context);
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::Wait(MojoHandle handle,
                       MojoHandleSignals signals,
                       MojoDeadline deadline,
                       MojoHandleSignalsState* signals_state) {
   uint32_t unused = static_cast<uint32_t>(-1);
   HandleSignalsState hss;
   MojoResult rv = WaitManyInternal(&handle, &signals, 1, deadline, &unused,
@@ skipping 23 matching lines @@
     // Note: The |reinterpret_cast| is safe, since |HandleSignalsState| is a
     // subclass of |MojoHandleSignalsState| that doesn't add any data members.
     rv = WaitManyInternal(handles, signals, num_handles, deadline, &index,
                           reinterpret_cast<HandleSignalsState*>(signals_state));
   }
   if (index != static_cast<uint32_t>(-1) && result_index)
     *result_index = index;
   return rv;
 }
 
+MojoResult Core::Watch(MojoHandle handle,
+                       MojoHandleSignals signals,
+                       MojoWatchCallback callback,
+                       uintptr_t context) {
+  scoped_refptr<Dispatcher> dispatcher = GetDispatcher(handle);
+  if (!dispatcher)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  Dispatcher::RequestContext request_context;
+  return dispatcher->Watch(
+      signals, base::Bind(&CallWatchCallback, callback, context), context,
+      &request_context);
+}
+
+MojoResult Core::CancelWatch(MojoHandle handle, uintptr_t context) {
+  scoped_refptr<Dispatcher> dispatcher = GetDispatcher(handle);
+  if (!dispatcher)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+  return dispatcher->CancelWatch(context);
+}
+
 MojoResult Core::CreateWaitSet(MojoHandle* wait_set_handle) {
   if (!wait_set_handle)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   scoped_refptr<WaitSetDispatcher> dispatcher = new WaitSetDispatcher();
   MojoHandle h = AddDispatcher(dispatcher);
   if (h == MOJO_HANDLE_INVALID) {
     LOG(ERROR) << "Handle table full";
-    dispatcher->Close();
+    Dispatcher::RequestContext request_context;
+    dispatcher->Close(&request_context);
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   *wait_set_handle = h;
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::AddHandle(MojoHandle wait_set_handle,
                            MojoHandle handle,
                            MojoHandleSignals signals) {
@@ skipping 64 matching lines @@
   uint64_t pipe_id = base::RandUint64();
 
   *message_pipe_handle0 = AddDispatcher(
       new MessagePipeDispatcher(GetNodeController(), port0, pipe_id, 0));
   if (*message_pipe_handle0 == MOJO_HANDLE_INVALID)
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
 
   *message_pipe_handle1 = AddDispatcher(
       new MessagePipeDispatcher(GetNodeController(), port1, pipe_id, 1));
   if (*message_pipe_handle1 == MOJO_HANDLE_INVALID) {
+    Dispatcher::RequestContext request_context;
     scoped_refptr<Dispatcher> unused;
-    unused->Close();
+    unused->Close(&request_context);
     handles_.GetAndRemoveDispatcher(*message_pipe_handle0, &unused);
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::WriteMessage(MojoHandle message_pipe_handle,
                               const void* bytes,
                               uint32_t num_bytes,
                               const MojoHandle* handles,
                               uint32_t num_handles,
                               MojoWriteMessageFlags flags) {
   auto dispatcher = GetDispatcher(message_pipe_handle);
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
-  if (num_handles == 0)  // Fast path: no handles.
-    return dispatcher->WriteMessage(bytes, num_bytes, nullptr, 0, flags);
+  Dispatcher::RequestContext request_context;
+  if (num_handles == 0) {
+    // Fast path: no handles.
+    return dispatcher->WriteMessage(
+        bytes, num_bytes, nullptr, 0, flags, &request_context);
+  }
 
   CHECK(handles);
 
   if (num_handles > kMaxHandlesPerMessage)
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
 
   for (size_t i = 0; i < num_handles; ++i) {
     if (message_pipe_handle == handles[i])
       return MOJO_RESULT_BUSY;
   }
 
   std::vector<Dispatcher::DispatcherInTransit> dispatchers;
   {
     base::AutoLock lock(handles_lock_);
-    MojoResult rv = handles_.BeginTransit(handles, num_handles, &dispatchers);
+    MojoResult rv = handles_.BeginTransit(
+        handles, num_handles, &dispatchers, &request_context);
     if (rv != MOJO_RESULT_OK) {
-      handles_.CancelTransit(dispatchers);
+      handles_.CancelTransit(dispatchers, &request_context);
       return rv;
     }
   }
   DCHECK_EQ(num_handles, dispatchers.size());
 
   MojoResult rv = dispatcher->WriteMessage(
-      bytes, num_bytes, dispatchers.data(), num_handles, flags);
+      bytes, num_bytes, dispatchers.data(), num_handles, flags,
+      &request_context);
 
   {
     base::AutoLock lock(handles_lock_);
     if (rv == MOJO_RESULT_OK) {
-      handles_.CompleteTransitAndClose(dispatchers);
+      handles_.CompleteTransitAndClose(dispatchers, &request_context);
     } else {
-      handles_.CancelTransit(dispatchers);
+      handles_.CancelTransit(dispatchers, &request_context);
     }
   }
 
   return rv;
 }
 
 MojoResult Core::ReadMessage(MojoHandle message_pipe_handle,
                              void* bytes,
                              uint32_t* num_bytes,
                              MojoHandle* handles,
                              uint32_t* num_handles,
                              MojoReadMessageFlags flags) {
   CHECK((!num_handles || !*num_handles || handles) &&
         (!num_bytes || !*num_bytes || bytes));
   auto dispatcher = GetDispatcher(message_pipe_handle);
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
-  return dispatcher->ReadMessage(bytes, num_bytes, handles, num_handles, flags);
+  Dispatcher::RequestContext request_context;
+  return dispatcher->ReadMessage(
+      bytes, num_bytes, handles, num_handles, flags, &request_context);
 }
 
 MojoResult Core::CreateDataPipe(
     const MojoCreateDataPipeOptions* options,
     MojoHandle* data_pipe_producer_handle,
     MojoHandle* data_pipe_consumer_handle) {
   if (options && options->struct_size != sizeof(MojoCreateDataPipeOptions))
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   MojoCreateDataPipeOptions create_options;
@@ skipping 28 matching lines @@
       true /* initialized */, pipe_id);
 
   *data_pipe_producer_handle = AddDispatcher(producer);
   *data_pipe_consumer_handle = AddDispatcher(consumer);
   if (*data_pipe_producer_handle == MOJO_HANDLE_INVALID ||
       *data_pipe_consumer_handle == MOJO_HANDLE_INVALID) {
     if (*data_pipe_producer_handle != MOJO_HANDLE_INVALID) {
       scoped_refptr<Dispatcher> unused;
       handles_.GetAndRemoveDispatcher(*data_pipe_producer_handle, &unused);
     }
-    producer->Close();
-    consumer->Close();
+
+    Dispatcher::RequestContext request_context;
+    producer->Close(&request_context);
+    consumer->Close(&request_context);
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::WriteData(MojoHandle data_pipe_producer_handle,
                            const void* elements,
                            uint32_t* num_bytes,
                            MojoWriteDataFlags flags) {
@@ skipping 75 matching lines @@
   result =
       SharedBufferDispatcher::Create(validated_options, num_bytes, &dispatcher);
   if (result != MOJO_RESULT_OK) {
     DCHECK(!dispatcher);
     return result;
   }
 
   *shared_buffer_handle = AddDispatcher(dispatcher);
   if (*shared_buffer_handle == MOJO_HANDLE_INVALID) {
     LOG(ERROR) << "Handle table full";
-    dispatcher->Close();
+    Dispatcher::RequestContext request_context;
+    dispatcher->Close(&request_context);
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::DuplicateBufferHandle(
     MojoHandle buffer_handle,
     const MojoDuplicateBufferHandleOptions* options,
     MojoHandle* new_buffer_handle) {
   scoped_refptr<Dispatcher> dispatcher(GetDispatcher(buffer_handle));
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   // Don't verify |options| here; that's the dispatcher's job.
   scoped_refptr<Dispatcher> new_dispatcher;
   MojoResult result =
       dispatcher->DuplicateBufferHandle(options, &new_dispatcher);
   if (result != MOJO_RESULT_OK)
     return result;
 
   *new_buffer_handle = AddDispatcher(new_dispatcher);
   if (*new_buffer_handle == MOJO_HANDLE_INVALID) {
     LOG(ERROR) << "Handle table full";
-    dispatcher->Close();
+    Dispatcher::RequestContext request_context;
+    dispatcher->Close(&request_context);
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::MapBuffer(MojoHandle buffer_handle,
                            uint64_t offset,
                            uint64_t num_bytes,
                            void** buffer,
@@ skipping 103 matching lines @@
     scoped_ptr<NodeController> node_controller) {
   // It's OK to leak this reference. At this point we know the IO loop is still
   // running, and we know the NodeController will observe its eventual
   // destruction. This tells the NodeController to delete itself when that
   // happens.
   node_controller.release()->DestroyOnIOThreadShutdown();
 }
 
 }  // namespace edk
 }  // namespace mojo