<commit_before>#include <string.h>

#include "transport.h"
#include "core.h"
#include "network.h"
#include "log.h"

using namespace std;

// FIXME: Rework netCreateBuf and netExPacket. We don't need to
// duplicate the sender/receiver info the packet. This should be known
// by the transport layer and given to us. We also should be more
// intelligent about the time stamps, right now the method is very
// ugly.

Network::Network(Core *core)
      : _core(core)
{
   LOG_ASSERT_ERROR(sizeof(g_type_to_static_network_map) / sizeof(EStaticNetwork) == NUM_PACKET_TYPES,
                    "Static network type map has incorrect number of entries.");

   _numMod = Config::getSingleton()->getTotalCores();
   _tid = _core->getId();

   _transport = Transport::getSingleton()->createNode(_core->getId());

   _callbacks = new NetworkCallback [NUM_PACKET_TYPES];
   _callbackObjs = new void* [NUM_PACKET_TYPES];
   for (SInt32 i = 0; i < NUM_PACKET_TYPES; i++)
      _callbacks[i] = NULL;

   UInt32 modelTypes[NUM_STATIC_NETWORKS];
   Config::getSingleton()->getNetworkModels(modelTypes);

   for (SInt32 i = 0; i < NUM_STATIC_NETWORKS; i++)
      _models[i] = NetworkModel::createModel(this, modelTypes[i], (EStaticNetwork)i);

   LOG_PRINT("Initialized.");
}

Network::~Network()
{
   for (SInt32 i = 0; i < NUM_STATIC_NETWORKS; i++)
      delete _models[i];

   delete [] _callbackObjs;
   delete [] _callbacks;

   delete _transport;

   LOG_PRINT("Destroyed.");
}

void Network::registerCallback(PacketType type, NetworkCallback callback, void *obj)
{
   assert((UInt32)type < NUM_PACKET_TYPES);
   _callbacks[type] = callback;
   _callbackObjs[type] = obj;
}

void Network::unregisterCallback(PacketType type)
{
   assert((UInt32)type < NUM_PACKET_TYPES);
   _callbacks[type] = NULL;
}

void Network::outputSummary(std::ostream &out) const
{
   out << "Network summary:\n";
   for (UInt32 i = 0; i < NUM_STATIC_NETWORKS; i++)
   {
      out << "  Network model " << i << ":\n";
      _models[i]->outputSummary(out);
   }
}

// Polling function that performs background activities, such as
// pulling from the physical transport layer and routing packets to
// the appropriate queues.

void Network::netPullFromTransport()
{
   do
   {
      LOG_PRINT("Entering netPullFromTransport");

      NetPacket packet(_transport->recv());
 
      LOG_PRINT("Pull packet : type %i, from %i, time %llu", (SInt32)packet.type, packet.sender, packet.time);
      assert(0 <= packet.sender && packet.sender < _numMod);
      LOG_ASSERT_ERROR(0 <= packet.type && packet.type < NUM_PACKET_TYPES, "Packet type: %d not between 0 and %d", packet.type, NUM_PACKET_TYPES);

      // was this packet sent to us, or should it just be forwarded?
      if (packet.receiver != _core->getId())
      {
         // Disable this feature now. None of the network models use it
         LOG_PRINT("Forwarding packet : type %i, from %i, to %i, core_id %i, time %llu.", 
               (SInt32)packet.type, packet.sender, packet.receiver, _core->getId(), packet.time);
         forwardPacket(packet);

         // if this isn't a broadcast message, then we shouldn't process it further
         if (packet.receiver != NetPacket::BROADCAST)
            continue;
      }

      // asynchronous I/O support
      NetworkCallback callback = _callbacks[packet.type];

      if (callback != NULL)
      {
         LOG_PRINT("Executing callback on packet : type %i, from %i, to %i, core_id %i, time %llu", 
               (SInt32)packet.type, packet.sender, packet.receiver, _core->getId(), packet.time);
         assert(0 <= packet.sender && packet.sender < _numMod);
         assert(0 <= packet.type && packet.type < NUM_PACKET_TYPES);

         callback(_callbackObjs[packet.type], packet);

         if (packet.length > 0)
            delete [] (Byte*) packet.data;
      }

      // synchronous I/O support
      else
      {
         LOG_PRINT("Enqueuing packet : type %i, from %i, to %i, core_id %i, time %llu.", 
               (SInt32)packet.type, packet.sender, packet.receiver, _core->getId(), packet.time);
         _netQueueLock.acquire();
         _netQueue.push_back(packet);
         _netQueueLock.release();
         _netQueueCond.broadcast();
      }
   }
   while (_transport->query());
}

