Starting debugger rework, adding base async socket, removing flatbuffers.

src/xenia/base/socket.h

@@ -0,0 +1,108 @@
+/**
+ ******************************************************************************
+ * Xenia : Xbox 360 Emulator Research Project                                 *
+ ******************************************************************************
+ * Copyright 2015 Ben Vanik. All rights reserved.                             *
+ * Released under the BSD license - see LICENSE in the root for more details. *
+ ******************************************************************************
+ */
+
+#ifndef XENIA_BASE_SOCKET_H_
+#define XENIA_BASE_SOCKET_H_
+
+#include <cstdint>
+#include <functional>
+#include <memory>
+
+#include "xenia/base/threading.h"
+
+namespace xe {
+
+// An asynchronous bidirectional socket.
+// Sockets are wait-based and operate in a thread-free manner. Data may be
+// read from or written to a socket from any thread with the wait handle as
+// a mechanism to efficiently wait until new data is available.
+// Consumers should wait on the wait handle and when signalled poll to see
+// if the socket is still connected or new data is available. When reading
+// data Receive should be called until it returns 0 new bytes and then the
+// wait handle should be waited on until new data arrives.
+class Socket {
+ public:
+  // TODO(benvanik): client socket static Connect method.
+
+  virtual ~Socket() = default;
+
+  // TODO(benvanik): socket info? remote IP? etc?
+
+  // Returns a wait handle that can be used to wait on incoming client data or
+  // status updates. When this handle is signalled consumers should poll for
+  // new data and react accordingly.
+  virtual xe::threading::WaitHandle* wait_handle() = 0;
+
+  // Returns true if the client is connected and can send/receive data.
+  virtual bool is_connected() = 0;
+
+  // Closes the socket.
+  // This will signal the wait handle.
+  virtual void Close() = 0;
+
+  // Receives incoming data up to the given capacity and returns the total
+  // number of bytes received.
+  // 0 is returned if there is no data available. -1 is returned if the client
+  // has been closed.
+  virtual size_t Receive(void* buffer, size_t buffer_capacity) = 0;
+
+  // Asynchronously sends a set of buffers.
+  // The buffers are copied and may be modified immediately after the function
+  // returns.
+  // Returns false if the socket is disconnected or the data cannot be sent.
+  virtual bool Send(const std::pair<const void*, size_t>* buffers,
+                    size_t buffer_count) = 0;
+
+  // Asynchronously sends a set of buffers.
+  // The buffers are copied and may be modified immediately after the function
+  // returns.
+  // Returns false if the socket is disconnected or the data cannot be sent.
+  bool Send(std::vector<std::pair<const void*, size_t>> buffers) {
+    return Send(buffers.data(), buffers.size());
+  }
+
+  // Asynchronously sends a buffer of data.
+  // The buffer is copied and may be modified immediately after the function
+  // returns.
+  // Returns false if the socket is disconnected or the data cannot be sent.
+  bool Send(const void* buffer, size_t buffer_length) {
+    auto buffer_list = std::make_pair(buffer, buffer_length);
+    return Send(&buffer_list, 1);
+  }
+
+  // Asynchronously sends a buffer of data.
+  // The buffer is copied and may be modified immediately after the function
+  // returns.
+  // Returns false if the socket is disconnected or the data cannot be sent.
+  bool Send(const std::vector<uint8_t>& buffer) {
+    auto buffer_list = std::make_pair(buffer.data(), buffer.size());
+    return Send(&buffer_list, 1);
+  }
+};
+
+// Runs a socket server on the specified local port.
+// The server accepts clients from a background thread until closed (by
+// deletion). Clients are not tied to their creating server and the server
+// may be closed while clients remain connected.
+class SocketServer {
+ public:
+  // Creates a new socket server bound to the given local port.
+  // The accept callback will be called from a random thread when a new client
+  // connects.
+  // Returns null if the server cannot be bound to the given port (in use, etc).
+  static std::unique_ptr<SocketServer> Create(
+      uint16_t port,
+      std::function<void(std::unique_ptr<Socket> client)> accept_callback);
+
+  virtual ~SocketServer() = default;
+};
+
+}  // namespace xe
+
+#endif  // XENIA_BASE_SOCKET_H_

src/xenia/base/socket_win.cc

