controlsocket.cc

COPE the first practical network coding scheme which is developped on click

/*
 * controlsocket.{cc,hh} -- element listens to TCP/IP or Unix-domain sockets
 * Eddie Kohler
 *
 * Copyright (c) 2000 Massachusetts Institute of Technology
 * Copyright (c) 2001-3 International Computer Science Institute
 * Copyright (c) 2004 Regents of the University of California
 *
 * Permission is hereby granted, free of charge, to any person obtaining a
 * copy of this software and associated documentation files (the "Software"),
 * to deal in the Software without restriction, subject to the conditions
 * listed in the Click LICENSE file. These conditions include: you must
 * preserve this copyright notice, and you cannot mention the copyright
 * holders in advertising related to the Software without their permission.
 * The Software is provided WITHOUT ANY WARRANTY, EXPRESS OR IMPLIED. This
 * notice is a summary of the Click LICENSE file; the license in that file is
 * legally binding.
 */

#include <click/config.h>
#include "controlsocket.hh"
#include <click/confparse.hh>
#include <click/error.hh>
#include <click/timer.hh>
#include <click/router.hh>
#include <click/straccum.hh>
#include <click/llrpc.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <arpa/inet.h>
#include <fcntl.h>
CLICK_DECLS

const char ControlSocket::protocol_version[] = "1.1";

struct ControlSocketErrorHandler : public BaseErrorHandler { public:

  ControlSocketErrorHandler()		{ _error_code = ControlSocket::CSERR_OK; }

  const Vector<String> &messages() const { return _messages; }
  int error_code() const		{ return _error_code; }
  
  void handle_text(Seriousness, const String &);

  void set_error_code(int c)		{ _error_code = c; }
  
 private:

  Vector<String> _messages;
  int _error_code;

};

void
ControlSocketErrorHandler::handle_text(Seriousness, const String &m)
{
  const char *begin = m.begin();
  const char *end = m.end();
  while (begin < end) {
    const char *nl = find(begin, end, '\n');
    _messages.push_back(m.substring(begin, nl));
    begin = nl + 1;
  }
}


ControlSocket::ControlSocket()
  : _socket_fd(-1), _proxy(0), _full_proxy(0), _retry_timer(0)
{
}

ControlSocket::~ControlSocket()
{
}

int
ControlSocket::configure(Vector<String> &conf, ErrorHandler *errh)
{
  String socktype;
  if (cp_va_parse(conf, this, errh,
		  cpString, "type of socket ('TCP' or 'UNIX')", &socktype,
		  cpIgnoreRest, cpEnd) < 0)
    return -1;

  // remove keyword arguments
  bool read_only = false, verbose = false, retry_warnings = true;
  _retries = 0;
  if (cp_va_parse_remove_keywords(conf, 2, this, errh,
		"READONLY", cpBool, "read-only?", &read_only,
		"PROXY", cpElement, "handler proxy", &_proxy,
		"VERBOSE", cpBool, "be verbose?", &verbose,
		"RETRIES", cpInteger, "number of retries", &_retries,
		"RETRY_WARNINGS", cpBool, "warn on unsuccessful socket attempt?", &retry_warnings,
		cpEnd) < 0)
    return -1;
  _read_only = read_only;
  _verbose = verbose;
  _retry_warnings = retry_warnings;
  
  socktype = socktype.upper();
  if (socktype == "TCP") {
    _tcp_socket = true;
    unsigned short portno;
    if (cp_va_parse(conf, this, errh,
		    cpIgnore, cpUnsignedShort, "port number", &portno, cpEnd) < 0)
      return -1;
    _unix_pathname = String(portno);

  } else if (socktype == "UNIX") {
    _tcp_socket = false;
    if (cp_va_parse(conf, this, errh,
		    cpIgnore, cpString, "filename", &_unix_pathname, cpEnd) < 0)
      return -1;
    if (_unix_pathname.length() >= (int)sizeof(((struct sockaddr_un *)0)->sun_path))
      return errh->error("filename too long");

  } else
    return errh->error("unknown socket type '%s'", socktype.cc());
  
  return 0;
}


