Multicast DNS implementation (initial)

net/dns/mdns_client_impl.cc

+// Copyright (c) 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "net/dns/mdns_client_impl.h"
+
+#include "base/bind.h"
+#include "base/message_loop_proxy.h"
+#include "base/stl_util.h"
+#include "base/time/default_clock.h"
+#include "net/base/dns_util.h"
+#include "net/base/net_errors.h"
+#include "net/base/net_log.h"
+#include "net/base/rand_callback.h"
+#include "net/dns/dns_protocol.h"
+#include "net/dns/mdns_query.h"
+#include "net/udp/datagram_socket.h"
+
+namespace net {
+
+static const char kMDNSMulticastGroupIPv4[] = "224.0.0.251";
+static const char kMDNSMulticastGroupIPv6[] = "FF02::FB";
+
+static const unsigned kMDnsTransactionNoResultsSeconds = 3;
+
+MDnsClientImpl::Core::Core(MDnsClientImpl* client,
+                           MDnsConnectionFactory* connection_factory)
+    : client_(client), connection_(connection_factory->CreateConnection(this)) {
+}
+
+MDnsClientImpl::Core::~Core() {
+  cleanup_callback_.Cancel();
+  STLDeleteValues(&listeners_);
+}
+
+bool MDnsClientImpl::Core::Init() {
+  return connection_->Init();
+}
+
+bool MDnsClientImpl::Core::SendQuery(uint16 rrtype, std::string name) {
+  std::string name_dns;
+  if (!DNSDomainFromDot(name, &name_dns))
+    return false;
+
+  MDnsQuery query(name_dns, rrtype);
+
+  connection_->Send(query.io_buffer(), query.size());
+
+  return true;
+}
+
+void MDnsClientImpl::Core::HandlePacket(DnsResponse* response,
+                                        int bytes_read) {
+  unsigned offset;
+
+  if (!response->InitParseWithoutQuery(bytes_read)) {
+    LOG(WARNING) << "Could not understand an mDNS packet.";
+    return;  // Message is unreadable.
+  }
+
+  // TODO(noamsml): duplicate query suppression.
+  if (!(response->flags() & dns_protocol::kFlagResponse)) {
+    return;  // Message is a query. ignore it.
+  }
+  DnsRecordParser parser = response->Parser();
+  unsigned answer_count = response->answer_count() +
+      response->additional_answer_count();
+
+  for (unsigned i = 0; i < answer_count; i++) {
+    offset = parser.GetOffset();
+    scoped_ptr<const RecordParsed> scoped_record = RecordParsed::CreateFrom(
+        &parser, base::Time::Now());
+
+    if (!scoped_record) {
+      LOG(WARNING) << "Could not understand an mDNS record.";
+
+      if (offset == parser.GetOffset()) {
+        LOG(WARNING) << "Abandoned parsing the rest of the packet.";
+        return;  // The parser did not advance, abort reading the packet.
+      } else {
+        continue;  // We may be able to extract other records from the packet.
+      }
+    }
+
+    if ((scoped_record->klass() & dns_protocol::kMDnsClassMask) !=
+        dns_protocol::kClassIN) {
+      LOG(WARNING) << "Received an mDNS record with non-IN class. Ignoring.";
+      continue;  // Ignore all records not in the IN class.
+    }
+
+    // We want to retain a copy of the record pointer for updating listeners
+    // but we are passing ownership to the cache.
+    const RecordParsed* record = scoped_record.get();
+    MDnsCache::UpdateType update = cache_.UpdateDnsRecord(scoped_record.Pass());
+
+    // Cleanup time may have changed.
+    ScheduleCleanup(cache_.next_expiration());
+
+    if (update != MDnsCache::NoChange) {
+      MDnsUpdateType update_external;
+
+      switch (update) {
+        case MDnsCache::RecordAdded:
+          update_external = kMDnsRecordAdded;
+          break;
+        case MDnsCache::RecordChanged:
+          update_external = kMDnsRecordChanged;
+          break;
+        case MDnsCache::NoChange:
+          NOTREACHED();
+          // Dummy assignment to suppress compiler warning.
+          update_external = kMDnsRecordChanged;
+          break;
+      }
+
+      AlertListeners(update_external,
+                     ListenerKey(record->type(), record->name()), record);
+      // Alert listeners listening only for rrtype and not for name.
+      AlertListeners(update_external, ListenerKey(record->type(), ""), record);
+    }
+  }
+}
+
+void MDnsClientImpl::Core::AlertListeners(
+    MDnsUpdateType update_type,
+    const ListenerKey& key,
+    const RecordParsed* record) {
+  ListenerMap::iterator listener_map_iterator = listeners_.find(key);
+  if (listener_map_iterator == listeners_.end()) return;
+
+  FOR_EACH_OBSERVER(MDnsListenerImpl, *listener_map_iterator->second,
+                    AlertDelegate(update_type, record));
+}
+
+void MDnsClientImpl::Core::AddListener(
+    MDnsListenerImpl* listener) {
+  ListenerKey key(listener->GetType(), listener->GetName());
+  std::pair<ListenerMap::iterator, bool> observer_insert_result =
+      listeners_.insert(
+          make_pair(key, static_cast<ObserverList<MDnsListenerImpl>*>(NULL)));
+
+  // If an equivalent key does not exist, actually create the observer list.
+  if (observer_insert_result.second) {
+    observer_insert_result.first->second = new ObserverList<MDnsListenerImpl>();
+  }
+
+  ObserverList<MDnsListenerImpl>* observer_list =
+      observer_insert_result.first->second;
+
+  observer_list->AddObserver(listener);
+}
+
+void MDnsClientImpl::Core::RemoveListener(MDnsListenerImpl* listener) {
+  ListenerKey key(listener->GetType(), listener->GetName());
+  ListenerMap::iterator observer_list_iterator = listeners_.find(key);
+
+  DCHECK(observer_list_iterator != listeners_.end());
+  DCHECK(observer_list_iterator->second->HasObserver(listener));
+
+  observer_list_iterator->second->RemoveObserver(listener);
+
+  // Remove the observer list from the map if it is empty
+  if (observer_list_iterator->second->size() == 0) {
+    delete observer_list_iterator->second;
+    listeners_.erase(observer_list_iterator);
+  }
+}
+
+void MDnsClientImpl::Core::ScheduleCleanup(base::Time cleanup) {
+  // Cleanup is already scheduled, no need to do anything.
+  if (cleanup == scheduled_cleanup_) return;
+  scheduled_cleanup_ = cleanup;
+
+  // This line has the effect of cancelling the previously scheduled cleanup.
+  cleanup_callback_.Reset(base::Bind(
+      &MDnsClientImpl::Core::DoCleanup, base::Unretained(this)));
+
+  // cleanup == base::Time means no cleanup necessary.
+  if (cleanup != base::Time()) {
+    base::MessageLoop::current()->PostDelayedTask(
+        FROM_HERE,
+        cleanup_callback_.callback(),
+        cleanup - base::Time::Now());
+  }
+}
+
+void MDnsClientImpl::Core::DoCleanup() {
+  cache_.CleanupRecords(base::Time::Now(), base::Bind(
+      &MDnsClientImpl::Core::OnRecordRemoved, base::Unretained(this)));
+
+  ScheduleCleanup(cache_.next_expiration());
+}
+
+void MDnsClientImpl::Core::OnRecordRemoved(
+    const RecordParsed* record) {
+  AlertListeners(kMDnsRecordRemoved,
+                 ListenerKey(record->type(), record->name()), record);
+  // Alert listeners listening only for rrtype and not for name.
+  AlertListeners(kMDnsRecordRemoved, ListenerKey(record->type(), ""),
+                 record);
+}
+
+void MDnsClientImpl::Core::QueryCache(
+    uint16 rrtype, const std::string& name,
+    std::vector<const RecordParsed*>* records) const {
+  cache_.FindDnsRecords(rrtype, name, records, base::Time::Now());
+}
+
+MDnsClientImpl::MDnsClientImpl()
+    : listen_refs_(0), connection_factory_owned_(
+          new MDnsConnectionImplFactory()) {
+  connection_factory_ = connection_factory_owned_.get();
+}
+
+MDnsClientImpl::MDnsClientImpl(MDnsConnectionFactory* connection_factory)
+    : listen_refs_(0), connection_factory_(connection_factory) {
+}
+
+MDnsClientImpl::~MDnsClientImpl() {
+}
+
+bool MDnsClientImpl::AddListenRef() {
+  if (!core_.get()) {
+    core_.reset(new Core(this, connection_factory_));
+    if (!core_->Init()) {
+      core_.reset();
+      return false;
+    }
+  }
+  listen_refs_++;
+  return true;
+}
+
+void MDnsClientImpl::SubtractListenRef() {
+  listen_refs_--;
+  if (listen_refs_ == 0) {
+    base::MessageLoop::current()->PostTask(FROM_HERE, base::Bind(
+        &MDnsClientImpl::Shutdown, base::Unretained(this)));
+  }
+}
+
+void MDnsClientImpl::Shutdown() {
+  // We need to check that new listeners haven't been created.
+  if (listen_refs_ == 0) {
+    core_.reset();
+  }
+}
+
+bool MDnsClientImpl::IsListeningForTests() {
+  return core_.get() != NULL;
+}
+
+scoped_ptr<MDnsListener> MDnsClientImpl::CreateListener(
+    uint16 rrtype,
+    const std::string& name,
+    MDnsListener::Delegate* delegate) {
+  return scoped_ptr<net::MDnsListener>(
+      new MDnsListenerImpl(rrtype, name, delegate, this));
+}
+
+scoped_ptr<MDnsTransaction> MDnsClientImpl::CreateTransaction(
+    uint16 rrtype,
+    const std::string& name,
+    int flags,
+    const MDnsTransaction::ResultCallback& callback) {
+  return scoped_ptr<MDnsTransaction>(
+      new MDnsTransactionImpl(rrtype, name, flags, callback, this));
+}
+
+MDnsListenerImpl::MDnsListenerImpl(
+    uint16 rrtype,
+    const std::string& name,
+    MDnsListener::Delegate* delegate,
+    MDnsClientImpl* client)
+    : rrtype_(rrtype), name_(name), client_(client), delegate_(delegate),
+      started_(false) {
+}
+
+bool MDnsListenerImpl::Start() {
+  DCHECK(!started_);
+
+  if (!client_->AddListenRef()) return false;
+  started_ = true;
+
+  client_->core()->AddListener(this);
+
+  return true;
+}
+
+MDnsListenerImpl::~MDnsListenerImpl() {
+  if (started_) {
+    client_->core()->RemoveListener(this);
+    client_->SubtractListenRef();
+  }
+}
+
+const std::string& MDnsListenerImpl::GetName() const {
+  return name_;
+}
+
+uint16 MDnsListenerImpl::GetType() const {
+  return rrtype_;
+}
+
+void MDnsListenerImpl::AlertDelegate(MDnsUpdateType update_type,
+                                     const RecordParsed* record) {
+  DCHECK(started_);
+  delegate_->OnRecordUpdate(update_type, record);
+}
+
+MDnsTransactionImpl::MDnsTransactionImpl(
+    uint16 rrtype,
+    const std::string& name,
+    int flags,
+    const MDnsTransaction::ResultCallback& callback,
+    MDnsClientImpl* client)
+    : rrtype_(rrtype), name_(name), callback_(callback), client_(client),
+      started_(false), flags_(flags) {
+  DCHECK((flags_ & kMDnsTransactionFlagMask) == flags_);
+  DCHECK(flags_ & kMDnsTransactionQueryCache ||
+         flags_ & kMDnsTransactionQueryNetwork);
+}
+
+MDnsTransactionImpl::~MDnsTransactionImpl() {
+  timeout_.Cancel();
+}
+
+bool MDnsTransactionImpl::Start() {
+  DCHECK(!started_);
+  started_ = true;
+  std::vector<const RecordParsed*> records;
+  base::WeakPtr<MDnsTransactionImpl> weak_this = AsWeakPtr();
+
+  if (flags_ & kMDnsTransactionQueryCache) {
+    if (client_->core()) {
+      client_->core()->QueryCache(rrtype_, name_, &records);
+      for (std::vector<const RecordParsed*>::iterator i = records.begin();
+           i != records.end() && weak_this; ++i) {
+        weak_this->TriggerCallback(kMDnsTransactionRecord, records.front());
+      }
+    }
+  }
+
+  if (!weak_this) return true;
+
+  if (is_valid() && flags_ & kMDnsTransactionQueryNetwork) {
+    listener_ = client_->CreateListener(rrtype_, name_, this);
+    if (!listener_->Start()) return false;
+
+    DCHECK(client_->core());
+    if (!client_->core()->SendQuery(rrtype_, name_)) {
+      return false;
+    }
+
+    timeout_.Reset(base::Bind(&MDnsTransactionImpl::SignalTransactionOver,
+                              weak_this));
+    base::MessageLoop::current()->PostDelayedTask(
+        FROM_HERE,
+        timeout_.callback(),
+        base::TimeDelta::FromSeconds(kMDnsTransactionNoResultsSeconds));
+
+    return listener_.get() != NULL;
+  } else {
+    // If this is a cache only query, signal that the transaction is over
+    // immediately.
+    SignalTransactionOver();
+  }
+
+  return true;
+}
+
+const std::string& MDnsTransactionImpl::GetName() const {
+  return name_;
+}
+
+uint16 MDnsTransactionImpl::GetType() const {
+  return rrtype_;
+}
+
+void MDnsTransactionImpl::CacheRecordFound(const RecordParsed* record) {
+  DCHECK(started_);
+  OnRecordUpdate(kMDnsRecordAdded, record);
+}
+
+void MDnsTransactionImpl::TriggerCallback(MDnsTransactionResult result,
+                                          const RecordParsed* record) {
+  DCHECK(started_);
+  if (!is_valid()) return;
+
+  // Ensure callback is run after touching all class state, so that
+  // the callback can delete the transaction.
+  MDnsTransaction::ResultCallback callback = callback_;
+
+  if (flags_ & kMDnsTransactionSingleResult) {
+    Reset();
+  }
+
+  callback.Run(result, record);
+}
+
+void MDnsTransactionImpl::Reset() {
+  callback_.Reset();
+  listener_.reset();
+  timeout_.Cancel();
+}
+
+void MDnsTransactionImpl::OnRecordUpdate(MDnsUpdateType update,
+                                         const RecordParsed* record) {
+  DCHECK(started_);
+  if (update ==  kMDnsRecordAdded || update == kMDnsRecordChanged) {
+    TriggerCallback(kMDnsTransactionRecord, record);
+  }
+}
+
+void MDnsTransactionImpl::SignalTransactionOver() {
+  DCHECK(started_);
+  base::WeakPtr<MDnsTransactionImpl> weak_this = AsWeakPtr();
+
+  if (flags_ & kMDnsTransactionSingleResult) {
+    TriggerCallback(kMDnsTransactionNoResults, NULL);
+  } else {
+    TriggerCallback(kMDnsTransactionDone, NULL);
+  }
+
+  if (weak_this) {
+    weak_this->Reset();
+  }
+}
+
+void MDnsTransactionImpl::OnNsecRecord(const std::string& name, unsigned type) {
+  // TODO(noamsml): NSEC records not yet implemented
+}
+
+void MDnsTransactionImpl::OnCachePurged() {
+  // TODO(noamsml): Cache purge situations not yet implemented
+}
+
+MDnsConnectionImpl::RecvLoop::RecvLoop(
+    DatagramServerSocket* socket, MDnsConnectionImpl* connection)
+    : socket_(socket), connection_(connection),
+      response_(new DnsResponse(dns_protocol::kMaxMulticastSize)) {
+}
+
+MDnsConnectionImpl::RecvLoop::~RecvLoop() {
+}
+
+void MDnsConnectionImpl::RecvLoop::DoLoop(int rv) {
+  do {
+    if (rv > 0) {
+      connection_->OnDatagramReceived(socket_,
+                                      response_.get(), &recv_addr_, rv);
+    }
+
+    rv = socket_->RecvFrom(
+        response_->io_buffer(),
+        response_->io_buffer()->size(),
+        &recv_addr_,
+        base::Bind(&MDnsConnectionImpl::RecvLoop::OnDatagramReceived,
+                   base::Unretained(this)));
+  } while (rv > 0);
+
+  if (rv != ERR_IO_PENDING) {
+    connection_->OnError(this, socket_, rv);

[szym, 2013/06/07 16:40:02]

Ok, but be careful that this can be called inside Init().

Normally, you'd do return rv and in both Init and OnDatagramReceived.
if (DoLoop(rv) != OK)
  OnError(rv);

Right now your OnError does nothing.

[Noam Samuel, 2013/06/07 23:54:43]

OnError now tunnels up to client, still does nothing. Will add retry/flush logic
in subsequent CL.

On 2013/06/07 16:40:02, szym wrote:

+  }
+}
+
+void MDnsConnectionImpl::RecvLoop::OnDatagramReceived(int rv) {
+  DoLoop(rv);
+}
+
+MDnsConnectionImpl::MDnsConnectionImpl(MDnsConnection::Delegate* delegate)
+    : socket_ipv4_(new UDPServerSocket(NULL, NetLog::Source())),
+      socket_ipv6_(new UDPServerSocket(NULL, NetLog::Source())),
+      loop_ipv4_(socket_ipv4_.get(), this),
+      loop_ipv6_(socket_ipv6_.get(), this),
+  delegate_(delegate) {
+}
+
+MDnsConnectionImpl::~MDnsConnectionImpl() {
+  socket_ipv4_->Close();
+  socket_ipv6_->Close();

[szym, 2013/06/07 16:40:02]

No need for either.

[Noam Samuel, 2013/06/07 23:54:43]

Done.

+}
+
+bool MDnsConnectionImpl::Init() {
+  if (!BindSocket(socket_ipv4_.get(), kIPv4AddressSize,
+                  kMDNSMulticastGroupIPv4))
+    return false;
+
+  if (!BindSocket(socket_ipv6_.get(), kIPv6AddressSize,
+                  kMDNSMulticastGroupIPv6))
+    return false;
+
+  loop_ipv4_.DoLoop(0);
+  loop_ipv6_.DoLoop(0);
+
+  return true;
+}
+
+bool MDnsConnectionImpl::Send(IOBuffer* buffer, unsigned size) {
+  int rv = socket_ipv4_->SendTo(
+      buffer,
+      size,
+      GetIPv4SendEndpoint(),
+      base::Bind(&MDnsConnectionImpl::SendDone,
+                 base::Unretained(this) ));
+  if (rv < OK && rv != ERR_IO_PENDING) return false;
+
+  rv = socket_ipv6_->SendTo(
+      buffer,
+      size,
+      GetIPv6SendEndpoint(),
+      base::Bind(&MDnsConnectionImpl::SendDone,
+                 base::Unretained(this) ));
+  if (rv < OK && rv != ERR_IO_PENDING) return false;
+
+  return true;
+}
+
+void MDnsConnectionImpl::SendDone(int sent) {
+  // TODO(noamsml): Queueing and retry logic.
+}
+
+void MDnsConnectionImpl::OnError(RecvLoop* loop,
+                                 DatagramServerSocket* socket,
+                                 int error) {
+  // TODO(noamsml): Error handling.
+}
+
+bool MDnsConnectionImpl::BindSocket(
+    DatagramServerSocket* socket,
+    int addr_size,
+    const char* multicast_group) {
+  IPAddressNumber address_any;
+  address_any.resize(addr_size, 0);
+
+  IPAddressNumber multicast_group_number;
+
+  IPEndPoint bind_endpoint(address_any, dns_protocol::kDefaultPortMulticast);
+
+  bool success = ParseIPLiteralToNumber(multicast_group,
+                                        &multicast_group_number);
+  DCHECK(success);
+
+  socket->AllowAddressReuse();
+  int status = socket->Listen(bind_endpoint);

[szym, 2013/06/07 16:40:02]

nit: suggest rv or result instead of status.

[Noam Samuel, 2013/06/07 23:54:43]

Done.

+
+  if (status < 0)
+    return false;
+
+  socket->SetMulticastLoopbackMode(false);
+
+  status = socket->JoinGroup(multicast_group_number);
+
+  if (status < 0)
+    return false;
+
+  return true;
+}
+
+IPEndPoint MDnsConnectionImpl::GetIPv4SendEndpoint() {
+  IPAddressNumber multicast_group_number;
+  bool success = ParseIPLiteralToNumber(kMDNSMulticastGroupIPv4,
+                                        &multicast_group_number);

[szym, 2013/06/07 16:40:02]

You are parsing the constant on every sent packet. Could you instead initialize
both in the constructor of MDnsConnectionImpl and just pass IPEndPoint around?

You could also use it in BindSocket (IPEndPoint::address()).

[Noam Samuel, 2013/06/07 23:54:43]

Done.

+  DCHECK(success);
+  return IPEndPoint(multicast_group_number,
+                    dns_protocol::kDefaultPortMulticast);
+}
+
+IPEndPoint MDnsConnectionImpl::GetIPv6SendEndpoint() {
+  IPAddressNumber multicast_group_number;
+  bool success = ParseIPLiteralToNumber(kMDNSMulticastGroupIPv6,
+                                        &multicast_group_number);
+  DCHECK(success);
+  return IPEndPoint(multicast_group_number,
+                    dns_protocol::kDefaultPortMulticast);
+}
+
+void MDnsConnectionImpl::OnDatagramReceived(
+    DatagramServerSocket* socket,
+    DnsResponse* response,
+    IPEndPoint* recv_addr,
+    int bytes_read) {
+  // TODO(noamsml): More sophisticated error handling.
+  DCHECK_GT(bytes_read, 0);
+  delegate_->HandlePacket(response, bytes_read);
+}
+
+MDnsConnectionImplFactory::MDnsConnectionImplFactory() {
+}
+
+MDnsConnectionImplFactory::~MDnsConnectionImplFactory() {
+}
+
+scoped_ptr<MDnsConnection> MDnsConnectionImplFactory::CreateConnection(
+    MDnsConnection::Delegate* delegate) {
+  return scoped_ptr<MDnsConnection>(new MDnsConnectionImpl(delegate));
+}
+
+}  // namespace net