49#include "debug/DistEthernet.hh"
50#include "debug/DistEthernetPkt.hh"
73 inform(
"Next dist synchronisation tick is changed to %lu.\n",
nextAt);
77 panic(
"Dist synchronisation interval must be greater than zero");
81 inform(
"Dist synchronisation interval is changed to %lu.\n",
89 std::unique_lock<std::mutex> sync_lock(
lock);
106 nextAt = std::numeric_limits<Tick>::max();
107 nextRepeat = std::numeric_limits<Tick>::max();
114 needExit = ReqType::none;
115 needCkpt = ReqType::none;
116 needStopSync = ReqType::none;
120 nextAt = std::numeric_limits<Tick>::max();
121 nextRepeat = std::numeric_limits<Tick>::max();
128 std::unique_lock<std::mutex> sync_lock(
lock);
131 assert(waitNum == 0);
135 header.msgType = MsgType::cmdSyncReq;
137 header.syncRepeat = nextRepeat;
138 header.needCkpt = needCkpt;
139 header.needStopSync = needStopSync;
140 if (needCkpt != ReqType::none)
141 needCkpt = ReqType::pending;
142 header.needExit = needExit;
143 if (needExit != ReqType::none)
144 needExit = ReqType::pending;
145 if (needStopSync != ReqType::none)
146 needStopSync = ReqType::pending;
149 auto lf = [
this]{
return waitNum == 0; };
150 cv.wait(sync_lock, lf);
152 assert(isAbort || !same_tick || (nextAt ==
curTick()));
160 std::unique_lock<std::mutex> sync_lock(
lock);
164 auto lf = [
this]{
return waitNum == 0; };
165 cv.wait(sync_lock, lf);
167 assert(waitNum == 0);
170 assert(!same_tick || (nextAt ==
curTick()));
173 header.msgType = MsgType::cmdSyncAck;
175 header.syncRepeat = nextRepeat;
176 if (doCkpt || numCkptReq == numNodes) {
178 header.needCkpt = ReqType::immediate;
181 header.needCkpt = ReqType::none;
183 if (doExit || numExitReq == numNodes) {
185 header.needExit = ReqType::immediate;
187 header.needExit = ReqType::none;
189 if (doStopSync || numStopSyncReq == numNodes) {
192 header.needStopSync = ReqType::immediate;
194 header.needStopSync = ReqType::none;
207 std::unique_lock<std::mutex> sync_lock(
lock);
212 if (send_tick > nextAt)
214 if (nextRepeat > sync_repeat)
215 nextRepeat = sync_repeat;
217 if (need_ckpt == ReqType::collective)
219 else if (need_ckpt == ReqType::immediate)
221 if (need_exit == ReqType::collective)
223 else if (need_exit == ReqType::immediate)
225 if (need_stop_sync == ReqType::collective)
227 else if (need_stop_sync == ReqType::immediate)
248 std::unique_lock<std::mutex> sync_lock(
lock);
253 nextAt = max_send_tick;
254 nextRepeat = next_repeat;
255 doCkpt = (do_ckpt != ReqType::none);
256 doExit = (do_exit != ReqType::none);
257 doStopSync = (do_stop_sync != ReqType::none);
272 std::lock_guard<std::mutex> sync_lock(
lock);
273 assert(req != ReqType::none);
274 if (needCkpt != ReqType::none)
275 warn(
"Ckpt requested multiple times (req:%d)\n",
static_cast<int>(req));
276 if (needCkpt == ReqType::none || req == ReqType::immediate)
283 std::lock_guard<std::mutex> sync_lock(
lock);
284 assert(req != ReqType::none);
285 if (needExit != ReqType::none)
286 warn(
"Exit requested multiple times (req:%d)\n",
static_cast<int>(req));
287 if (needExit == ReqType::none || req == ReqType::immediate)
308 int need_exit =
static_cast<int>(needExit);
317 needExit =
static_cast<ReqType>(need_exit);
344 panic(
"DistIface::SyncEvent::start() aborted\n");
353 assert(!scheduled());
362 inform(
"Dist sync scheduled at %lu and repeats %lu\n", when(),
380 "Distributed sync is hit while draining");
419 warn_once(
"Tried to wake up thread in dist-gem5, but it "
420 "was already awake!\n");
438 recvDone = recv_done;
439 linkDelay = link_delay;
447 Tick recv_tick = send_tick + send_delay + linkDelay;
449 assert(recv_tick >= prev_recv_tick + send_delay);
450 panic_if(prev_recv_tick + send_delay > recv_tick,
451 "Receive window is smaller than send delay");
453 "Simulators out of sync - missed packet receive by %llu ticks"
454 "(rev_recv_tick: %lu send_tick: %lu send_delay: %lu "
456 curTick() - recv_tick, prev_recv_tick, send_tick, send_delay,
475 while (!descQueue.empty()) {
476 Desc d = descQueue.front();
479 d.sendDelay =
d.packet->simLength;
486 if (recvDone->scheduled()) {
487 assert(!descQueue.empty());
488 eventManager->reschedule(recvDone,
curTick());
490 assert(descQueue.empty() &&
v.empty());
502 Tick recv_tick = calcReceiveTick(send_tick, send_delay, prevRecvTick);
504 DPRINTF(DistEthernetPkt,
"DistIface::recvScheduler::pushPacket "
505 "send_tick:%llu send_delay:%llu link_delay:%llu recv_tick:%llu\n",
506 send_tick, send_delay, linkDelay, recv_tick);
520 descQueue.emplace(new_packet, send_tick, send_delay);
521 if (descQueue.size() == 1) {
522 assert(!recvDone->scheduled());
523 eventManager->schedule(recvDone, recv_tick);
525 assert(recvDone->scheduled());
526 panic_if(descQueue.front().sendTick + descQueue.front().sendDelay > recv_tick,
527 "Out of order packet received (recv_tick: %lu top(): %lu\n",
528 recv_tick, descQueue.front().sendTick + descQueue.front().sendDelay);
542 if (descQueue.size() > 0) {
543 Tick recv_tick = calcReceiveTick(descQueue.front().sendTick,
544 descQueue.front().sendDelay,
546 eventManager->schedule(recvDone, recv_tick);
557 packet->serialize(
"rxPacket", cp);
565 packet = std::make_shared<EthPacketData>();
566 packet->unserialize(
"rxPacket", cp);
574 std::queue<Desc> tmp_queue(descQueue);
575 unsigned n_desc_queue = descQueue.size();
576 assert(tmp_queue.size() == descQueue.size());
578 for (
int i = 0;
i < n_desc_queue;
i++) {
579 tmp_queue.front().serializeSection(cp,
csprintf(
"rxDesc_%d",
i));
582 assert(tmp_queue.empty());
588 assert(descQueue.size() == 0);
589 assert(!recvDone->scheduled());
590 assert(!ckptRestore);
594 unsigned n_desc_queue;
596 for (
int i = 0;
i < n_desc_queue;
i++) {
599 descQueue.push(recv_desc);
610 bool is_switch,
int num_nodes) :
615 DPRINTF(DistEthernet,
"DistIface() ctor rank:%d\n",dist_rank);
618 assert(
sync ==
nullptr);
657 header.sendDelay = send_delay;
659 header.dataPacketLength = pkt->length;
660 header.simLength = pkt->simLength;
666 "DistIface::sendDataPacket() done size:%d send_delay:%llu\n",
667 pkt->length, send_delay);
698 if (
header.msgType == MsgType::dataDescriptor) {
723 const_cast<Event *
>(recv_done),
731 DPRINTF(DistEthernet,
"DistIFace::drain() called\n");
740 DPRINTF(DistEthernet,
"DistIFace::drainResume() called\n");
768 unsigned rank_orig, dist_iface_id_orig;
772 panic_if(
rank != rank_orig,
"Rank mismatch at resume (rank=%d, orig=%d)",
775 "at resume (distIfaceId=%d, orig=%d)",
distIfaceId,
801 assert(
sync !=
nullptr);
808 rng->init(5489 * (
rank+1) + 257);
814 DPRINTF(DistEthernet,
"DistIface::startup() started\n");
818 DPRINTF(DistEthernet,
"DistIface::startup() done\n");
825 DPRINTF(DistEthernet,
"DistIface::readyToCkpt() called, delay:%lu "
826 "period:%lu\n", delay, period);
829 inform(
"m5 checkpoint called with zero delay => triggering collaborative "
833 inform(
"m5 checkpoint called with non-zero delay => triggering immediate "
834 "checkpoint (at the next sync)\n");
838 inform(
"Non-zero period for m5_ckpt is ignored in "
839 "distributed gem5 runs\n");
848 std::lock_guard<std::mutex> sync_lock(
lock);
863 inform(
"Request toggling syncronization off\n");
875 inform(
"Request toggling syncronization on\n");
893 DPRINTF(DistEthernet,
"DistIface::readyToExit() called, delay:%lu\n",
901 inform(
"m5 exit called with zero delay => triggering collaborative "
905 inform(
"m5 exit called with non-zero delay => triggering immediate "
906 "exit (at the next sync)\n");
921 warn(
"Dist-rank parameter is queried in single gem5 simulation.");
934 warn(
"Dist-size parameter is queried in single gem5 simulation.");
void unserialize(CheckpointIn &cp) override
Unserialize an object.
void serialize(CheckpointOut &cp) const override
Serialize an object.
void init(Event *recv_done, Tick link_delay)
Initialize network link parameters.
void resumeRecvTicks()
Adjust receive ticks for pending packets when restoring from a checkpoint.
EthPacketPtr popPacket()
Fetch the next packet that is to be received by the simulated network link.
void pushPacket(EthPacketPtr new_packet, Tick send_tick, Tick send_delay)
Push a newly arrived packet into the desc queue.
Tick calcReceiveTick(Tick send_tick, Tick send_delay, Tick prev_recv_tick)
Calculate the tick to schedule the next receive done event.
The global event to schedule periodic dist sync.
void start()
Schedule the first periodic sync event.
void process() override
This is a global event so process() will only be called by exactly one simulation thread.
void requestExit(ReqType req) override
void serialize(CheckpointOut &cp) const override
Serialize an object.
ReqType needStopSync
Sync stop requested.
bool progress(Tick max_req_tick, Tick next_repeat, ReqType do_ckpt, ReqType do_exit, ReqType do_stop_sync) override
Callback when the receiver thread gets a sync ack message.
void requestCkpt(ReqType req) override
bool run(bool same_tick) override
Core method to perform a full dist sync.
void unserialize(CheckpointIn &cp) override
Unserialize an object.
void requestStopSync(ReqType req) override
bool run(bool same_tick) override
Core method to perform a full dist sync.
SyncSwitch(int num_nodes)
void serialize(CheckpointOut &cp) const override
Serialize an object.
void unserialize(CheckpointIn &cp) override
Unserialize an object.
bool progress(Tick max_req_tick, Tick next_repeat, ReqType do_ckpt, ReqType do_exit, ReqType do_stop_sync) override
Callback when the receiver thread gets a sync ack message.
Tick nextRepeat
The repeat value for the next periodic sync.
bool doStopSync
Flag is set if sync is to stop upon sync completion.
void init(Tick start, Tick repeat)
Initialize periodic sync params.
Tick nextAt
Tick for the next periodic sync (if the event is not scheduled yet)
virtual void requestStopSync(ReqType req)=0
virtual void requestExit(ReqType req)=0
virtual void requestCkpt(ReqType req)=0
std::mutex lock
The lock to protect access to the Sync object.
virtual bool progress(Tick send_tick, Tick next_repeat, ReqType do_ckpt, ReqType do_exit, ReqType do_stop_sync)=0
Callback when the receiver thread gets a sync ack message.
void abort()
Abort processing an on-going sync event (in case of an error, e.g.
static uint64_t rankParam()
Getter for the dist rank param.
virtual void initTransport()=0
Init hook for the underlaying transport.
RecvScheduler recvScheduler
Meta information about data packets received.
static bool readyToExit(Tick delay)
Initiate the exit from the simulation.
bool syncStartOnPseudoOp
Use pseudoOp to start synchronization.
static void toggleSync(ThreadContext *tc)
Trigger the primary to start/stop synchronization.
static bool readyToCkpt(Tick delay, Tick period)
Initiate taking a checkpoint.
void init(const Event *e, Tick link_delay)
unsigned rank
The rank of this process among the gem5 peers.
unsigned distIfaceId
Unique id for the dist link.
static DistIface * primary
The very first DistIface object created becomes the primary interface.
static uint64_t sizeParam()
Getter for the dist size param.
DrainState drain() override
Draining is the process of clearing out the states of SimObjects.These are the SimObjects that are pa...
std::thread * recvThread
Receiver thread pointer.
static SyncEvent * syncEvent
The singleton SyncEvent object to schedule periodic dist sync.
void unserialize(CheckpointIn &cp) override
Unserialize an object.
static System * sys
System pointer used to wakeup sleeping threads when stopping sync.
void recvThreadFunc(Event *recv_done, Tick link_delay)
The function executed by a receiver thread.
virtual void sendCmd(const Header &header)=0
Send out a control command to the remote end.
virtual bool recvHeader(Header &header)=0
Receive a header (i.e.
static Sync * sync
The singleton Sync object to perform dist synchronisation.
virtual void sendPacket(const Header &header, const EthPacketPtr &packet)=0
Send out a data packet to the remote end.
static unsigned distIfaceNum
Number of DistIface objects (i.e.
void drainResume() override
Resume execution after a successful drain.
void spawnRecvThread(const Event *recv_done, Tick link_delay)
spawn the receiver thread.
static bool isSwitch
Is this node a switch?
void serialize(CheckpointOut &cp) const override
Serialize an object.
virtual void recvPacket(const Header &header, EthPacketPtr &packet)=0
Receive a packet from the remote end.
Tick syncRepeat
Frequency of dist sync events in ticks.
Tick syncStart
Tick to schedule the first dist sync event.
static unsigned recvThreadsNum
Number of receiver threads (in this gem5 process)
DistIface(unsigned dist_rank, unsigned dist_size, Tick sync_start, Tick sync_repeat, EventManager *em, bool use_pseudo_op, bool is_switch, int num_nodes)
ctor
void packetOut(EthPacketPtr pkt, Tick send_delay)
Send out an Ethernet packet.
unsigned size
The number of gem5 processes comprising this dist simulation.
void lock()
Provide an interface for locking/unlocking the event queue.
ThreadContext is the external interface to all thread state for anything outside of the CPU.
virtual System * getSystemPtr()=0
void quiesceTick(Tick resume)
Quiesce, suspend, and schedule activate at resume.
void quiesce()
Quiesce thread context.
@ Suspended
Temporarily inactive.
virtual Status status() const =0
DrainState
Object drain/handover states.
@ Drained
Buffers drained, ready for serialization/handover.
#define panic(...)
This implements a cprintf based panic() function.
#define panic_if(cond,...)
Conditional panic macro that checks the supplied condition and only panics if the condition is true a...
void serializeSection(CheckpointOut &cp, const char *name) const
Serialize an object into a new section.
void unserializeSection(CheckpointIn &cp, const char *name)
Unserialize an a child object.
Copyright (c) 2024 Arm Limited All rights reserved.
Tick curTick()
The universal simulation clock.
std::ostream CheckpointOut
uint64_t Tick
Tick count type.
void exitSimLoop(const std::string &message, int exit_code, Tick when, Tick repeat, bool serialize)
Schedule an event to exit the simulation loop (returning to Python) at the end of the current cycle (...
std::string csprintf(const char *format, const Args &...args)
EventQueue * curEventQueue()
std::shared_ptr< EthPacketData > EthPacketPtr
#define UNSERIALIZE_SCALAR(scalar)
#define SERIALIZE_SCALAR(scalar)
Received packet descriptor.
void serialize(CheckpointOut &cp) const override
Serialize an object.
void unserialize(CheckpointIn &cp) override
Unserialize an object.