controlsocket.cc

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

{
  Element *e;
  const Handler* h = parse_handler(fd, handlername, &e);
  if (!h)
    return ANY_ERR;
  else if (!h->read_visible())
    return message(fd, CSERR_PERMISSION, "Handler '" + handlername + "' write-only");

  // collect errors from proxy
  ControlSocketErrorHandler errh;
  _proxied_handler = h->name();
  _proxied_errh = &errh;
  
  String data = h->call_read(e, param);

  // did we get an error message?
  if (errh.nerrors() > 0)
    return transfer_messages(fd, CSERR_UNSPECIFIED, String(), &errh);
  
  message(fd, CSERR_OK, "Read handler '" + handlername + "' OK");
  _out_texts[fd] += "DATA " + String(data.length()) + "\r\n";
  _out_texts[fd] += data;
  return 0;
}

int
ControlSocket::write_command(int fd, const String &handlername, const String &data)
{
  Element *e;
  const Handler* h = parse_handler(fd, handlername, &e);
  if (!h)
    return ANY_ERR;
  else if (!h->writable())
    return message(fd, CSERR_PERMISSION, "Handler '" + handlername + "' read-only");

  if (_read_only)
    return message(fd, CSERR_PERMISSION, "Permission denied for '" + handlername + "'");

#ifdef LARGEST_HANDLER_WRITE
  if (data.length() > LARGEST_HANDLER_WRITE)
    return message(fd, CSERR_DATA_TOO_BIG, "Data too large for write handler '" + handlername + "'");
#endif
  
  ControlSocketErrorHandler errh;
  
  // call handler
  int result = h->call_write(data, e, &errh);

  // add a generic error message for certain handler codes
  int code = errh.error_code();
  if (code == CSERR_OK) {
    if (errh.nerrors() > 0 || result < 0)
      code = CSERR_HANDLER_ERROR;
    else if (errh.nwarnings() > 0)
      code = CSERR_OK_HANDLER_WARNING;
  }

  String msg;
  if (code == CSERR_OK)
    msg = "Write handler '" + handlername + "' OK";
  else if (code == CSERR_OK_HANDLER_WARNING)
    msg = "Write handler '" + handlername + "' OK with warnings";
  else if (code == CSERR_HANDLER_ERROR)
    msg = "Write handler '" + handlername + "' error";
  transfer_messages(fd, code, msg, &errh);
  return 0;
}

int
ControlSocket::check_command(int fd, const String &hname, bool write)
{
  int ok = 0;
  int any_visible = 0;
  ControlSocketErrorHandler errh;

  if (_full_proxy) {
    String phname = canonical_handler_name(hname);
    if (find(phname, '.') == phname.end())
      phname = "0." + phname;
    ok = _full_proxy->check_handler(phname, write, &errh);
  } else {
    Element *e;
    const Handler* h = parse_handler(fd, hname, &e);
    if (!h)
      return 0;			// error messages already reported
    ok = (h->visible() && (write ? h->write_visible() : h->read_visible()));
    any_visible = h->visible();
  }

  // remember _read_only!
  if (write && _read_only && ok)
    return message(fd, CSERR_PERMISSION, "Permission denied for '" + hname + "'");
  else if (errh.messages().size() > 0)
    transfer_messages(fd, CSERR_OK, String(), &errh);
  else if (ok)
    message(fd, CSERR_OK, String(write ? "Write" : "Read") + " handler '" + hname + "' OK");
  else if (any_visible)
    message(fd, CSERR_NO_SUCH_HANDLER, "Handler '" + hname + (write ? "' not writable" : "' not readable"));
  else
    message(fd, CSERR_NO_SUCH_HANDLER, "No " + String(write ? "write" : "read") + " handler named '" + hname + "'");
  return 0;
}