int
ControlSocket::initialize_socket_error(ErrorHandler *errh, const char *syscall)
{
  int e = errno;		// preserve errno

  if (_socket_fd >= 0) {
    close(_socket_fd);
    _socket_fd = -1;
  }

  if (_retries >= 0) {
    if (_retry_warnings)
      errh->warning("%s: %s (%d %s left)", syscall, strerror(e), _retries + 1, (_retries == 0 ? "try" : "tries"));
    return -EINVAL;
  } else
    return errh->error("%s: %s", syscall, strerror(e));
}

int
ControlSocket::initialize_socket(ErrorHandler *errh)
{
  _retries--;

  // open socket, set options, bind to address
  if (_tcp_socket) {
    _socket_fd = socket(PF_INET, SOCK_STREAM, 0);
    if (_socket_fd < 0)
      return initialize_socket_error(errh, "socket");
    int sockopt = 1;
    if (setsockopt(_socket_fd, SOL_SOCKET, SO_REUSEADDR, (void *)&sockopt, sizeof(sockopt)) < 0)
      errh->warning("setsockopt: %s", strerror(errno));

    // bind to port
    int portno;
    (void) cp_integer(_unix_pathname, &portno);
    struct sockaddr_in sa;
    sa.sin_family = AF_INET;
    sa.sin_port = htons(portno);
    sa.sin_addr = inet_makeaddr(0, 0);
    if (bind(_socket_fd, (struct sockaddr *)&sa, sizeof(sa)) < 0)
      return initialize_socket_error(errh, "bind");

  } else {
    _socket_fd = socket(PF_UNIX, SOCK_STREAM, 0);
    if (_socket_fd < 0)
      return initialize_socket_error(errh, "socket");

    // bind to port
    struct sockaddr_un sa;
    sa.sun_family = AF_UNIX;
    memcpy(sa.sun_path, _unix_pathname.cc(), _unix_pathname.length() + 1);
    if (bind(_socket_fd, (struct sockaddr *)&sa, sizeof(sa)) < 0)
      return initialize_socket_error(errh, "bind");
  }

  // start listening
  if (listen(_socket_fd, 2) < 0)
    return initialize_socket_error(errh, "listen");
  
  // nonblocking I/O and close-on-exec for the socket
  fcntl(_socket_fd, F_SETFL, O_NONBLOCK);
  fcntl(_socket_fd, F_SETFD, FD_CLOEXEC);

  add_select(_socket_fd, SELECT_READ);
  return 0;
}

void
ControlSocket::retry_hook(Timer *t, void *thunk)
{
  ControlSocket *cs = (ControlSocket *)thunk;
  if (cs->_socket_fd >= 0)
    /* nada */;
  else if (cs->initialize_socket(ErrorHandler::default_handler()) >= 0)
    /* nada */;
  else if (cs->_retries >= 0)
    t->reschedule_after_s(1);
  else
    cs->router()->please_stop_driver();
}

int
ControlSocket::initialize(ErrorHandler *errh)
{
  // check for a full proxy
  if (_proxy)
    _full_proxy = static_cast<HandlerProxy *>(_proxy->cast("HandlerProxy"));
  
  // ask the proxy to send us errors
  if (_full_proxy)
    _full_proxy->add_error_receiver(proxy_error_function, this);

  if (initialize_socket(errh) >= 0)
    return 0;
  else if (_retries >= 0) {
    _retry_timer = new Timer(retry_hook, this);
    _retry_timer->initialize(this);
    _retry_timer->schedule_after_s(1);
    return 0;
  } else
    return -1;
}

void
ControlSocket::take_state(Element *e, ErrorHandler *errh)
{
  ControlSocket *cs = (ControlSocket *)e->cast("ControlSocket");
  if (!cs)
    return;

  if (_socket_fd >= 0) {
    errh->error("already initialized, can't take state");
    return;
  } else if (_tcp_socket != cs->_tcp_socket
	     || _unix_pathname != cs->_unix_pathname) {
    errh->error("incompatible ControlSockets");
    return;
  }

  _socket_fd = cs->_socket_fd;
  cs->_socket_fd = -1;
  _in_texts.swap(cs->_in_texts);
  _out_texts.swap(cs->_out_texts);
  _flags.swap(cs->_flags);

  if (_socket_fd >= 0)
    add_select(_socket_fd, SELECT_READ);
  for (int i = 0; i < _flags.size(); i++)
    if (_flags[i] >= 0)
      add_select(i, SELECT_READ | SELECT_WRITE);
}

