#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <zlib.h>

#include "../shared/GlobalConfigFile.h"
#include "../shared/SimpleLogFile.h"
#include "../shared/Marshal.h"
#include "../shared/MultiCast.h"
#include "../shared/CommandDispatcher.h"
#include "../shared/ReplyToClient.h"

#include "AlertRecordMultiCast.h"

// TODO set socket options, non blocking!
//      I did some.  I need to double check to make sure I got it all.



/////////////////////////////////////////////////////////////////////
// AlertRecordMultiCastListener
//
// This is a helper for AlertRecordMultiCastReceiver.  We want to
// grab packets from the network as quickly as possible and move them
// into our own queue before they get dropped.
//
// Originally we were reading the packets and parsing them in the
// same thread.  We spent a lot of time uncompressing and otherwise
// processing the data before handing it off to the various
// listeners.
//
// Hopefully this new version will be more responsive and will lose
// fewer packets.
/////////////////////////////////////////////////////////////////////

static const std::string s_X_peek = "X peek";
static const std::string s_X_read = "X read";
static const std::string s_X_process = "X process";

class AlertRecordMultiCastListener : private ForeverThreadUser
{
public:
  typedef SmarterP< std::string > Data;
  typedef std::function< void(Data const &) > Callback;
  
private:
  const int _fd;
  const Callback _callback;
  std::string baseName() { return getContainer()->getThreadName(); }

  // For IContainerThreadUser
  virtual void beforeSleep(IBeforeSleepCallbacks &callbacks) override
  {
    ThreadMonitor &tm = ThreadMonitor::find();
    int count = 0;
    while (true)
    {
      tm.setState(s_X_peek);
      //sendToLogFile(TclList()<<FLF<<"ABOUT TO recv()");
      // TODO the peek trick is nice, but it always requires two kernel crossings
      // per packet.  We know the max size, wouldn't it be faster to always
      // make a buffer that size on the stack?  It's tempting to guess a good
      // size and still grow if we have to, but unnecessary and complicated.
      // TODO Consider recvmmsg().  Receive multiple messages at once.  Maybe
      // reduce the number of kernel crossings a lot more, especially at our
      // busiest times.
      // Note:  I wrote the previous two TODO items before profiling told
      // me to ignore these and issues and put the recv() in a separate thread
      // from the processing.  That change made a huge improvement.  So these
      // TODO items seem lees important now.
      // Interesting.  We are only waiting on this one incoming socket.  We
      // don't need this whole setup where we are doing a non blocking read
      // followed by a poll on that same socket.  It's unnecessarily
      // complicated, but the extra work is done when there is no real work
      // to do, so maybe it's not terrible.
      ssize_t nbytes =
	recv(_fd, NULL, 0, MSG_DONTWAIT | MSG_PEEK | MSG_TRUNC);
      if (nbytes < 0)
      {
	if (!((errno == EAGAIN ) || (errno == EWOULDBLOCK)))
	  // Not sure what would cause this.  Give me info.
	  // TODO tempting to add an explain() here.
	  sendToLogFile(TclList()<<FLF<<baseName()<<"recv"<<errno<<errorString());
	else
	  // The common and expected case.
	  tm.increment("EAGAIN");
	break;
      }
      
      tm.setState(s_X_read);
      SmarterP< std::string > bytes(NULL, nbytes, 'P');
      nbytes = recv(_fd, &(*bytes)[0], nbytes, MSG_TRUNC);
      if (nbytes != (ssize_t)bytes->length())
      {
	TclList msg;
	msg<<FLF<<baseName()<<"recv 2"<<nbytes<<bytes->length();
	if (nbytes < 0)
	  msg<<errno<<errorString();
	sendToLogFile(msg);
	break;
      }
      tm.setState(s_X_process);
      _callback(bytes);
      count++;
    }
    tm.increment("X count", count);
    callbacks.waitForRead(_fd);    
  }
  
public:
  AlertRecordMultiCastListener(int fd, Callback callback,
			       IContainerThread *thread) :
    ForeverThreadUser(thread), _fd(fd), _callback(callback)
  {
    assert(fd >= 0);
    start();
  }
};


/////////////////////////////////////////////////////////////////////
// AlertRecordMultiCastBase
/////////////////////////////////////////////////////////////////////

AlertRecordMultiCastBase::AlertRecordMultiCastBase
(std::string const &baseName) :
  _baseName(baseName + "_multicast"),
  _port(-1),
  _okay(false)
{ // getConfigItem:  alerts_multicast_addr=239.0.1.0:2000
  // getConfigItem:  top_list_multicast_addr=239.0.2.0:2000
  // getConfigItem:  This setting is used by both the master and the slaves
  // getConfigItem:  so they know where to get data.
  // getConfigItem:  Use other settings to turn the data on or off.
  // getConfigItem:  Consider alerts_multicast_feed_next_server=1 or
  // getConfigItem:  top_list_master_info=multicast
  // getConfigItem:  Recommend using values in the range 239.0.0.0 - 239.255.255.255.
  // getConfigItem:  This is called "Organization-Local Scope".
  // getConfigItem:  https://www.iana.org/assignments/multicast-addresses/multicast-addresses.xhtml
  // getConfigItem:  Ideally all the groups would be different even if you
  // getConfigItem:  ignored the least significant 5 bits.  An unmanaged switch
  // getConfigItem:  ignores those 5 bits when delivering messages so it might
  // getConfigItem:  send you lots of data you don't need.
  _configInput = getConfigItem(configKey());
  if (_configInput != "")
  {
    std::vector< std::string > pieces = explode(":", _configInput);
    if (pieces.size() != 2)
    {
      TclList msg;
      msg<<FLF<<"INVALID"<<_baseName<<configKey()<<_configInput;
      sendToLogFile(msg);
    }
    else
    {
      _group = pieces[0];
      _port = strtolDefault(pieces[1], -1);
      _okay = (!_group.empty()) && (_port <= 0xffff) && (_port > 0);
    }
  }
}


/////////////////////////////////////////////////////////////////////
// No class
/////////////////////////////////////////////////////////////////////

enum class ItemType : unsigned char
  { // A single alert or top list row.
    Record,
    
    // This comes from each alert server to say it is alive.  It gets
    // passed along a similar route the the live data, as a test.  Currently
    // the C++ side doesn't use this.  We send these records to the database,
    // and the database side somehow tickles the monitor_alive table.  (I
    // don't remember the details, but when the database was down and the
    // C++ stuff was all up, all the entries like "fogell:1202 -> top_list"
    // and "jules:1203 -> alerts" went down in the back office.
    Status,
    
    // This is the periodic message that we send.  It is internal to
    // AlertRecordMultiCast.  Among other things, it's a constant ping,
    // so it can provide status 24/7.
    LinkStatus,
    
    // We tried to decode the item type and failed.
    Error
  };

inline void add(std::string &bytes, ItemType itemType)
{
  bytes += (char)itemType;
}

inline ItemType takeItemType(std::string &bytes)
{
  if (bytes.empty())
    return ItemType::Error;
  else
  {
    const ItemType result = (ItemType)bytes[bytes.size() - 1];
    bytes.resize(bytes.size() - 1);
    if (result > ItemType::Error)
      return ItemType::Error;
    else
      return result;
  }
}

static SmarterP< std::string > compressString(std::string const &input)
{
  ThreadMonitor::SetState tm("compress");
  tm.increment("compress");
  size_t compressedLength = compressBound(input.length());
  SmarterP< std::string > result(NULL, compressedLength, '\0');
  const int error = compress((unsigned char *)&(*result)[0],
			     &compressedLength,
			     (unsigned char const *)&input[0],
			     input.length());
  if (error)
  { // This should never happen.  Effectively an assertion failed.
    TclList msg;
    msg<<FLF<<"error";
    if (error == Z_MEM_ERROR)
      msg<<"Z_MEM_ERROR";
    else if (error == Z_BUF_ERROR)
      msg<<Z_BUF_ERROR;
    else
      msg<<error;
    msg<<"input.length()"<<input.length()
       <<"compressBound()"<<compressBound(input.length());
    scheduleSyncrhonizedShutdown();
  }
  result->resize(compressedLength);
  tm.increment("compress in", input.size());
  tm.increment("compress out", compressedLength);
  return result;
}

