Scale Stack 0.3 Developer Documentation

scalestack/network/datagram.cc

Go to the documentation of this file.

/*
 * Scale Stack
 *
 * Copyright 2010 Chris Behrens
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "config.h"

#include <cerrno>
#include <cstring>
#include <netdb.h>
#include <sys/socket.h>
#include <sys/types.h>

#include <scalestack/kernel/module.h>
#include <scalestack/network/datagram.h>
#include <scalestack/network/datagram_service.h>

using namespace std;

namespace scalestack
{
namespace network
{

/*
 * Public methods.
 */

datagram_provider::datagram_provider(event::handler_service& handler_service,
                                     datagram_service& datagram_service,
                                     const struct addrinfo* local,
                                     const struct addrinfo* peer):
  event::handler(handler_service),
  _file_descriptor(-1),
  _peer_size(),
  _receive_buffer_size(),
  _datagram(datagram_service.add_datagram()),
  _datagram_service(datagram_service),
  _peer(reinterpret_cast<struct sockaddr*>(&_peer_storage)),
  _peer_storage()
{
  _datagram->_datagram_provider = this;

  int family = 0;

  if (local != NULL)
    family = local->ai_family;
  else if (peer != NULL)
    family = peer->ai_family;
  else
    _module.log_fatal(_("Must specify local or peer address for datagram"));

  int file_descriptor = socket(family, SOCK_DGRAM, 0);
  if (file_descriptor == -1)
    _module.log_fatal(_("Failed socket: %s:%d"), strerror(errno), errno);

  if (local != NULL)
  {
    int return_code = ::bind(file_descriptor, local->ai_addr,
                             local->ai_addrlen);
    if (return_code == -1)
    {
      close(file_descriptor);
      if (errno == EADDRINUSE)
      {
        _module.log_info(_("Bind failed, local address already in use"));
        throw exception();
      }

      _module.log_fatal(_("Failed bind: %s:%d"), strerror(errno), errno);
    }
  }

  if (peer != NULL)
  {
    memcpy(_peer, peer->ai_addr, peer->ai_addrlen);
    _peer_size = peer->ai_addrlen;
  }

  _file_descriptor = file_descriptor;
}

datagram_provider::~datagram_provider()
{
}

void datagram_provider::started(void)
{
  set_file_descriptor(_file_descriptor);
  _datagram->started();
}

void datagram_provider::set_file_descriptor(int file_descriptor)
{
  _file_descriptor = file_descriptor;
  event::handler::set_file_descriptor(file_descriptor);
  if (file_descriptor != -1)
    watch_read();
}

void datagram_provider::consume(size_t size)
{
  if (size == 0)
    return;

  if (size > _receive_buffer_size)
  {
    _module.log_error(_("Consumed more data than is in buffer: %u > %u"),
                      size, _receive_buffer_size);
    shutdown();
  }
  else if (size == _receive_buffer_size)
  {
    _receive_buffer_size = 0;
    watch_read();
  }
  else
    _receive_buffer_size -= size;
}

size_t datagram_provider::send(uint8_t* buffer,
                               size_t size,
                               const struct sockaddr& peer,
                               socklen_t peer_size)
{
  if (size == 0)
    return 0;

  ssize_t sent = ::sendto(_file_descriptor, buffer, size, 0, &peer, peer_size);
  if (sent > 0)
  {
    if (sent != static_cast<ssize_t>(size))
    {
      _module.log_error(_("Datagram sendto() only sent %d of %u"), sent, size);
      shutdown();
      return 0;
    }

    _module.log_debug(_("Sent %d bytes"), sent);
    watch_read();
    return sent;
  }

  if (sent == 0)
    return 0;

  if (errno == EAGAIN)
  {
    watch_write();
    return 0;
  }

  _module.log_error(_("Failed sendto: %s:%d"), strerror(errno), errno);
  shutdown();
  return 0;
}

size_t datagram_provider::send(uint8_t* buffer, size_t size)
{
  if (_peer_size == 0)
    _module.log_fatal(_("No peer address was set."));

  return send(buffer, size, *_peer, _peer_size);
}

/*
 * Private methods.
 */

void datagram_provider::read_ready(int)
{
  while (_receive_buffer_size == 0)
  {
    _peer_size = sizeof(_peer_storage);

    ssize_t received = ::recvfrom(_file_descriptor,
                                  _receive_buffer,
                                  _receive_buffer_max_size,
                                  0,
                                  _peer,
                                  &_peer_size);
    if (received == 0)
      return;
    else if (received == -1)
    {
      if (errno == EAGAIN)
      {
        watch_read();
        return;
      }
      else if (errno == EINTR)
        continue;

      _module.log_error(_("Failed recvfrom: %s:%d"), strerror(errno), errno);
      shutdown();
      return;
    }

    _module.log_debug(_("Received %d bytes"), received);
    _receive_buffer_size = received;
    consume(_datagram->receive(_receive_buffer, _receive_buffer_size,
                               *_peer, _peer_size));
  }
}

} /* namespace network */
} /* namespace scalestack */

---