// FIXME: Can forwardPacket be subsumed by netSend?

void Network::forwardPacket(const NetPacket &packet)
{
   netSend(packet);
}

SInt32 Network::netSend(const NetPacket& packet)
{
   assert(packet.type >= 0 && packet.type < NUM_PACKET_TYPES);

   NetworkModel *model = _models[g_type_to_static_network_map[packet.type]];

   vector<NetworkModel::Hop> hopVec;
   model->routePacket(packet, hopVec);

   Byte *buffer = packet.makeBuffer();

   for (UInt32 i = 0; i < hopVec.size(); i++)
   {
      LOG_PRINT("Send packet : type %i, from %i, to %i, next_hop %i, core_id %i, time %llu", 
            (SInt32) packet.type, packet.sender, hopVec[i].final_dest, hopVec[i].next_dest, _core->getId(), hopVec[i].time);
      LOG_ASSERT_ERROR(hopVec[i].time >= packet.time, "hopVec[%d].time(%llu) < packet.time(%llu)", i, hopVec[i].time, packet.time);

      NetPacket* buff_pkt = (NetPacket*) buffer;
      buff_pkt->time = hopVec[i].time;
      buff_pkt->receiver = hopVec[i].final_dest;

      _transport->send(hopVec[i].next_dest, buffer, packet.bufferSize());
      
      LOG_PRINT("Sent packet");
   }

   delete [] buffer;


   return packet.length;
}

// Stupid helper class to eliminate special cases for empty
// sender/type vectors in a NetMatch
class NetRecvIterator
{
   public:
      NetRecvIterator(UInt32 i)
            : _mode(INT)
            , _max(i)
            , _i(0)
      {
      }
      NetRecvIterator(const std::vector<SInt32> &v)
            : _mode(SENDER_VECTOR)
            , _senders(&v)
            , _i(0)
      {
      }
      NetRecvIterator(const std::vector<PacketType> &v)
            : _mode(TYPE_VECTOR)
            , _types(&v)
            , _i(0)
      {
      }

      inline UInt32 get()
      {
         switch (_mode)
         {
         case INT:
            return _i;
         case SENDER_VECTOR:
            return (UInt32)_senders->at(_i);
         case TYPE_VECTOR:
            return (UInt32)_types->at(_i);
         default:
            assert(false);
            return (UInt32)-1;
         };
      }

      inline Boolean done()
      {
         switch (_mode)
         {
         case INT:
            return _i >= _max;
         case SENDER_VECTOR:
            return _i >= _senders->size();
         case TYPE_VECTOR:
            return _i >= _types->size();
         default:
            assert(false);
            return true;
         };
      }

      inline void next()
      {
         ++_i;
      }

      inline void reset()
      {
         _i = 0;
      }

   private:
      enum
      {
         INT, SENDER_VECTOR, TYPE_VECTOR
      } _mode;

      union
      {
         UInt32 _max;
         const std::vector<SInt32> *_senders;
         const std::vector<PacketType> *_types;
      };

      UInt32 _i;
};