// Given the compressed data and the size of the original uncompressed data,
// recreate the original uncompressed data.  If there is any problem return
// the empty string and record the details in ThreadMonitor.
static std::string uncompressString(std::string const &compressed,
				    size_t uncompressedSize)
{
  ThreadMonitor::SetState tm("uncompress");
  tm.increment("uncompress");
  std::string result(uncompressedSize, '\0');
  size_t destLen = uncompressedSize;
  const int error = uncompress((unsigned char *)&result[0],
			       &destLen,
			       (unsigned char const *)&compressed[0],
			       compressed.length());
  switch (error)
  {
  case Z_OK:
    if (destLen != uncompressedSize)
    {
      tm.increment("uncompress size mismatch");
      return "";
    }
    else
      return result;
  case Z_MEM_ERROR: // if there was not enough memory, maybe an internal error?
    tm.increment("Z_MEM_ERROR");
    return "";
  case Z_BUF_ERROR: // if there was not enough room in the output buffer
    // So basically the same as uncompress size mismatch.
    tm.increment("Z_BUF_ERROR");
    return "";
  case Z_DATA_ERROR: // if the input data was corrupted or incomplete
    tm.increment("Z_DATA_ERROR");
    return "";
  default:
    tm.increment("unexpected uncompress error " + ntoa(error));
    return "";
  }
}


// You can uncompress data without knowing the uncompressedSize in advance.
// But it's easier if you know the size.  And in this context we know we can
// store the size in 2 bytes.  So every packet contains the compressed data
// followed by two bytes indicating the size of the original data.
static SmarterCP< std::string > compressPacket(std::string const &original)
{
  SmarterP< std::string > result = compressString(original);
  marshal(*result, (uint16_t)original.length());
  // It's tempting to skip the following line.  We'll use more memory
  // but save time.  Why are we bothering with smart pointers in the first
  // if not to avoid one extra copy?
  //result->shrink_to_fit();  
  return result;
}

// Returns true on success and puts the uncompressed data into packet.
// Returns false on failure and packet is in an unknown state.
static bool uncompressPacket(std::string &packet)
{  
  uint16_t size;
  try
  {
    mPop(packet, size);
  }
  catch (MarshallingException &ex)
  {
    return false;
  }
  packet = uncompressString(packet, size);
  if (packet.size() == size)
    return true;
  return false;
}

static std::string s_STATUS_TIME = "time";
static std::string s_STATUS_FILL_IN = "fill in";

static std::string s_ID = "id";


/////////////////////////////////////////////////////////////////////
// AlertRecordMultiCastSender
/////////////////////////////////////////////////////////////////////

void AlertRecordMultiCastSender::handleRequestInThread(Request *original)
{
  switch (original->callbackId)
  {
  case mtFillIn:
    {
      ThreadMonitor::SetState tm("mtFillIn");
      ExternalRequest *current = dynamic_cast< ExternalRequest * >(original);
      const int64_t id = strtolDefault(current->getProperty(s_ID), -1);
      if (PacketAndTime *packetAndtime = getProperty(_recentData, id))
      {
	addToOutputQueue(current->getSocketInfo(), packetAndtime->packet,
			 current->getResponseMessageId());
	tm.increment("mtFillIn good");
      }
      else
      {
	addToOutputQueue(current->getSocketInfo(), "",
			 current->getResponseMessageId());
	tm.increment("mtFillIn bad");
      }
      /*
	TclList msg;
	msg<<FLF<<"requested id"<<id;
	if (_recentData.empty())
	msg<<"empty";
	else
	msg<<"oldest"<<_recentData.begin()->first
	<<"newest"<<_recentData.rbegin()->first;
	sendToLogFile(msg);
      */
      // To test this code from the command line:
      // 1) Uncomment the logging code above.
      // 2) telnet 127.0.0.1 7899
      // 3) command=top_list_multicast_fill_in&message_id=12
      // 4) command=top_list_multicast_fill_in&id=294&message_id=12
      // 5) command=top_list_multicast_fill_in&id=304&message_id=12
      // I was watching the log file from this process while I was telnetting in.
      // The first time I didn't even specify an id and I saw this in the log:
      // {Sun Aug 16 19:59:35 2020} 0 AlertRecordMultiCast.C 345 handleRequestInThread {requested id} -1 oldest 292 newest 293
      // The second time I wasn't fast enough and I saw this:
      // {Sun Aug 16 20:00:11 2020} 0 AlertRecordMultiCast.C 345 handleRequestInThread {requested id} 294 oldest 296 newest 303
      // The third time I saw a good response on telnet and nothing in the log.
      break;
    }
  case mtDebug:
    {
      ThreadMonitor::SetState tm("mtDebug");
      ExternalRequest *current = dynamic_cast< ExternalRequest * >(original);
      TclList msg;
      msg<<FLF<<"mtDebug";
      
      const int skip = strtolDefault(current->getProperty("skip"), -1);
      if (skip >= 0)
      {
	msg<<TclList("old skip", _debugSkipCount, "new skip", skip);
	_debugSkipCount = skip;
      }
      else
	msg<<TclList("skip", _debugSkipCount);
      
      const std::string replay = current->getProperty("replay");
      if (!replay.empty())
      {
	std::vector< std::string > pieces = explode(",", replay);
	TclList r;
	r<<"replay";
	for (std::string const &piece : pieces)
	{
	  bool found = false;
	  if (const int64_t id = strtolDefault(piece, 0))
	    if (PacketAndTime *const packetAndTime =
		getProperty(_recentData, id))
	    {
	      _packetsToSend.push_back(packetAndTime->packet);
	      r<<id;
	      found = true;
	    }
	  if (!found)
	    r<<TclList("Not found", piece);	    
	}
	trySendNow();
	msg<<r;
      }
      
      // This is probably redundant, but I wanted to be sure.
      msg<<TclList("_lastPacketId", _lastPacketId);
      
      const int maxAge = strtolDefault(current->getProperty("max_age"), -1);
      if (maxAge >= 0)
      {
	msg<<TclList("_maxFillInAgeMicroSeconds",
		     "old", _maxFillInAgeMicroSeconds, "new", maxAge);
	_maxFillInAgeMicroSeconds = maxAge;
      }
      else
	msg<<TclList("_maxFillInAgeMicroSeconds", _maxFillInAgeMicroSeconds);
      
      TclList available;
      available<<"_recentData";
      if (_recentData.empty())	
	available<<"empty";
      else
      {
	const int64_t now = getMicroTime();
	auto const dump = [&](decltype(*_recentData.begin()) &kvp) {
	  const int64_t age = now - kvp.second.submitted;
	  return TclList("id", kvp.first, "age μs", age,
			 "expires in μs", _maxFillInAgeMicroSeconds - age);
	};
	available<<"oldest"<<dump(*_recentData.begin())
		 <<"newest"<<dump(*_recentData.rbegin());
      }
      msg<<available;
      
      // TODO should link this log message to the socket.
      // I.e not SimpleLogFile.h
      sendToLogFile(msg);
      addToOutputQueue(current->getSocketInfo(), msg,
		       current->getResponseMessageId());   
      break;

/* Here's a transcript of a recent debugging session.
 *
I started by logging into the forwarding server and setting max age to 1,000
seconds and telling it to skip the next 2,000 normal outgoing messages:

[phil@becca Logs]$ telnet joey-mousepad 7899
Trying 192.168.1.238...
Connected to joey-mousepad.
Escape character is '^]'.
command=alerts_multicast_debug_send&max_age=1000000000&skip=2000
== MESSAGE 0 ========== 299 ==
AlertRecordMultiCast.C 377 handleRequestInThread mtDebug {{old skip} 0 {new skip} 2000} {_lastPacketId 335} {_maxFillInAgeMicroSeconds old 15000000 new 1000000000} {_recentData oldest {id 334 {age μs} 10988091 {expires in μs} 989011909} newest {id 335 {age μs} 977975 {expires in μs} 999022025}}

I saw that the highest id number was 335, so I should send test packets
starting with 336 if I wanted to keep things in order.

I asked for a status update to see if new packets had arrived yet.  You can't
send packets just any time you want.  You can only send existing packets, so
you have to wait for some external event.

command=alerts_multicast_debug_send
== MESSAGE 0 ========== 264 ==
AlertRecordMultiCast.C 377 handleRequestInThread mtDebug {skip 1991} {_lastPacketId 344} {_maxFillInAgeMicroSeconds 1000000000} {_recentData oldest {id 334 {age μs} 56193974 {expires in μs} 943806026} newest {id 344 {age μs} 6202243 {expires in μs} 993797757}}

We have data up to 344, so there are 9 new packets that the slaves haven't seen
yet.  (Alternatively you could have looked at the skip count to see that the
server has skipped sending 9 packets.)  Send the next 4 packets, but slightly
out of order.

command=alerts_multicast_debug_send&replay=337,339,336,338
== MESSAGE 0 ========== 290 ==
AlertRecordMultiCast.C 377 handleRequestInThread mtDebug {skip 1968} {replay 337 339 336 338} {_lastPacketId 367} {_maxFillInAgeMicroSeconds 1000000000} {_recentData oldest {id 334 {age μs} 162487871 {expires in μs} 837512129} newest {id 367 {age μs} 1505373 {expires in μs} 998494627}}

The forwarding server shows success.  I was also doing a tail -f on a top
list slave server's log.  This is what I saw:

{Mon Aug 17 21:58:07 2020} 0 AlertRecordMultiCast.C 1084 onMessageReceived {WRONG PACKET ORDER} alerts_multicast expected 336 found 337 skipped 1
{Mon Aug 17 21:58:07 2020} 0 AlertRecordMultiCast.C 1084 onMessageReceived {WRONG PACKET ORDER} alerts_multicast expected 336 found 339 skipped 3

This is exactly what I expected.  Notice that the top list server received
packets in the wrong order, but it put them back in order before processing
them.  There were no warnings, that's how I knew it worked.

Finally I set skip to 0 to return to normal operations.  

command=alerts_multicast_debug_send&skip=0
== MESSAGE 0 ========== 284 ==
AlertRecordMultiCast.C 377 handleRequestInThread mtDebug {{old skip} 1964 {new skip} 0} {_lastPacketId 371} {_maxFillInAgeMicroSeconds 1000000000} {_recentData oldest {id 334 {age μs} 205732059 {expires in μs} 794267941} newest {id 371 {age μs} 5211941 {expires in μs} 994788059}}

Soon after I sent that command I saw the following in the top list server's
log file:

{Mon Aug 17 21:58:55 2020} 0 AlertRecordMultiCast.C 1084 onMessageReceived {WRONG PACKET ORDER} alerts_multicast expected 340 found 372 skipped 32
{Mon Aug 17 21:58:55 2020} 0 AlertRecordMultiCast.C 938 sendRequest {{command alerts_multicast_fill_in} {id 340}}
{Mon Aug 17 21:58:55 2020} 0 AlertRecordMultiCast.C 1294 beforeSleep sendFillInRequest(340) {after 1501μs}
{Mon Aug 17 21:58:55 2020} 0 AlertRecordMultiCast.C 938 sendRequest {{command alerts_multicast_fill_in} {id 341}}
{Mon Aug 17 21:58:55 2020} 0 AlertRecordMultiCast.C 1294 beforeSleep sendFillInRequest(341) {after 1501μs}
...
{Mon Aug 17 21:58:55 2020} 0 AlertRecordMultiCast.C 938 sendRequest {{command alerts_multicast_fill_in} {id 371}}
{Mon Aug 17 21:58:55 2020} 0 AlertRecordMultiCast.C 1294 beforeSleep sendFillInRequest(371) {after 1501μs}
{Mon Aug 17 21:58:55 2020} 0 ZLibMalloc.C 70 fixedMalloc reserving 5944 in alerts_multicast_parser
{Mon Aug 17 21:58:55 2020} 0 ZLibMalloc.C 70 fixedMalloc reserving 65544 in alerts_multicast_parser
{Mon Aug 17 21:58:55 2020} 0 ZLibMalloc.C 70 fixedMalloc reserving 7160 in alerts_multicast_parser
{Mon Aug 17 21:58:55 2020} 0 ZLibMalloc.C 70 fixedMalloc reserving 32776 in alerts_multicast_parser
{Mon Aug 17 21:58:55 2020} 0 AlertRecordMultiCast.C 977 onMessage {fill in received} 340 use
{Mon Aug 17 21:58:55 2020} 0 AlertRecordMultiCast.C 977 onMessage {fill in received} 341 use
...
{Mon Aug 17 21:58:55 2020} 0 AlertRecordMultiCast.C 977 onMessage {fill in received} 370 use
{Mon Aug 17 21:58:55 2020} 0 AlertRecordMultiCast.C 977 onMessage {fill in received} 371 use

In the previous test the top list server only had to reorder the packets I gave
it.  In this test there were missing packets so it had to request them from
the fill in server and merge them in with the live data.  As with the previous
test, everything went according to plan.  Everything was recovered and put
back into the right order.

By the way, this is what an error looks like.  I wasn't typing fast enough.  I 
fixed this by raising the max_age and trying again.

command=alerts_multicast_debug_send&replay=394,393,392,391,390
== MESSAGE 0 ========== 334 ==
AlertRecordMultiCast.C 377 handleRequestInThread mtDebug {skip 460} {replay 394 393 {{Not found} 392} {{Not found} 391} {{Not found} 390}} {_lastPacketId 397} {_maxFillInAgeMicroSeconds 1000000000} {_recentData oldest {id 393 {age μs} 42123006 {expires in μs} 957876994} newest {id 397 {age μs} 2656770 {expires in μs} 997343230}}

       */
    }
  default:
    abort();
  }
}

