[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 13 matching lines @@
 #include "mojo/edk/system/async_waiter.h"
 #include "mojo/edk/system/channel.h"
 #include "mojo/edk/system/configuration.h"
 #include "mojo/edk/system/data_pipe_consumer_dispatcher.h"
 #include "mojo/edk/system/data_pipe_producer_dispatcher.h"
 #include "mojo/edk/system/handle_signals_state.h"
 #include "mojo/edk/system/message_pipe_dispatcher.h"
 #include "mojo/edk/system/platform_handle_dispatcher.h"
 #include "mojo/edk/system/ports/node.h"
 #include "mojo/edk/system/remote_message_pipe_bootstrap.h"
+#include "mojo/edk/system/request_context.h"
 #include "mojo/edk/system/shared_buffer_dispatcher.h"
 #include "mojo/edk/system/wait_set_dispatcher.h"
 #include "mojo/edk/system/waiter.h"
 
 namespace mojo {
 namespace edk {
 
 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 201 matching lines @@
   if (rv == MOJO_RESULT_OK)
     ignore_result(waiter.release());
   return rv;
 }
 
 MojoTimeTicks Core::GetTimeTicksNow() {
   return base::TimeTicks::Now().ToInternalValue();
 }
 
 MojoResult Core::Close(MojoHandle handle) {
+  RequestContext request_context;
   scoped_refptr<Dispatcher> dispatcher;
   {
     base::AutoLock lock(handles_lock_);
     MojoResult rv = handles_.GetAndRemoveDispatcher(handle, &dispatcher);
     if (rv != MOJO_RESULT_OK)
       return rv;
   }
   dispatcher->Close();
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::Wait(MojoHandle handle,
                       MojoHandleSignals signals,
                       MojoDeadline deadline,
                       MojoHandleSignalsState* signals_state) {
+  RequestContext request_context;
   uint32_t unused = static_cast<uint32_t>(-1);
   HandleSignalsState hss;
   MojoResult rv = WaitManyInternal(&handle, &signals, 1, deadline, &unused,
                                    signals_state ? &hss : nullptr);
   if (rv != MOJO_RESULT_INVALID_ARGUMENT && signals_state)
     *signals_state = hss;
   return rv;
 }
 
 MojoResult Core::WaitMany(const MojoHandle* handles,
                           const MojoHandleSignals* signals,
                           uint32_t num_handles,
                           MojoDeadline deadline,
                           uint32_t* result_index,
                           MojoHandleSignalsState* signals_state) {
+  RequestContext request_context;
   if (num_handles < 1)
     return MOJO_RESULT_INVALID_ARGUMENT;
   if (num_handles > GetConfiguration().max_wait_many_num_handles)
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
 
   uint32_t index = static_cast<uint32_t>(-1);
   MojoResult rv;
   if (!signals_state) {
     rv = WaitManyInternal(handles, signals, num_handles, deadline, &index,
                           nullptr);
   } else {
     // 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) {
+  RequestContext request_context;
+  scoped_refptr<Dispatcher> dispatcher = GetDispatcher(handle);
+  if (!dispatcher)
+    return MOJO_RESULT_INVALID_ARGUMENT;
+  return dispatcher->Watch(
+      signals, base::Bind(&CallWatchCallback, callback, context), context);
+}
+
+MojoResult Core::CancelWatch(MojoHandle handle, uintptr_t context) {
+  RequestContext request_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) {
+  RequestContext request_context;
   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();
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   *wait_set_handle = h;
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::AddHandle(MojoHandle wait_set_handle,
                            MojoHandle handle,
                            MojoHandleSignals signals) {
+  RequestContext request_context;
   scoped_refptr<Dispatcher> wait_set_dispatcher(GetDispatcher(wait_set_handle));
   if (!wait_set_dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   scoped_refptr<Dispatcher> dispatcher(GetDispatcher(handle));
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return wait_set_dispatcher->AddWaitingDispatcher(dispatcher, signals, handle);
 }
 
 MojoResult Core::RemoveHandle(MojoHandle wait_set_handle,
                               MojoHandle handle) {
+  RequestContext request_context;
   scoped_refptr<Dispatcher> wait_set_dispatcher(GetDispatcher(wait_set_handle));
   if (!wait_set_dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   scoped_refptr<Dispatcher> dispatcher(GetDispatcher(handle));
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return wait_set_dispatcher->RemoveWaitingDispatcher(dispatcher);
 }
 
 MojoResult Core::GetReadyHandles(MojoHandle wait_set_handle,
                                  uint32_t* count,
                                  MojoHandle* handles,
                                  MojoResult* results,
                                  MojoHandleSignalsState* signals_states) {
+  RequestContext request_context;
   if (!handles || !count || !(*count) || !results)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   scoped_refptr<Dispatcher> wait_set_dispatcher(GetDispatcher(wait_set_handle));
   if (!wait_set_dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   DispatcherVector awoken_dispatchers;
   base::StackVector<uintptr_t, 16> contexts;
   contexts->assign(*count, MOJO_HANDLE_INVALID);
 
   MojoResult result = wait_set_dispatcher->GetReadyDispatchers(
       count, &awoken_dispatchers, results, contexts->data());
 
   if (result == MOJO_RESULT_OK) {
     for (size_t i = 0; i < *count; i++) {
       handles[i] = static_cast<MojoHandle>(contexts[i]);
       if (signals_states)
         signals_states[i] = awoken_dispatchers[i]->GetHandleSignalsState();
     }
   }
 
   return result;
 }
 
 MojoResult Core::CreateMessagePipe(
     const MojoCreateMessagePipeOptions* options,
     MojoHandle* message_pipe_handle0,
     MojoHandle* message_pipe_handle1) {
+  RequestContext request_context;
   ports::PortRef port0, port1;
   GetNodeController()->node()->CreatePortPair(&port0, &port1);
 
   CHECK(message_pipe_handle0);
   CHECK(message_pipe_handle1);
 
   uint64_t pipe_id = base::RandUint64();
 
   *message_pipe_handle0 = AddDispatcher(
       new MessagePipeDispatcher(GetNodeController(), port0, pipe_id, 0));
@@ skipping 11 matching lines @@
 
   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) {
+  RequestContext request_context;
   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);
 
   CHECK(handles);
 
   if (num_handles > kMaxHandlesPerMessage)
@@ skipping 29 matching lines @@
 
   return rv;
 }
 
 MojoResult Core::ReadMessage(MojoHandle message_pipe_handle,
                              void* bytes,
                              uint32_t* num_bytes,
                              MojoHandle* handles,
                              uint32_t* num_handles,
                              MojoReadMessageFlags flags) {
+  RequestContext request_context;
   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);
 }
 
 MojoResult Core::CreateDataPipe(
     const MojoCreateDataPipeOptions* options,
     MojoHandle* data_pipe_producer_handle,
     MojoHandle* data_pipe_consumer_handle) {
+  RequestContext request_context;
   if (options && options->struct_size != sizeof(MojoCreateDataPipeOptions))
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   MojoCreateDataPipeOptions create_options;
   create_options.struct_size = sizeof(MojoCreateDataPipeOptions);
   create_options.flags = options ? options->flags : 0;
   create_options.element_num_bytes = options ? options->element_num_bytes : 1;
   // TODO: Use Configuration to get default data pipe capacity.
   create_options.capacity_num_bytes =
       options && options->capacity_num_bytes ? options->capacity_num_bytes
@@ skipping 34 matching lines @@
     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) {
+  RequestContext request_context;
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_producer_handle));
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return dispatcher->WriteData(elements, num_bytes, flags);
 }
 
 MojoResult Core::BeginWriteData(MojoHandle data_pipe_producer_handle,
                                 void** buffer,
                                 uint32_t* buffer_num_bytes,
                                 MojoWriteDataFlags flags) {
+  RequestContext request_context;
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_producer_handle));
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return dispatcher->BeginWriteData(buffer, buffer_num_bytes, flags);
 }
 
 MojoResult Core::EndWriteData(MojoHandle data_pipe_producer_handle,
                               uint32_t num_bytes_written) {
+  RequestContext request_context;
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_producer_handle));
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return dispatcher->EndWriteData(num_bytes_written);
 }
 
 MojoResult Core::ReadData(MojoHandle data_pipe_consumer_handle,
                           void* elements,
                           uint32_t* num_bytes,
                           MojoReadDataFlags flags) {
+  RequestContext request_context;
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_consumer_handle));
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return dispatcher->ReadData(elements, num_bytes, flags);
 }
 
 MojoResult Core::BeginReadData(MojoHandle data_pipe_consumer_handle,
                                const void** buffer,
                                uint32_t* buffer_num_bytes,
                                MojoReadDataFlags flags) {
+  RequestContext request_context;
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_consumer_handle));
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return dispatcher->BeginReadData(buffer, buffer_num_bytes, flags);
 }
 
 MojoResult Core::EndReadData(MojoHandle data_pipe_consumer_handle,
                              uint32_t num_bytes_read) {
+  RequestContext request_context;
   scoped_refptr<Dispatcher> dispatcher(
       GetDispatcher(data_pipe_consumer_handle));
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return dispatcher->EndReadData(num_bytes_read);
 }
 
 MojoResult Core::CreateSharedBuffer(
     const MojoCreateSharedBufferOptions* options,
     uint64_t num_bytes,
     MojoHandle* shared_buffer_handle) {
+  RequestContext request_context;
   MojoCreateSharedBufferOptions validated_options = {};
   MojoResult result = SharedBufferDispatcher::ValidateCreateOptions(
       options, &validated_options);
   if (result != MOJO_RESULT_OK)
     return result;
 
   scoped_refptr<SharedBufferDispatcher> dispatcher;
   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();
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::DuplicateBufferHandle(
     MojoHandle buffer_handle,
     const MojoDuplicateBufferHandleOptions* options,
     MojoHandle* new_buffer_handle) {
+  RequestContext request_context;
   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();
     return MOJO_RESULT_RESOURCE_EXHAUSTED;
   }
 
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::MapBuffer(MojoHandle buffer_handle,
                            uint64_t offset,
                            uint64_t num_bytes,
                            void** buffer,
                            MojoMapBufferFlags flags) {
+  RequestContext request_context;
   scoped_refptr<Dispatcher> dispatcher(GetDispatcher(buffer_handle));
   if (!dispatcher)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   scoped_ptr<PlatformSharedBufferMapping> mapping;
   MojoResult result = dispatcher->MapBuffer(offset, num_bytes, flags, &mapping);
   if (result != MOJO_RESULT_OK)
     return result;
 
   DCHECK(mapping);
   void* address = mapping->GetBase();
   {
     base::AutoLock locker(mapping_table_lock_);
     result = mapping_table_.AddMapping(std::move(mapping));
   }
   if (result != MOJO_RESULT_OK)
     return result;
 
   *buffer = address;
   return MOJO_RESULT_OK;
 }
 
 MojoResult Core::UnmapBuffer(void* buffer) {
+  RequestContext request_context;
   base::AutoLock lock(mapping_table_lock_);
   return mapping_table_.RemoveMapping(buffer);
 }
 
 void Core::GetActiveHandlesForTest(std::vector<MojoHandle>* handles) {
   base::AutoLock lock(handles_lock_);
   handles_.GetActiveHandlesForTest(handles);
 }
 
 MojoResult Core::WaitManyInternal(const MojoHandle* handles,
@@ skipping 69 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