NetPacket Network::netRecv(const NetMatch &match)
{
   LOG_PRINT("Entering netRecv.");

   // Track via iterator to minimize copying
   NetQueue::iterator itr;
   Boolean found;

   found = false;

   NetRecvIterator sender = match.senders.empty()
                            ? NetRecvIterator(_numMod)
                            : NetRecvIterator(match.senders);

   NetRecvIterator type = match.types.empty()
                          ? NetRecvIterator((UInt32)NUM_PACKET_TYPES)
                          : NetRecvIterator(match.types);

   LOG_ASSERT_ERROR(_core && _core->getPerformanceModel(),
                    "Core and/or performance model not initialized.");
   UInt64 start_time = _core->getPerformanceModel()->getCycleCount();

   _netQueueLock.acquire();

   while (!found)
   {
      itr = _netQueue.end();

      // check every entry in the queue
      for (NetQueue::iterator i = _netQueue.begin();
            i != _netQueue.end();
            i++)
      {
         // only find packets that match
         for (sender.reset(); !sender.done(); sender.next())
         {
            if (i->sender != (SInt32)sender.get())
               continue;

            for (type.reset(); !type.done(); type.next())
            {
               if (i->type != (PacketType)type.get())
                  continue;

               found = true;

               // find the earliest packet
               if (itr == _netQueue.end() ||
                   itr->time > i->time)
               {
                  itr = i;
               }
            }
         }
      }

      // go to sleep until a packet arrives if none have been found
      if (!found)
      {
         _netQueueCond.wait(_netQueueLock);
      }
   }

   assert(found == true && itr != _netQueue.end());
   assert(0 <= itr->sender && itr->sender < _numMod);
   assert(0 <= itr->type && itr->type < NUM_PACKET_TYPES);
   assert(itr->receiver == _core->getId());

   // Copy result
   NetPacket packet = *itr;
   _netQueue.erase(itr);
   _netQueueLock.release();

   LOG_PRINT("packet.time(%llu), start_time(%llu)", packet.time, start_time);<commit_msg>[network] Augmented netRecv() with a BROADCAST<commit_after>#include <string.h>

#include "transport.h"
#include "core.h"
#include "network.h"
#include "log.h"

using namespace std;

// FIXME: Rework netCreateBuf and netExPacket. We don't need to
// duplicate the sender/receiver info the packet. This should be known
// by the transport layer and given to us. We also should be more
// intelligent about the time stamps, right now the method is very
// ugly.

Network::Network(Core *core)
      : _core(core)
{
   LOG_ASSERT_ERROR(sizeof(g_type_to_static_network_map) / sizeof(EStaticNetwork) == NUM_PACKET_TYPES,
                    "Static network type map has incorrect number of entries.");

   _numMod = Config::getSingleton()->getTotalCores();
   _tid = _core->getId();

   _transport = Transport::getSingleton()->createNode(_core->getId());

   _callbacks = new NetworkCallback [NUM_PACKET_TYPES];
   _callbackObjs = new void* [NUM_PACKET_TYPES];
   for (SInt32 i = 0; i < NUM_PACKET_TYPES; i++)
      _callbacks[i] = NULL;

   UInt32 modelTypes[NUM_STATIC_NETWORKS];
   Config::getSingleton()->getNetworkModels(modelTypes);

   for (SInt32 i = 0; i < NUM_STATIC_NETWORKS; i++)
      _models[i] = NetworkModel::createModel(this, modelTypes[i], (EStaticNetwork)i);

   LOG_PRINT("Initialized.");
}

Network::~Network()
{
   for (SInt32 i = 0; i < NUM_STATIC_NETWORKS; i++)
      delete _models[i];

   delete [] _callbackObjs;
   delete [] _callbacks;

   delete _transport;

   LOG_PRINT("Destroyed.");
}

void Network::registerCallback(PacketType type, NetworkCallback callback, void *obj)
{
   assert((UInt32)type < NUM_PACKET_TYPES);
   _callbacks[type] = callback;
   _callbackObjs[type] = obj;
}

void Network::unregisterCallback(PacketType type)
{
   assert((UInt32)type < NUM_PACKET_TYPES);
   _callbacks[type] = NULL;
}

void Network::outputSummary(std::ostream &out) const
{
   out << "Network summary:\n";
   for (UInt32 i = 0; i < NUM_STATIC_NETWORKS; i++)
   {
      out << "  Network model " << i << ":\n";
      _models[i]->outputSummary(out);
   }
}

// Polling function that performs background activities, such as
// pulling from the physical transport layer and routing packets to
// the appropriate queues.

void Network::netPullFromTransport()
{
   do
   {
      LOG_PRINT("Entering netPullFromTransport");

      NetPacket packet(_transport->recv());
 
      LOG_PRINT("Pull packet : type %i, from %i, time %llu", (SInt32)packet.type, packet.sender, packet.time);
      assert(0 <= packet.sender && packet.sender < _numMod);
      LOG_ASSERT_ERROR(0 <= packet.type && packet.type < NUM_PACKET_TYPES, "Packet type: %d not between 0 and %d", packet.type, NUM_PACKET_TYPES);

      // was this packet sent to us, or should it just be forwarded?
      if (packet.receiver != _core->getId())
      {
         // Disable this feature now. None of the network models use it
         LOG_PRINT("Forwarding packet : type %i, from %i, to %i, core_id %i, time %llu.", 
               (SInt32)packet.type, packet.sender, packet.receiver, _core->getId(), packet.time);
         forwardPacket(packet);

         // if this isn't a broadcast message, then we shouldn't process it further
         if (packet.receiver != NetPacket::BROADCAST)
            continue;
      }

      // asynchronous I/O support
      NetworkCallback callback = _callbacks[packet.type];

      if (callback != NULL)
      {
         LOG_PRINT("Executing callback on packet : type %i, from %i, to %i, core_id %i, time %llu", 
               (SInt32)packet.type, packet.sender, packet.receiver, _core->getId(), packet.time);
         assert(0 <= packet.sender && packet.sender < _numMod);
         assert(0 <= packet.type && packet.type < NUM_PACKET_TYPES);

         callback(_callbackObjs[packet.type], packet);

         if (packet.length > 0)
            delete [] (Byte*) packet.data;
      }

      // synchronous I/O support
      else
      {
         LOG_PRINT("Enqueuing packet : type %i, from %i, to %i, core_id %i, time %llu.", 
               (SInt32)packet.type, packet.sender, packet.receiver, _core->getId(), packet.time);
         _netQueueLock.acquire();
         _netQueue.push_back(packet);
         _netQueueLock.release();
         _netQueueCond.broadcast();
      }
   }
   while (_transport->query());
}

// FIXME: Can forwardPacket be subsumed by netSend?

void Network::forwardPacket(const NetPacket &packet)
{
   netSend(packet);
}

SInt32 Network::netSend(const NetPacket& packet)
{
   assert(packet.type >= 0 && packet.type < NUM_PACKET_TYPES);

   NetworkModel *model = _models[g_type_to_static_network_map[packet.type]];

   vector<NetworkModel::Hop> hopVec;
   model->routePacket(packet, hopVec);

   Byte *buffer = packet.makeBuffer();

   for (UInt32 i = 0; i < hopVec.size(); i++)
   {
      LOG_PRINT("Send packet : type %i, from %i, to %i, next_hop %i, core_id %i, time %llu", 
            (SInt32) packet.type, packet.sender, hopVec[i].final_dest, hopVec[i].next_dest, _core->getId(), hopVec[i].time);
      LOG_ASSERT_ERROR(hopVec[i].time >= packet.time, "hopVec[%d].time(%llu) < packet.time(%llu)", i, hopVec[i].time, packet.time);

      NetPacket* buff_pkt = (NetPacket*) buffer;
      buff_pkt->time = hopVec[i].time;
      buff_pkt->receiver = hopVec[i].final_dest;

      _transport->send(hopVec[i].next_dest, buffer, packet.bufferSize());
      
      LOG_PRINT("Sent packet");
   }

   delete [] buffer;


   return packet.length;
}

// Stupid helper class to eliminate special cases for empty
// sender/type vectors in a NetMatch
class NetRecvIterator
{
   public:
      NetRecvIterator(UInt32 i)
            : _mode(INT)
            , _max(i)
            , _i(0)
      {
      }
      NetRecvIterator(const std::vector<SInt32> &v)
            : _mode(SENDER_VECTOR)
            , _senders(&v)
            , _i(0)
      {
      }
      NetRecvIterator(const std::vector<PacketType> &v)
            : _mode(TYPE_VECTOR)
            , _types(&v)
            , _i(0)
      {
      }

      inline UInt32 get()
      {
         switch (_mode)
         {
         case INT:
            return _i;
         case SENDER_VECTOR:
            return (UInt32)_senders->at(_i);
         case TYPE_VECTOR:
            return (UInt32)_types->at(_i);
         default:
            assert(false);
            return (UInt32)-1;
         };
      }

      inline Boolean done()
      {
         switch (_mode)
         {
         case INT:
            return _i >= _max;
         case SENDER_VECTOR:
            return _i >= _senders->size();
         case TYPE_VECTOR:
            return _i >= _types->size();
         default:
            assert(false);
            return true;
         };
      }

      inline void next()
      {
         ++_i;
      }

      inline void reset()
      {
         _i = 0;
      }

   private:
      enum
      {
         INT, SENDER_VECTOR, TYPE_VECTOR
      } _mode;

      union
      {
         UInt32 _max;
         const std::vector<SInt32> *_senders;
         const std::vector<PacketType> *_types;
      };

      UInt32 _i;
};

NetPacket Network::netRecv(const NetMatch &match)
{
   LOG_PRINT("Entering netRecv.");

   // Track via iterator to minimize copying
   NetQueue::iterator itr;
   Boolean found;

   found = false;

   NetRecvIterator sender = match.senders.empty()
                            ? NetRecvIterator(_numMod)
                            : NetRecvIterator(match.senders);

   NetRecvIterator type = match.types.empty()
                          ? NetRecvIterator((UInt32)NUM_PACKET_TYPES)
                          : NetRecvIterator(match.types);

   LOG_ASSERT_ERROR(_core && _core->getPerformanceModel(),
                    "Core and/or performance model not initialized.");
   UInt64 start_time = _core->getPerformanceModel()->getCycleCount();

   _netQueueLock.acquire();

   while (!found)
   {
      itr = _netQueue.end();

      // check every entry in the queue
      for (NetQueue::iterator i = _netQueue.begin();
            i != _netQueue.end();
            i++)
      {
         // only find packets that match
         for (sender.reset(); !sender.done(); sender.next())
         {
            if (i->sender != (SInt32)sender.get())
               continue;

            for (type.reset(); !type.done(); type.next())
            {
               if (i->type != (PacketType)type.get())
                  continue;

               found = true;

               // find the earliest packet
               if (itr == _netQueue.end() ||
                   itr->time > i->time)
               {
                  itr = i;
               }
            }
         }
      }

      // go to sleep until a packet arrives if none have been found
      if (!found)
      {
         _netQueueCond.wait(_netQueueLock);
      }
   }

   assert(found == true && itr != _netQueue.end());
   assert(0 <= itr->sender && itr->sender < _numMod);
   assert(0 <= itr->type && itr->type < NUM_PACKET_TYPES);
   assert((itr->receiver == _core->getId()) || (itr->receiver == NetPacket::BROADCAST));

   // Copy result
   NetPacket packet = *itr;
   _netQueue.erase(itr);
   _netQueueLock.release();

   LOG_PRINT("packet.time(%llu), start_time(%llu)", packet.time, start_time);
{'fraction_non_alphanumeric': 0.12860863902756675, 'fraction_numerical': 0.007705665291947037, 'mean_word_length': 2.3080236941303176, 'pattern_counts': {'":': 2, '<': 73, '<?xml version=': 0, '>': 98, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5664025', 'n_tokens_mistral': 5919, 'n_tokens_neox': 5733, 'n_words': 1352}