int
ControlSocket::llrpc_command(int fd, const String &llrpcname, String data)
{
  const char *octothorp = find(llrpcname, '#');
  uint32_t command;
  if (!cp_unsigned(llrpcname.substring(octothorp + 1, llrpcname.end()), 16, &command))
    return message(fd, CSERR_SYNTAX, "Syntax error in LLRPC name '" + llrpcname + "'");
  // transform net LLRPC id into host LLRPC id
  command = CLICK_LLRPC_NTOH(command);
  
  Element *e;
  const Handler* h = parse_handler(fd, llrpcname.substring(llrpcname.begin(), octothorp) + ".name", &e);
  if (!h)
    return ANY_ERR;

  int size = _CLICK_IOC_SIZE(command);
  if (!size || !(command & (_CLICK_IOC_IN | _CLICK_IOC_OUT)) || !(command & _CLICK_IOC_FLAT))
    return message(fd, CSERR_UNIMPLEMENTED, "Cannot call LLRPC '" + llrpcname + "' remotely");

  if (_read_only)		// can't tell whether an LLRPC is read-only;
    				// so disallow them all
    return message(fd, CSERR_PERMISSION, "Permission denied for '" + llrpcname + "'");

  if ((command & _CLICK_IOC_IN) && data.length() != size)
    return message(fd, CSERR_LLRPC_ERROR, "LLRPC '" + llrpcname + "' requires " + String(size) + " bytes input data");
  else if (command & _CLICK_IOC_OUT)
    data = String::garbage_string(size);

  // collect errors from proxy
  ControlSocketErrorHandler errh;
  _proxied_handler = llrpcname;
  _proxied_errh = &errh;
  
  int retval;
  if (_proxy) {
    struct click_llrpc_proxy_st pst;
    pst.proxied_handler = (void*) h;
    pst.proxied_command = command;
    pst.proxied_data = data.mutable_data();
    retval = _proxy->llrpc(CLICK_LLRPC_PROXY, &pst);
  } else
    retval = e->llrpc(command, data.mutable_data());
  
  // did we get an error message?
  String msg;
  if (retval < 0)
    msg = "LLRPC '" + llrpcname + "' error: " + String(strerror(-retval));
  else if (errh.nerrors() > 0)
    msg = "LLRPC '" + llrpcname + "' error";
  else
    msg = "LLRPC '" + llrpcname + "' OK";
  int code = (retval < 0 || errh.nerrors() > 0 ? CSERR_LLRPC_ERROR : CSERR_OK);
  transfer_messages(fd, code, msg, &errh);

  if (code == CSERR_OK) {
    if (!(command & _CLICK_IOC_OUT))
      data = String();
    _out_texts[fd] += "DATA " + String(data.length()) + "\r\n";
    _out_texts[fd] += data;
  }
  return 0;
}

int
ControlSocket::parse_command(int fd, const String &line)
{
  // split 'line' into words; don't use cp_ functions since they strip comments
  Vector<String> words;
  const char *data = line.data();
  int len = line.length();
  for (int pos = 0; pos < len; ) {
      while (pos < len && isspace((unsigned char) data[pos]))
      pos++;
    int first = pos;
    while (pos < len && !isspace((unsigned char) data[pos]))
      pos++;
    if (first < pos)
      words.push_back(line.substring(first, pos - first));
  }
  if (words.size() == 0)
    return 0;

  // branch on command
  String command = words[0].upper();
  if (command == "READ" || command == "GET") {
      if (words.size() < 2)
	  return message(fd, CSERR_SYNTAX, "Wrong number of arguments");
      String param;
      if (words.size() > 2)
	  param = line.substring(words[2].begin(), words.back().end());
      return read_command(fd, words[1], param);
    
  } else if (command == "WRITE" || command == "SET") {
      if (words.size() < 2)
	  return message(fd, CSERR_SYNTAX, "Wrong number of arguments");
      String data;
      if (words.size() > 2)
	  data = line.substring(words[2].begin(), words.back().end());
      return write_command(fd, words[1], data);
    
  } else if (command == "WRITEDATA" || command == "SETDATA") {
    if (words.size() != 3)
      return message(fd, CSERR_SYNTAX, "Wrong number of arguments");
    int datalen;
    if (!cp_integer(words[2], &datalen) || datalen < 0)
      return message(fd, CSERR_SYNTAX, "Syntax error in 'writedata'");
    if (_in_texts[fd].length() < datalen) {
      if (_flags[fd] & READ_CLOSED)
	return message(fd, CSERR_SYNTAX, "Not enough data");
      else			// retry
	return 1;
    }
    String data = _in_texts[fd].substring(0, datalen);
    _in_texts[fd] = _in_texts[fd].substring(datalen);
    return write_command(fd, words[1], data);

  } else if (command == "CHECKREAD") {
    if (words.size() != 2)
      return message(fd, CSERR_SYNTAX, "Wrong number of arguments");
    return check_command(fd, words[1], false);
    
  } else if (command == "CHECKWRITE") {
    if (words.size() != 2)
      return message(fd, CSERR_SYNTAX, "Wrong number of arguments");
    return check_command(fd, words[1], true);

  } else if (command == "LLRPC") {
    if (words.size() != 2 && words.size() != 3)
      return message(fd, CSERR_SYNTAX, "Wrong number of arguments");
    int datalen = 0;
    if (words.size() == 3 && (!cp_integer(words[2], &datalen) || datalen < 0))
      return message(fd, CSERR_SYNTAX, "Syntax error in 'llrpc'");
    if (_in_texts[fd].length() < datalen) {
      if (_flags[fd] & READ_CLOSED)
	return message(fd, CSERR_SYNTAX, "Not enough data");
      else			// retry
	return 1;
    }
    String data = _in_texts[fd].substring(0, datalen);
    _in_texts[fd] = _in_texts[fd].substring(datalen);
    return llrpc_command(fd, words[1], data);

  } else if (command == "CLOSE" || command == "QUIT") {
    if (words.size() != 1)
      message(fd, CSERR_SYNTAX, "Bad command syntax");
    message(fd, CSERR_OK, "Goodbye!");
    _flags[fd] |= READ_CLOSED;
    _in_texts[fd] = String();
    return 0;
    
  } else
    return message(fd, CSERR_UNIMPLEMENTED, "Command '" + command + "' unimplemented");
}

