udptest.erl

-module(udptest).

-export([start_sender/2, start_reader/2]).


start_sender(Addr, Port) ->
  Options0 = [{broadcast, true}, {reuseaddr, true}, {active, once}, {multicast_loop, true},
             {multicast_ttl, 4}, {high_msgq_watermark, 262144}, {low_msgq_watermark, 65536}, {sndbuf, 262144}],
  % {multicast_if, {127,0,0,1}}],
  Options = Options0,
  PPS = 1000, % 1000 packets per second
  spawn(fun() -> send_loop(Options, Addr, Port, PPS) end).


send_loop(Options, Addr, Port, PPS) ->
  {ok, S} = gen_udp:open(0, Options),
  Time0 = erlang:system_time(PPS),
  send_loop(S, Addr, Port, PPS, Time0).

send_loop(S, Addr, Port, PPS, PrevTime) ->
  Time = erlang:system_time(PPS),
  PacketIDs = lists:seq(PrevTime + 1, Time),
  [send_packet(ID, S, Addr, Port) || ID <- PacketIDs],
  timer:sleep(1),
  send_loop(S, Addr, Port, PPS, Time).

send_packet(ID, S, Addr, Port) ->
  % 8 + 1200 + 8 + 200 = 1416 bytes
  Packet = <<ID:64, 0:1200/integer-unit:8, ID:64, 0:200/integer-unit:8>>,
  ok = gen_udp:send(S, Addr, Port, Packet).



start_reader(Addr, Port) ->
  GwIP = {0,0,0,0}, % {127,0,0,1},
  Common = [binary,{reuseaddr,true},{recbuf,2*1024*1024},inet,{read_packets,100},{active,500}],
  Multicast = [{multicast_ttl,4},{multicast_loop,true},{ip,Addr},{add_membership,{Addr,GwIP}}],
  Options = Common ++ Multicast,
  spawn(fun() -> run_reader(Port, Options) end).


run_reader(Port, Options) ->
  {ok, S} = gen_udp:open(Port, Options),
  read_loop(S, 0).

read_loop(S, Inactive) when Inactive > 100 ->
  inet:setopts(S, [{active, Inactive}]),
  read_loop(S, 0);

read_loop(S, Inactive) ->
  receive
    {udp, S, _IP, _InPortNo, _Bin} ->
      read_loop(S, Inactive + 1);
    Other ->
      io:format("reader ~w/~w unexpected: ~P", [self(), S, Other, 15]),
      read_loop(S, Inactive)
  end.