@@ -0,0 +1,249 @@
+/**
+ ******************************************************************************
+ * Xenia : Xbox 360 Emulator Research Project                                 *
+ ******************************************************************************
+ * Copyright 2015 Ben Vanik. All rights reserved.                             *
+ * Released under the BSD license - see LICENSE in the root for more details. *
+ ******************************************************************************
+ */
+
+#include "xenia/base/socket.h"
+
+#include <mutex>
+#include <thread>
+
+#include "xenia/base/logging.h"
+
+// platform_win.h must come first to include Windows headers.
+#include "xenia/base/platform_win.h"
+#include <mstcpip.h>
+#include <winsock2.h>
+#include <ws2tcpip.h>
+
+namespace xe {
+
+void InitializeWinsock() {
+  static bool has_initialized = false;
+  if (!has_initialized) {
+    has_initialized = true;
+    WSADATA wsa_data;
+    WSAStartup(MAKEWORD(2, 2), &wsa_data);
+  }
+}
+
+class Win32Socket : public Socket {
+ public:
+  Win32Socket() = default;
+  ~Win32Socket() override { Close(); }
+
+  bool Accept(SOCKET socket) {
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
+
+    socket_ = socket;
+
+    // Create event and bind to the socket, waiting for read/close
+    // notifications.
+    event_ = xe::threading::Event::CreateManualResetEvent(false);
+    WSAEventSelect(socket_, event_->native_handle(), FD_READ | FD_CLOSE);
+
+    // Keepalive for a looong time, as we may be paused by the debugger/etc.
+    struct tcp_keepalive alive;
+    alive.onoff = TRUE;
+    alive.keepalivetime = 7200000;
+    alive.keepaliveinterval = 6000;
+    DWORD bytes_returned;
+    WSAIoctl(socket_, SIO_KEEPALIVE_VALS, &alive, sizeof(alive), nullptr, 0,
+             &bytes_returned, nullptr, nullptr);
+
+    return true;
+  }
+
+  xe::threading::WaitHandle* wait_handle() override { return event_.get(); }
+
+  bool is_connected() override {
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
+    return socket_ != INVALID_SOCKET;
+  }
+
+  void Close() override {
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
+
+    if (socket_ != INVALID_SOCKET) {
+      SOCKET socket = socket_;
+      socket_ = INVALID_SOCKET;
+      shutdown(socket, SD_SEND);
+      closesocket(socket);
+    }
+
+    if (event_) {
+      // Set event so any future waits will immediately succeed.
+      event_->Set();
+    }
+  }
+
+  size_t Receive(void* buffer, size_t buffer_capacity) override {
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
+
+    if (socket_ == INVALID_SOCKET) {
+      return -1;
+    }
+    int ret =
+        recv(socket_, reinterpret_cast<char*>(buffer), int(buffer_capacity), 0);
+    if (ret == SOCKET_ERROR) {
+      int e = WSAGetLastError();
+      if (e == WSAEWOULDBLOCK) {
+        // Ok - no more data to read.
+        // Reset our event so we'll block next time.
+        event_->Reset();
+        return 0;
+      }
+      XELOGE("Socket send error: %d", e);
+      Close();
+      return -1;
+    } else if (ret == 0) {
+      // Socket gracefully closed.
+      Close();
+      return -1;
+    }
+    return ret;
+  }
+
+  bool Send(const std::pair<const void*, size_t>* buffers,
+            size_t buffer_count) override {
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
+
+    if (socket_ == INVALID_SOCKET) {
+      return false;
+    }
+    std::vector<WSABUF> buffer_data_list;
+    std::vector<DWORD> buffer_data_sent;
+    for (size_t i = 0; i < buffer_count; ++i) {
+      WSABUF buf;
+      buf.len = ULONG(buffers[i].second);
+      buf.buf = reinterpret_cast<char*>(const_cast<void*>(buffers[i].first));
+      buffer_data_list.emplace_back(buf);
+      buffer_data_sent.emplace_back(0);
+    }
+    int ret = WSASend(socket_, buffer_data_list.data(), DWORD(buffer_count),
+                      buffer_data_sent.data(), 0, nullptr, nullptr);
+    if (ret == SOCKET_ERROR) {
+      int e = WSAGetLastError();
+      XELOGE("Socket send error: %d", e);
+      Close();
+      return false;
+    }
+    return true;
+  }
+
+ private:
+  std::recursive_mutex mutex_;
+  SOCKET socket_ = INVALID_SOCKET;
+  std::unique_ptr<xe::threading::Event> event_;
+};
+
+class Win32SocketServer : public SocketServer {
+ public:
+  Win32SocketServer(
+      std::function<void(std::unique_ptr<Socket> client)> accept_callback)
+      : accept_callback_(std::move(accept_callback)) {}
+  ~Win32SocketServer() override {
+    if (socket_ != INVALID_SOCKET) {
+      SOCKET socket = socket_;
+      socket_ = INVALID_SOCKET;
+      linger so_linger;
+      so_linger.l_onoff = TRUE;
+      so_linger.l_linger = 30;
+      setsockopt(socket, SOL_SOCKET, SO_LINGER,
+                 reinterpret_cast<const char*>(&so_linger), sizeof(so_linger));
+      shutdown(socket, SD_SEND);
+      closesocket(socket);
+    }
+
+    if (accept_thread_) {
+      // Join thread, which should die because the socket is now invalid.
+      xe::threading::Wait(accept_thread_.get(), true);
+      accept_thread_.reset();
+    }
+  }
+
+  bool Bind(uint16_t port) {
+    socket_ = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+    if (socket_ < 1) {
+      XELOGE("Unable to create listen socket");
+      return false;
+    }
+    struct tcp_keepalive alive;
+    alive.onoff = TRUE;
+    alive.keepalivetime = 7200000;
+    alive.keepaliveinterval = 6000;
+    DWORD bytes_returned;
+    WSAIoctl(socket_, SIO_KEEPALIVE_VALS, &alive, sizeof(alive), nullptr, 0,
+             &bytes_returned, nullptr, nullptr);
+    int opt_value = 1;
+    setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR,
+               reinterpret_cast<const char*>(&opt_value), sizeof(opt_value));
+    opt_value = 1;
+    setsockopt(socket_, IPPROTO_TCP, TCP_NODELAY,
+               reinterpret_cast<const char*>(&opt_value), sizeof(opt_value));
+
+    sockaddr_in socket_addr = {0};
+    socket_addr.sin_family = AF_INET;
+    socket_addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    socket_addr.sin_port = htons(port);
+    if (bind(socket_, reinterpret_cast<sockaddr*>(&socket_addr),
+             sizeof(socket_addr)) == SOCKET_ERROR) {
+      int e = WSAGetLastError();
+      XELOGE("Unable to bind debug socket: %d", e);
+      return false;
+    }
+    if (listen(socket_, 5) == SOCKET_ERROR) {
+      int e = WSAGetLastError();
+      XELOGE("Unable to listen on accept socket %d", e);
+      return false;
+    }
+
+    accept_thread_ = xe::threading::Thread::Create({}, [this, port]() {
+      xe::threading::set_name(std::string("xe::SocketServer localhost:") +
+                              std::to_string(port));
+      while (socket_ != INVALID_SOCKET) {
+        sockaddr_in6 client_addr;
+        int client_count = sizeof(client_addr);
+        SOCKET client_socket = accept(
+            socket_, reinterpret_cast<sockaddr*>(&client_addr), &client_count);
+        if (client_socket == INVALID_SOCKET) {
+          continue;
+        }
+
+        auto client = std::make_unique<Win32Socket>();
+        if (!client->Accept(client_socket)) {
+          XELOGE("Unable to accept socket; ignoring");
+          continue;
+        }
+        accept_callback_(std::move(client));
+      }
+    });
+
+    return true;
+  }
+
+ private:
+  std::function<void(std::unique_ptr<Socket> client)> accept_callback_;
+  std::unique_ptr<xe::threading::Thread> accept_thread_;
+
+  SOCKET socket_ = INVALID_SOCKET;
+};
+
+std::unique_ptr<SocketServer> SocketServer::Create(
+    uint16_t port,
+    std::function<void(std::unique_ptr<Socket> client)> accept_callback) {
+  InitializeWinsock();
+
+  auto socket_server =
+      std::make_unique<Win32SocketServer>(std::move(accept_callback));
+  if (!socket_server->Bind(port)) {
+    return nullptr;
+  }
+  return std::unique_ptr<SocketServer>(socket_server.release());
+}
+
+}  // namespace xe