fromipsumdump.cc

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

		break;

	      case W_TCP_SACK:
		if (*data == '.')
		    data++;
		else if (*data != '-') {
		    have_tcp_opt = true;
		    data = parse_tcp_opt_ascii(data, end, &tcp_opt, DO_TCPOPT_SACK);
		}
		break;
		
	      case W_TCP_NTOPT:
		if (*data == '.')
		    data++;
		else if (*data != '-') {
		    have_tcp_opt = true;
		    data = parse_tcp_opt_ascii(data, end, &tcp_opt, DO_TCPOPT_NTALL);
		}
		break;
		
	      case W_TCP_OPT:
		if (*data == '.')
		    data++;
		else if (*data != '-') {
		    have_tcp_opt = true;
		    data = parse_tcp_opt_ascii(data, end, &tcp_opt, DO_TCPOPT_ALL);
		}
		break;
		
	      case W_LINK:
		if (*data == '>' || *data == 'L') {
		    u1 = 0;
		    data++;
		} else if (*data == '<' || *data == 'X' || *data == 'R') {
		    u1 = 1;
		    data++;
		} else
		    data = cp_unsigned(data, end, 0, &u1);
		break;

	      case W_PAYLOAD:
		if (*data == '\"') {
		    payload.clear();
		    const unsigned char *fdata = data + 1;
		    for (data++; data < end && *data != '\"'; data++)
			if (*data == '\\' && data < end - 1) {
			    payload.append((const char *) fdata, (const char *) data);
			    fdata = (const unsigned char *) cp_process_backslash((const char *) data, (const char *) end, payload);
			    data = fdata - 1; // account for loop increment
			}
		    payload.append((const char *) fdata, (const char *) data);
		    // bag payload if it didn't parse correctly
		    if (data >= end || *data != '\"')
			data = original_data;
		    else {
			have_payload = have_payload_len = true;
			payload_len = payload.length();
		    }
		}
		break;

	    }

	    // check whether we correctly parsed something
	    {
		bool this_ok = (data > original_data && (data >= end || isspace(*data)));
		while (data < end && !isspace(*data))
		    data++;
		while (data < end && isspace(*data))
		    data++;
		if (!this_ok)
		    continue;
	    }

	    // store contents
	  store_contents:
	    switch (_contents[i]) {

	      case W_TIMESTAMP:
	      case W_NTIMESTAMP:
		if (u2 < 1000000000)
		    q->timestamp_anno().set_nsec(u1, u2), ok++;
		break;

	      case W_TIMESTAMP_SEC:
		q->timestamp_anno().set_sec(u1), ok++;
		break;

	      case W_TIMESTAMP_USEC:
		if (u1 < 1000000)
		    q->timestamp_anno().set_subsec(Timestamp::usec_to_subsec(u1)), ok++;
		break;

	      case W_TIMESTAMP_USEC1:
		if (u1 == 0 && u2 < 1000000)
		    q->timestamp_anno().set_usec(0, u2), ok++;
		else if (u1 == 0)
		    q->timestamp_anno().set_usec(u2/1000000, u2%1000000), ok++;
#if HAVE_INT64_TYPES
		else {
		    uint64_t uu = ((uint64_t)u1 << 32) | u2;
		    q->timestamp_anno().set_usec(uu/1000000, uu%1000000), ok++;
		}
#endif
		break;
		
	      case W_IP_SRC:
		iph->ip_src.s_addr = htonl(u1), ip_ok++;
		break;

	      case W_IP_DST:
		iph->ip_dst.s_addr = htonl(u1), ip_ok++;
		break;
		
	      case W_IP_LEN:
		if (u1 <= 0xFFFF)
		    byte_count = u1, ok++;
		break;
		
	      case W_PAYLOAD_LEN:
		if (u1 <= 0xFFFF)
		    payload_len = u1, have_payload_len = true, ok++;
		break;

	      case W_IP_CAPTURE_LEN:
		/* XXX do nothing with this for now */
		ok++;
		break;
		
	      case W_IP_PROTO:
		if (u1 <= 255)
		    iph->ip_p = u1, ip_ok++;
		break;

	      case W_IP_TOS:
		if (u1 <= 255)
		    iph->ip_tos = u1, ip_ok++;
		break;

	      case W_IP_TTL:
		if (u1 <= 255)
		    iph->ip_ttl = u1, ip_ok++;
		break;

	      case W_IP_ID:
		if (u1 <= 0xFFFF)
		    iph->ip_id = htons(u1), ip_ok++;
		break;

	      case W_IP_FRAG:
		iph->ip_off = u1, ip_ok++;
		break;

	      case W_IP_FRAGOFF:
		if ((u1 & ~IP_MF) <= IP_OFFMASK)
		    iph->ip_off = htons(u1), ip_ok++;
		break;

	      case W_SPORT:
		if (u1 <= 0xFFFF)
		    q->udp_header()->uh_sport = htons(u1), ip_ok++;
		break;

	      case W_DPORT:
		if (u1 <= 0xFFFF)
		    q->udp_header()->uh_dport = htons(u1), ip_ok++;
		break;

	      case W_TCP_SEQ:
		q->tcp_header()->th_seq = htonl(u1), ip_ok++;
		break;

	      case W_TCP_ACK:
		q->tcp_header()->th_ack = htonl(u1), ip_ok++;
		break;

	      case W_TCP_FLAGS:
		if (u1 <= 0xFF)
		    q->tcp_header()->th_flags = u1, ip_ok++;
		else if (u1 <= 0xFFF)
		    // th_off will be set later
		    *reinterpret_cast<uint16_t *>(q->transport_header() + 12) = htons(u1), ip_ok++;
		break;

	      case W_TCP_WINDOW:
		if (u1 <= 0xFFFF)
		    q->tcp_header()->th_win = htons(u1), ip_ok++;
		break;
		
	      case W_TCP_URP:
		if (u1 <= 0xFFFF)
		    q->tcp_header()->th_urp = htons(u1), ip_ok++;
		break;
		
	      case W_COUNT:
		if (u1)
		    SET_EXTRA_PACKETS_ANNO(q, u1 - 1), ok++;
		break;

	      case W_LINK:
		SET_PAINT_ANNO(q, u1), ok++;
		break;

	      case W_AGGREGATE:
		SET_AGGREGATE_ANNO(q, u1), ok++;
		break;

	      case W_FIRST_TIMESTAMP:
	      case W_FIRST_NTIMESTAMP:
		if (u2 < 1000000000) {
		    SET_FIRST_TIMESTAMP_ANNO(q, Timestamp::make_nsec(u1, u2));
		    ok++;
		}
		break;

	    }
	}

	if (!ok && !ip_ok)
	    break;

	// append IP options if any
	if (have_ip_opt && ip_opt) {
	    if (!(q = handle_ip_opt(q, ip_opt)))
		return 0;
	    else		// iph may have changed!! (don't use tcph etc.)
		iph = q->ip_header();
	}
	
	// set TCP offset to a reasonable value; possibly reduce packet length
	if (iph->ip_p == IP_PROTO_TCP && IP_FIRSTFRAG(iph)) {
	    if (!have_tcp_opt || !tcp_opt)
		q->tcp_header()->th_off = sizeof(click_tcp) >> 2;
	    else if (!(q = handle_tcp_opt(q, tcp_opt)))
		return 0;
	    else		// iph may have changed!!
		iph = q->ip_header();
	} else if (iph->ip_p == IP_PROTO_UDP && IP_FIRSTFRAG(iph))
	    q->take(sizeof(click_tcp) - sizeof(click_udp));
	else
	    q->take(sizeof(click_tcp));

	// set IP length
	if (have_payload) {	// XXX what if byte_count indicates IP options?
	    int old_length = q->length();
	    iph->ip_len = ntohs(old_length + payload.length());
	    if ((q = q->put(payload.length()))) {
		memcpy(q->data() + old_length, payload.data(), payload.length());
		// iph may have changed!!
		iph = q->ip_header();
	    }
	} else if (byte_count) {
	    iph->ip_len = ntohs(byte_count);
	    SET_EXTRA_LENGTH_ANNO(q, byte_count - q->length());
	} else if (have_payload_len) {
	    iph->ip_len = ntohs(q->length() + payload_len);
	    SET_EXTRA_LENGTH_ANNO(q, payload_len);
	} else
	    iph->ip_len = ntohs(q->length());

	// set UDP length (after IP length is available)
	if (iph->ip_p == IP_PROTO_UDP && IP_FIRSTFRAG(iph))
	    q->udp_header()->uh_ulen = htons(ntohs(iph->ip_len) - (iph->ip_hl << 2));
	
	// set data from flow ID
	if (_use_flowid) {
	    IPFlowID flowid = (PAINT_ANNO(q) & 1 ? _flowid.rev() : _flowid);
	    if (flowid.saddr())
		iph->ip_src = flowid.saddr();
	    if (flowid.daddr())
		iph->ip_dst = flowid.daddr();
	    if (flowid.sport() && IP_FIRSTFRAG(iph))
		q->tcp_header()->th_sport = flowid.sport();
	    if (flowid.dport() && IP_FIRSTFRAG(iph))
		q->tcp_header()->th_dport = flowid.dport();
	    if (_use_aggregate)
		SET_AGGREGATE_ANNO(q, _aggregate);
	} else if (!ip_ok)
	    q->set_network_header(0, 0);

	// set destination IP address annotation
	q->set_dst_ip_anno(iph->ip_dst);

	// set checksum
	if (_checksum && ip_ok)
	    set_checksums(q, iph);
	
	return q;
    }

    // bad format if we get here
    if (!_format_complaint) {
	// don't complain if the line was all blank
	if ((int) strspn(line.data(), " \t\n\r") != line.length()) {
	    if (_contents.size() == 0)
		_ff.error(errh, "no '!data' provided");
	    else
		_ff.error(errh, "packet parse error");
	    _format_complaint = true;
	}
    }
    if (q)
	q->kill();
    return 0;
}