void AlertRecordMultiCastSender::initializeInThread()
{
  ThreadMonitor::find().add(this);
}

SmarterCP< std::string >
AlertRecordMultiCastSender::compress(std::string const &plaintext)
{ 
  return compressPacket(plaintext);
}

AlertRecordMultiCastSender::Sent
AlertRecordMultiCastSender::trySendNow(std::string const &packet) const
{
  ThreadMonitor::SetState tm("trySendNow");
  if (_fd == -1)
    return Sent::FailForGood;
  int nbytes =
    sendto(_fd, packet.c_str(), packet.length(), MSG_DONTWAIT,
	   (struct sockaddr*) &_addr, sizeof(_addr));
  if (nbytes >= 0)
  {
    tm.increment("sendto Success");
    return Sent::Success;
  }
  else if ((errno == EAGAIN) || (errno == EWOULDBLOCK) || (errno == EINTR))
  {
    tm.increment("sendto Retry");
    return Sent::ShouldRetry;
  }
  else
  {
    tm.increment("sendto Fail");
    return Sent::FailForGood;
  }
}

void AlertRecordMultiCastSender::trySendNow()
{
  const auto initial = _packetsToSend.size();
  if (initial)
  {
    while (!_packetsToSend.empty())
    {
      std::string const &packet = *_packetsToSend.front();
      const Sent sent = trySendNow(packet);
      if (sent == Sent::ShouldRetry)
	break;
      _packetsToSend.pop_front();
    }
    const auto sent = initial - _packetsToSend.size();
    std::string counter;
    switch (sent)
    {
    case 0:
      counter = "trySendNow 0";
      break;
    case 1:
      counter = "trySendNow 1";
      break;
    default:
      counter = "trySendNow " + ntoa(sent);
      break;
    }
    ThreadMonitor::find().increment(counter);
  }
}

void AlertRecordMultiCastSender::beforeSleep(IBeforeSleepCallbacks &callbacks)
{
  if (time(NULL) >= _nextLinkStatusTime)
    sendLinkStatusMessageNow();
  callbacks.wakeAtTime(_nextLinkStatusTime);
  trySendNow();
  if (_packetsToSend.empty() && !_packetBeingBuilt.empty())
  {
    ThreadMonitor::find().increment("packet partial");
    pushPacketBeingBuilt();
    trySendNow();
  }
  if (!_packetsToSend.empty())
    callbacks.waitForWrite(_fd);
}

void AlertRecordMultiCastSender::pushPacketBeingBuilt()
{
  auto const packet = compress(_packetBeingBuilt);
  _packetBeingBuilt.clear();
  if (_debugSkipCount)
  {
    _debugSkipCount--;
    ThreadMonitor::find().increment("pushPacketBeingBuilt() SKIP");
  }
  else
    _packetsToSend.push_back(packet);  
  const int64_t now = getMicroTime();
  _recentData.emplace(_lastPacketId, PacketAndTime(now, packet));
  const int64_t cutOff = now - _maxFillInAgeMicroSeconds;
  while (_recentData.begin()->second.submitted < cutOff)
    _recentData.erase(_recentData.begin());
}

