X-Git-Url: https://git.syn-net.org/debian/?a=blobdiff_plain;f=src%2FpacketSource.cpp;h=bd3590d2e9cbfe0ab3ba960285e8fade0ec07d97;hb=bd02a5f6be4815d46524d613f727199ca756c971;hp=0882de52f0952832bb64443e0987abffe9408f2f;hpb=bb834fe0ed7a38b724f49b944adb801634eb6194;p=anytun.git

diff --git a/src/packetSource.cpp b/src/packetSource.cpp
index 0882de5..bd3590d 100644
--- a/src/packetSource.cpp
+++ b/src/packetSource.cpp
@@ -11,7 +11,7 @@
  *  tunneling and relaying of packets of any protocol.
  *
  *
- *  Copyright (C) 2007-2009 Othmar Gsenger, Erwin Nindl, 
+ *  Copyright (C) 2007-2009 Othmar Gsenger, Erwin Nindl,
  *                          Christian Pointner <satp@wirdorange.org>
  *
  *  This file is part of Anytun.
@@ -32,6 +32,7 @@
 
 #include <boost/asio.hpp>
 #include <boost/bind.hpp>
+#include <boost/thread.hpp>
 
 #include "datatypes.h"
 #include "packetSource.h"
@@ -39,22 +40,78 @@
 #include "resolver.h"
 #include "options.h"
 #include "signalController.h"
+#include "anytunError.h"
 
 void PacketSource::waitUntilReady()
 {
   ready_sem_.down();
 }
 
-UDPPacketSource::UDPPacketSource(std::string localaddr, std::string port) : sock_(io_service_)
+UDPPacketSource::UDPPacketSource(std::string localaddr, std::string port)
 {
   gResolver.resolveUdp(localaddr, port, boost::bind(&UDPPacketSource::onResolve, this, _1), boost::bind(&UDPPacketSource::onError, this, _1), gOpt.getResolvAddrType());
 }
 
-void UDPPacketSource::onResolve(const boost::asio::ip::udp::endpoint& e)
+UDPPacketSource::~UDPPacketSource()
 {
-  cLog.msg(Log::PRIO_NOTICE) << "opening socket: " << e;
-  sock_.open(e.protocol());
-  sock_.bind(e);
+  std::list<SocketsElement>::iterator it = sockets_.begin();
+  for(; it != sockets_.end(); ++it) {
+    /// this might be a needed by the receiver thread, TODO cleanup
+    //    delete[](it->buf_);
+    //    delete(it->sem_);
+    //    delete(it->sock_);
+  }
+}
+
+void UDPPacketSource::onResolve(PacketSourceResolverIt& it)
+{
+  while(it != PacketSourceResolverIt()) {
+    PacketSourceEndpoint e = *it;
+    cLog.msg(Log::PRIO_NOTICE) << "opening socket: " << e;
+
+    SocketsElement sock;
+    sock.buf_ = NULL;
+    sock.len_ = 0;
+    sock.sem_ = NULL;
+    sock.sock_ = new proto::socket(io_service_);
+    if(!sock.sock_) {
+      AnytunError::throwErr() << "memory error";
+    }
+
+    sock.sock_->open(e.protocol());
+#ifndef _MSC_VER
+    if(e.protocol() == proto::v6()) {
+      sock.sock_->set_option(boost::asio::ip::v6_only(true));
+    }
+#endif
+    sock.sock_->bind(e);
+    sockets_.push_back(sock);
+
+    it++;
+  }
+
+  // prepare multi-socket recv
+  if(sockets_.size() > 1) {
+    std::list<SocketsElement>::iterator it = sockets_.begin();
+    for(; it != sockets_.end(); ++it) {
+      it->len_ = MAX_PACKET_LENGTH;
+      it->buf_ = new uint8_t[it->len_];
+      if(!it->buf_) {
+        AnytunError::throwErr() << "memory error";
+      }
+
+      it->sem_ = new Semaphore();
+      if(!it->sem_) {
+        delete[](it->buf_);
+        AnytunError::throwErr() << "memory error";
+      }
+
+      boost::thread(boost::bind(&UDPPacketSource::recv_thread, this, it));
+      it->sem_->up();
+    }
+
+  }
+
   ready_sem_.up();
 }
 
@@ -63,13 +120,53 @@ void UDPPacketSource::onError(const std::runtime_error& e)
   gSignalController.inject(SIGERROR, e.what());
 }
 
-u_int32_t UDPPacketSource::recv(u_int8_t* buf, u_int32_t len, PacketSourceEndpoint& remote)
+void UDPPacketSource::recv_thread(std::list<SocketsElement>::iterator it)
+{
+  cLog.msg(Log::PRIO_INFO) << "started receiver thread for " << it->sock_->local_endpoint();
+
+  ThreadResult result;
+  result.it_ = it;
+  for(;;) {
+    it->sem_->down();
+    result.len_ = static_cast<uint32_t>(it->sock_->receive_from(boost::asio::buffer(it->buf_, it->len_), result.remote_));
+    {
+      Lock lock(thread_result_mutex_);
+      thread_result_queue_.push(result);
+    }
+    thread_result_sem_.up();
+  }
+}
+
+uint32_t UDPPacketSource::recv(uint8_t* buf, uint32_t len, PacketSourceEndpoint& remote)
 {
-  return static_cast<u_int32_t>(sock_.receive_from(boost::asio::buffer(buf, len), remote));
+  if(sockets_.size() == 1) {
+    return static_cast<uint32_t>(sockets_.front().sock_->receive_from(boost::asio::buffer(buf, len), remote));
+  }
+
+  thread_result_sem_.down();
+  ThreadResult result;
+  {
+    Lock lock(thread_result_mutex_);
+    result = thread_result_queue_.front();
+    thread_result_queue_.pop();
+  }
+  remote = result.remote_;
+  std::memcpy(buf, result.it_->buf_, (len < result.len_) ? len : result.len_);
+  len = (len < result.len_) ? len : result.len_;
+  result.it_->sem_->up();
+
+  return len;
 }
 
-void UDPPacketSource::send(u_int8_t* buf, u_int32_t len, PacketSourceEndpoint remote)
+void UDPPacketSource::send(uint8_t* buf, uint32_t len, PacketSourceEndpoint remote)
 {
-  sock_.send_to(boost::asio::buffer(buf, len), remote);
+  std::list<SocketsElement>::iterator it = sockets_.begin();
+  for(; it != sockets_.end(); ++it) {
+    if(it->sock_->local_endpoint().protocol() == remote.protocol()) {
+      it->sock_->send_to(boost::asio::buffer(buf, len), remote);
+      return;
+    }
+  }
+  cLog.msg(Log::PRIO_WARNING) << "no suitable socket found for remote endpoint protocol: " << remote;
 }