inline Packet *
set_packet_lengths(Packet *p, uint32_t extra_length)
{
    uint32_t length = p->length() + extra_length;
    if (htons(length) != p->ip_header()->ip_len) {
	if (WritablePacket *q = p->uniqueify()) {
	    click_ip *ip = q->ip_header();
	    ip->ip_len = htons(length);
	    if (ip->ip_p == IP_PROTO_UDP)
		q->udp_header()->uh_ulen = htons(length - (ip->ip_hl << 2));
	    return q;
	} else
	    return 0;
    } else
	return p;
}

Packet *
FromIPSummaryDump::handle_multipacket(Packet *p)
{
    assert(!_work_packet || _work_packet == p);
    
    if (!p || !EXTRA_PACKETS_ANNO(p)) {
	_work_packet = 0;
	return p;
    }

    uint32_t count = 1 + EXTRA_PACKETS_ANNO(p);

    // set up _multipacket variables on new packets (_work_packet == 0)
    if (!_work_packet) {
	assert(count > 1);
	// set length of all but the last packet
	_multipacket_length = (p->length() + EXTRA_LENGTH_ANNO(p)) / count;
	// beware if there isn't enough EXTRA_LENGTH to cover all the packets
	if (_multipacket_length < p->length()) {
	    _multipacket_length = p->length();
	    SET_EXTRA_LENGTH_ANNO(p, _multipacket_length * (count - 1));
	}
	// set timestamps
	_multipacket_end_timestamp = p->timestamp_anno();
	if (FIRST_TIMESTAMP_ANNO(p)) {
	    _multipacket_timestamp_delta = (p->timestamp_anno() - FIRST_TIMESTAMP_ANNO(p)) / (count - 1);
	    p->timestamp_anno() = FIRST_TIMESTAMP_ANNO(p);
	} else
	    _multipacket_timestamp_delta = Timestamp();
	// prepare IP lengths for _multipacket_extra_length
	_work_packet = set_packet_lengths(p, _multipacket_length - p->length());
	if (!_work_packet)
	    return 0;
    }

    // prepare packet to return
    if ((p = p->clone())) {
	SET_EXTRA_PACKETS_ANNO(p, 0);
	SET_EXTRA_LENGTH_ANNO(p, _multipacket_length - p->length());
    }

    // reduce weight of _work_packet 
    SET_EXTRA_PACKETS_ANNO(_work_packet, count - 2);
    SET_EXTRA_LENGTH_ANNO(_work_packet, EXTRA_LENGTH_ANNO(_work_packet) - _multipacket_length);
    if (count == 2) {
	_work_packet->timestamp_anno() = _multipacket_end_timestamp;
	_work_packet = set_packet_lengths(_work_packet, EXTRA_LENGTH_ANNO(_work_packet));
    } else
	_work_packet->timestamp_anno() += _multipacket_timestamp_delta;

    return p;
}

