/*
* Copyright (c) 2015 ARM Limited
* All rights reserved
*
* The license below extends only to copyright in the software and shall
* not be construed as granting a license to any other intellectual
* property including but not limited to intellectual property relating
* to a hardware implementation of the functionality of the software
* licensed hereunder. You may use the software subject to the license
* terms below provided that you ensure that this notice is replicated
* unmodified and in its entirety in all distributions of the software,
* modified or unmodified, in source code or in binary form.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met: redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer;
* redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution;
* neither the name of the copyright holders nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
* Authors: Gabor Dozsa
*/
/* @file
* The interface class for multi gem5 simulations.
*/
#include "dev/multi_iface.hh"
#include <queue>
#include <thread>
#include "base/random.hh"
#include "base/trace.hh"
#include "debug/MultiEthernet.hh"
#include "debug/MultiEthernetPkt.hh"
#include "dev/etherpkt.hh"
#include "sim/sim_exit.hh"
#include "sim/sim_object.hh"
MultiIface::Sync *MultiIface::sync = nullptr;
MultiIface::SyncEvent *MultiIface::syncEvent = nullptr;
unsigned MultiIface::recvThreadsNum = 0;
MultiIface *MultiIface::master = nullptr;
bool
MultiIface::Sync::run(SyncTrigger t, Tick sync_tick)
{
std::unique_lock<std::mutex> sync_lock(lock);
trigger = t;
if (trigger != SyncTrigger::periodic) {
DPRINTF(MultiEthernet,"MultiIface::Sync::run() trigger:%d\n",
(unsigned)trigger);
}
switch (state) {
case SyncState::asyncCkpt:
switch (trigger) {
case SyncTrigger::ckpt:
assert(MultiIface::syncEvent->interrupted == false);
state = SyncState::busy;
break;
case SyncTrigger::periodic:
if (waitNum == 0) {
// So all recv threads got an async checkpoint request already
// and a simExit is scheduled at the end of the current tick
// (i.e. it is a periodic sync scheduled at the same tick as
// the simExit).
state = SyncState::idle;
DPRINTF(MultiEthernet,"MultiIface::Sync::run() interrupted "
"due to async ckpt scheduled\n");
return false;
} else {
// we still need to wait for some receiver thread to get the
// aysnc ckpt request. We are going to proceed as 'interrupted'
// periodic sync.
state = SyncState::interrupted;
DPRINTF(MultiEthernet,"MultiIface::Sync::run() interrupted "
"due to ckpt request is coming in\n");
}
break;
case SyncTrigger::atomic:
assert(trigger != SyncTrigger::atomic);
}
break;
case SyncState::idle:
state = SyncState::busy;
break;
// Only one sync can be active at any time
case SyncState::interrupted:
case SyncState::busy:
assert(state != SyncState::interrupted);
assert(state != SyncState::busy);
break;
}
// Kick-off the sync unless we are in the middle of an interrupted
// periodic sync
if (state != SyncState::interrupted) {
assert(waitNum == 0);
waitNum = MultiIface::recvThreadsNum;
// initiate the global synchronisation
assert(MultiIface::master != nullptr);
MultiIface::master->syncRaw(triggerToMsg[(unsigned)trigger], sync_tick);
}
// now wait until all receiver threads complete the synchronisation
auto lf = [this]{ return waitNum == 0; };
cv.wait(sync_lock, lf);
// we are done
assert(state == SyncState::busy || state == SyncState::interrupted);
bool ret = (state != SyncState::interrupted);
state = SyncState::idle;
return ret;
}
void
MultiIface::Sync::progress(MsgType msg)
{
std::unique_lock<std::mutex> sync_lock(lock);
switch (msg) {
case MsgType::cmdAtomicSyncAck:
assert(state == SyncState::busy && trigger == SyncTrigger::atomic);
break;
case MsgType::cmdPeriodicSyncAck:
assert(state == SyncState::busy && trigger == SyncTrigger::periodic);
break;
case MsgType::cmdCkptSyncAck:
assert(state == SyncState::busy && trigger == SyncTrigger::ckpt);
break;
case MsgType::cmdCkptSyncReq:
switch (state) {
case SyncState::busy:
if (trigger == SyncTrigger::ckpt) {
// We are already in a checkpoint sync but got another ckpt
// sync request. This may happen if two (or more) peer gem5
// processes try to start a ckpt nearly at the same time.
// Incrementing waitNum here (before decrementing it below)
// effectively results in ignoring this new ckpt sync request.
waitNum++;
break;
}
assert (waitNum == recvThreadsNum);
state = SyncState::interrupted;
// we need to fall over here to handle "recvThreadsNum == 1" case
case SyncState::interrupted:
assert(trigger == SyncTrigger::periodic);
assert(waitNum >= 1);
if (waitNum == 1) {
exitSimLoop("checkpoint");
}
break;
case SyncState::idle:
// There is no on-going sync so we got an async ckpt request. If we
// are the only receiver thread then we need to schedule the
// checkpoint. Otherwise, only change the state to 'asyncCkpt' and
// let the last receiver thread to schedule the checkpoint at the
// 'asyncCkpt' case.
// Note that a periodic or resume sync may start later and that can
// trigger a state change to 'interrupted' (so the checkpoint may
// get scheduled at 'interrupted' case finally).
assert(waitNum == 0);
state = SyncState::asyncCkpt;
waitNum = MultiIface::recvThreadsNum;
// we need to fall over here to handle "recvThreadsNum == 1" case
case SyncState::asyncCkpt:
assert(waitNum >= 1);
if (waitNum == 1)
exitSimLoop("checkpoint");
break;
default:
panic("Unexpected state for checkpoint request message");
break;
}
break;
default:
panic("Unknown msg type");
break;
}
waitNum--;
assert(state != SyncState::idle);
// Notify the simultaion thread if there is an on-going sync.
if (state != SyncState::asyncCkpt) {
sync_lock.unlock();
cv.notify_one();
}
}
void MultiIface::SyncEvent::start(Tick start, Tick interval)
{
assert(!scheduled());
if (interval == 0)
panic("Multi synchronisation period must be greater than zero");
repeat = interval;
schedule(start);
}
void
MultiIface::SyncEvent::adjust(Tick start_tick, Tick repeat_tick)
{
// The new multi interface may require earlier start of the
// synchronisation.
assert(scheduled() == true);
if (start_tick < when())
reschedule(start_tick);
// The new multi interface may require more frequent synchronisation.
if (repeat == 0)
panic("Multi synchronisation period must be greater than zero");
if (repeat < repeat_tick)
repeat = repeat_tick;
}
void
MultiIface::SyncEvent::process()
{
/*
* Note that this is a global event so this process method will be called
* by only exactly one thread.
*/
// if we are draining the system then we must not start a periodic sync (as
// it is not sure that all peer gem5 will reach this tick before taking
// the checkpoint).
if (isDraining == true) {
assert(interrupted == false);
interrupted = true;
DPRINTF(MultiEthernet,"MultiIface::SyncEvent::process() interrupted "
"due to draining\n");
return;
}
if (interrupted == false)
scheduledAt = curTick();
/*
* We hold the eventq lock at this point but the receiver thread may
* need the lock to schedule new recv events while waiting for the
* multi sync to complete.
* Note that the other simulation threads also release their eventq
* locks while waiting for us due to the global event semantics.
*/
curEventQueue()->unlock();
// we do a global sync here
interrupted = !MultiIface::sync->run(SyncTrigger::periodic, scheduledAt);
// Global sync completed or got interrupted.
// we are expected to exit with the eventq lock held
curEventQueue()->lock();
// schedule the next global sync event if this one completed. Otherwise
// (i.e. this one was interrupted by a checkpoint request), we will
// reschedule this one after the draining is complete.
if (!interrupted)
schedule(scheduledAt + repeat);
}
void MultiIface::SyncEvent::resume()
{
Tick sync_tick;
assert(!scheduled());
if (interrupted) {
assert(curTick() >= scheduledAt);
// We have to complete the interrupted periodic sync asap.
// Note that this sync might be interrupted now again with a checkpoint
// request from a peer gem5...
sync_tick = curTick();
schedule(sync_tick);
} else {
// So we completed the last periodic sync, let's find out the tick for
// next one
assert(curTick() > scheduledAt);
sync_tick = scheduledAt + repeat;
if (sync_tick < curTick())
panic("Cannot resume periodic synchronisation");
schedule(sync_tick);
}
DPRINTF(MultiEthernet,
"MultiIface::SyncEvent periodic sync resumed at %lld "
"(curTick:%lld)\n", sync_tick, curTick());
}
void MultiIface::SyncEvent::serialize(const std::string &base,
CheckpointOut &cp) const
{
// Save the periodic multi sync schedule information
paramOut(cp, base + ".periodicSyncRepeat", repeat);
paramOut(cp, base + ".periodicSyncInterrupted", interrupted);
paramOut(cp, base + ".periodicSyncAt", scheduledAt);
}
void MultiIface::SyncEvent::unserialize(const std::string &base,
CheckpointIn &cp)
{
paramIn(cp, base + ".periodicSyncRepeat", repeat);
paramIn(cp, base + ".periodicSyncInterrupted", interrupted);
paramIn(cp, base + ".periodicSyncAt", scheduledAt);
}
MultiIface::MultiIface(unsigned multi_rank,
Tick sync_start,
Tick sync_repeat,
EventManager *em) :
syncStart(sync_start), syncRepeat(sync_repeat),
recvThread(nullptr), eventManager(em), recvDone(nullptr),
scheduledRecvPacket(nullptr), linkDelay(0), rank(multi_rank)
{
DPRINTF(MultiEthernet, "MultiIface() ctor rank:%d\n",multi_rank);
if (master == nullptr) {
assert(sync == nullptr);
assert(syncEvent == nullptr);
sync = new Sync();
syncEvent = new SyncEvent();
master = this;
}
}
MultiIface::~MultiIface()
{
assert(recvThread);
delete recvThread;
if (this == master) {
assert(syncEvent);
delete syncEvent;
assert(sync);
delete sync;
}
}
void
MultiIface::packetOut(EthPacketPtr pkt, Tick send_delay)
{
MultiHeaderPkt::Header header_pkt;
unsigned address_length = MultiHeaderPkt::maxAddressLength();
// Prepare a multi header packet for the Ethernet packet we want to
// send out.
header_pkt.msgType = MsgType::dataDescriptor;
header_pkt.sendTick = curTick();
header_pkt.sendDelay = send_delay;
// Store also the source and destination addresses.
pkt->packAddress(header_pkt.srcAddress, header_pkt.dstAddress,
address_length);
header_pkt.dataPacketLength = pkt->size();
// Send out the multi hedare packet followed by the Ethernet packet.
sendRaw(&header_pkt, sizeof(header_pkt), header_pkt.dstAddress);
sendRaw(pkt->data, pkt->size(), header_pkt.dstAddress);
DPRINTF(MultiEthernetPkt,
"MultiIface::sendDataPacket() done size:%d send_delay:%llu "
"src:0x%02x%02x%02x%02x%02x%02x "
"dst:0x%02x%02x%02x%02x%02x%02x\n",
pkt->size(), send_delay,
header_pkt.srcAddress[0], header_pkt.srcAddress[1],
header_pkt.srcAddress[2], header_pkt.srcAddress[3],
header_pkt.srcAddress[4], header_pkt.srcAddress[5],
header_pkt.dstAddress[0], header_pkt.dstAddress[1],
header_pkt.dstAddress[2], header_pkt.dstAddress[3],
header_pkt.dstAddress[4], header_pkt.dstAddress[5]);
}
bool
MultiIface::recvHeader(MultiHeaderPkt::Header &header_pkt)
{
// Blocking receive of an incoming multi header packet.
return recvRaw((void *)&header_pkt, sizeof(header_pkt));
}
void
MultiIface::recvData(const MultiHeaderPkt::Header &header_pkt)
{
// We are here beacuse a header packet has been received implying
// that an Ethernet (data) packet is coming in next.
assert(header_pkt.msgType == MsgType::dataDescriptor);
// Allocate storage for the incoming Ethernet packet.
EthPacketPtr new_packet(new EthPacketData(header_pkt.dataPacketLength));
// Now execute the blocking receive and store the incoming data directly
// in the new EthPacketData object.
if (! recvRaw((void *)(new_packet->data), header_pkt.dataPacketLength))
panic("Missing data packet");
new_packet->length = header_pkt.dataPacketLength;
// Grab the event queue lock to schedule a new receive event for the
// data packet.
curEventQueue()->lock();
// Compute the receive tick. It includes the send delay and the
// simulated link delay.
Tick recv_tick = header_pkt.sendTick + header_pkt.sendDelay + linkDelay;
DPRINTF(MultiEthernetPkt, "MultiIface::recvThread() packet receive, "
"send_tick:%llu send_delay:%llu link_delay:%llu recv_tick:%llu\n",
header_pkt.sendTick, header_pkt.sendDelay, linkDelay, recv_tick);
if (recv_tick <= curTick()) {
panic("Simulators out of sync - missed packet receive by %llu ticks",
curTick() - recv_tick);
}
// Now we are about to schedule a recvDone event for the new data packet.
// We use the same recvDone object for all incoming data packets. If
// that is already scheduled - i.e. a receive event for a previous
// data packet is already pending - then we have to check whether the
// receive tick for the new packet is earlier than that of the currently
// pending event. Packets may arrive out-of-order with respect to
// simulated receive time. If that is the case, we need to re-schedule the
// recvDone event for the new packet. Otherwise, we save the packet
// pointer and the recv tick for the new packet in the recvQueue. See
// the implementation of the packetIn() method for comments on how this
// information is retrieved from the recvQueue by the simulation thread.
if (!recvDone->scheduled()) {
assert(recvQueue.size() == 0);
assert(scheduledRecvPacket == nullptr);
scheduledRecvPacket = new_packet;
eventManager->schedule(recvDone, recv_tick);
} else if (recvDone->when() > recv_tick) {
recvQueue.emplace(scheduledRecvPacket, recvDone->when());
eventManager->reschedule(recvDone, recv_tick);
scheduledRecvPacket = new_packet;
} else {
recvQueue.emplace(new_packet, recv_tick);
}
curEventQueue()->unlock();
}
void
MultiIface::recvThreadFunc()
{
EthPacketPtr new_packet;
MultiHeaderPkt::Header header;
// The new receiver thread shares the event queue with the simulation
// thread (associated with the simulated Ethernet link).
curEventQueue(eventManager->eventQueue());
// Main loop to wait for and process any incoming message.
for (;;) {
// recvHeader() blocks until the next multi header packet comes in.
if (!recvHeader(header)) {
// We lost connection to the peer gem5 processes most likely
// because one of them called m5 exit. So we stop here.
exit_message("info", 0, "Message server closed connection, "
"simulation is exiting");
}
// We got a valid multi header packet, let's process it
if (header.msgType == MsgType::dataDescriptor) {
recvData(header);
} else {
// everything else must be synchronisation related command
sync->progress(header.msgType);
}
}
}
EthPacketPtr
MultiIface::packetIn()
{
// We are called within the process() method of the recvDone event. We
// return the packet that triggered the current receive event.
// If there is further packets in the recvQueue, we also have to schedule
// the recvEvent for the next packet with the smallest receive tick.
// The priority queue container ensures that smallest receive tick is
// always on the top of the queue.
assert(scheduledRecvPacket != nullptr);
EthPacketPtr next_packet = scheduledRecvPacket;
if (! recvQueue.empty()) {
eventManager->schedule(recvDone, recvQueue.top().second);
scheduledRecvPacket = recvQueue.top().first;
recvQueue.pop();
} else {
scheduledRecvPacket = nullptr;
}
return next_packet;
}
void
MultiIface::spawnRecvThread(Event *recv_done, Tick link_delay)
{
assert(recvThread == nullptr);
// all receive thread must be spawned before simulation starts
assert(eventManager->eventQueue()->getCurTick() == 0);
recvDone = recv_done;
linkDelay = link_delay;
recvThread = new std::thread(&MultiIface::recvThreadFunc, this);
recvThreadsNum++;
}
DrainState
MultiIface::drain()
{
DPRINTF(MultiEthernet,"MultiIFace::drain() called\n");
// This can be called multiple times in the same drain cycle.
if (master == this) {
syncEvent->isDraining = true;
}
return DrainState::Drained;
}
void MultiIface::drainDone() {
if (master == this) {
assert(syncEvent->isDraining == true);
syncEvent->isDraining = false;
// We need to resume the interrupted periodic sync here now that the
// draining is done. If the last periodic sync completed before the
// checkpoint then the next one is already scheduled.
if (syncEvent->interrupted)
syncEvent->resume();
}
}
void MultiIface::serialize(const std::string &base, CheckpointOut &cp) const
{
// Drain the multi interface before the checkpoint is taken. We cannot call
// this as part of the normal drain cycle because this multi sync has to be
// called exactly once after the system is fully drained.
// Note that every peer will take a checkpoint but they may take it at
// different ticks.
// This sync request may interrupt an on-going periodic sync in some peers.
sync->run(SyncTrigger::ckpt, curTick());
// Save the periodic multi sync status
syncEvent->serialize(base, cp);
unsigned n_rx_packets = recvQueue.size();
if (scheduledRecvPacket != nullptr)
n_rx_packets++;
paramOut(cp, base + ".nRxPackets", n_rx_packets);
if (n_rx_packets > 0) {
assert(recvDone->scheduled());
scheduledRecvPacket->serialize(base + ".rxPacket[0]", cp);
}
for (unsigned i=1; i < n_rx_packets; i++) {
const RecvInfo recv_info = recvQueue.impl().at(i-1);
recv_info.first->serialize(base + csprintf(".rxPacket[%d]", i), cp);
Tick rx_tick = recv_info.second;
paramOut(cp, base + csprintf(".rxTick[%d]", i), rx_tick);
}
}
void MultiIface::unserialize(const std::string &base, CheckpointIn &cp)
{
assert(recvQueue.size() == 0);
assert(scheduledRecvPacket == nullptr);
assert(recvDone->scheduled() == false);
// restore periodic sync info
syncEvent->unserialize(base, cp);
unsigned n_rx_packets;
paramIn(cp, base + ".nRxPackets", n_rx_packets);
if (n_rx_packets > 0) {
scheduledRecvPacket = std::make_shared<EthPacketData>(16384);
scheduledRecvPacket->unserialize(base + ".rxPacket[0]", cp);
// Note: receive event will be scheduled when the link is unserialized
}
for (unsigned i=1; i < n_rx_packets; i++) {
EthPacketPtr rx_packet = std::make_shared<EthPacketData>(16384);
rx_packet->unserialize(base + csprintf(".rxPacket[%d]", i), cp);
Tick rx_tick = 0;
paramIn(cp, base + csprintf(".rxTick[%d]", i), rx_tick);
assert(rx_tick > 0);
recvQueue.emplace(rx_packet,rx_tick);
}
}
void MultiIface::initRandom()
{
// Initialize the seed for random generator to avoid the same sequence
// in all gem5 peer processes
assert(master != nullptr);
if (this == master)
random_mt.init(5489 * (rank+1) + 257);
}
void MultiIface::startPeriodicSync()
{
DPRINTF(MultiEthernet, "MultiIface:::initPeriodicSync started\n");
// Do a global sync here to ensure that peer gem5 processes are around
// (actually this may not be needed...)
sync->run(SyncTrigger::atomic, curTick());
// Start the periodic sync if it is a fresh simulation from scratch
if (curTick() == 0) {
if (this == master) {
syncEvent->start(syncStart, syncRepeat);
inform("Multi synchronisation activated: start at %lld, "
"repeat at every %lld ticks.\n",
syncStart, syncRepeat);
} else {
// In case another multiIface object requires different schedule
// for periodic sync than the master does.
syncEvent->adjust(syncStart, syncRepeat);
}
} else {
// Schedule the next periodic sync if resuming from a checkpoint
if (this == master)
syncEvent->resume();
}
DPRINTF(MultiEthernet, "MultiIface::initPeriodicSync done\n");
}