void
ControlSocket::flush_write(int fd, bool read_needs_processing)
{
    assert(_flags[fd] >= 0);
    if (!(_flags[fd] & WRITE_CLOSED)) {
	int w = 0;
	while (_out_texts[fd].length()) {
	    const char *x = _out_texts[fd].data();
	    w = write(fd, x, _out_texts[fd].length());
	    if (w < 0 && errno != EINTR)
		break;
	    if (w > 0)
		_out_texts[fd] = _out_texts[fd].substring(w);
	}
	if (w < 0 && errno == EPIPE)
	    _flags[fd] |= WRITE_CLOSED;
	// don't select writes unless we have data to write (or read needs more
	// processing)
	if (_out_texts[fd].length() || read_needs_processing)
	    add_select(fd, SELECT_WRITE);
	else
	    remove_select(fd, SELECT_WRITE);
    }
}

void
ControlSocket::selected(int fd)
{
    if (fd == _socket_fd) {
	union { struct sockaddr_in in; struct sockaddr_un un; } sa;
#ifdef __APPLE__
	int sa_len;
#else
	socklen_t sa_len;
#endif
	sa_len = sizeof(sa);
	int new_fd = accept(_socket_fd, (struct sockaddr *)&sa, &sa_len);

	if (new_fd < 0) {
	    if (errno != EAGAIN)
		click_chatter("%s: accept: %s", declaration().cc(), strerror(errno));
	    return;
	}

	if (_verbose) {
	    if (_tcp_socket)
		click_chatter("%s: opened connection %d from %s.%d", declaration().cc(), new_fd, IPAddress(sa.in.sin_addr).unparse().cc(), ntohs(sa.in.sin_port));
	    else
		click_chatter("%s: opened connection %d", declaration().cc(), new_fd);
	}

	fcntl(new_fd, F_SETFL, O_NONBLOCK);
	fcntl(new_fd, F_SETFD, FD_CLOEXEC);
	add_select(new_fd, SELECT_READ | SELECT_WRITE);
	
	while (new_fd >= _in_texts.size()) {
	    _in_texts.push_back(String());
	    _out_texts.push_back(String());
	    _flags.push_back(-1);
	}
	_in_texts[new_fd] = String();
	_out_texts[new_fd] = String();
	_flags[new_fd] = 0;

	fd = new_fd;
	_out_texts[new_fd] = "Click::ControlSocket/" + String(protocol_version) + "\r\n";
    }

    // find file descriptor
    if (fd >= _in_texts.size() || _flags[fd] < 0)
	return;

    // read commands from socket (but only a bit on each select)
    if (!(_flags[fd] & READ_CLOSED)) {
	char buf[2048];
	int r = read(fd, buf, 2048);
	if (r > 0)
	    _in_texts[fd].append(buf, r);
	else if (r == 0 || (r < 0 && errno != EAGAIN && errno != EINTR))
	    _flags[fd] |= READ_CLOSED;
    }
  
    // parse commands
    // 16.Jun.2004: process only one command each time through
    bool blocked = false;
    if (_in_texts[fd].length()) {
	const char *in_text = _in_texts[fd].data();
	int len = _in_texts[fd].length();
	int pos = 0;
	while (pos < len && in_text[pos] != '\r' && in_text[pos] != '\n')
	    pos++;
	if (pos < len || (_flags[fd] & READ_CLOSED)) {
	    // have a complete command, parse it

	    // include end of line
	    if (pos < len - 1 && in_text[pos] == '\r' && in_text[pos+1] == '\n')
		pos += 2;
	    else if (pos < len)	// '\r' or '\n' alone
		pos++;
    
	    // grab string
	    String old_text = _in_texts[fd];
	    String line = old_text.substring(0, pos);
	    _in_texts[fd] = old_text.substring(pos);

	    // parse each individual command
	    if (parse_command(fd, line) > 0) {
		// more data to come, so wait
		_in_texts[fd] = old_text;
		blocked = true;
	    }
	}
    }

    // write data until blocked
    // The 2nd argument causes write events to remain selected when commands
    // remain to be processed (whether or not CS has data to write).
    flush_write(fd, _in_texts[fd].length() && !blocked);

    // maybe close out
    if (((_flags[fd] & READ_CLOSED) && !_in_texts[fd].length() && !_out_texts[fd].length())
	|| (_flags[fd] & WRITE_CLOSED)) {
	remove_select(fd, SELECT_READ | SELECT_WRITE);
	close(fd);
	if (_verbose)
	    click_chatter("%s: closed connection %d", declaration().cc(), fd);
	_flags[fd] = -1;
    }
}


ErrorHandler *
ControlSocket::proxy_error_function(const String &h, void *thunk)
{
  ControlSocket *cs = static_cast<ControlSocket *>(thunk);
  return (h == cs->_proxied_handler ? cs->_proxied_errh : 0);
}

CLICK_ENDDECLS
ELEMENT_REQUIRES(userlevel)
EXPORT_ELEMENT(ControlSocket)