Scale Stack 0.3 Developer Documentation

scalestack/network/stream.cc

Go to the documentation of this file.

/*
 * Scale Stack
 *
 * Copyright 2010 Eric Day
 *
 * 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/stream.h>
#include <scalestack/network/stream_service.h>

using namespace std;

namespace scalestack
{
namespace network
{

/*
 * Public methods.
 */

stream_provider::stream_provider(event::handler_service& handler_service,
                                 stream_service& stream_service,
                                 int file_descriptor):
  event::handler(handler_service),
  _addresses(),
  _address(),
  _connecting(),
  _waiting_for_consume(),
  _call_flush_write(),
  _shutdown_write(),
  _connection_dead(),
  _file_descriptor(file_descriptor),
  _read_buffer_offset(),
  _read_buffer_size(),
  _write_buffer_offset(),
  _write_buffer_size(),
  _stream(stream_service.add_stream()),
  _stream_service(stream_service)
{
  _stream->_stream_provider = this;
}

stream_provider::stream_provider(event::handler_service& handler_service,
                                 stream_service& stream_service,
                                 const struct addrinfo* addresses):
  event::handler(handler_service),
  _addresses(addresses),
  _address(),
  _connecting(),
  _waiting_for_consume(),
  _call_flush_write(),
  _shutdown_write(),
  _connection_dead(),
  _file_descriptor(-1),
  _read_buffer_offset(),
  _read_buffer_size(),
  _write_buffer_offset(),
  _write_buffer_size(),
  _stream(stream_service.add_stream()),
  _stream_service(stream_service)
{
  _stream->_stream_provider = this;
}

stream_provider::~stream_provider()
{
}

void stream_provider::started(void)
{
  if (_addresses == NULL)
  {
    set_file_descriptor(_file_descriptor);
    watch_read();
    _stream->connected();
  }
  else
    reconnect();
}

void stream_provider::run(void)
{
  if (_shutdown_write && _write_buffer_size == 0)
  {
    int return_code = ::shutdown(_file_descriptor, SHUT_WR);
    if (return_code == -1)
    {
      _module.log_error(_("Failed write shutdown: %s:%d"), strerror(errno),
                        errno);
      shutdown();
      return;
    }
  }

  _stream->run();
}

void stream_provider::set_file_descriptor(int file_descriptor)
{
  _file_descriptor = file_descriptor;
  event::handler::set_file_descriptor(file_descriptor);
}

void stream_provider::reconnect(void)
{
  set_file_descriptor(-1);
  _address = _addresses;
  _connect();
}

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

  _consume(size);
  watch_read();
}

size_t stream_provider::write(uint8_t* buffer, size_t size, bool flush)
{
  if (size == 0 || _connecting)
    return 0;

  size_t written;

  /* Clear this because a forced write could remove the need for flush. */
  _call_flush_write = false;

  if (_write_buffer_size == 0 && (flush || size >= _write_buffer_max_size))
    written = 0;
  else if (size <= (_write_buffer_max_size -
                    (_write_buffer_offset + _write_buffer_size)))
  {
    /* Everything fits in the write buffer, so copy it in there. */
    memcpy(_write_buffer + _write_buffer_offset + _write_buffer_size,
           buffer,
           size);
    _write_buffer_size += size;
    if (flush)
      _flush();

    return size;
  }
  else
  {
    /* Not everything could fit in the buffer. */
    if (size < _write_buffer_max_size)
    {
      /* Copy as much as we can in if the requested amount isn't too large. */
      written = _write_buffer_max_size -
                (_write_buffer_offset + _write_buffer_size);
      if (written > 0)
      {
        memcpy(_write_buffer + _write_buffer_offset + _write_buffer_size,
               buffer,
               written);
        _write_buffer_size += written;
      }
    }
    else
      written = 0;

    _flush();
  }

  if (_write_buffer_size == 0 &&
      (flush || (size - written) >= _write_buffer_max_size))
  {
    /*
     * There is nothing in the buffer and either a flush was requested or
     * the size is larger than the buffer, so just try writing directly.
     */
    written += _write(buffer + written, size - written);
    if (written == size)
      return size;
  }

  /* If the remaining is larger than the buffer, don't bother buffering it. */
  if ((size - written) >= _write_buffer_max_size)
  {
    _call_flush_write = true;
    return written;
  }

  size_t fill_size =
    _write_buffer_max_size - (_write_buffer_offset + _write_buffer_size);
  if (fill_size == 0)
  {
    _call_flush_write = true;
    return written;
  }

  if ((size - written) <= fill_size)
    fill_size = size - written;
  else
    _call_flush_write = true;

  /* Fill the buffer with what we can. */
  memcpy(_write_buffer + _write_buffer_offset + _write_buffer_size,
         buffer + written,
         fill_size);
  _write_buffer_size += fill_size;

  return written + fill_size;
}

void stream_provider::shutdown_write(void)
{
  _shutdown_write = true;
  run_now();
}

/*
 * Private methods.
 */