void AlertRecordMultiCastSender::sendInThread(std::string const &bytes)
{
  static const size_t MAX_SINGLE_ITEM_SIZE = 7500;
  static const size_t PACKET_BREAK_SIZE = 10000;
  if (bytes.length() > MAX_SINGLE_ITEM_SIZE)
  {
    sendToLogFile(TclList()<<FLF<<getBaseName()<<"DROPPING crazy big request"
		  <<"MAX_SINGLE_ITEM_SIZE"<<MAX_SINGLE_ITEM_SIZE
		  <<"PACKET_BREAK_SIZE"<<PACKET_BREAK_SIZE
		  <<"bytes"<<bytes.length()
		  <<"_packetBeingBuilt"<<_packetBeingBuilt.length());
  }
  else
  {
    ThreadMonitor &tm = ThreadMonitor::find();
    if (bytes.length() + _packetBeingBuilt.length() > PACKET_BREAK_SIZE)
    {
      tm.increment("packet full");
      pushPacketBeingBuilt();
      if (const auto initialSize = _packetsToSend.size())
      { 
	trySendNow();
	// {pf Q progress} 244 {pf Q stuck} 2829 {pf new queue} 1 {pf new send} 194829
	// This is a sample of the data we see from the increment()'s below.
	// Most of the time we see only "pf new send", i.e. nothing queued,
	// everything processed immediately.  And when we do see something,
	// "pf new queue" is never higher than 3 and is typically 1.  I.e.
	// one burst of time between when we got behind and when we caught up.
	// The example above is from when the server first started.  When we
	// start with a Q, we typically see progress a little over 10% of the
	// time, so it's good that we are checking and sending data a little
	// sooner.
	const bool progress = _packetsToSend.size() < initialSize;
	if (initialSize == 1)
	  // The queue was empty before we received this packet.  We
	  // immediately tried to send the packet.
	  if (progress)
	    // There was only one item in the queue and we sent it.
	    tm.increment("pf new send");
	  else
	    // This new packet will sit in the queue.  This is the start
	    // of a delay.  We know this happens sometimes.
	    tm.increment("pf new queue");
	else
	  // There were already packets in the queue before we added a new
	  // one.  So at least one call to sendto() failed but we plan to
	  // retry.  Is there value in calling trySendNow() again,
	  // or should we have skipped it and waited for beforeSleep()?
	  // That's what the following two statuses are trying to prove.
	  if (progress)
	    // Good, we sent more to the o/s.  Data is getting out sooner.
	    tm.increment("pf Q progress");
	  else
	    // This last call to trySendNow() was unnecessary, and possibly
	    // wasteful.
	    tm.increment("pf Q stuck");
      }
    }
    else if (!_packetBeingBuilt.empty())
      tm.increment("packet share");
    if (_packetBeingBuilt.empty())
    { // Starting a new packet.  Print a short header.
      marshal(_packetBeingBuilt, _sessionId);
      _lastPacketId++;
      marshal(_packetBeingBuilt, _lastPacketId);
    }
    marshal(_packetBeingBuilt, bytes);
    // Request a wakup soon.  This call is advertised as cheap, so don't
    // overthink it.  Call this any time we get a valid request.
    requestBeforeSleep();
  }
}

void AlertRecordMultiCastSender::sendLinkStatusMessageNow()
{
  std::map< std::string, std::string > map;
  map[s_STATUS_TIME] = marshal(getMicroTime());
  if (!_fillInAddress.empty())
    map[s_STATUS_FILL_IN] = _fillInAddress;
  /*
  const std::string command = _baseName + "_fill_in";
  sendToLogFile(TclList()<<FLF<<"command"<<command);  // DO NOT COMMIT!
  map[s_STATUS_COMMAND] = _baseName + "_fill_in";
  */
  std::string status = marshal(map);  
  add(status, ItemType::LinkStatus);
  // TODO is there a way to mark this message as not interesting?  I don't
  // want someone to replay a bunch of heartbeats.
  sendInThread(status);
  _nextLinkStatusTime = time(NULL) + 10;  // Again in 10 seconds.
}

void AlertRecordMultiCastSender::send(Record::Ref const &record)
{
  std::string encoded = record->getEncoded();
  add(encoded, ItemType::Record);
  invokeIfRequired([=](){ sendInThread(encoded); });
}

void AlertRecordMultiCastSender::send(std::string status)
{
  add(status, ItemType::Status);
  invokeIfRequired([=](){ sendInThread(status); });
}

bool AlertRecordMultiCastSender::isOkay() const
{
  return _fd >= 0;
}

std::string AlertRecordMultiCastSender::_fillInAddress;

void AlertRecordMultiCastSender::setListenPort(int port)
{
  _fillInAddress = getShortHostName() + ':' + ntoa(port);
}

std::string AlertRecordMultiCastSender::getInfoForThreadMonitor()
{
  TclList msg;
  msg<<"_lastPacketId"<<_lastPacketId<<"_recentData"<<_recentData.size();
  return msg;
}

AlertRecordMultiCastSender::AlertRecordMultiCastSender
(std::string const &baseName, IContainerThread *thread) :
  AlertRecordMultiCastBase(baseName),
  ForeverThreadUser(thread),
  _fd(-1),
  _sessionId(getMicroTime()),
  _lastPacketId(0),
  _debugSkipCount(0),
  _maxFillInAgeMicroSeconds(15000000L),
  _nextLinkStatusTime(0)
{
  if (okay())
  {
    memset(&_addr, 0, sizeof(_addr));
    _addr.sin_family = AF_INET;
    _addr.sin_addr.s_addr = inet_addr(group());
    _addr.sin_port = htons(port());
    _fd = socket(AF_INET, SOCK_DGRAM, 0);
    if (_fd < 0)
    {
      TclList msg;
      msg<<FLF<<"ERROR"<<configKey()<<configInput()<<errorString();
      sendToLogFile(msg);
    }
    else
    {
      TclList msg;
      msg<<FLF<<"success"
	 <<"baseName"<<baseName<<"getBaseName()"<<getBaseName()
	 <<"configKey"<<configKey()<<"configInput"<<configInput()
	 <<"group"<<group()<<"port"<<port();
      sendToLogFile(msg);
      
      // Add something like
      // top_list_multicast_addr=239.0.2.0:2002
      // to the multicast info.  To say that we are the master
      // and we are broadcasting on this channel.
      //
      // MultiCast refers to the way we can request info about all
      // servers on the network.  I generally like to keep this
      // list short, but this is exactly the type of information
      // that is perfect for MultiCast.  If two different servers
      // are both posting to the same multicast channel, that's
      // bad.  You want to know what those machines are ASAP,
      // and there's no other easy way to get this info.
      MultiCast::getInstance().addInfo(configKey(), configInput());

      CommandDispatcher *cd = CommandDispatcher::getInstance();
      cd->listenForCommand(commandName(), this, mtFillIn);
      const std::string debugCommand = getBaseName() + "_debug_send";
      cd->listenForCommand(debugCommand, this, mtDebug);
      
      start();
    }
  }
}


/////////////////////////////////////////////////////////////////////
// AlertRecordMultiCastReceiver::ParsedPacket
/////////////////////////////////////////////////////////////////////

AlertRecordMultiCastReceiver::ParsedPacket::ParsedPacket
(std::string &bytes,
 AlertRecordMultiCastReceiver *owner) :
  _owner(NULL)
{
  ThreadMonitor::SetState tm("ParsedPacket");
  try
  {
    decompress(bytes);
    size_t offset = 0;
    unmarshal(bytes, offset, _sessionId);
    unmarshal(bytes, offset, _packetId);
    //TclList msg;
    //msg<<FLF<<"_sessionId"<<_sessionId<<"_packetId"<<_packetId<<"types";
    while (!unmarshalFinished(bytes, offset))
    {
      std::string item;
      unmarshal(bytes, offset, item);
      ItemType itemType = takeItemType(item);
      //msg<<(int)itemType;
      switch (itemType)
      {
      case ItemType::Status:
	//msg<<item;
	// About once a minute I get blasted with the same 7 status messages:
	// jules:1201, jules:1202, jules:1203, fogell:1202, fogell:1203,
	// fogell:1205, and fogell:1201
	_actions.push_back([=]() {
	    for (const auto &callback : owner->_statusListeners)
	      callback(item);
	  });
	break;
      case ItemType::Record:
	{
	  Record::Ref record = Record::create(item);
	  if (!record)
	    throw MarshallingException("Corrupt data (create).");
	  else
	    _actions.push_back([=]() {
		for (const auto &callback : owner->_recordListeners)
		  callback(record);
	      });
	  break;
	}
      case ItemType::LinkStatus:
	_actions.push_back([=]() {
	    owner->_linkStatus.update(item);
	  });
	break;
      default:
	// Interesting.  New item types will cause us to throw out the entire
	// packet.  It might make more sense to ignore any unknown item
	// types and assume that was part of a future upgrade.
	throw MarshallingException("Corrupt data (itemType).");
      }
    }
    //sendToLogFile(msg);
    // Success!
    _owner = owner;
  }
  catch (MarshallingException &ex)
  {
    sendToLogFile(TclList()<<FLF<<"MarshallingException"<<ex.what()<<"input length"<<bytes.length());
  }
}