bool
FromIPSummaryDump::run_task()
{
    if (!_active)
	return false;
    Packet *p;

    while (1) {
	p = (_work_packet ? _work_packet : read_packet(0));
	if (!p) {
	    if (_stop)
		router()->please_stop_driver();
	    return false;
	}
	if (_multipacket)
	    p = handle_multipacket(p);
	// check sampling probability
	if (_sampling_prob >= (1 << SAMPLING_SHIFT)
	    || (uint32_t)(random() & ((1 << SAMPLING_SHIFT) - 1)) < _sampling_prob)
	    break;
	if (p)
	    p->kill();
    }

    if (p)
	output(0).push(p);
    _task.fast_reschedule();
    return true;
}

Packet *
FromIPSummaryDump::pull(int)
{
    if (!_active)
	return 0;
    Packet *p;

    while (1) {
	p = (_work_packet ? _work_packet : read_packet(0));
	if (!p) {
	    if (_stop)
		router()->please_stop_driver();
	    _notifier.set_listeners(false);
	    return 0;
	}
	if (_multipacket)
	    p = handle_multipacket(p);
	// check sampling probability
	if (_sampling_prob >= (1 << SAMPLING_SHIFT)
	    || (uint32_t)(random() & ((1 << SAMPLING_SHIFT) - 1)) < _sampling_prob)
	    break;
	if (p)
	    p->kill();
    }

    _notifier.set_listeners(true);
    return p;
}


