source/net/resolve.cpp

   1 #include "platform_api.h"
   2 #include <msp/core/systemerror.h>
   3 #include <msp/strings/format.h>
   4 #include "sockaddr_private.h"
   5 #include "socket.h"
   6 #include "resolve.h"
   7
   8 using namespace std;
   9
  10 namespace {
  11
  12 void parse_host_serv(const string &str, string &host, string &serv)
  13 {
  14         if(str[0]=='[')
  15         {
  16                 string::size_type bracket = str.find(']');
  17                 host = str.substr(1, bracket-1);
  18                 string::size_type colon = str.find(':', bracket);
  19                 if(colon!=string::npos)
  20                         serv = str.substr(colon+1);
  21         }
  22         else
  23         {
  24                 string::size_type colon = str.find(':');
  25                 if(colon!=string::npos)
  26                 {
  27                         host = str.substr(0, colon);
  28                         serv = str.substr(colon+1);
  29                 }
  30                 else
  31                         host = str;
  32         }
  33 }
  34
  35 }
  36
  37
  38 namespace Msp {
  39 namespace Net {
  40
  41 SockAddr *resolve(const string &host, const string &serv, Family family)
  42 {
  43         const char *chost = (host.empty() ? 0 : host.c_str());
  44         const char *cserv = (serv.empty() ? 0 : serv.c_str());
  45         int flags = 0;
  46         if(host=="*")
  47         {
  48                 flags = AI_PASSIVE;
  49                 chost = 0;
  50         }
  51
  52         addrinfo hints = { flags, family_to_sys(family), 0, 0, 0, 0, 0, 0 };
  53         addrinfo *res;
  54
  55         int err = getaddrinfo(chost, cserv, &hints, &res);
  56         if(err==0)
  57         {
  58                 SockAddr::SysAddr sa;
  59                 sa.size = res->ai_addrlen;
  60                 const char *sptr = reinterpret_cast<const char *>(res->ai_addr);
  61                 char *dptr = reinterpret_cast<char *>(&sa.addr);
  62                 copy(sptr, sptr+res->ai_addrlen, dptr);
  63                 SockAddr *addr = SockAddr::new_from_sys(sa);
  64                 freeaddrinfo(res);
  65                 return addr;
  66         }
  67         else
  68 #ifdef _WIN32
  69                 throw system_error("getaddrinfo", WSAGetLastError());
  70 #else
  71                 throw system_error("getaddrinfo", gai_strerror(err));
  72 #endif
  73 }
  74
  75 SockAddr *resolve(const string &str, Family family)
  76 {
  77         string host, serv;
  78         parse_host_serv(str, host, serv);
  79
  80         return resolve(host, serv, family);
  81 }
  82
  83
  84 Resolver::Resolver():
  85         event_disp(0),
  86         next_tag(1)
  87 {
  88         thread.get_notify_pipe().signal_data_available.connect(sigc::mem_fun(this, &Resolver::task_done));
  89 }
  90
  91 void Resolver::use_event_dispatcher(IO::EventDispatcher *ed)
  92 {
  93         if(event_disp)
  94                 event_disp->remove(thread.get_notify_pipe());
  95         event_disp = ed;
  96         if(event_disp)
  97                 event_disp->add(thread.get_notify_pipe());
  98 }
  99
 100 unsigned Resolver::resolve(const string &host, const string &serv, Family family)
 101 {
 102         Task task;
 103         task.tag = next_tag++;
 104         task.host = host;
 105         task.serv = serv;
 106         task.family = family;
 107         thread.add_task(task);
 108         return task.tag;
 109 }
 110
 111 unsigned Resolver::resolve(const string &str, Family family)
 112 {
 113         string host, serv;
 114         parse_host_serv(str, host, serv);
 115
 116         return resolve(host, serv, family);
 117 }
 118
 119 void Resolver::tick()
 120 {
 121         if(IO::poll(thread.get_notify_pipe(), IO::P_INPUT, Time::zero))
 122                 task_done();
 123 }
 124
 125 void Resolver::task_done()
 126 {
 127         char buf[64];
 128         thread.get_notify_pipe().read(buf, sizeof(buf));
 129
 130         while(Task *task = thread.get_complete_task())
 131         {
 132                 if(task->addr)
 133                         signal_address_resolved.emit(task->tag, *task->addr);
 134                 else if(task->error)
 135                 {
 136                         if(signal_resolve_failed.empty())
 137                         {
 138                                 RefPtr<runtime_error> err = task->error;
 139                                 task->error = 0;
 140                                 thread.pop_complete_task();
 141                                 throw *err;
 142                         }
 143                         signal_resolve_failed.emit(task->tag, *task->error);
 144                 }
 145                 thread.pop_complete_task();
 146         }
 147 }
 148
 149
 150 Resolver::Task::Task():
 151         tag(0),
 152         family(UNSPEC),
 153         addr(0),
 154         error(0)
 155 { }
 156
 157
 158 Resolver::WorkerThread::WorkerThread():
 159         Thread("Resolver"),
 160         sem(1),
 161         done(false)
 162 {
 163         launch();
 164 }
 165
 166 Resolver::WorkerThread::~WorkerThread()
 167 {
 168         done = true;
 169         sem.signal();
 170         join();
 171 }
 172
 173 void Resolver::WorkerThread::add_task(const Task &t)
 174 {
 175         MutexLock lock(queue_mutex);
 176         bool was_starved = (queue.empty() || queue.back().is_complete());
 177         queue.push_back(t);
 178         if(was_starved)
 179                 sem.signal();
 180 }
 181
 182 Resolver::Task *Resolver::WorkerThread::get_complete_task()
 183 {
 184         MutexLock lock(queue_mutex);
 185         if(!queue.empty() && queue.front().is_complete())
 186                 return &queue.front();
 187         else
 188                 return 0;
 189 }
 190
 191 void Resolver::WorkerThread::pop_complete_task()
 192 {
 193         MutexLock lock(queue_mutex);
 194         if(!queue.empty() && queue.front().is_complete())
 195         {
 196                 delete queue.front().addr;
 197                 delete queue.front().error;
 198                 queue.pop_front();
 199         }
 200 }
 201
 202 void Resolver::WorkerThread::main()
 203 {
 204         bool wait = true;
 205         while(!done)
 206         {
 207                 if(wait)
 208                         sem.wait();
 209                 wait = false;
 210
 211                 Task *task = 0;
 212                 {
 213                         MutexLock lock(queue_mutex);
 214                         for(auto i=queue.begin(); (!task && i!=queue.end()); ++i)
 215                                 if(!i->is_complete())
 216                                         task = &*i;
 217                 }
 218
 219                 if(task)
 220                 {
 221                         try
 222                         {
 223                                 SockAddr *addr = Net::resolve(task->host, task->serv, task->family);
 224                                 {
 225                                         MutexLock lock(queue_mutex);
 226                                         task->addr = addr;
 227                                 }
 228                         }
 229                         catch(const runtime_error &e)
 230                         {
 231                                 MutexLock lock(queue_mutex);
 232                                 task->error = new runtime_error(e);
 233                         }
 234                         notify_pipe.put(1);
 235                 }
 236                 else
 237                         wait = true;
 238         }
 239 }
 240
 241 } // namespace Net
 242 } // namespace Msp