void AlertRecordMultiCastReceiver::ParsedPacket::deliverPayload()
{
  assert(isOkay());

  // Someone else is responsible for putting packets in order.  This
  // is a good place to double check.
  if (_owner->_lastPacketIdDelivered + 1 != _packetId)
  {
    TclList msg;
    msg<<FLF;
    if (_packetId == 1)
      msg<<"session restarted";
    else if (_owner->_lastPacketIdDelivered == 0)
      msg<<"joining session in progress";
    else
      msg<<"DELIVERED OUT OF ORDER";
    msg<<"_lastPacketIdDelivered"<<_owner->_lastPacketIdDelivered
       <<"_packetId"<<_packetId;
    sendToLogFile(msg);
  }
  _owner->_lastPacketIdDelivered = _packetId;

  _owner = NULL;  // Don't call deliverPayload() twice!
  
  // Do everything at the end, only after we've successfully decoded
  // everything.  It's like a transaction.  Everything succeeds or
  // everything fails.  If the first half of the message looks good,
  // but the second half is corrupt, assume it's all corrupt.
  for (auto &action : _actions)
    action();
  ThreadMonitor::find().increment("action", _actions.size());
}


/////////////////////////////////////////////////////////////////////
// AlertRecordMultiCastReceiver::LinkStatus
/////////////////////////////////////////////////////////////////////

void AlertRecordMultiCastReceiver::LinkStatus::resetDelay()
{
  _delayCount = 0;
  _totalDelay = 0;
  maximize(_smallestDelay);
  minimize(_biggestDelay);
}

void AlertRecordMultiCastReceiver::LinkStatus::reset()
{
  _fillInAddress.clear();
  _owner._fillInConnection.clearAddress();
  resetDelay();  
}

AlertRecordMultiCastReceiver::LinkStatus::LinkStatus
(AlertRecordMultiCastReceiver &owner) :
  _owner(owner)
{
  reset();
}

void
AlertRecordMultiCastReceiver::LinkStatus::update(std::string const &message)
{
  // ThreadMonitor::find().increment("LinkStatus");
  try
  {
    std::map< std::string, std::string > map;
    unmarshal(message, map);
    std::string const &newFillInAddress = map[s_STATUS_FILL_IN];
    if (_fillInAddress != newFillInAddress)
    {
      sendToLogFile(TclList()<<FLF<<getBaseName()
		    <<"_fillInAddress"<<_fillInAddress
		    <<"newFillInAddress"<<newFillInAddress);
      // {Tue Aug 25 18:59:50 2020} 0 AlertRecordMultiCast.C 961 update alerts_multicast _fillInAddress {} newFillInAddress id-10-t:7897
      // {Tue Aug 25 18:59:51 2020} 0 AlertRecordMultiCast.C 961 update top_list_multicast _fillInAddress {} newFillInAddress id-10-t:7891
      _fillInAddress = newFillInAddress;
      _owner._fillInConnection.setAddress(_fillInAddress);
    }
    std::string marshalledTime = map[s_STATUS_TIME];
    int64_t remoteTime;
    unmarshal(marshalledTime, remoteTime);
    const int64_t localTime = getMicroTime();
    const int64_t delay = localTime - remoteTime;
    _delayCount++;
    _totalDelay += delay;
    _smallestDelay = std::min(_smallestDelay, delay);
    _biggestDelay = std::max(_biggestDelay, delay);
  }
  catch (MarshallingException const &ex)
  {
    ThreadMonitor::find().increment("LinkStatus::update ERROR");
  }
}

std::string AlertRecordMultiCastReceiver::LinkStatus::forThreadMonitor() 
{
  TclList result;
  if (_delayCount)
  {
    result<<"LinkStatus"<<"_smallestDelay"<<_smallestDelay
	  <<"average delay"<<(_totalDelay/_delayCount)
	  <<"_biggestDelay"<<_biggestDelay;
    resetDelay();
  }
  return result;
}


/////////////////////////////////////////////////////////////////////
// AlertRecordMultiCastReceiver::FillInConnection
/////////////////////////////////////////////////////////////////////

bool AlertRecordMultiCastReceiver::FillInConnection::shouldTryToConnect()
{ // _highestFillInRequested is initially 0, and set back to 0 any time we are
  // not waiting for old data.  It is set to the highest value we've
  // requested recently, so it will be non-0 if we've sent anything recently.
  return _owner->_highestFillInRequested
    && !(_serverName.empty() || _serverPort.empty());
}

AlertRecordMultiCastReceiver::FillInConnection::FillInConnection
(AlertRecordMultiCastReceiver *owner) :
  ServerConnection64(owner->commandName(), owner->getContainer()),
  _owner(owner)
{
}

void AlertRecordMultiCastReceiver::FillInConnection::setAddress
(std::string const &serverName, std::string const &serverPort)
{
  if ((serverName != _serverName) || (serverPort != _serverPort))
  {
    _serverName = serverName;
    _serverPort = serverPort;
    reset();
  }
}

void AlertRecordMultiCastReceiver::FillInConnection::setAddress
(std::string const &configString)
{ // TODO Why is this parsing code copied everywhere?  I got it from
  // ServerConnection64::getAndParseAddress() but I know that's not the only
  // other place it exists.
  const std::vector< std::string > pieces = explode(":", configString);
  const bool success = pieces.size() == 2;
  if (!success)
    clearAddress();
  else
    setAddress(pieces[0], pieces[1]);
}

void AlertRecordMultiCastReceiver::FillInConnection::clearAddress()
{
  setAddress("", "");
}

void AlertRecordMultiCastReceiver::FillInConnection::sendRequest(int64_t id)
{
  Message message;
  message["command"] = _owner->commandName();
  message[s_ID] = ntoa(id);
  sendMessage(message, _owner, id);
  //sendToLogFile(TclList()<<FLF<<message);
  // The previous line was redundant.  In the logs you would see
  // {Tue Aug 25 07:10:00 2020} 0 AlertRecordMultiCast.C 1030 sendRequest {{command alerts_multicast_fill_in} {id 1050746}}
  // (created here) followed by 
  // {Tue Aug 25 07:10:00 2020} 0 AlertRecordMultiCast.C 1430 beforeSleep sendFillInRequest id 1050746 {after μs} 3001 now 1598364600890954 flagged 1598364600887953
}

void AlertRecordMultiCastReceiver::FillInConnection::start()
{
  ForeverThreadUser::start();
}


/////////////////////////////////////////////////////////////////////
// AlertRecordMultiCastReceiver
/////////////////////////////////////////////////////////////////////

bool AlertRecordMultiCastReceiver::fillInIsStillRelevant(int64_t id)
{
  if (id <= _lastPacketIdDelivered)
    // We've already delivered this, or we skipped it and delivered a packet
    // from later in the stream.  We never go back.  This one was too slow so
    // it was skipped.
    return false;
  else if (id > _highestFillInRequested)
    // Unexpected.  Maybe from a previous session?
    return false;
  else if (_deferredPackets.count(id))
    // We already have a copy of this.  Perhaps we had given up on the
    // original, so we requested a duplicate, then the original finally
    // arrived, so the duplicate is no longer relevant.
    return false;
  else
    return true;
}

void AlertRecordMultiCastReceiver::onMessage
(std::string bytes,
 int64_t clientId,
 ServerConnection64::MessageId messageId)
{
  ThreadMonitor::SetState tm("fill in recv");
  const bool useIt = fillInIsStillRelevant(clientId);
  sendToLogFile(TclList()<<FLF<<"fill in received"<<getBaseName()
		<<"id"<<clientId
		<<"now"<<getMicroTime()
		<<(useIt?"use":"DISCARD"));  // TODO noisy
  if (useIt)
  {
    tm.increment("fill in found");
    // Process the packet almost identically to a packet we received from the
    // multicast group.
    onMessageReceived(bytes, clientId);
    // This is exactly the strange case requestBeforeSleep() was designed for.
    // Normally if we are in any callback but beforeSleep(), we know there
    // will be a beforeSleep() soon.  However, method is being called by
    // a method in a different ForeverThreadUser in the same thread.  We
    // don't know for sure, but it's possible that our beforeSleep() was
    // already called, and the other ContainerThreadUser is calling us from
    // his beforeSleep() call.  In that case we want our beforeSleep() to be
    // called a second time, sometime soon, sometime before we go to sleep!
    requestBeforeSleep();
  }
  else
    tm.increment("fill in skip");
}

