/* * 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 dist-gem5 simulations. */ #include "dev/net/dist_iface.hh" #include #include #include "base/random.hh" #include "base/trace.hh" #include "debug/DistEthernet.hh" #include "debug/DistEthernetPkt.hh" #include "dev/net/etherpkt.hh" #include "sim/sim_exit.hh" #include "sim/sim_object.hh" using namespace std; DistIface::Sync *DistIface::sync = nullptr; DistIface::SyncEvent *DistIface::syncEvent = nullptr; unsigned DistIface::distIfaceNum = 0; unsigned DistIface::recvThreadsNum = 0; DistIface *DistIface::master = nullptr; void DistIface::Sync::init(Tick start_tick, Tick repeat_tick) { if (start_tick < firstAt) { firstAt = start_tick; inform("Next dist synchronisation tick is changed to %lu.\n", nextAt); } if (repeat_tick == 0) panic("Dist synchronisation interval must be greater than zero"); if (repeat_tick < nextRepeat) { nextRepeat = repeat_tick; inform("Dist synchronisation interval is changed to %lu.\n", nextRepeat); } } DistIface::SyncSwitch::SyncSwitch(int num_nodes) { numNodes = num_nodes; waitNum = num_nodes; numExitReq = 0; numCkptReq = 0; doExit = false; doCkpt = false; firstAt = std::numeric_limits::max(); nextAt = 0; nextRepeat = std::numeric_limits::max(); } DistIface::SyncNode::SyncNode() { waitNum = 0; needExit = ReqType::none; needCkpt = ReqType::none; doExit = false; doCkpt = false; firstAt = std::numeric_limits::max(); nextAt = 0; nextRepeat = std::numeric_limits::max(); } void DistIface::SyncNode::run(bool same_tick) { std::unique_lock sync_lock(lock); Header header; assert(waitNum == 0); waitNum = DistIface::recvThreadsNum; // initiate the global synchronisation header.msgType = MsgType::cmdSyncReq; header.sendTick = curTick(); header.syncRepeat = nextRepeat; header.needCkpt = needCkpt; if (needCkpt != ReqType::none) needCkpt = ReqType::pending; header.needExit = needExit; if (needExit != ReqType::none) needExit = ReqType::pending; DistIface::master->sendCmd(header); // now wait until all receiver threads complete the synchronisation auto lf = [this]{ return waitNum == 0; }; cv.wait(sync_lock, lf); // global synchronisation is done assert(!same_tick || (nextAt == curTick())); } void DistIface::SyncSwitch::run(bool same_tick) { std::unique_lock sync_lock(lock); Header header; // Wait for the sync requests from the nodes if (waitNum > 0) { auto lf = [this]{ return waitNum == 0; }; cv.wait(sync_lock, lf); } assert(waitNum == 0); assert(!same_tick || (nextAt == curTick())); waitNum = numNodes; // Complete the global synchronisation header.msgType = MsgType::cmdSyncAck; header.sendTick = nextAt; header.syncRepeat = nextRepeat; if (doCkpt || numCkptReq == numNodes) { doCkpt = true; header.needCkpt = ReqType::immediate; numCkptReq = 0; } else { header.needCkpt = ReqType::none; } if (doExit || numExitReq == numNodes) { doExit = true; header.needExit = ReqType::immediate; } else { header.needExit = ReqType::none; } DistIface::master->sendCmd(header); } void DistIface::SyncSwitch::progress(Tick send_tick, Tick sync_repeat, ReqType need_ckpt, ReqType need_exit) { std::unique_lock sync_lock(lock); assert(waitNum > 0); if (send_tick > nextAt) nextAt = send_tick; if (nextRepeat > sync_repeat) nextRepeat = sync_repeat; if (need_ckpt == ReqType::collective) numCkptReq++; else if (need_ckpt == ReqType::immediate) doCkpt = true; if (need_exit == ReqType::collective) numExitReq++; else if (need_exit == ReqType::immediate) doExit = true; waitNum--; // Notify the simulation thread if the on-going sync is complete if (waitNum == 0) { sync_lock.unlock(); cv.notify_one(); } } void DistIface::SyncNode::progress(Tick max_send_tick, Tick next_repeat, ReqType do_ckpt, ReqType do_exit) { std::unique_lock sync_lock(lock); assert(waitNum > 0); nextAt = max_send_tick; nextRepeat = next_repeat; doCkpt = (do_ckpt != ReqType::none); doExit = (do_exit != ReqType::none); waitNum--; // Notify the simulation thread if the on-going sync is complete if (waitNum == 0) { sync_lock.unlock(); cv.notify_one(); } } void DistIface::SyncNode::requestCkpt(ReqType req) { std::lock_guard sync_lock(lock); assert(req != ReqType::none); if (needCkpt != ReqType::none) warn("Ckpt requested multiple times (req:%d)\n", static_cast(req)); if (needCkpt == ReqType::none || req == ReqType::immediate) needCkpt = req; } void DistIface::SyncNode::requestExit(ReqType req) { std::lock_guard sync_lock(lock); assert(req != ReqType::none); if (needExit != ReqType::none) warn("Exit requested multiple times (req:%d)\n", static_cast(req)); if (needExit == ReqType::none || req == ReqType::immediate) needExit = req; } void DistIface::Sync::drainComplete() { if (doCkpt) { // The first DistIface object called this right before writing the // checkpoint. We need to drain the underlying physical network here. // Note that other gem5 peers may enter this barrier at different // ticks due to draining. run(false); // Only the "first" DistIface object has to perform the sync doCkpt = false; } } void DistIface::SyncNode::serialize(CheckpointOut &cp) const { int need_exit = static_cast(needExit); SERIALIZE_SCALAR(need_exit); } void DistIface::SyncNode::unserialize(CheckpointIn &cp) { int need_exit; UNSERIALIZE_SCALAR(need_exit); needExit = static_cast(need_exit); } void DistIface::SyncSwitch::serialize(CheckpointOut &cp) const { SERIALIZE_SCALAR(numExitReq); } void DistIface::SyncSwitch::unserialize(CheckpointIn &cp) { UNSERIALIZE_SCALAR(numExitReq); } void DistIface::SyncEvent::start() { // Note that this may be called either from startup() or drainResume() // At this point, all DistIface objects has already called Sync::init() so // we have a local minimum of the start tick and repeat for the periodic // sync. Tick firstAt = DistIface::sync->firstAt; repeat = DistIface::sync->nextRepeat; // Do a global barrier to agree on a common repeat value (the smallest // one from all participating nodes DistIface::sync->run(curTick() == 0); assert(!DistIface::sync->doCkpt); assert(!DistIface::sync->doExit); assert(DistIface::sync->nextAt >= curTick()); assert(DistIface::sync->nextRepeat <= repeat); // if this is called at tick 0 then we use the config start param otherwise // the maximum of the current tick of all participating nodes if (curTick() == 0) { assert(!scheduled()); assert(DistIface::sync->nextAt == 0); schedule(firstAt); } else { if (scheduled()) reschedule(DistIface::sync->nextAt); else schedule(DistIface::sync->nextAt); } inform("Dist sync scheduled at %lu and repeats %lu\n", when(), DistIface::sync->nextRepeat); } void DistIface::SyncEvent::process() { // We may not start a global periodic sync while draining before taking a // checkpoint. This is due to the possibility that peer gem5 processes // may not hit the same periodic sync before they complete draining and // that would make this periodic sync clash with sync called from // DistIface::serialize() by other gem5 processes. // We would need a 'distributed drain' solution to eliminate this // restriction. // Note that if draining was not triggered by checkpointing then we are // fine since no extra global sync will happen (i.e. all peer gem5 will // hit this periodic sync eventually). panic_if(_draining && DistIface::sync->doCkpt, "Distributed sync is hit while draining"); /* * Note that this is a global event so this process method will be called * by only exactly one thread. */ /* * 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 * dist sync to complete. * Note that the other simulation threads also release their eventq * locks while waiting for us due to the global event semantics. */ { EventQueue::ScopedRelease sr(curEventQueue()); // we do a global sync here that is supposed to happen at the same // tick in all gem5 peers DistIface::sync->run(true); // global sync completed } if (DistIface::sync->doCkpt) exitSimLoop("checkpoint"); if (DistIface::sync->doExit) exitSimLoop("exit request from gem5 peers"); // schedule the next periodic sync repeat = DistIface::sync->nextRepeat; schedule(curTick() + repeat); } void DistIface::RecvScheduler::init(Event *recv_done, Tick link_delay) { // This is called from the receiver thread when it starts running. The new // receiver thread shares the event queue with the simulation thread // (associated with the simulated Ethernet link). curEventQueue(eventManager->eventQueue()); recvDone = recv_done; linkDelay = link_delay; } Tick DistIface::RecvScheduler::calcReceiveTick(Tick send_tick, Tick send_delay, Tick prev_recv_tick) { Tick recv_tick = send_tick + send_delay + linkDelay; // sanity check (we need atleast a send delay long window) assert(recv_tick >= prev_recv_tick + send_delay); panic_if(prev_recv_tick + send_delay > recv_tick, "Receive window is smaller than send delay"); panic_if(recv_tick <= curTick(), "Simulators out of sync - missed packet receive by %llu ticks" "(rev_recv_tick: %lu send_tick: %lu send_delay: %lu " "linkDelay: %lu )", curTick() - recv_tick, prev_recv_tick, send_tick, send_delay, linkDelay); return recv_tick; } void DistIface::RecvScheduler::resumeRecvTicks() { // Schedule pending packets asap in case link speed/delay changed when // restoring from the checkpoint. // This may be done during unserialize except that curTick() is unknown // so we call this during drainResume(). // If we are not restoring from a checkppint then link latency could not // change so we just return. if (!ckptRestore) return; std::vector v; while (!descQueue.empty()) { Desc d = descQueue.front(); descQueue.pop(); d.sendTick = curTick(); d.sendDelay = d.packet->size(); // assume 1 tick/byte max link speed v.push_back(d); } for (auto &d : v) descQueue.push(d); if (recvDone->scheduled()) { assert(!descQueue.empty()); eventManager->reschedule(recvDone, curTick()); } else { assert(descQueue.empty() && v.empty()); } ckptRestore = false; } void DistIface::RecvScheduler::pushPacket(EthPacketPtr new_packet, Tick send_tick, Tick send_delay) { // Note : this is called from the receiver thread curEventQueue()->lock(); Tick recv_tick = calcReceiveTick(send_tick, send_delay, prevRecvTick); DPRINTF(DistEthernetPkt, "DistIface::recvScheduler::pushPacket " "send_tick:%llu send_delay:%llu link_delay:%llu recv_tick:%llu\n", send_tick, send_delay, linkDelay, recv_tick); // Every packet must be sent and arrive in the same quantum assert(send_tick > master->syncEvent->when() - master->syncEvent->repeat); // No packet may be scheduled for receive in the arrival quantum assert(send_tick + send_delay + linkDelay > master->syncEvent->when()); // 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. Packet // descriptors are saved in the ordered queue. The currently scheduled // packet is always on the top of the queue. // NOTE: we use the event queue lock to protect the receive desc queue, // too, which is accessed both by the receiver thread and the simulation // thread. descQueue.emplace(new_packet, send_tick, send_delay); if (descQueue.size() == 1) { assert(!recvDone->scheduled()); eventManager->schedule(recvDone, recv_tick); } else { assert(recvDone->scheduled()); panic_if(descQueue.front().sendTick + descQueue.front().sendDelay > recv_tick, "Out of order packet received (recv_tick: %lu top(): %lu\n", recv_tick, descQueue.front().sendTick + descQueue.front().sendDelay); } curEventQueue()->unlock(); } EthPacketPtr DistIface::RecvScheduler::popPacket() { // Note : this is called from the simulation thread when a receive done // event is being processed for the link. We assume that the thread holds // the event queue queue lock when this is called! EthPacketPtr next_packet = descQueue.front().packet; descQueue.pop(); if (descQueue.size() > 0) { Tick recv_tick = calcReceiveTick(descQueue.front().sendTick, descQueue.front().sendDelay, curTick()); eventManager->schedule(recvDone, recv_tick); } prevRecvTick = curTick(); return next_packet; } void DistIface::RecvScheduler::Desc::serialize(CheckpointOut &cp) const { SERIALIZE_SCALAR(sendTick); SERIALIZE_SCALAR(sendDelay); packet->serialize("rxPacket", cp); } void DistIface::RecvScheduler::Desc::unserialize(CheckpointIn &cp) { UNSERIALIZE_SCALAR(sendTick); UNSERIALIZE_SCALAR(sendDelay); packet = std::make_shared(16384); packet->unserialize("rxPacket", cp); } void DistIface::RecvScheduler::serialize(CheckpointOut &cp) const { SERIALIZE_SCALAR(prevRecvTick); // serialize the receive desc queue std::queue tmp_queue(descQueue); unsigned n_desc_queue = descQueue.size(); assert(tmp_queue.size() == descQueue.size()); SERIALIZE_SCALAR(n_desc_queue); for (int i = 0; i < n_desc_queue; i++) { tmp_queue.front().serializeSection(cp, csprintf("rxDesc_%d", i)); tmp_queue.pop(); } assert(tmp_queue.empty()); } void DistIface::RecvScheduler::unserialize(CheckpointIn &cp) { assert(descQueue.size() == 0); assert(!recvDone->scheduled()); assert(!ckptRestore); UNSERIALIZE_SCALAR(prevRecvTick); // unserialize the receive desc queue unsigned n_desc_queue; UNSERIALIZE_SCALAR(n_desc_queue); for (int i = 0; i < n_desc_queue; i++) { Desc recv_desc; recv_desc.unserializeSection(cp, csprintf("rxDesc_%d", i)); descQueue.push(recv_desc); } ckptRestore = true; } DistIface::DistIface(unsigned dist_rank, unsigned dist_size, Tick sync_start, Tick sync_repeat, EventManager *em, bool is_switch, int num_nodes) : syncStart(sync_start), syncRepeat(sync_repeat), recvThread(nullptr), recvScheduler(em), rank(dist_rank), size(dist_size) { DPRINTF(DistEthernet, "DistIface() ctor rank:%d\n",dist_rank); isMaster = false; if (master == nullptr) { assert(sync == nullptr); assert(syncEvent == nullptr); if (is_switch) sync = new SyncSwitch(num_nodes); else sync = new SyncNode(); syncEvent = new SyncEvent(); master = this; isMaster = true; } distIfaceId = distIfaceNum; distIfaceNum++; } DistIface::~DistIface() { assert(recvThread); delete recvThread; if (this == master) { assert(syncEvent); delete syncEvent; assert(sync); delete sync; master = nullptr; } } void DistIface::packetOut(EthPacketPtr pkt, Tick send_delay) { Header header; // Prepare a dist header packet for the Ethernet packet we want to // send out. header.msgType = MsgType::dataDescriptor; header.sendTick = curTick(); header.sendDelay = send_delay; header.dataPacketLength = pkt->size(); // Send out the packet and the meta info. sendPacket(header, pkt); DPRINTF(DistEthernetPkt, "DistIface::sendDataPacket() done size:%d send_delay:%llu\n", pkt->size(), send_delay); } void DistIface::recvThreadFunc(Event *recv_done, Tick link_delay) { EthPacketPtr new_packet; DistHeaderPkt::Header header; // Initialize receive scheduler parameters recvScheduler.init(recv_done, link_delay); // Main loop to wait for and process any incoming message. for (;;) { // recvHeader() blocks until the next dist 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. // Grab the eventq lock to stop the simulation thread curEventQueue()->lock(); exit_message("info", 0, "Message server closed connection, " "simulation is exiting"); } // We got a valid dist header packet, let's process it if (header.msgType == MsgType::dataDescriptor) { recvPacket(header, new_packet); recvScheduler.pushPacket(new_packet, header.sendTick, header.sendDelay); } else { // everything else must be synchronisation related command sync->progress(header.sendTick, header.syncRepeat, header.needCkpt, header.needExit); } } } void DistIface::spawnRecvThread(const Event *recv_done, Tick link_delay) { assert(recvThread == nullptr); recvThread = new std::thread(&DistIface::recvThreadFunc, this, const_cast(recv_done), link_delay); recvThreadsNum++; } DrainState DistIface::drain() { DPRINTF(DistEthernet,"DistIFace::drain() called\n"); // This can be called multiple times in the same drain cycle. if (this == master) syncEvent->draining(true); return DrainState::Drained; } void DistIface::drainResume() { DPRINTF(DistEthernet,"DistIFace::drainResume() called\n"); if (this == master) syncEvent->draining(false); recvScheduler.resumeRecvTicks(); } void DistIface::serialize(CheckpointOut &cp) const { // Drain the dist interface before the checkpoint is taken. We cannot call // this as part of the normal drain cycle because this dist sync has to be // called exactly once after the system is fully drained. sync->drainComplete(); unsigned rank_orig = rank, dist_iface_id_orig = distIfaceId; SERIALIZE_SCALAR(rank_orig); SERIALIZE_SCALAR(dist_iface_id_orig); recvScheduler.serializeSection(cp, "recvScheduler"); if (this == master) { sync->serializeSection(cp, "Sync"); } } void DistIface::unserialize(CheckpointIn &cp) { unsigned rank_orig, dist_iface_id_orig; UNSERIALIZE_SCALAR(rank_orig); UNSERIALIZE_SCALAR(dist_iface_id_orig); panic_if(rank != rank_orig, "Rank mismatch at resume (rank=%d, orig=%d)", rank, rank_orig); panic_if(distIfaceId != dist_iface_id_orig, "Dist iface ID mismatch " "at resume (distIfaceId=%d, orig=%d)", distIfaceId, dist_iface_id_orig); recvScheduler.unserializeSection(cp, "recvScheduler"); if (this == master) { sync->unserializeSection(cp, "Sync"); } } void DistIface::init(const Event *done_event, Tick link_delay) { // Init hook for the underlaying message transport to setup/finalize // communication channels initTransport(); // Spawn a new receiver thread that will process messages // coming in from peer gem5 processes. // The receive thread will also schedule a (receive) doneEvent // for each incoming data packet. spawnRecvThread(done_event, link_delay); // Adjust the periodic sync start and interval. Different DistIface // might have different requirements. The singleton sync object // will select the minimum values for both params. assert(sync != nullptr); sync->init(syncStart, syncRepeat); // 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 DistIface::startup() { DPRINTF(DistEthernet, "DistIface::startup() started\n"); if (this == master) syncEvent->start(); DPRINTF(DistEthernet, "DistIface::startup() done\n"); } bool DistIface::readyToCkpt(Tick delay, Tick period) { bool ret = true; DPRINTF(DistEthernet, "DistIface::readyToCkpt() called, delay:%lu " "period:%lu\n", delay, period); if (master) { if (delay == 0) { inform("m5 checkpoint called with zero delay => triggering collaborative " "checkpoint\n"); sync->requestCkpt(ReqType::collective); } else { inform("m5 checkpoint called with non-zero delay => triggering immediate " "checkpoint (at the next sync)\n"); sync->requestCkpt(ReqType::immediate); } if (period != 0) inform("Non-zero period for m5_ckpt is ignored in " "distributed gem5 runs\n"); ret = false; } return ret; } bool DistIface::readyToExit(Tick delay) { bool ret = true; DPRINTF(DistEthernet, "DistIface::readyToExit() called, delay:%lu\n", delay); if (master) { if (delay == 0) { inform("m5 exit called with zero delay => triggering collaborative " "exit\n"); sync->requestExit(ReqType::collective); } else { inform("m5 exit called with non-zero delay => triggering immediate " "exit (at the next sync)\n"); sync->requestExit(ReqType::immediate); } ret = false; } return ret; } uint64_t DistIface::rankParam() { uint64_t val; if (master) { val = master->rank; } else { warn("Dist-rank parameter is queried in single gem5 simulation."); val = 0; } return val; } uint64_t DistIface::sizeParam() { uint64_t val; if (master) { val = master->size; } else { warn("Dist-size parameter is queried in single gem5 simulation."); val = 1; } return val; }