/*-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 <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <errno.h>
#include <oop.h>
#include <assert.h>

#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;
}