/*-GNU-GPL-BEGIN-* nepim - network pipemeter Copyright (C) 2005 Everton da Silva Marques nepim is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 2, or (at your option) any later version. nepim is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with nepim; see the file COPYING. If not, write to the Free Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *-GNU-GPL-END-*/ /* $Id: udp_server.c,v 1.67 2005/10/13 19:02:20 evertonm Exp $ */ #include #include #include #include #include #include #include #include #include #include #include "conf.h" #include "sock.h" #include "common.h" #include "slot.h" #include "usock.h" #include "udp_header.h" #include "tcpwrap.h" extern nepim_slot_set_t slots; /* from server.c */ extern nepim_usock_set_t udp_tab; /* from server.c */ static void schedule_stat_interval(int local_slot); static void udp_slot_cancel_stat_timers(nepim_slot_t *slot); static void *on_udp_rate_delay(oop_source *src, struct timeval tv, void *user); static void *on_udp_write(oop_source *src, int sd, oop_event event, void *user); static void cancel_slot_rate_write(nepim_slot_t *slot); static void cancel_slot_write(nepim_slot_t *slot); static void udp_slot_cancel_writers(nepim_slot_t *slot); static void *on_udp_keepalive_time(oop_source *src, struct timeval tv, void *user); static void *on_udp_keepalive_require(oop_source *src, struct timeval tv, void *user); static void udp_slot_kill(nepim_slot_t *slot); static void *on_udp_interval(oop_source *src, struct timeval tv, void *user) { nepim_slot_t *slot = user; nepim_session_t *session = &slot->session; nepim_slot_stat(stdout, NEPIM_LABEL_PARTIAL, slot->udp_sd, slot->index, slot->index_remote, session->byte_interval_recv, session->byte_interval_sent, session->stat_interval, session->tv_start.tv_sec, session->test_duration, session->interval_reads, session->interval_writes, slot->interval_pkt_lost, slot->interval_pkt_dup); session->byte_interval_recv = 0; session->byte_interval_sent = 0; session->interval_reads = 0; session->interval_writes = 0; slot->total_pkt_lost += slot->interval_pkt_lost; slot->total_pkt_dup += slot->interval_pkt_dup; slot->interval_pkt_lost = 0; slot->interval_pkt_dup = 0; schedule_stat_interval(slot->index); return OOP_CONTINUE; } static void schedule_stat_interval(int local_slot) { nepim_slot_t *slot = nepim_slot_set_get(&slots, local_slot); nepim_session_t *session; assert(slot); session = &slot->session; { int result = gettimeofday(&session->tv_interval, 0); assert(!result); } session->tv_interval.tv_sec += session->stat_interval; nepim_global.oop_src->on_time(nepim_global.oop_src, session->tv_interval, on_udp_interval, slot); } static void *on_udp_duration(oop_source *src, struct timeval tv, void *user) { nepim_slot_t *slot = user; nepim_session_t *session = &slot->session; int sd = slot->udp_sd; slot->total_pkt_lost += slot->interval_pkt_lost; slot->total_pkt_dup += slot->interval_pkt_dup; nepim_slot_stat(stdout, NEPIM_LABEL_TOTAL, sd, slot->index, slot->index_remote, session->byte_total_recv, session->byte_total_sent, session->test_duration, session->tv_start.tv_sec, session->test_duration, session->total_reads, session->total_writes, slot->total_pkt_lost, slot->total_pkt_dup); session->byte_total_recv = 0; session->byte_total_sent = 0; session->total_reads = 0; session->total_writes = 0; session->duration_done = 1; #if 0 udp_slot_cancel_stat_timers(slot); udp_slot_cancel_writers(slot); fprintf(stderr, "DEBUG FIXME %s %s: (%d-%d) finished, " "not killed (beware possibly active remote sender)\n", __FILE__, __PRETTY_FUNCTION__, slot->index, slot->index_remote); #else udp_slot_kill(slot); #endif return OOP_CONTINUE; } static void udp_slot_cancel_stat_timers(nepim_slot_t *slot) { nepim_session_t *session = &slot->session; nepim_global.oop_src->cancel_time(nepim_global.oop_src, session->tv_duration, on_udp_duration, slot); nepim_global.oop_src->cancel_time(nepim_global.oop_src, session->tv_interval, on_udp_interval, slot); } static void udp_slot_cancel_writers(nepim_slot_t *slot) { nepim_session_t *session = &slot->session; if (session->must_send) { if ((session->max_bit_rate > 0) || (session->max_pkt_rate > 0)) { /* stop current writing, if any */ if (slot->want_write) { assert(slot->want_write == 1); cancel_slot_rate_write(slot); /* rate write */ } /* stop periodic write scheduler */ nepim_global.oop_src->cancel_time(nepim_global.oop_src, session->tv_send_rate, on_udp_rate_delay, slot); } else { cancel_slot_write(slot); /* full write */ } } else { if (session->udp_keepalive_send) { if (slot->want_write) { assert(slot->want_write == 1); nepim_cancel_slot_keepalive(slot); /* keepalive write */ } /* stop periodic keepalive scheduler */ nepim_global.oop_src->cancel_time(nepim_global.oop_src, session->tv_keepalive, on_udp_keepalive_time, slot); } } assert(!slot->want_write); } static void udp_slot_cancel_io(nepim_slot_t *slot) { udp_slot_cancel_writers(slot); } static void udp_slot_kill(nepim_slot_t *slot) { int slot_index = slot->index; udp_slot_cancel_stat_timers(slot); udp_slot_cancel_io(slot); if (slot->session.udp_keepalive_require) nepim_global.oop_src->cancel_time(nepim_global.oop_src, slot->session.tv_keepalive_timer, on_udp_keepalive_require, slot); nepim_sock_show_opt(stderr, slot->udp_sd); nepim_slot_set_del(&slots, slot_index); } static void will_slot_rate_write(nepim_slot_t *slot) { nepim_session_t *session = &slot->session; int sd = slot->udp_sd; #ifndef NDEBUG { if (session->max_bit_rate <= 0) { assert(session->max_pkt_rate > 0); } if (session->max_pkt_rate <= 0) { assert(session->max_bit_rate > 0); } } #endif /* NDEBUG */ assert(!slot->want_write); ++slot->want_write; if (nepim_usock_writer_add(&udp_tab, sd)) return; nepim_global.oop_src->on_fd(nepim_global.oop_src, sd, OOP_WRITE, on_udp_write, 0); } static void cancel_slot_rate_write(nepim_slot_t *slot) { nepim_session_t *session = &slot->session; int sd = slot->udp_sd; #ifndef NDEBUG { if (session->max_bit_rate <= 0) { assert(session->max_pkt_rate > 0); } if (session->max_pkt_rate <= 0) { assert(session->max_bit_rate > 0); } } #endif /* NDEBUG */ assert(slot->want_write == 1); --slot->want_write; if (nepim_usock_writer_del(&udp_tab, sd)) return; nepim_global.oop_src->cancel_fd(nepim_global.oop_src, sd, OOP_WRITE); } static void will_slot_write(nepim_slot_t *slot) { nepim_session_t *session = &slot->session; int sd = slot->udp_sd; assert(session->max_bit_rate < 1); assert(session->max_pkt_rate < 1); assert(!slot->want_write); ++slot->want_write; if (nepim_usock_writer_add(&udp_tab, sd)) return; nepim_global.oop_src->on_fd(nepim_global.oop_src, sd, OOP_WRITE, on_udp_write, 0); } static void cancel_slot_write(nepim_slot_t *slot) { nepim_session_t *session = &slot->session; int sd = slot->udp_sd; assert(session->max_bit_rate < 1); assert(session->max_pkt_rate < 1); assert(slot->want_write == 1); --slot->want_write; if (nepim_usock_writer_del(&udp_tab, sd)) return; nepim_global.oop_src->cancel_fd(nepim_global.oop_src, sd, OOP_WRITE); } static void *on_udp_rate_delay(oop_source *src, struct timeval tv, void *user) { nepim_slot_t *slot = user; nepim_session_t *session = &slot->session; assert(timercmp(&tv, &session->tv_send_rate, ==)); #ifndef NDEBUG { if (session->max_bit_rate <= 0) { assert(session->max_pkt_rate > 0); } if (session->max_pkt_rate <= 0) { assert(session->max_bit_rate > 0); } } #endif /* NDEBUG */ /* save next scheduling time */ { int result = gettimeofday(&session->tv_send_rate, 0); assert(!result); } nepim_timer_usec_add(&session->tv_send_rate, session->write_delay); /* calculate bytes to be written from rate */ session->rate_remaining = nepim_bps2bytes(session->max_bit_rate, session->write_delay); session->pkt_remaining = nepim_pps2packets(session->max_pkt_rate, session->write_delay); #ifndef NDEBUG { if (session->rate_remaining <= 0) { assert(session->pkt_remaining > 0); } if (session->pkt_remaining <= 0) { assert(session->rate_remaining > 0); } } #endif /* NDEBUG */ /* start to write */ will_slot_rate_write(slot); return OOP_CONTINUE; } static void udp_write_rate(nepim_slot_t *slot) { char buf[nepim_global.udp_write_size]; int wr; nepim_session_t *session; int to_write; int sd; assert(nepim_global.udp_write_size == sizeof(buf)); assert(slot); sd = slot->udp_sd; session = &slot->session; #ifndef NDEBUG { if (session->max_bit_rate <= 0) { assert(session->max_pkt_rate > 0); } if (session->max_pkt_rate <= 0) { assert(session->max_bit_rate > 0); } } #endif /* NDEBUG */ #ifndef NDEBUG { if (session->rate_remaining <= 0) { assert(session->pkt_remaining > 0); } if (session->pkt_remaining <= 0) { assert(session->rate_remaining > 0); } } #endif /* NDEBUG */ if (session->rate_remaining > 0) to_write = NEPIM_RANGE(session->rate_remaining, UDP_HEADER_LEN, sizeof(buf)); else to_write = sizeof(buf); assert(to_write >= 0); assert(to_write >= UDP_HEADER_LEN); assert(to_write <= sizeof(buf)); wr = nepim_slot_buf_sendto(slot, buf, to_write, UDP_TYPE_DATA); if (wr < 0) { switch (errno) { case EINTR: case EAGAIN: nepim_usock_write_error(&udp_tab, sd, slot->index, slot->index_remote, errno); return; case EPIPE: fprintf(stderr, "rate_write: EPIPE on UDP socket %d\n", sd); break; } fprintf(stderr, "rate_write: connection lost on UDP socket %d\n", sd); if (!session->duration_done) report_broken_slot_stat(stdout, slot); udp_slot_kill(slot); return; } assert(wr >= 0); assert(wr <= to_write); nepim_usock_write_good(&udp_tab, sd); session->rate_remaining -= wr; --session->pkt_remaining; nepim_session_write_add(session, wr); /* finished ? */ if ( ((session->max_bit_rate > 0) && (session->rate_remaining < 1)) || ((session->max_pkt_rate > 0) && (session->pkt_remaining < 1)) ){ /* stop writing */ cancel_slot_rate_write(slot); /* schedule for next time saved by on_udp_rate_delay() */ nepim_global.oop_src->on_time(nepim_global.oop_src, session->tv_send_rate, on_udp_rate_delay, slot); } } static void udp_write_full(nepim_slot_t *slot) { char buf[nepim_global.udp_write_size]; int wr; nepim_session_t *session; int sd; assert(nepim_global.udp_write_size == sizeof(buf)); assert(slot); sd = slot->udp_sd; session = &slot->session; assert(session->max_bit_rate < 1); assert(session->max_pkt_rate < 1); wr = nepim_slot_buf_sendto(slot, buf, sizeof(buf), UDP_TYPE_DATA); if (wr < 0) { switch (errno) { case EINTR: case EAGAIN: nepim_usock_write_error(&udp_tab, sd, slot->index, slot->index_remote, errno); return; case EPIPE: fprintf(stderr, "write: EPIPE on UDP socket %d\n", sd); break; } fprintf(stderr, "write: connection lost on UDP socket %d\n", sd); if (!session->duration_done) report_broken_slot_stat(stdout, slot); udp_slot_kill(slot); return; } assert(wr >= 0); assert(wr <= sizeof(buf)); nepim_usock_write_good(&udp_tab, sd); nepim_session_write_add(session, wr); } static void *on_udp_write(oop_source *src, int sd, oop_event event, void *user) { nepim_slot_t *slot; nepim_session_t *session; assert(!user); assert(event == OOP_WRITE); slot = nepim_slot_find_next_writer(&slots, sd); assert(slot); session = &slot->session; if (!session->must_send) { if (session->udp_keepalive_send) { nepim_udp_write_keepalive(slot, nepim_slot_buf_sendto, on_udp_keepalive_time, udp_slot_kill); return OOP_CONTINUE; } } if ((session->max_bit_rate > 0) || (session->max_pkt_rate > 0)) { udp_write_rate(slot); return OOP_CONTINUE; } udp_write_full(slot); return OOP_CONTINUE; } static void *on_udp_keepalive_time(oop_source *src, struct timeval tv, void *user) { nepim_slot_t *slot = user; nepim_session_t *session = &slot->session; assert(timercmp(&tv, &session->tv_keepalive, ==)); assert(!session->must_send); assert(session->udp_keepalive_send); /* save next scheduling time */ { int result = gettimeofday(&session->tv_keepalive, 0); assert(!result); } nepim_timer_usec_add(&session->tv_keepalive, nepim_global.udp_keepalive_delay); /* start to write */ nepim_will_slot_keepalive(slot, on_udp_write); return OOP_CONTINUE; } static void *on_udp_keepalive_require(oop_source *src, struct timeval tv, void *user) { nepim_slot_t *slot = user; nepim_session_t *session = &slot->session; assert(session->udp_keepalive_require); assert(slot->keepalives_recv >= 0); if (slot->keepalives_recv < 1) { fprintf(stderr, "%d %d-%d: broken slot: peer not sending keepalives?\n", slot->udp_sd, slot->index, slot->index_remote); report_broken_slot_stat(stdout, slot); udp_slot_kill(slot); return OOP_CONTINUE; } slot->keepalives_recv = 0; nepim_schedule_keepalive_timer(slot, on_udp_keepalive_require); return OOP_CONTINUE; } static void udp_slot_start(int local_slot) { nepim_slot_t *slot = nepim_slot_set_get(&slots, local_slot); nepim_session_t *session; int sd; assert(slot); session = &slot->session; sd = slot->udp_sd; if (session->must_send) { if ((session->max_bit_rate > 0) || (session->max_pkt_rate > 0)) { session->tv_send_rate = OOP_TIME_NOW; nepim_global.oop_src->on_time(nepim_global.oop_src, session->tv_send_rate, on_udp_rate_delay, slot); } else { will_slot_write(slot); } } else { if (session->udp_keepalive_send) { fprintf(stderr, "DEBUG %s %s: %d %d-%d: scheduling sending of keepalives\n", __FILE__, __PRETTY_FUNCTION__, slot->udp_sd, slot->index, slot->index_remote); session->tv_keepalive = OOP_TIME_NOW; nepim_global.oop_src->on_time(nepim_global.oop_src, session->tv_keepalive, on_udp_keepalive_time, slot); } } { int result = gettimeofday(&session->tv_start, 0); assert(!result); session->tv_duration = session->tv_start; } session->tv_duration.tv_sec += session->test_duration; nepim_global.oop_src->on_time(nepim_global.oop_src, session->tv_duration, on_udp_duration, slot); nepim_sock_show_opt(stderr, sd); schedule_stat_interval(local_slot); if (session->udp_keepalive_require) nepim_schedule_keepalive_timer(slot, on_udp_keepalive_require); } static int parse_greetings(nepim_greet_t *opt, int sd, int local_slot, int remote_slot, char *buf, const int buf_len, const struct sockaddr *remote, socklen_t remote_len) { int result; char *eos; char *past_end; char peer[100]; char password_buf[100]; eos = memchr(buf, '\n', buf_len); if (eos) past_end = eos; else past_end = buf + buf_len; assert(past_end >= buf); assert(past_end <= (buf + buf_len)); *past_end = '\0'; opt->password_buf = password_buf; opt->password_buf_size = sizeof(password_buf); result = nepim_parse_greetings(opt, buf, past_end); if (result) return result; nepim_sock_dump_addr(peer, sizeof(peer), (const struct sockaddr *) remote); fprintf(stderr, "%d %d-%d: UDP from (%s,%d): send=%d bps=%lld pps=%d " "interval=%d duration=%d delay=%ld ka_send=%d ka_req=%d " "password=%s\n", sd, local_slot, remote_slot, peer, nepim_sock_get_port((struct sockaddr *) remote), opt->must_send, opt->bit_rate, opt->pkt_rate, opt->stat_interval, opt->test_duration, opt->write_delay, opt->udp_keepalive_send, opt->udp_keepalive_require, password_buf); if (nepim_global.tcpwrap) { if (!nepim_hosts_ctl(nepim_global.tcpwrap, nepim_global.prog_name, peer)) { fprintf(stderr, "%d: %d-%d: %s: %s: TCP wrapper denied access from client=[%s] to service=[%s]\n", sd, local_slot, remote_slot, __FILE__, __PRETTY_FUNCTION__, peer, nepim_global.prog_name); return 1; } } if (nepim_global.password) if (strcmp(nepim_global.password, password_buf)) { fprintf(stderr, "%d: bad client password=%s\n", sd, password_buf); return 1; } return 0; } static void udp_consume(nepim_slot_t *slot, size_t len, uint32_t seq) { nepim_session_t *session = &slot->session; /* any valid traffic accounted as keepalive */ ++slot->keepalives_recv; nepim_session_read_add(session, len); if (nepim_global.udp_exp_stats) nepim_win_add(&slot->win_recv_seq, seq, &slot->interval_pkt_lost, &slot->interval_pkt_dup); } static void *on_udp_segment(oop_source *src, int sd, oop_event event, void *unnused) { int rd; char buf[nepim_global.udp_read_size]; union { struct sockaddr_in inet; struct sockaddr_in6 inet6; } from; socklen_t fromlen = sizeof(from); nepim_udp_hdr_t hdr; assert(sizeof(buf) == nepim_global.udp_read_size); assert(event == OOP_READ); assert(!unnused); rd = recvfrom(sd, buf, nepim_global.udp_read_size, 0, (struct sockaddr *) &from, &fromlen); if (rd < 0) { switch (errno) { case EINTR: case EAGAIN: case ECONNREFUSED: nepim_usock_read_error(&udp_tab, sd, errno); return OOP_CONTINUE; } fprintf(stderr, "%d: recvfrom: unexpected failure: errno=%d: %s\n", sd, errno, strerror(errno)); assert(0); return OOP_CONTINUE; } assert(rd >= 0); assert(rd <= sizeof(buf)); nepim_usock_read_good(&udp_tab, sd); if (nepim_udp_hdr_parse(&hdr, buf, rd)) { fprintf(stderr, "%d: recvfrom: could not parse header from UDP segment: %d/%d bytes\n", sd, rd, nepim_global.udp_read_size); return OOP_CONTINUE; } if (hdr.version) { fprintf(stderr, "%d: recvfrom: ignoring unknown UDP header version: %d\n", sd, hdr.version); return OOP_CONTINUE; } /* request for new slot? */ if (hdr.type == UDP_TYPE_HELLO) { int result; int local_slot; nepim_greet_t opt; #if 0 fprintf(stderr, "DEBUG %s %s: new slot request: %d (%d-%d)\n", __FILE__, __PRETTY_FUNCTION__, sd, nepim_uint16_read(buf + 1), nepim_uint16_read(buf + 3)); #endif if (nepim_slot_find_addr(&slots, hdr.src_slot, (const struct sockaddr *) &from, fromlen)) { fprintf(stderr, "%d %d-%d: normal behavior: ignoring existing remote src_slot/addr pair\n", sd, hdr.dst_slot, hdr.src_slot); return OOP_CONTINUE; } result = parse_greetings(&opt, sd, hdr.dst_slot, hdr.src_slot, buf + UDP_HEADER_LEN, rd - UDP_HEADER_LEN, (const struct sockaddr *) &from, fromlen); if (result) { fprintf(stderr, "%d %d-%d: recvfrom: bad greetings: %d\n", sd, hdr.dst_slot, hdr.src_slot, result); return OOP_CONTINUE; } local_slot = nepim_slot_find_free(&slots); assert(local_slot < 0xFFFF); nepim_slot_set_add(&slots, sd, local_slot, hdr.src_slot, (const struct sockaddr *) &from, fromlen, &opt); udp_slot_start(local_slot); return OOP_CONTINUE; } /* new slot request */ { nepim_slot_t *slot = nepim_slot_set_search(&slots, hdr.dst_slot); /* multicast client can't know the server slot index, since the multicast server never send replies. so, the server finds the slot by socket address. */ if (!slot) slot = nepim_slot_set_search_remote(&slots, (const struct sockaddr *) &from, fromlen); if (!slot) { fprintf(stderr, "%d: recvfrom: unknown local slot: %d\n", sd, hdr.dst_slot); return OOP_CONTINUE; } if (slot->index_remote != hdr.src_slot) { fprintf(stderr, "%d %d: recvfrom: remote index mismatch: expected=%d got=%d\n", sd, hdr.dst_slot, slot->index_remote, hdr.src_slot); return OOP_CONTINUE; } if (slot->session.remote_len != fromlen) { fprintf(stderr, "%d %d-%d: recvfrom: bad address length: expected=%d got=%d\n", sd, hdr.dst_slot, hdr.src_slot, slot->session.remote_len, fromlen); return OOP_CONTINUE; } if (memcmp(&slot->session.remote, &from, fromlen)) { fprintf(stderr, "%d %d-%d: recvfrom: address mismatch\n", sd, hdr.dst_slot, hdr.src_slot); return OOP_CONTINUE; } udp_consume(slot, rd, hdr.seq); } return OOP_CONTINUE; } int nepim_udp_listener(const char *hostname, const char *portname, int mcast_join) { struct addrinfo hints; struct addrinfo *ai_res; struct addrinfo *ai; int result; int udp_listeners = 0; hints.ai_socktype = SOCK_DGRAM; hints.ai_protocol = IPPROTO_UDP; hints.ai_flags = AI_PASSIVE | AI_CANONNAME; hints.ai_family = PF_UNSPEC; hints.ai_addrlen = 0; hints.ai_addr = 0; hints.ai_canonname = 0; result = getaddrinfo(hostname, portname, &hints, &ai_res); if (result) { fprintf(stderr, "%s: getaddrinfo(%s,%s): %s\n", __PRETTY_FUNCTION__, hostname, portname, gai_strerror(result)); return 0; } for (ai = ai_res; ai; ai = ai->ai_next) { int sd; #if 0 fprintf(stderr, "UDP listener: host=%s,%s(%d) len=%d family=%d type=%d proto=%d\n", hostname, portname, nepim_sock_get_port(ai->ai_addr), ai->ai_addrlen, ai->ai_family, ai->ai_socktype, ai->ai_protocol); #endif if (nepim_global.no_inet6 && (ai->ai_family == PF_INET6)) continue; if (nepim_global.no_inet4 && (ai->ai_family == PF_INET)) continue; sd = nepim_create_socket(ai->ai_addr, ai->ai_addrlen, ai->ai_family, ai->ai_socktype, ai->ai_protocol, nepim_global.pmtu_mode, nepim_global.ttl, mcast_join, nepim_global.win_recv, nepim_global.win_send); if (sd < 0) { fprintf(stderr, "spawn_udp_listener: UDP listener socket failed for %s,%s(%d): %d\n", hostname, portname, nepim_sock_get_port(ai->ai_addr), sd); break; } nepim_global.oop_src->on_fd(nepim_global.oop_src, sd, OOP_READ, on_udp_segment, 0); nepim_usock_set_add(&udp_tab, sd); ++udp_listeners; fprintf(stderr, "%d: UDP socket listening on %s,%s(%d)\n", sd, hostname, portname, nepim_sock_get_port(ai->ai_addr)); nepim_sock_show_opt(stderr, sd); } freeaddrinfo(ai_res); return udp_listeners; }