void
AlertRecordMultiCastReceiver::onAbort(int64_t clientId,
				      ServerConnection64::MessageId messageId)
{
  ThreadMonitor::SetState tm("fill in abort");
  if (fillInIsStillRelevant(clientId))
  {
    tm.increment("fill in abort retry");
    _fillInConnection.sendRequest(clientId);    
  }
  else
    tm.increment("fill in abort abandon");
}

void AlertRecordMultiCastReceiver::initializeInThread()
{
  ThreadMonitor::find().add(this);
}

std::string AlertRecordMultiCastReceiver::getInfoForThreadMonitor()
{
  TclList result;
  result<<"_nextPacketId"<<_nextPacketId;
  const std::string linkStatus = _linkStatus.forThreadMonitor();
  if (!linkStatus.empty())
    result<<linkStatus;
  return result;
}

bool AlertRecordMultiCastReceiver::isOkay() const
{
  return _isOkay;
}

void AlertRecordMultiCastReceiver::decompress(std::string &data)
{
  const bool success = uncompressPacket(data);
  if (!success)
    throw MarshallingException("Corrupt data (decompress).");
}

// fillInId, the second argument, is the expected packet id.
// It defaults to 0, meaning we read this from the multicast stream,
// and we expect the new packet to have the next sequential id.
// Otherwise we know the packet is from the fill in server, and
// fillInId was the requested packet id.
// If the packet ids or session ids don't match for a fill in response,
// notify ThreadMonitor, otherwise ignore the message.
void AlertRecordMultiCastReceiver::onMessageReceived(std::string &bytes,
						     int64_t fillInId)
{
  ThreadMonitor::SetState tm("AlertRecordMultiCastReceiver::onMessageReceived");
  ParsedPacket::Ref parsed(NULL, bytes, this);
  if (!parsed->isOkay())
    // Note:  This is what we expect to happen if we ask for fill in data
    // which has already expired.  Of course, we probably won't be asking
    // for expired data, but we should not panic if we see this.
    return;
  auto const packetSessionId = parsed->getSessionId();
  auto const packetId = parsed->getPacketId();
  if (fillInId)
  {
    if (packetSessionId != _sessionId)
      // Fill in data from a previous session.  Toss it!
      return;
    if (packetId != fillInId)
      // Something strange.  We got a valid packet but not the one we were
      // expecting.  Looks like a bug.
      // TODO report this!
      return;
  }
  else if (packetSessionId != _sessionId)
  { // A multicast packet has a new session id, so we discard everything from
    // the previous session.
    _sessionId = packetSessionId;
    tm.increment("new session " + ntoa(_sessionId));
    _linkStatus.reset();
    _nextPacketId = 0; // Wildcard!
    _arrivalTime.clear();
    _highestFillInRequested = 0;  // No requests sent this server session.
    for (auto const &kvp : _deferredPackets)
      kvp.second->deliverPayload();
    _deferredPackets.clear();
  }
  else if (packetId != _nextPacketId)
  {
    if ((!_deferredPackets.empty())
	&& (_deferredPackets.rbegin()->first + 1 == packetId))
      // One or more packets went missing.  Then a bunch more came in a row.
      // For example:  1, 3, 4, 5, 6, 2, 7
      // When we see #3 we report lots of details with sendToLogFile().
      // We we see 4, 5 and 6 we just increment() this counter.
      tm.increment("WPO more");
    else
    {
      tm.increment("WPO log");
      TclList msg;
      msg<<FLF<<"WRONG PACKET ORDER"<<getBaseName()
	 <<"expected"<<_nextPacketId
	 <<"found"<<packetId;
      if (packetId > _nextPacketId)
	msg<<"skipped"<<(packetId - _nextPacketId);
      else
	msg<<"duplicated"<<(_nextPacketId - packetId);
      msg<<"_deferredPackets"<<_deferredPackets.size()
	 <<"_arrivalTime"<<_arrivalTime
	 <<"now"<<getMicroTime();
      sendToLogFile(msg);
    }
  }

  // At this point we have several possibilities.
  // 1) We just connected to a new master.  Immediately throw out any history
  //    (or process it) and process this packet.
  // 2) This is the packet we are expecting and there are is no history.
  //    Immediately process this packet.
  // 3) This packet id is lower than the next packet we were expecting, and
  //    is not a hole that we were hoping to fill.  I.e. this packet was
  //    unexpected.  Maybe a duplicate.  Maybe two packets were delivered
  //    out of order and they were the first two from a new master.
  //    Throw out the packet.  Log to to ThreadMonitor, but do nothing else.
  // 4) This packet id is higher than the next packet we were expecting.  I.e.
  //    we are missing some packets.  Save the packet to a list so we can
  //    process it later.  And make a note of the missing items.  And the time.
  // 5) This packet is on the list of missing items.  Remove the id number from
  //    the list of missing items.  Add the packet to the list of packets we
  //    plan to process soon.  Check if some or all of the saved packets can
  //    be sent now, i.e. there are no missing packets before these packets.
  //
  // On a timer we need to check on missing items.  Maybe send a request
  // to a fill in server.  Maybe abandon the request and processing the
  // items that we've saved.
  //
  // We need to keep track of the next expected packet id number.
  // o Normally that's simple.  We see that the current packet's id number
  //   is what we are expecting, so we ++ what we are expecting.
  // o New session is a wild card.  Just assume the current packet's id
  //   number is correct and next time we expect a value one higher.
  // o If we get a packet with an id number higher than expected, save
  //   that packet for later.  We'll have a std::map from id numbers to
  //   packets.  Keep the expect packed id number the same.
  // o If we get a packet with an id number lower than expected, toss it,
  //   log it, no change to the expected id number.
  // o If we have some saved packets, and we have a new packet with
  //   id >= what we expect, store that with the other packets.  We might
  //   have unblocked some messages, so check for that now or soon.
  //
  // In the map we also record the precise time when the packet arrived.
  // Based on how old the packet is we might want to (a) request another
  // copy of any packets blocking this packet or (b) skip the missing
  // packets and move on.  (If we see a missing packet our first thought
  // should be out of order, not dead.  Give it a little time before
  // asking for another copy.)
  //
  // If we are holding packets then we need to check on them periodically.
  // Check on them when we add a new packet, or when some time has
  // passed.  Here's the algorithm:
  // o Look at the lowest numbered packet we are holding and see if it
  //   matches the one we are looking for.
  // o If so, send it, remove it from the map, update the expected value,
  //   and jump to the top of the loop.
  //
  // Clear the list of packets when we start a new session.
  //
  // Maybe a different data structure to say when to request more data.
  // o If you are adding the first new packet, you need all the packets
  //   >= next expected packet id but < the id of the packet we just
  //   received.
  // o If you are adding a new packet, and we already have at least one
  //   packet with a higher id, ignore it.
  // o If you are adding a new packet, and it has the highest id value, and
  //   there is at least one missing id number between the previous highest
  //   id number
  // o When the timer goes off be sure to check again.  Some or all of
  //   the packets that were originally missing might have arrived.
  // o We need to keep a map from times to ranges.  (The lowest and
  //   highest ids that we might request.)  Right before we send the
  //   request we'll review that range and give a specific list of
  //   id numbers.
  // o Seems like this data structure would also be an efficient place to
  //   find the give up time, described above.  In fact, the map described
  //   above should only store the packets themselves, not the arrival
  //   time.

  if (packetId < _nextPacketId)
    tm.increment("duplicate ignored");
  else if (((packetId == _nextPacketId)
	    ||(_nextPacketId == 0))
	   && _deferredPackets.empty())
  { // This is an optimization for the common case.  
    _nextPacketId = packetId + 1;
    parsed->deliverPayload();
  }
  else if (_deferredPackets.count(packetId))
    tm.increment("duplicate ignored");
  else
  {
    tm.increment(_deferredPackets.empty()?"pause start":"pause more");
    _deferredPackets[packetId] = parsed;
    while ((!_deferredPackets.empty())
	   && (_deferredPackets.begin()->first == _nextPacketId))
    {
      _arrivalTime.erase(_nextPacketId);
      _deferredPackets.begin()->second->deliverPayload();
      _deferredPackets.erase(_deferredPackets.begin());
      _nextPacketId++;
      tm.increment("RO release"); // reorder
    }
    if ((!_deferredPackets.empty())
	&& (_deferredPackets.rbegin()->first == packetId))
    { // The new packet has the highest id number of any packet we have
      // deferred so far.
      bool skippedNewPackets;
      if (_deferredPackets.size() == 1)
	// The new packet is the only packet that was deferred.  So we know
	// we skipped something or we wouldn't have gotten here.
	skippedNewPackets = true;
      else
      {
	auto it = _deferredPackets.rbegin();
	it--;  // The entire right before the current packet.
	const auto previousPacketPresentId = it->first;
	// There is space between the current packet and the packet with
	// the second highest id.  So we are missing at least one packet.
	skippedNewPackets = previousPacketPresentId + 1 < packetId;
      }
      if (skippedNewPackets)
      { // The number of times, not the number of packets!
	// This is basically the number of holes in the data.
	tm.increment("skipped new packets");
	_arrivalTime[packetId] = getMicroTime();
      }
    }
  }
}

