[Fuchsia] EventHandler: epoll -> ports v2

runtime/bin/eventhandler_fuchsia.h

 // Copyright (c) 2016, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 #ifndef RUNTIME_BIN_EVENTHANDLER_FUCHSIA_H_
 #define RUNTIME_BIN_EVENTHANDLER_FUCHSIA_H_
 
 #if !defined(RUNTIME_BIN_EVENTHANDLER_H_)
 #error Do not include eventhandler_fuchsia.h directly; use eventhandler.h.
 #endif
 
 #include <errno.h>
+#include <magenta/status.h>
+#include <magenta/syscalls.h>
+#include <magenta/syscalls/object.h>
+#include <magenta/syscalls/port.h>
+#include <mxio/private.h>
 #include <sys/epoll.h>
 #include <sys/socket.h>
 #include <unistd.h>
 
+#include "bin/reference_counting.h"
+#include "bin/thread.h"
 #include "platform/signal_blocker.h"
 
 namespace dart {
 namespace bin {
 
+class DescriptorInfo;
+
+class IOHandle : public ReferenceCounted<IOHandle> {
+ public:
+  explicit IOHandle(intptr_t fd)
+      : ReferenceCounted(),
+        mutex_(new Mutex()),
+        write_events_enabled_(true),
+        read_events_enabled_(true),
+        fd_(fd),
+        handle_(MX_HANDLE_INVALID),
+        mxio_(__mxio_fd_to_io(fd)) {}
+
+  intptr_t fd() const { return fd_; }
+
+  // Called from SocketBase::{Read(), Write()} and ServerSocket::Accept() on
+  // the Dart thread.
+  intptr_t Read(void* buffer, intptr_t num_bytes);
+  intptr_t Write(const void* buffer, intptr_t num_bytes);
+  intptr_t Accept(struct sockaddr* addr, socklen_t* addrlen);
+
+  // Called from the EventHandler thread.
+  void Close();
+  uint32_t MaskToEpollEvents(intptr_t mask);
+  bool AsyncWait(mx_handle_t port, uint32_t events, uint64_t key);
+  void CancelWait(mx_handle_t port, uint64_t key);
+  uint32_t WaitEnd(mx_signals_t observed);
+  intptr_t ToggleEvents(intptr_t event_mask);
+
+  static intptr_t EpollEventsToMask(intptr_t events);
+
+ private:
+  ~IOHandle() {
+    if (mxio_ != NULL) {
+      __mxio_release(mxio_);
+    }
+    delete mutex_;
+  }
+
+  mx_status_t ReadySignalsLocked(mx_signals_t* ready_signals);
+
+  // Mutex that protects the state here.
+  Mutex* mutex_;
+  bool write_events_enabled_;
+  bool read_events_enabled_;
+  // TODO(zra): Add flag to enable/disable peer closed signal?
+  intptr_t fd_;
+  mx_handle_t handle_;
+  mxio_t* mxio_;
+
+  friend class ReferenceCounted<IOHandle>;
+  DISALLOW_COPY_AND_ASSIGN(IOHandle);
+};
+
 class DescriptorInfo : public DescriptorInfoBase {
  public:
-  explicit DescriptorInfo(intptr_t fd) : DescriptorInfoBase(fd) {}
+  explicit DescriptorInfo(intptr_t fd) : DescriptorInfoBase(fd) {
+    IOHandle* handle = reinterpret_cast<IOHandle*>(fd);
+    handle->Retain();
+  }
 
-  virtual ~DescriptorInfo() {}
-
-  intptr_t GetPollEvents();
+  virtual ~DescriptorInfo() {
+    IOHandle* handle = reinterpret_cast<IOHandle*>(fd_);
+    handle->Release();
+  }
 
   virtual void Close() {
-    // Should be VOID_TEMP_FAILURE_RETRY
-    VOID_NO_RETRY_EXPECTED(close(fd_));
-    fd_ = -1;
+    IOHandle* handle = reinterpret_cast<IOHandle*>(fd_);
+    handle->Close();
   }
 
+  IOHandle* io_handle() const { return reinterpret_cast<IOHandle*>(fd_); }
+
  private:
   DISALLOW_COPY_AND_ASSIGN(DescriptorInfo);
 };
 
 class DescriptorInfoSingle : public DescriptorInfoSingleMixin<DescriptorInfo> {
  public:
   explicit DescriptorInfoSingle(intptr_t fd)
       : DescriptorInfoSingleMixin(fd, false) {}
   virtual ~DescriptorInfoSingle() {}
 
@@ skipping 10 matching lines @@
 
  private:
   DISALLOW_COPY_AND_ASSIGN(DescriptorInfoMultiple);
 };
 
 class EventHandlerImplementation {
  public:
   EventHandlerImplementation();
   ~EventHandlerImplementation();
 
-  void UpdateEpollInstance(intptr_t old_mask, DescriptorInfo* di);
+  void UpdatePort(intptr_t old_mask, DescriptorInfo* di);
 
   // Gets the socket data structure for a given file
   // descriptor. Creates a new one if one is not found.
   DescriptorInfo* GetDescriptorInfo(intptr_t fd, bool is_listening);
   void SendData(intptr_t id, Dart_Port dart_port, int64_t data);
   void Start(EventHandler* handler);
   void Shutdown();
 
  private:
+  static const uint64_t kInterruptPacketKey = 0xda47da47;

[abarth, 2017/06/19 19:02:45]

You might as well just use the value 0x1 for this purpose.  This value makes it
look like you're trying not to collide with other things, but the value appears
to be just used in a user packet, which you're fully in control of.

[zra, 2017/06/19 21:04:28]

Done.

+
   static void Poll(uword args);
   static void* GetHashmapKeyFromFd(intptr_t fd);
   static uint32_t GetHashmapHashFromFd(intptr_t fd);
+  static void AddToPort(mx_handle_t port_handle, DescriptorInfo* di);
+  static void RemoveFromPort(mx_handle_t port_handle, DescriptorInfo* di);
 
   int64_t GetTimeout() const;
-  void HandleEvents(struct epoll_event* events, int size);
+  void HandlePacket(mx_port_packet_t* pkt);
   void HandleTimeout();
   void WakeupHandler(intptr_t id, Dart_Port dart_port, int64_t data);
-  intptr_t GetPollEvents(intptr_t events, DescriptorInfo* di);
-  void HandleInterruptFd();
+  intptr_t GetPollEvents(intptr_t events);
+  void HandleInterrupt(InterruptMessage* msg);
 
   HashMap socket_map_;
   TimeoutQueue timeout_queue_;
   bool shutdown_;
-  int interrupt_fds_[2];
-  int epoll_fd_;
+  mx_handle_t port_handle_;
 
   DISALLOW_COPY_AND_ASSIGN(EventHandlerImplementation);
 };
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // RUNTIME_BIN_EVENTHANDLER_FUCHSIA_H_