Move decryption workers into their own thread

Author: oskirby

src/platforms/macos/daemon/wgsessionmacos.cpp

@@ -74,13 +74,13 @@ WgSessionMacos::~WgSessionMacos() {
     close(m_netSocket);
   }
   if (m_encryptWorker) {
-    m_encryptWorker->shutdown();
+    m_encryptWorker->stop();
     m_encryptWorker->wait();
   }
-
-  // Shut down the worker pools.
-  m_decryptPool.clear();
-  m_decryptPool.waitForDone();
+  if (m_decryptWorker) {
+    m_decryptWorker->stop();
+    m_decryptWorker->wait();
+  }
 }
 
 void WgSessionMacos::processResult(int op, const QByteArray& buf) const {
@@ -137,22 +137,6 @@ void WgSessionMacos::renegotiate() {
   processResult(res.op, output.first(res.size));
 }
 
-void WgSessionMacos::mhopInput(const QByteArray& data) {
-  //logger.debug() << name() << "mhop:" << QString(data.toBase64());
-  quint32 family;
-  if (data.length() <= sizeof(family)) {
-    return;
-  }
-
-  QByteArray packet = data.mid(sizeof(family));
-  quint8 version = (packet[0] >> 4);
-  if (version == 4) {
-    mhopInputV4(packet);
-  } else if (version == 6) {
-    mhopInputV6(packet);
-  }
-}
-
 WireguardUtils::PeerStatus WgSessionMacos::status() const {
   WireguardUtils::PeerStatus result(m_config.m_serverPublicKey);
 
@@ -239,33 +223,51 @@ QByteArray WgSessionMacos::mhopHeader(const QByteArray& packet) const {
   return result;
 }
 
-void WgSessionMacos::mhopInputV4(const QByteArray& packet) {
+QByteArray WgSessionMacos::mhopUnwrap(const QByteArray& packet) {
+  if (packet.length() <= sizeof(quint32)) {
+    return QByteArray();
+  }
+
+  quint32 header = qFromBigEndian<quint32>(packet.constData());
+  if (header == AF_INET) {
+    return mhopInputV4(packet.sliced(sizeof(header)));
+  } else if (header == AF_INET6) {
+    return mhopInputV6(packet.sliced(sizeof(header)));
+  } else {
+    return QByteArray();
+  }
+}
+
+QByteArray WgSessionMacos::mhopInputV4(const QByteArray& packet) {
   // Parse the IPv4 header
   auto header = reinterpret_cast<const struct ipv4header*>(packet.constData());
   quint16 hlen = (header->ihl & 0xF) * 4;
+  if ((header->ihl >> 4) != 4) {
+    return QByteArray();
+  }
   if ((hlen < sizeof(struct ipv4header)) || (hlen > packet.length())) {
-    return;
+    return QByteArray();
   }
 
   // Validate the header.
   if ((qFromBigEndian(header->source) != m_serverIpv4.toIPv4Address()) ||
       (qFromBigEndian(header->dest) != m_innerIpv4.toIPv4Address()) ||
       (header->proto != IPPROTO_UDP) || (header->ttl == 0) ||
       inetChecksum(header, sizeof(struct ipv4header)) != 0x0000) {
-    return;
+    return QByteArray();
   }
 
   // Handle IPv4 defragmentation
-  QByteArray dgram = packet.mid(hlen);
+  QByteArray dgram = packet.sliced(hlen);
   if (header->frag & qToBigEndian<quint16>(0x3fff)) {
     dgram = mhopDefragV4(header, dgram);
     if (dgram.isEmpty()) {
-      return;
+      return QByteArray();
     }
   }
 
   // Process the UDP header
-  mhopInputUDP(m_serverIpv4, m_innerIpv4, dgram);
+  return mhopInputUDP(m_serverIpv4, m_innerIpv4, dgram);
 }
 
 QByteArray WgSessionMacos::mhopDefragV4(const struct ipv4header* header,
@@ -301,33 +303,32 @@ QByteArray WgSessionMacos::mhopDefragV4(const struct ipv4header* header,
   return result;
 }
 
-void WgSessionMacos::mhopInputV6(const QByteArray& packet) {
+QByteArray WgSessionMacos::mhopInputV6(const QByteArray& packet) {
   // TODO: Implement Me!
+  return QByteArray();
 }
 
-void WgSessionMacos::mhopInputUDP(const QHostAddress& src,
-                                  const QHostAddress& dst,
-                                  const QByteArray& dgram) {
+QByteArray WgSessionMacos::mhopInputUDP(const QHostAddress& src,
+                                        const QHostAddress& dst,
+                                        const QByteArray& dgram) {
   const quint16* hdr = reinterpret_cast<const quint16*>(dgram.constData());
   if ((dgram.length() < 8) || (hdr[0] != htons(m_serverPort)) ||
       (hdr[1] != htons(m_innerPort)) || (htons(hdr[2]) > dgram.length())) {
     logger.debug() << "mhop drop udp:" << dgram.toHex();
-    return;
+    return QByteArray();
   }
 
-  QByteArray data = dgram.mid(8);
+  QByteArray data = dgram.sliced(8);
   if (hdr[3] != 0x0000) {
     // Validate the checksum
     quint16 cksum = udpChecksum(src, dst, htons(hdr[0]), htons(hdr[1]), data);
     if (hdr[3] != cksum) {
       logger.debug() << "mhop drop cksum:" << dgram.toHex();
-      return;
+      return QByteArray();
     }
   }
 
-  // At last - we can handle the payload.
-  auto worker = new WgDecryptWorker(this, data);
-  m_decryptPool.start(worker);
+  return data;
 }
 
 bool WgSessionMacos::start(const struct sockaddr* addr, int len) {
@@ -354,9 +355,15 @@ void WgSessionMacos::start(qintptr sd) {
   fcntl(sd, F_SETFL, flags | O_NONBLOCK);
 
   m_netSocket = sd;
-  auto notifier = new QSocketNotifier(m_netSocket, QSocketNotifier::Read, this);
-  connect(notifier, &QSocketNotifier::activated, this,
-          &WgSessionMacos::netReadyRead);
+
+  // Start the decryption worker.
+  if (m_config.m_hopType == InterfaceConfig::MultiHopExit) {
+    m_decryptWorker = new WgMultihopWorker(this, sd);
+  } else {
+    m_decryptWorker = new WgDecryptWorker(this, sd);
+  }
+  m_decryptWorker->setMtu(m_tunmtu);
+  m_decryptWorker->start();
 
   renegotiate();
 }
@@ -382,6 +389,13 @@ void WgSessionMacos::setMtu(int mtu) {
   }
   m_tunmtu = mtu;
 
+  if (m_encryptWorker) {
+    m_encryptWorker->setMtu(mtu);
+  }
+  if (m_decryptWorker) {
+    m_decryptWorker->setMtu(mtu);
+  }
+
   // Set the MTU if it's a utun device.
   struct ifreq ifr;
   socklen_t ifnamesize = sizeof(ifr.ifr_name);
@@ -394,29 +408,6 @@ void WgSessionMacos::setMtu(int mtu) {
   }
 }
 
-void WgSessionMacos::netReadyRead() {
-  QByteArray rxbuf;
-  rxbuf.resize(m_tunmtu + WG_MTU_OVERHEAD);
-
-  while (true) {
-    // Try to read a packet from the network.
-    int rxlen = recv(m_netSocket, (void*)rxbuf.data(), rxbuf.length(), MSG_DONTWAIT);
-    if (rxlen < 0) {
-      if (errno == EAGAIN) return;
-      logger.debug() << "Recv error:" << strerror(errno);
-      return;
-    }
-  
-    QByteArray packet = rxbuf.first(rxlen);
-    if (m_config.m_hopType == InterfaceConfig::MultiHopExit) {
-      mhopInput(packet);
-    } else {
-      auto worker = new WgDecryptWorker(this, packet);
-      m_decryptPool.start(worker);
-    }
-  }
-}
-
 void WgSessionMacos::tunWrite(qintptr fd, const QByteArray& packet, const QByteArray& append) const {
   //logger.debug() << name() << "decrypt:" << QString(packet.toBase64());
   quint32 family = ((packet.at(0) >> 4) == 4) ? AF_INET : AF_INET6;

src/platforms/macos/daemon/wgsessionmacos.h

@@ -18,8 +18,7 @@ struct wireguard_tunnel;
 struct ipv4header;
 struct sockaddr;
 
-class WgEncryptWorker;
-class WgDecryptWorker;
+class WgSessionWorker;
 
 class WgSessionMacos final : public QObject {
   Q_OBJECT
@@ -42,22 +41,22 @@ class WgSessionMacos final : public QObject {
   static inline constexpr int WG_MTU_OVERHEAD = 80;
   static inline constexpr int WG_PACKET_OVERHEAD = 32;
 
- protected slots:
-  void netReadyRead();
+ protected:
+  QByteArray mhopUnwrap(const QByteArray& packet);
+  void processResult(int op, const QByteArray& buf) const;
 
  private:
   void timeout();
-  void processResult(int op, const QByteArray& buf) const;
 
   void tunWrite(qintptr fd, const QByteArray& packet,
                 const QByteArray& append = QByteArray()) const;
 
   QByteArray mhopHeader(const QByteArray& packet) const;
-  void mhopInput(const QByteArray& packet);
-  void mhopInputV4(const QByteArray& packet);
-  void mhopInputV6(const QByteArray& packet);
-  void mhopInputUDP(const QHostAddress& src, const QHostAddress& dst,
-                    const QByteArray& packet);
+
+  QByteArray mhopInputV4(const QByteArray& packet);
+  QByteArray mhopInputV6(const QByteArray& packet);
+  QByteArray mhopInputUDP(const QHostAddress& src, const QHostAddress& dst,
+                          const QByteArray& packet);
 
   QByteArray mhopDefragV4(const struct ipv4header* header,
                           const QByteArray& packet);
@@ -94,11 +93,12 @@ class WgSessionMacos final : public QObject {
 
  protected:
   struct wireguard_tunnel* m_tunnel = nullptr;
-  WgEncryptWorker* m_encryptWorker = nullptr;
-  QThreadPool m_decryptPool;
+  WgSessionWorker* m_encryptWorker = nullptr;
+  WgSessionWorker* m_decryptWorker = nullptr;
 
   friend class WgEncryptWorker;
   friend class WgDecryptWorker;
+  friend class WgMultihopWorker;
 };
 
 #endif  // WGSESSIONMACOS_H

src/platforms/macos/daemon/wgsessionworker.cpp

@@ -23,23 +23,37 @@ namespace {
 Logger logger("WgSessionWorker");
 };  // namespace
 
-WgEncryptWorker::WgEncryptWorker(WgSessionMacos* session, qintptr socket)
+WgSessionWorker::WgSessionWorker(WgSessionMacos* session, qintptr socket)
     : QThread(session), m_session(session) {
   MZ_COUNT_CTOR(WgSessionMacos);
+  logger.debug() << metaObject()->className() << "created.";
+
   m_socket = dup(socket);
   m_mtu = IPV6_MMTU;
 
   int flags = fcntl(m_socket, F_GETFL, 0);
   fcntl(m_socket, F_SETFL, flags & ~O_NONBLOCK);
 }
 
-WgEncryptWorker::~WgEncryptWorker() {
+WgSessionWorker::~WgSessionWorker() {
   MZ_COUNT_DTOR(WgSessionMacos);
+  logger.debug() << metaObject()->className() << "destroyed.";
+
   if (m_socket >= 0) {
     close(m_socket);
   }
 }
 
+void WgSessionWorker::setMtu(int mtu) {
+  logger.debug() << "set mtu:" << mtu;
+  m_mtu = mtu;
+}
+
+void WgSessionWorker::stop() {
+  requestInterruption();
+  shutdown(m_socket, SHUT_RD);
+}
+
 void WgEncryptWorker::run() {
   quint32 header = 0;
   QByteArray packet;
@@ -90,18 +104,56 @@ void WgEncryptWorker::run() {
   }
 }
 
-void WgEncryptWorker::shutdown() {
-  requestInterruption();
-  ::shutdown(m_socket, SHUT_RD);
+void WgDecryptWorker::run() {
+  QByteArray dgram;
+  QByteArray decrypt;
+
+  while (!isInterruptionRequested()) {
+    // Resize in case the MTU changed.
+    int mtu = m_mtu.loadAcquire();
+    dgram.resize(mtu + WgSessionMacos::WG_PACKET_OVERHEAD);
+    decrypt.resize(mtu + WgSessionMacos::WG_PACKET_OVERHEAD);
+
+    // Try to read a packet from the network.
+    uint8_t* ptr = reinterpret_cast<uint8_t*>(dgram.data());
+    int len = recv(m_socket, ptr, dgram.length(), 0);
+    if (len < 0) {
+      if (errno == EAGAIN) continue;
+      if (errno == EINTR) continue;
+      logger.debug() << "Recv error:" << strerror(errno);
+      return;
+    }
+
+    uint8_t* dec = reinterpret_cast<uint8_t*>(decrypt.data());
+    auto res = wireguard_read(m_session->m_tunnel, ptr, len, dec, decrypt.size());
+    m_session->processResult(res.op, decrypt.first(res.size));
+  }
 }
 
-void WgDecryptWorker::run() {
-  QByteArray output;
-  output.resize(m_packet.size());
-
-  const uint8_t* ptr = reinterpret_cast<const uint8_t*>(m_packet.constData());
-  uint8_t* outptr = reinterpret_cast<uint8_t*>(output.data());
-  auto res = wireguard_read(m_session->m_tunnel, ptr, m_packet.size(), outptr,
-                            output.size());
-  m_session->processResult(res.op, output.first(res.size));
+void WgMultihopWorker::run() {
+  QByteArray dgram;
+  QByteArray decrypt;
+
+  while (!isInterruptionRequested()) {
+    // Resize in case the MTU changed.
+    int mtu = m_mtu.loadAcquire();
+    dgram.resize(mtu + WgSessionMacos::WG_PACKET_OVERHEAD);
+    decrypt.resize(mtu + WgSessionMacos::WG_PACKET_OVERHEAD);
+
+    // Try to read a packet from the network.
+    int len = recv(m_socket, (void*)dgram.data(), dgram.length(), 0);
+    if (len < 0) {
+      if (errno == EAGAIN) continue;
+      if (errno == EINTR) continue;
+      logger.debug() << "Recv error:" << strerror(errno);
+      return;
+    }
+
+    QByteArray packet = m_session->mhopUnwrap(dgram.first(len));
+    const uint8_t* ptr = reinterpret_cast<const uint8_t*>(packet.constData());
+    uint8_t* dec = reinterpret_cast<uint8_t*>(decrypt.data());
+    auto res = wireguard_read(m_session->m_tunnel, ptr, packet.size(), dec,
+                              decrypt.size());
+    m_session->processResult(res.op, decrypt.first(res.size));
+  }
 }