void
AlertRecordMultiCastReceiver::beforeSleep(IBeforeSleepCallbacks &callbacks)
{
  class ConfigInfo
  {
  public:
    int64_t requestFillInAfter;
    int64_t abandonAfter;
    ConfigInfo() :
      // getConfigItem: multicast_receiver_fill_in_after:
      // getConfigItem:   Wait this many μs after noticing a missing packet
      // getConfigItem:   before asking the server to send a second copy.
      // getConfigItem:   We don't do this immediately because sometimes
      // getConfigItem:   the packets come out of order.  You'll get packet
      // getConfigItem:   #1, then packet #3, then after a short pause, you
      // getConfigItem:   finally get packet #2.  If this number is too low
      // getConfigItem:   we send extra requests to the master server.  If this
      // getConfigItem:   number is too high we wait too long to send the
      // getConfigItem:   replacement request and lots of packets are all
      // getConfigItem:   held up a little longer.
      requestFillInAfter(strtolDefault(getConfigItem("multicast_receiver_fill_in_after"), 3000)),
      // getConfigItem: multicast_receiver_abandon_after:
      // getConfigItem:   Wait this many μs after noticing a missing packet
      // getConfigItem:   before giving up on it.  At this point we just
      // getConfigItem:   skip some data and process the rest in order.  We
      // getConfigItem:   never use the missing data, even if it eventually
      // getConfigItem:   shows up, because it would be out of order.
      // getConfigItem:   Make this too small and you might lose some data.
      // getConfigItem:   Make this too big and you might cause all the data
      // getConfigItem:   to be held up for a while.
      // getConfigItem:   The default is currently 1.5 seconds.  We keep
      // getConfigItem:   raising that value.  Becca missed two packets at
      // getConfigItem:   5 seconds after the close.  They came in after
      // getConfigItem:   1.135231 seconds, shortly after we'd given up
      // getConfigItem:   on them.
      abandonAfter(strtolDefault(getConfigItem("multicast_receiver_abandon_after"), 1500000))
    {
      const bool invalid = (requestFillInAfter < 0) || (abandonAfter < 0);
      TclList msg;
      msg<<FLF
	 <<"multicast_receiver_fill_in_after"<<requestFillInAfter<<(requestFillInAfter/1000000.0)
	 <<"multicast_receiver_abandon_after"<<abandonAfter<<(abandonAfter/1000000.0);
      if (invalid)
	msg<<"INVALID";
      sendToLogFile(msg);
      // {Tue Aug 25 18:56:52 2020} 0 AlertRecordMultiCast.C 1374 ConfigInfo multicast_receiver_fill_in_after 3000 0.003 multicast_receiver_abandon_after 1500000 1.5
      if (invalid)
	scheduleSyncrhonizedShutdown();
    }
  } static const configInfo;
    
  if (!_arrivalTime.empty())
  {
    ThreadMonitor &tm = ThreadMonitor::find();
    const auto now = getMicroTime();
    // If a packet arrived more than configInfo.abandonAfter microseconds ago,
    // we want to send it now.  We will abandon any missing packets that were
    // holding us up and just send the data that we have.
    const auto abandonCutOff = now - configInfo.abandonAfter;
    int64_t abandonCutOffId = 0;
    while (!_arrivalTime.empty())
    {
      auto const begin = _arrivalTime.begin();
      auto const time = begin->second;
      if (time > abandonCutOff)
	break;
      auto const id = begin->first;
      abandonCutOffId = id;
      _arrivalTime.erase(begin);
    }
    if (abandonCutOffId)
      while (!_deferredPackets.empty())
      {
	auto const begin = _deferredPackets.begin();
	auto const id = begin->first;
	if (id > abandonCutOffId)
	  break;
	begin->second->deliverPayload();
	_deferredPackets.erase(begin);
	_nextPacketId = id + 1;
	tm.increment("TO release 1"); // timeout
      }
    while ((!_deferredPackets.empty())
           && (_deferredPackets.begin()->first == _nextPacketId))
    {
      _arrivalTime.erase(_nextPacketId);
      _deferredPackets.begin()->second->deliverPayload();
      _deferredPackets.erase(_deferredPackets.begin());
      _nextPacketId++;
      tm.increment("TO release 2"); // time out
    }

    // See what's been missing long enough that we should send out a fill in
    // request.
    const auto requestFillInCutOff = now - configInfo.requestFillInAfter;
    int64_t requestFillInCutOffId = 0;
    for (auto &kvp : _arrivalTime)
    {
      int64_t const &id = kvp.first;
      int64_t const &time = kvp.second;
      if (time < requestFillInCutOff)
	// Everything requested before this id is old enough that we should
	// send a duplicate request, if we have not alreay.
	requestFillInCutOffId = id;
      else
      { // We're past all of the old requests.  We'll deal with this next
	// request soon.
	callbacks.wakeAfterMicroSeconds(time - requestFillInCutOff);
	break;
      }
    }
    if (_highestFillInRequested < requestFillInCutOffId)
    {
      // Should we set _highestFillInRequested = std::max(1, _highestFillInRequested) here?  Does FillInConnection::shouldTryToConnect() get called before we change _highestFillInRequested, below?
      for (int64_t id = std::max(_highestFillInRequested + 1, _nextPacketId);
	   id < requestFillInCutOffId;
	   id++)
	if (!_deferredPackets.count(id))
	{
	  _fillInConnection.sendRequest(id);
	  tm.increment("fill in req");
	  // TODO this next sendToLogFile() is very noisy!
	  const auto flagged = _arrivalTime[requestFillInCutOffId];
	  sendToLogFile(TclList()<<FLF<<"sendFillInRequest"<<"id"<<id
			<<"after μs"<<(now - flagged)
			<<"now"<<now
			<<"flagged"<<flagged);
	}
      // Next time start from here.  Don't send anything with an id <
      // requestFillInCutOffId because it's already been sent.  And while
      // we're at it, skip past requestFillInCutOffId, too.  That came from
      // a packet that we've already received, so we're not going to
      // request it again.
      _highestFillInRequested = requestFillInCutOffId;
    }
    
    if (!_arrivalTime.empty())
    { // Wake up when it is time to give up on a missing packet and process
      // the packets which come after that and we are already holding onto.
      auto const oldestTime = _arrivalTime.begin()->second;
      auto const wakeupTime = oldestTime + 10000;
      auto const usUntilWake = wakeupTime - getMicroTime();
      callbacks.wakeAfterMicroSeconds(usUntilWake);
    }
  }
  if (_arrivalTime.empty())
    // _arrivalTime.empty() => Nothing is queued up and we are not waiting for
    //                         any fill in response.
    // (!_arrivalTime.empty()) && _highestFillInRequested =>
    //    Something is queued up and we are waiting for at least one fill in
    //    response.
    // (!_arrivalTime.empty()) && (!_highestFillInRequested) =>
    //    Something is queued up but we have not yet have any outstanding
    //    fill in requests.
    _highestFillInRequested = 0;
}

void AlertRecordMultiCastReceiver::listen
(std::function< void(Record::Ref const &) > const &f)
{
  invokeIfRequired([=](){ _recordListeners.push_back(f); });
}

void AlertRecordMultiCastReceiver::listen
(std::function< void(std::string const &) > const &f)
{
  invokeIfRequired([=](){ _statusListeners.push_back(f); });
}