enum { H_SAMPLING_PROB, H_ACTIVE, H_ENCAP, H_STOP };

String
FromIPSummaryDump::read_handler(Element *e, void *thunk)
{
    FromIPSummaryDump *fd = static_cast<FromIPSummaryDump *>(e);
    switch ((intptr_t)thunk) {
      case H_SAMPLING_PROB:
	return cp_unparse_real2(fd->_sampling_prob, SAMPLING_SHIFT) + "\n";
      case H_ACTIVE:
	return cp_unparse_bool(fd->_active) + "\n";
      case H_ENCAP:
	return "IP\n";
      default:
	return "<error>\n";
    }
}

int
FromIPSummaryDump::write_handler(const String &s_in, Element *e, void *thunk, ErrorHandler *errh)
{
    FromIPSummaryDump *fd = static_cast<FromIPSummaryDump *>(e);
    String s = cp_uncomment(s_in);
    switch ((intptr_t)thunk) {
      case H_ACTIVE: {
	  bool active;
	  if (cp_bool(s, &active)) {
	      fd->_active = active;
	      if (fd->output_is_push(0) && active && !fd->_task.scheduled())
		  fd->_task.reschedule();
	      else if (!fd->output_is_push(0))
		  fd->_notifier.set_listeners(active);
	      return 0;
	  } else
	      return errh->error("'active' should be Boolean");
      }
      case H_STOP:
	fd->_active = false;
	fd->router()->please_stop_driver();
	return 0;
      default:
	return -EINVAL;
    }
}

void
FromIPSummaryDump::add_handlers()
{
    add_read_handler("sampling_prob", read_handler, (void *)H_SAMPLING_PROB);
    add_read_handler("active", read_handler, (void *)H_ACTIVE);
    add_write_handler("active", write_handler, (void *)H_ACTIVE);
    add_read_handler("encap", read_handler, (void *)H_ENCAP);
    add_write_handler("stop", write_handler, (void *)H_STOP);
    _ff.add_handlers(this);
    if (output_is_push(0))
	add_task_handlers(&_task);
}

ELEMENT_REQUIRES(userlevel FromFile IPSummaryDumpInfo)
EXPORT_ELEMENT(FromIPSummaryDump)
CLICK_ENDDECLS