void
ControlSocket::cleanup(CleanupStage)
{
  if (_full_proxy)
    _full_proxy->remove_error_receiver(proxy_error_function, this);
  if (_socket_fd >= 0) {
    // shut down the listening socket in case we forked
#ifdef SHUT_RDWR
    shutdown(_socket_fd, SHUT_RDWR);
#else
    shutdown(_socket_fd, 2);
#endif
    close(_socket_fd);
    if (!_tcp_socket)
      unlink(_unix_pathname.c_str());
    _socket_fd = -1;
  }
  for (int i = 0; i < _flags.size(); i++)
    if (_flags[i] >= 0) {
      flush_write(i, false);	// try one last time to emit all data
      close(i);
      _flags[i] = -1;
    }
  if (_retry_timer) {
    _retry_timer->cleanup();
    delete _retry_timer;
    _retry_timer = 0;
  }
}

int
ControlSocket::message(int fd, int code, const String &s, bool continuation)
{
  assert(code >= 100 && code <= 999);
  if (fd >= 0 && !(_flags[fd] & WRITE_CLOSED))
    _out_texts[fd] += String(code) + (continuation ? "-" : " ") + s.printable() + "\r\n";
  return ANY_ERR;
}

int
ControlSocket::transfer_messages(int fd, int default_code, const String &msg,
				 ControlSocketErrorHandler *errh)
{
  int code = errh->error_code();
  if (code == CSERR_OK)
    code = default_code;
  const Vector<String> &messages = errh->messages();
  
  if (msg) {
    if (messages.size() > 0)
      message(fd, code, msg + ":", true);
    else
      message(fd, code, msg, false);
  }
  
  for (int i = 0; i < messages.size(); i++)
    message(fd, code, messages[i], i < messages.size() - 1);
  
  return ANY_ERR;
}

static String
canonical_handler_name(const String &n)
{
  const char *dot = find(n, '.');
  if (dot == n.begin() || (dot == n.begin() + 1 && n.front() == '0'))
    return n.substring(dot + 1, n.end());
  else
    return n;
}

String
ControlSocket::proxied_handler_name(const String &n) const
{
  if (_full_proxy && find(n, '.') == n.end())
    return "0." + n;
  else
    return n;
}

const Handler*
ControlSocket::parse_handler(int fd, const String &full_name, Element **es)
{
  // Parse full_name into element_name and handler_name.
  String canonical_name = canonical_handler_name(full_name);

  // Check for proxy.
  if (_proxy) {
    // collect errors from proxy
    ControlSocketErrorHandler errh;
    _proxied_handler = proxied_handler_name(canonical_name);
    _proxied_errh = &errh;
    
    const Handler* h = Router::handler(_proxy, _proxied_handler);
    
    if (errh.nerrors() > 0) {
      transfer_messages(fd, CSERR_NO_SUCH_HANDLER, String(), &errh);
      return 0;
    } else if (!h) {
      message(fd, CSERR_NO_SUCH_HANDLER, "No proxied handler named '" + full_name + "'");
      return 0;
    } else {
      *es = _proxy;
      return h;
    }
  }

  // Otherwise, find element.
  Element *e;
  const char *dot = find(canonical_name, '.');
  String hname;
  
  if (dot != canonical_name.end()) {
    String ename = canonical_name.substring(canonical_name.begin(), dot);
    e = router()->find(ename);
    if (!e) {
      int num;
      if (cp_integer(ename, &num) && num > 0 && num <= router()->nelements())
	e = router()->element(num - 1);
    }
    if (!e) {
      message(fd, CSERR_NO_SUCH_ELEMENT, "No element named '" + ename + "'");
      return 0;
    }
    hname = canonical_name.substring(dot + 1, canonical_name.end());
  } else {
    e = router()->root_element();
    hname = canonical_name;
  }

  // Then find handler.
  const Handler* h = Router::handler(e, hname);
  if (h && h->visible()) {
    *es = e;
    return h;
  } else {
    message(fd, CSERR_NO_SUCH_HANDLER, "No handler named '" + full_name + "'");
    return 0;
  }
}

int
ControlSocket::read_command(int fd, const String &handlername, const String &param)