AlertRecordMultiCastReceiver::AlertRecordMultiCastReceiver
(std::string const &baseName) :
  AlertRecordMultiCastBase(baseName),
  ForeverThreadUser(IContainerThread::create(getBaseName() + "_parser")),
  _isOkay(false),
  _lastPacketIdDelivered(0),
  _sessionId(0),
  _nextPacketId(0),
  _highestFillInRequested(0),
  _fillInConnection(this),
  _linkStatus(*this)
{
  if(okay())
  {
    int fd = socket(AF_INET, SOCK_DGRAM, 0);
    if (fd < 0)
      sendToLogFile(TclList()<<FLF<<errorString()<<getBaseName()<<"socket()"<<"recv");
    else
    { // allow multiple sockets to use the same PORT number
      u_int yes = 1;
      if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
		     (char*) &yes, sizeof(yes)) < 0)
	sendToLogFile(TclList()<<FLF<<errorString()<<getBaseName()<<"SOL_SOCKETSOL_SOCKET"<<"SO_REUSEADDR");
      else
      { // set up destination address
	struct sockaddr_in addr;
	memset(&addr, 0, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_addr.s_addr = htonl(INADDR_ANY); // differs from sender
	addr.sin_port = htons(port());
	// bind to receive address
	if (bind(fd, (struct sockaddr*) &addr, sizeof(addr)) < 0)
	  sendToLogFile(TclList()<<FLF<<errorString()<<getBaseName()<<"bind()");
	else
	{ // use setsockopt() to request that the kernel join a multicast group
	  struct ip_mreq mreq;
	  mreq.imr_multiaddr.s_addr = inet_addr(group());
	  mreq.imr_interface.s_addr = htonl(INADDR_ANY);
	  if (setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
			 (char*) &mreq, sizeof(mreq)) < 0)
	    sendToLogFile(TclList()<<FLF<<errorString()<<getBaseName()
			  <<"setsockopt()"<<"IPPROTO_IP"<<"IP_ADD_MEMBERSHIP");
	  else
	  { // use setsockopt() to request only the requested group's
	    // messages show up.  By default if someone else connects
	    // to the same port but a different group, this socket
	    // would also get packets aimed at the other group.
	    // https://stackoverflow.com/a/20919920/971955
	    //
	    // Note:  If any process on the machine starting listening to a
	    // different multicast group with that port number, you will get
	    // cross talk.  The second subscription can come from a different
	    // process!  Previous comments speculated on this fact, and I
	    // have now confirmed that!
	    //
	    // Cross talk was causing the following messages in the log:
	    //{Tue Aug 11 11:44:25 2020} 0 AlertRecordMultiCast.C 638 operator() ItemType::LinkStatus placeholder
	    //
	    // Here's another thing I saw in the log.
	    // {Tue Aug 11 11:50:45 2020} 0 ThreadMonitor.C alerts_multicast_parser total_time 300 states {poll 94.61% ... counters {action 353533 {new session 1597170945561824} 16606 {new session 1597170945562102} 16606 
	    // From this you can see that we are switching between two
	    // session ids.  In the last 5 minutes we switched to each one
	    // of them 16606 times.
	    //
	    // In practice we know that the top list data would get filtered
	    // out from the alerts process.  It's still wasting RAM and CPU.
	    // But top list records have no alert type field.  And every
	    // alert request requires an alert type.  So the user won't see
	    // strange data, it will get filtered out before it gets to him.
	    int mc_all = 0;
	    if ((setsockopt(fd, IPPROTO_IP, IP_MULTICAST_ALL,
			    (void*) &mc_all, sizeof(mc_all))) < 0)
	      sendToLogFile(TclList()<<FLF<<errorString()<<getBaseName()
			    <<"setsockopt()"
			    <<"IPPROTO_IP"<<"IP_MULTICAST_ALL");
	    else
	    { // Success!
	      _isOkay = true;
	      TclList msg;
	      msg<<FLF<<"success"<<"baseName"<<baseName
		 <<"getBaseName()"<<getBaseName()
		 <<"configKey"<<configKey()
		 <<"configInput"<<configInput()
		 <<"group"<<group()<<"port"<<port();
	      sendToLogFile(msg);
	      // {Sat Aug  8 16:06:16 2020} 0 AlertRecordMultiCast.C 696 AlertRecordMultiCastReceiver success baseName alerts _baseName alerts_multicast configKey alerts_multicast_addr input 239.0.1.0:2000 group 239.0.1.0 port 2000
	      
	      // Create a second thread.  The listener thread does nothing but
	      // grab packets out of the air as quickly as it can and queue
	      // them for the parser thread.
	      IContainerThread *const listenerThread =
		IContainerThread::create(getBaseName() + "_listener");
	      
	      // We created the parser thread at the top of this constructor.
	      // I.e. getContainer() returns the parser thread.
	      //
	      // The parser thread is much more CPU intensive than the
	      // listener thread. The parser thread uncompresses the data,
	      // splits it into individual records, checks for missing or
	      // out of order packets, etc.  Except as noted, all private
	      // methods of AlertRecordMultiCastReceiver only run in this
	      // parser thread.
	      assert(getContainer()->getThreadName() ==
		     getBaseName() + "_parser");
	      
	      // The listener thread only knows about the parser thread
	      // via this callback.  The listener only deals in strings of
	      // bytes.  The listener does not try to interpret or change
	      // those bytes.
	      auto const callback = [=](SmarterP< std::string > data) {
		doWorkInThread([=]() {
		    onMessageReceived(*data);
		  });
	      };
	      
	      // And we start listening now.
	      new AlertRecordMultiCastListener(fd, callback, listenerThread);
	      start();
	      _fillInConnection.start();
	    }
	  }
	}
      }
    }
  }
}



/*
Test results based on the last 3,000 alerts from Friday night.  8/7/2020
(A bug / feature of a current setup, each time I restart my forwarding server it copies the last 3000 records from the server and forwards those.  I have outstanding TODO items to look at that, but for now that's just the fastest easiest way to get test data on the weekend.)
These were delivered in 322 packets.  0 packet loss.
uncompress:  0.029919855 seconds
AlertRecordMultiCastReceiver::onMessageReceived:  0.010738825 seconds -- each packet unmarshal and examine id numbers.  Loop through the individual messages, copy/extract the body of each individual record or status message, create and verify records, forward records and status messages to listeners .  Listeners can provide arbitrary callbacks; in fact RecordDispatcher is the only listener and it just sends a message to its own thread.
{X read} 0.0040441777 seconds total.
{X peek} 0.0012810036 seconds total.
{X null} 0.00013001421 seconds total, 130 microseconds total, 404 nanoseconds per call to ThreadMonitor::increment();  (plus other minor overhead)
Total bytes before compression:  3,029,473
Total bytes after compression:  1,111,450
compression rate:  63.3%  (100% is ideal, 0% is none, like gzip)

top list results were similar.  only 281 packets required for 3000 records.  
uncompress:  0.030547251 seconds
AlertRecordMultiCastReceiver::onMessageReceived:  0.008254791 seconds
{X read} 0.004672368 seconds total.
{X peek} 0.0009258363 seconds total.
{X null} 0.00011800197 seconds total, 118 microseconds total, 420 nanoseconds per call to ThreadMonitor::increment(); 
Total bytes before compression:  2,688,922
Total bytes after compression:  1,485,014
compression rate:  44.8%

The timing difference between the alerts and the top lists are all within the margin of error and don't mean much.  Looking at the logs for a similar process I see that the numbers for these two types of data are closer than the examples shown here.  I.e. the difference between one test run and the next is bigger than the difference between alerts data and top list data.

The difference in compression is real!  The alert server often generates several alerts from the same print all at once.  All but a few fields will be identical.
*/

/*
Here are the results from the same test after I broke 
AlertRecordMultiCastReceiver into two threads.  This is just the alerts from
just one instance of the server.  The results seem generally similar to before.
Presumably the diffrences are just measurement error.  

The listener thread shows some interesting results.  The "X process" phase
takes slightly more time than the reading.  And some of the normal overhead
functions are getting close.  I was looking at optimizing the system calls, but
my normal library functions seem to be a better place to look for improvements.

{Sun Aug  9 11:04:28 2020} 0 ThreadMonitor.C alerts_multicast_listener total_time 300.5 states {
poll 100% 
{X process}     0.001057%  - 0.003176285 seconds  (Send to alerts_multicast_parser)
{X read}        0.001044%  - 0.00313722 seconds
{X peek}        0.0004369% - 0.0013128845 seconds
{poll() finish} 0.0003248% - 0.000976024 seconds
counters {EAGAIN 353 {X count} 352} 
{RequestQueue {read count} 1 {average time μs} 44 {worst time μs} 44}

{Sun Aug  9 11:04:28 2020} 0 ThreadMonitor.C alerts_multicast_parser total_time 300.5 states {
poll 99.98% 
uncompress                                      0.01041%   - 0.03128205 seconds
AlertRecordMultiCastReceiver::onMessageReceived 0.004369%  - 0.013128845 seconds
{Read from queue}                               0.000769%  - 0.002310845 seconds
{poll() finish}                                 0.0002316% - 0.000695958 seconds
counters {action 3030 {new session 1596996108088761} 1 uncompress 352} 
{RequestQueue {read count} 354 {average time μs} 33.0056497175141 {worst time μs} 740}
*/