void stream_provider::read_ready(int)
{
  while (!_connecting && _read_buffer_size != _read_buffer_max_size)
  {
    size_t free_size =
      _read_buffer_max_size - (_read_buffer_offset + _read_buffer_size);
    if (free_size <= (_read_buffer_max_size / 4))
    {
      /*
       * We don't want to move any buffer space if the derived class could
       * not consume it all because it probably has active pointers into
       * this buffer.
       */
      if (_waiting_for_consume)
        return;

      memmove(_read_buffer,
              _read_buffer + _read_buffer_offset,
              _read_buffer_size);
      _read_buffer_offset = 0;
      free_size = _read_buffer_max_size - _read_buffer_size;
    }

    ssize_t return_size = ::read(
      _file_descriptor,
      _read_buffer + _read_buffer_offset + _read_buffer_size,
      free_size);
    if (return_size == 0)
    {
      _module.log_info(_("Received EOF on read"));
      _shutdown_write = false;

      if (_write_buffer_size > 0)
        _flush();

      if (_write_buffer_size > 0 || _call_flush_write)
      {
        _connection_dead = true;
        return;
      }

      _stream->read_eof();
      return;
    }
    else if (return_size == -1)
    {
      if (errno == EAGAIN)
      {
        watch_read();
        return;
      }
      else if (errno == EINTR)
        continue;

      _module.log_error(_("Received error on read: %s:%d"),
                        strerror(errno), errno);
      _shutdown_write = false;
      shutdown();
      return;
    }

    _module.log_debug(_("[%X] Read %d bytes"), this, return_size);

    _read_buffer_size += return_size;
    _consume(_stream->read(_read_buffer + _read_buffer_offset,
                           _read_buffer_size));
  }
}

void stream_provider::write_ready(int)
{
  if (_connecting)
  {
    _connecting = false;
    _connect();
    return;
  }

  if (_write_buffer_size != 0)
    _flush();

  if (_write_buffer_size == 0)
  {
    if (_call_flush_write)
    {
      _call_flush_write = false;
      _stream->flush_write();

      if (!_call_flush_write)
        watch_read();
    }

    if (_connection_dead && _write_buffer_size == 0 && !_call_flush_write)
      shutdown();
  }
}

void stream_provider::_connect(void)
{
  while (true)
  {
    if (_file_descriptor == -1)
    {
      int file_descriptor = ::socket(_address->ai_family, SOCK_STREAM, 0);

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

      _connecting = true;
      set_file_descriptor(file_descriptor);
    }

    int return_code = ::connect(_file_descriptor, _address->ai_addr,
                                _address->ai_addrlen);
    if (return_code == 0 || errno == EISCONN)
    {
      _module.log_info(_("Connected"));
      _connecting = false;
      watch_read();
      _stream->connected();
      return;
    }

    if (errno == EAGAIN || errno == EINTR)
      continue;

    if (errno == EINPROGRESS || errno == EALREADY)
    {
      _module.log_debug(_("Connection in progress"));
      _connecting = true;
      watch_write();
      return;
    }

    if (errno == ECONNREFUSED || errno == ENETUNREACH || errno == ETIMEDOUT)
    {
      _address = _address->ai_next;
      if (_address == NULL)
      {
        _module.log_error(_("Could not connect"));
        shutdown();
        return;
      }

      set_file_descriptor(-1);
      continue;
    }

    _module.log_error(_("Could not connect: %s:%d"), strerror(errno), errno);
    shutdown();
    return;
  }
}

void stream_provider::_consume(size_t size)
{
  if (size == 0)
  {
    _waiting_for_consume = true;
    return;
  }

  if (size > _read_buffer_size)
  {
    _module.log_error(_("Consumed more data than is in buffer: %u > %u"),
                      size, _read_buffer_size);
    shutdown();
  }
  else if (size == _read_buffer_size)
  {
    _read_buffer_size = 0;
    _read_buffer_offset = 0;
    _waiting_for_consume = false;
  }
  else
  {
    _read_buffer_size -= size;
    _read_buffer_offset += size;
    _waiting_for_consume = true;
  }
}

void stream_provider::_flush(void)
{
  size_t return_size = _write(_write_buffer + _write_buffer_offset,
                              _write_buffer_size);
  if (return_size == 0)
    return;

  if (return_size == _write_buffer_size)
  {
    _write_buffer_size = 0;
    _write_buffer_offset = 0;

    if (_shutdown_write)
      run_now();
  }
  else
  {
    _write_buffer_size -= return_size;
    _write_buffer_offset += return_size;

    if ((_write_buffer_offset + _write_buffer_size) >=
        (_write_buffer_max_size / 2))
    {
      memmove(_write_buffer,
              _write_buffer + _write_buffer_offset,
              _write_buffer_size);
      _write_buffer_offset = 0;
    }
  }
}

size_t stream_provider::_write(uint8_t* buffer, size_t size)
{
  while (true)
  {
    ssize_t return_size = ::write(_file_descriptor, buffer, size);
    if (return_size == 0)
    {
      _module.log_info(_("Received EOF on write"));
      _shutdown_write = false;
      shutdown();
      return 0;
    }
    else if (return_size == -1)
    {
      if (errno == EAGAIN)
      {
        watch_write();
        return 0;
      }
      else if (errno == EINTR)
        continue;

      _module.log_error(_("Received error on write: %s:%d"),
                        strerror(errno), errno);
      _shutdown_write = false;
      shutdown();
      return 0;
    }

    _module.log_debug(_("[%X] Wrote %d bytes"), this, return_size);

    if (static_cast<size_t>(return_size) != size)
      watch_write();

    return return_size;
  }
}

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

---