diff --git a/src/realm/activemsg.cc b/src/realm/activemsg.cc index 671db37ced4..88948bf4eea 100644 --- a/src/realm/activemsg.cc +++ b/src/realm/activemsg.cc @@ -19,6 +19,7 @@ #include "realm/atomics.h" #include "realm/activemsg.h" +#include "realm/bgwork.h" #include "realm/mutex.h" #include "realm/cmdline.h" #include "realm/logging.h" @@ -190,6 +191,8 @@ namespace Realm { // at least one of the two above must be non-null assert((e.handler != 0) || (e.handler_notimeout != 0)); e.handler_inline = nextreg->get_handler_inline(); + e.profile_sub_item_id = 0; + e.profile_id_registered = false; handlers.push_back(e); } @@ -391,6 +394,10 @@ namespace Realm { #ifdef DEBUG_INCOMING printf("adding incoming message from %d\n", sender); #endif + // Record that we did work handling a message for any networks + if(ThreadLocal::bgwork_profstate) { + ThreadLocal::bgwork_profstate->set_worked(true); + } // look up which message this is ActiveMessageHandlerTable::HandlerEntry *handler = @@ -754,14 +761,29 @@ namespace Realm { long long t_start = 0; bool do_profile = Config::profile_activemsg_handlers; + // lazily register this handler for fine-grained profiling + if(bgwork_profiler.get_level() >= 2 && + !current_msg->handler->profile_id_registered) { + current_msg->handler->profile_sub_item_id = bgwork_profiler.register_sub_item( + BGWP_SUB_AM_HANDLER, current_msg->handler->name); + current_msg->handler->profile_id_registered = true; + } + // do we have a handler that understands time limits? if(current_msg->handler->handler != 0) { if(do_profile) t_start = Clock::current_time_in_nanoseconds(); + if(current_msg->handler->profile_id_registered) + ThreadLocal::bgwork_profstate->fine_begin( + current_msg->handler->profile_sub_item_id); + (current_msg->handler->handler)(current_msg->sender, current_msg->hdr, current_msg->payload, current_msg->payload_size, work_until); + + if(current_msg->handler->profile_id_registered) + ThreadLocal::bgwork_profstate->fine_end(); } else { // estimate how long this handler will take, clamping at a // semi-arbitrary 20us @@ -788,9 +810,16 @@ namespace Realm { do_profile = true; t_start = Clock::current_time_in_nanoseconds(); + if(current_msg->handler->profile_id_registered) + ThreadLocal::bgwork_profstate->fine_begin( + current_msg->handler->profile_sub_item_id); + (current_msg->handler->handler_notimeout)(current_msg->sender, current_msg->hdr, current_msg->payload, current_msg->payload_size); + + if(current_msg->handler->profile_id_registered) + ThreadLocal::bgwork_profstate->fine_end(); } long long t_end = 0; diff --git a/src/realm/activemsg.h b/src/realm/activemsg.h index 4d76f76fe30..b9eed548dfe 100644 --- a/src/realm/activemsg.h +++ b/src/realm/activemsg.h @@ -271,6 +271,9 @@ namespace Realm { ActiveMessageHandlerStats stats; std::optional extract_frag_info; + + uint16_t profile_sub_item_id; + bool profile_id_registered; }; HandlerEntry *lookup_message_handler(MessageID id); diff --git a/src/realm/bgwork.cc b/src/realm/bgwork.cc index 4990464374f..228174b12b3 100644 --- a/src/realm/bgwork.cc +++ b/src/realm/bgwork.cc @@ -20,12 +20,455 @@ #include "realm/bgwork.h" #include "realm/timers.h" #include "realm/logging.h" +#include "realm/network.h" #include "realm/utils.h" #include "realm/numa/numasysif.h" +#include +#include +#include +#include + +#ifdef REALM_ON_WINDOWS +#include +#include +#else +#include +#include +#endif + +#ifdef REALM_BGWORK_PROFILE_USE_ZLIB +#include +#endif + namespace Realm { Logger log_bgwork("bgwork"); + Logger log_bgwork_profile("bgwork_profile"); + + BgWorkProfileManager bgwork_profiler; + + //////////////////////////////////////////////////////////////////////// + // + // class BgWorkProfileManager + // + + BgWorkProfileManager::BgWorkProfileManager() + : profile_level(0) + , initialized(false) + , max_buffer_bytes(1ULL << 30) // 1GB default + , fd(-1) + , node_id(0) + , next_sub_item_id(0) + , free_blocks(nullptr) + , completed_head(nullptr) + , completed_tail(nullptr) + , buffered_bytes(0) + , next_sequence(0) + {} + + BgWorkProfileManager::~BgWorkProfileManager() + { + // free any remaining blocks in the free list + while(free_blocks) { + ProfileBlock *next = free_blocks->next; + delete free_blocks; + free_blocks = next; + } + } + + void BgWorkProfileManager::set_level(int level) { profile_level = level; } + + void BgWorkProfileManager::set_logfile(const std::string &filename) + { + logfile_pattern = filename; + } + + void BgWorkProfileManager::set_bufsize(size_t megabytes) + { + max_buffer_bytes = (megabytes == 0) ? SIZE_MAX : megabytes * (1ULL << 20); + } + + int BgWorkProfileManager::get_level() const { return profile_level; } + + void BgWorkProfileManager::initialize(uint32_t _node_id) + { + if(profile_level == 0) + return; + + node_id = _node_id; + + // determine output filename + std::string filename = logfile_pattern; + if(filename.empty()) + filename = "bgwork_profile_%.bin"; + + // replace % with node ID + size_t pct = filename.find('%'); + if(pct != std::string::npos) { + char buf[32]; + snprintf(buf, sizeof(buf), "%u", node_id); + filename.replace(pct, 1, buf); + } else if(Network::max_node_id > 0) { + log_bgwork_profile.fatal() + << "multi-node run requires '%' in bgwork profile filename: " << filename; + abort(); + } + + // open output file +#ifdef REALM_ON_WINDOWS + fd = _open(filename.c_str(), _O_WRONLY | _O_CREAT | _O_TRUNC | _O_BINARY, 0644); +#else + fd = open(filename.c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0644); +#endif + if(fd < 0) { + log_bgwork_profile.fatal() << "failed to open bgwork profile file: " << filename; + abort(); + } + + log_bgwork_profile.info() << "bgwork profiling enabled: level=" << profile_level + << " file=" << filename + << " bufsize=" << (max_buffer_bytes >> 20) << "MB"; + + // Write header with placeholder counts/offset. Data blocks are appended + // starting at offset HEADER_SIZE. Descriptor tables and final header + // patch happen at shutdown. + write_file_header(); + + initialized = true; + } + + void BgWorkProfileManager::shutdown() + { + if(!initialized) + return; + + // Flush any remaining blocks from registered profiling states. + // Dedicated worker threads flush their own blocks before exiting, + // but task scheduler bgworkers may still have data. + { + AutoLock<> al(thread_mutex); + for(BgWorkProfileState *state : thread_states) { + ProfileBlock *current_block = state->flush(); + if(current_block) { + if(current_block->num_records > 0) { + AutoLock<> bl(block_mutex); + ProfileBlock *block = current_block; + if(completed_tail) { + completed_tail->next = block; + } else { + completed_head = block; + } + completed_tail = block; + block->next = nullptr; + buffered_bytes += block->used; + } else { + AutoLock<> bl(block_mutex); + current_block->next = free_blocks; + free_blocks = current_block; + } + } + // Note: states are not heap-allocated, do not delete + } + thread_states.clear(); + } + + // Flush all remaining in-memory data blocks to disk + flush_all_blocks(); + + // Record current file position -- this is where descriptor tables start + uint64_t desc_offset = lseek(fd, 0, SEEK_CUR); + + // Write descriptor tables (now complete) at end of file + write_descriptor_tables(); + + // Patch header with final counts and descriptor offset + { + AutoLock<> al(desc_mutex); + uint32_t work_count = static_cast(work_item_descs.size()); + uint32_t sub_count = static_cast(sub_item_descs.size()); + lseek(fd, 20, SEEK_SET); + write(fd, &work_count, sizeof(work_count)); + write(fd, &sub_count, sizeof(sub_count)); + write(fd, &desc_offset, sizeof(desc_offset)); + lseek(fd, 0, SEEK_END); + } + + // close file + if(fd >= 0) { +#ifdef REALM_ON_WINDOWS + _close(fd); +#else + close(fd); +#endif + fd = -1; + } + + log_bgwork_profile.info() << "bgwork profiling shutdown complete"; + initialized = false; + } + + void BgWorkProfileManager::register_work_item(uint16_t slot, const std::string &name) + { + AutoLock<> al(desc_mutex); + // check for duplicate + for(const auto &d : work_item_descs) { + if(d.slot == slot) + return; + } + work_item_descs.push_back({slot, name}); + log_bgwork_profile.debug() << "registered work item: slot=" << slot + << " name=" << name; + } + + uint16_t BgWorkProfileManager::register_sub_item(uint8_t type, const std::string &name) + { + AutoLock<> al(desc_mutex); + uint16_t id = next_sub_item_id++; + sub_item_descs.push_back({id, type, name}); + log_bgwork_profile.debug() << "registered sub-item: id=" << id + << " type=" << (int)type << " name=" << name; + return id; + } + + void BgWorkProfileManager::register_existing_items(BackgroundWorkManager &mgr) + { + unsigned count = mgr.num_work_items.load(); + for(unsigned i = 0; i < count; i++) { + BackgroundWorkItem *item = mgr.work_items[i]; + if(item) + register_work_item(static_cast(i), item->name); + } + } + + ProfileBlock *BgWorkProfileManager::alloc_block(uint64_t thread_id) + { + AutoLock<> al(block_mutex); + + ProfileBlock *block; + if(free_blocks) { + block = free_blocks; + free_blocks = block->next; + } else { + block = new ProfileBlock; + } + + block->used = 0; + block->base_timestamp = 0; + block->num_records = 0; + block->thread_id = thread_id; + block->sequence = next_sequence++; + block->next = nullptr; + + return block; + } + + void BgWorkProfileManager::complete_block(ProfileBlock *block) + { + bool need_flush = false; + + { + AutoLock<> al(block_mutex); + + if(completed_tail) { + completed_tail->next = block; + } else { + completed_head = block; + } + completed_tail = block; + block->next = nullptr; + buffered_bytes += block->used; + + need_flush = (buffered_bytes >= max_buffer_bytes); + } + + // Flush half the buffer to keep memory bounded while retaining some + // buffering to reduce write syscall frequency + if(need_flush) + flush_blocks_to_disk(max_buffer_bytes / 2); + } + + void BgWorkProfileManager::register_thread_state(BgWorkProfileState *state) + { + AutoLock<> al(thread_mutex); + thread_states.push_back(state); + } + + void BgWorkProfileManager::unregister_thread_state(BgWorkProfileState *state) + { + AutoLock<> al(thread_mutex); + for(auto it = thread_states.begin(); it != thread_states.end(); ++it) { + if(*it == state) { + thread_states.erase(it); + return; + } + } + } + + void BgWorkProfileManager::write_file_header() + { + // header: magic(4) + version(2) + flags(2) + node_id(4) + zero_time(8) + + // work_item_count(4) + sub_item_count(4) + desc_offset(8) = 36 bytes + uint8_t header[HEADER_SIZE]; + uint8_t *p = header; + + memcpy(p, BGWP_MAGIC, 4); + p += 4; + + uint16_t version = BGWP_VERSION; + memcpy(p, &version, 2); + p += 2; + + uint16_t flags = 0; + if(profile_level >= 2) + flags |= BGWP_FLAG_HAS_FINE; + memcpy(p, &flags, 2); + p += 2; + + memcpy(p, &node_id, 4); + p += 4; + + int64_t zero_time = Clock::get_zero_time(); + memcpy(p, &zero_time, 8); + p += 8; + + // descriptor counts and offset will be patched at shutdown + uint32_t zero32 = 0; + uint64_t zero64 = 0; + memcpy(p, &zero32, 4); + p += 4; // work item count + memcpy(p, &zero32, 4); + p += 4; // sub item count + memcpy(p, &zero64, 8); + p += 8; // descriptor table offset + + ssize_t written = write(fd, header, sizeof(header)); + (void)written; + } + + void BgWorkProfileManager::write_descriptor_tables() + { + AutoLock<> al(desc_mutex); + + // write work item descriptors + for(const auto &d : work_item_descs) { + uint16_t slot = d.slot; + uint16_t name_len = static_cast(d.name.size()); + write(fd, &slot, sizeof(slot)); + write(fd, &name_len, sizeof(name_len)); + write(fd, d.name.data(), name_len); + } + + // write sub-item descriptors + for(const auto &d : sub_item_descs) { + uint16_t id = d.id; + uint8_t type = d.type; + uint16_t name_len = static_cast(d.name.size()); + write(fd, &id, sizeof(id)); + write(fd, &type, sizeof(type)); + write(fd, &name_len, sizeof(name_len)); + write(fd, d.name.data(), name_len); + } + } + + void BgWorkProfileManager::flush_blocks_to_disk(size_t target_size) + { + while(true) { + ProfileBlock *block = nullptr; + { + AutoLock<> al(block_mutex); + if(!completed_head || buffered_bytes <= target_size) + return; + block = completed_head; + completed_head = block->next; + if(!completed_head) + completed_tail = nullptr; + buffered_bytes -= block->used; + } + + // write block header fields individually to avoid padding + uint64_t bh_thread_id = block->thread_id; + uint32_t bh_sequence = block->sequence; + uint32_t bh_record_count = block->num_records; + int64_t bh_base_timestamp = block->base_timestamp; + uint32_t bh_data_size = block->used; + uint32_t bh_compressed_size = 0; + +#ifdef REALM_BGWORK_PROFILE_USE_ZLIB + // try to compress the block + uLongf compressed_bound = compressBound(block->used); + std::vector compressed(compressed_bound); + int zret = compress2(compressed.data(), &compressed_bound, block->data, block->used, + Z_DEFAULT_COMPRESSION); + if(zret == Z_OK && compressed_bound < block->used) { + bh_compressed_size = static_cast(compressed_bound); + } +#endif + + write(fd, &bh_thread_id, 8); + write(fd, &bh_sequence, 4); + write(fd, &bh_record_count, 4); + write(fd, &bh_base_timestamp, 8); + write(fd, &bh_data_size, 4); + write(fd, &bh_compressed_size, 4); + +#ifdef REALM_BGWORK_PROFILE_USE_ZLIB + if(bh_compressed_size > 0) + write(fd, compressed.data(), bh_compressed_size); + else + write(fd, block->data, block->used); +#else + write(fd, block->data, block->used); +#endif + + // return block to free list + { + AutoLock<> al(block_mutex); + block->next = free_blocks; + free_blocks = block; + } + } + } + + void BgWorkProfileManager::flush_all_blocks() { flush_blocks_to_disk(0); } + + //////////////////////////////////////////////////////////////////////// + // + // class BgWorkProfileState + // + + BgWorkProfileState::BgWorkProfileState(void) + : level(bgwork_profiler.get_level()) + , thread_id( + static_cast(std::hash{}(std::this_thread::get_id()))) + { + if(level > 0) { + bgwork_profiler.register_thread_state(this); + } + // Save ourselves into the thread local variable + ThreadLocal::bgwork_profstate = this; + } + + BgWorkProfileState::~BgWorkProfileState(void) + { + // flush any remaining profiling data and unregister before the + // stack-allocated state goes away + if(level > 0) { + if(current_block) { + if(current_block->num_records > 0) + bgwork_profiler.complete_block(current_block); + current_block = nullptr; + } + bgwork_profiler.unregister_thread_state(this); + } + } + + ProfileBlock *BgWorkProfileState::flush(void) + { + ProfileBlock *current = current_block; + current_block = nullptr; + return current; + } //////////////////////////////////////////////////////////////////////// // @@ -82,6 +525,9 @@ namespace Realm { worker.set_manager(manager); worker.set_numa_domain(numa_domain); + // set up per-thread profiling state (stack-allocated, lives for thread lifetime) + BgWorkProfileState profstate; + log_bgwork.info() << "dedicated worker starting - worker=" << this << " numa=" << numa_domain; @@ -378,6 +824,8 @@ namespace Realm { << " slot=" << index << " name=" << name << " domain=" << numa_domain << " timeslice=" << min_timeslice_needed; + if(bgwork_profiler.get_level() > 0) + bgwork_profiler.register_work_item(static_cast(index), name); } // mark this work item as active (i.e. having work to do) @@ -563,6 +1011,7 @@ namespace Realm { log_bgwork.debug() << "work claimed: manager=" << manager << " slot=" << slot << " worker=" << this; long long t_start = Clock::current_time_in_nanoseconds(true /*absolute*/); + ThreadLocal::bgwork_profstate->begin(static_cast(slot)); // don't spend more than 1ms on any single task before going on to the // next thing - TODO: pull this out as a config variable long long t_quantum = (manager->cfg.work_item_timeslice + t_start); @@ -580,8 +1029,9 @@ namespace Realm { #ifdef DEBUG_REALM item->make_inactive(); #endif + TimeLimit time_limit = TimeLimit::absolute(t_quantum, interrupt_flag); while(true) { - bool requeue = item->do_work(TimeLimit::absolute(t_quantum, interrupt_flag)); + bool requeue = item->do_work(time_limit); if(requeue) { // we can just call this item's work function again if we're not out // of time and if there's nothing else to do @@ -595,6 +1045,7 @@ namespace Realm { t_quantum = (manager->cfg.work_item_timeslice + now); if((work_until_time > 0) && (work_until_time < t_quantum)) t_quantum = work_until_time; + time_limit = TimeLimit::absolute(t_quantum, interrupt_flag); continue; } } @@ -605,13 +1056,8 @@ namespace Realm { } else break; } -#ifdef REALM_BGWORK_PROFILE - long long t_stop = Clock::current_time_in_nanoseconds(true /*absolute*/); - long long elapsed = t_stop - t_start; - long long overshoot = ((t_stop > t_quantum) ? (t_stop - t_quantum) : 0); - log_bgwork.print() << "work: slot=" << slot << " elapsed=" << elapsed - << " overshoot=" << overshoot; -#endif + ThreadLocal::bgwork_profstate->end( + time_limit); // end() checks did_work internally // we're done with this slot for now manager->work_item_usecounts[slot].fetch_sub_acqrel(1); diff --git a/src/realm/bgwork.h b/src/realm/bgwork.h index 1a5de477a14..0ff413c466a 100644 --- a/src/realm/bgwork.h +++ b/src/realm/bgwork.h @@ -26,12 +26,234 @@ #include "realm/cmdline.h" #include "realm/timers.h" +#include +#include #include +#include namespace Realm { class BackgroundWorkItem; class BackgroundWorkThread; + class BackgroundWorkManager; + + // Background work profiling for Realm + // + // Binary file format specification (RBWP = Realm Background Work Profile): + // + // FILE HEADER (36 bytes, written at start, counts/offset patched at shutdown): + // Magic: 4 bytes "RBWP" + // Version: uint16_t (currently 1) + // Flags: uint16_t (bit 0 = has fine-grained data) + // Node ID: uint32_t + // Clock zero time: int64_t (nanoseconds, absolute) + // Work item descriptor count: uint32_t (patched at shutdown) + // Sub-item descriptor count: uint32_t (patched at shutdown) + // Descriptor table offset: uint64_t (patched at shutdown) + // + // DATA BLOCKS (appended during run and at shutdown, starting at offset 36): + // Block header: + // Thread ID: uint64_t + // Block sequence: uint32_t + // Record count: uint32_t + // Base timestamp: int64_t + // Data size: uint32_t (uncompressed) + // Compressed size: uint32_t (0 = uncompressed) + // Block data: uint8_t[compressed_size or data_size] + // + // DESCRIPTOR TABLES (written at shutdown, at descriptor_table_offset): + // + // WORK ITEM DESCRIPTOR TABLE: + // For each work item: + // Slot: uint16_t + // Name length: uint16_t + // Name: char[name_length] (not null-terminated) + // + // SUB-ITEM DESCRIPTOR TABLE (follows work item table): + // For each sub-item: + // ID: uint16_t + // Type: uint8_t (0=AM_HANDLER, 1=XFER_CHANNEL, 2=DEPPART_OP, 3=GPU_REAP) + // Name length: uint16_t + // Name: char[name_length] (not null-terminated) + // + // RECORDS within a block (variable-length, packed): + // Timestamp delta: 2, 4, or 8 bytes (see encoding below) + // Record type: uint8_t + // Payload: depends on record type + // + // Timestamp delta encoding: + // If delta fits in 15 bits: 2 bytes, high bit 0: 0bbb bbbb bbbb bbbb + // If delta fits in 30 bits: 4 bytes, high bits 10: 10bb bbbb ... bbbb bbbb + // Otherwise: 8 bytes, high bits 11: 11xx xxxx + 7 more bytes + // (stores absolute timestamp, not delta) + // + // Record types and payloads: + // COARSE_BEGIN (0x01): uint8_t slot + // COARSE_END (0x02): (no payload) + // FINE_BEGIN (0x11): uint16_t sub_item_id + // FINE_END (0x12): (no payload) + // GPU_WORK (0x21): uint64_t proc_id, uint8_t slot, int64_t start, int64_t stop + + // Record type constants + enum BgWorkProfileRecordType : uint8_t + { + BGWP_COARSE_BEGIN = 0x01, + BGWP_COARSE_END = 0x02, + BGWP_FINE_BEGIN = 0x11, + BGWP_FINE_END = 0x12, + BGWP_GPU_WORK = 0x21, + }; + + // Sub-item type constants + enum BgWorkProfileSubItemType : uint8_t + { + BGWP_SUB_AM_HANDLER = 0, + BGWP_SUB_XFER_CHANNEL = 1, + BGWP_SUB_DEPPART_OP = 2, + BGWP_SUB_GPU_REAP = 3, + }; + + // File format constants + static const char BGWP_MAGIC[4] = {'R', 'B', 'W', 'P'}; + static const uint16_t BGWP_VERSION = 1; + static const uint16_t BGWP_FLAG_HAS_FINE = 0x0001; + + struct ProfileBlock { + static const size_t BLOCK_SIZE = 16384; // 16KB + uint8_t data[BLOCK_SIZE]; + uint32_t used; + int64_t base_timestamp; + uint32_t num_records; + uint64_t thread_id; + uint32_t sequence; + ProfileBlock *next; + }; + + class BgWorkProfileState { + private: + const int level; // 0 = disabled, 1 = coarse, 2 = fine + ProfileBlock *current_block = nullptr; + int64_t last_timestamp = 0; // for delta encoding + const uint64_t thread_id = 0; + + // did_work flag: set to true by fine_begin, fine_end, gpu_work, + // or explicitly by do_work implementations; checked by end() + // to decide whether to record or discard + bool did_work = false; + + // saved state for discard (set by begin()) + uint32_t begin_block_used = 0; + uint32_t begin_block_num_records = 0; + int64_t begin_last_timestamp = 0; + + public: + BgWorkProfileState(void); + ~BgWorkProfileState(void); + // recording methods (all no-op when level == 0) + inline void begin(uint8_t slot); + inline void end(const TimeLimit &time_limit); + inline void set_worked(bool worked) + { + if(level > 0) + did_work = worked; + } + inline void discard(void); + inline void fine_begin(uint16_t sub_item_id); + inline void fine_end(); + inline void gpu_work(uint64_t proc_id, uint8_t slot, int64_t start, int64_t stop); + ProfileBlock *flush(void); + + private: + uint8_t *ensure_space(size_t needed); + static size_t encode_timestamp(uint8_t *buf, int64_t delta, int64_t absolute); + }; + + namespace ThreadLocal { + inline thread_local BgWorkProfileState *bgwork_profstate = nullptr; + }; + + struct BgWorkItemDescriptor { + uint16_t slot; + std::string name; + }; + + struct BgWorkSubItemDescriptor { + uint16_t id; + uint8_t type; + std::string name; + }; + + class BgWorkProfileManager { + public: + BgWorkProfileManager(); + ~BgWorkProfileManager(); + + // configuration (called before initialize) + void set_level(int level); + void set_logfile(const std::string &filename); + void set_bufsize(size_t megabytes); + + // returns the configured profiling level (0, 1, or 2) + int get_level() const; + + // lifecycle + void initialize(uint32_t node_id); + void shutdown(); + + // descriptor registration (called during module init, before recording starts) + void register_work_item(uint16_t slot, const std::string &name); + uint16_t register_sub_item(uint8_t type, const std::string &name); + + // retroactively register any work items that were added to the manager + // before the profiler was configured (e.g., network layer items) + void register_existing_items(BackgroundWorkManager &mgr); + + // block management (called by recording functions) + ProfileBlock *alloc_block(uint64_t thread_id); + void complete_block(ProfileBlock *block); + + // state management: Workers register their profstate so shutdown can flush + void register_thread_state(BgWorkProfileState *state); + void unregister_thread_state(BgWorkProfileState *state); + + private: + void write_file_header(); + void write_descriptor_tables(); + void flush_blocks_to_disk(size_t target_size); + void flush_all_blocks(); + + static const size_t HEADER_SIZE = 36; + + int profile_level; + std::string logfile_pattern; + bool initialized; + size_t max_buffer_bytes; + + // file state + int fd; + uint32_t node_id; + + // descriptors + Mutex desc_mutex; + std::vector work_item_descs; + std::vector sub_item_descs; + uint16_t next_sub_item_id; + + // block pool and completed list + Mutex block_mutex; + ProfileBlock *free_blocks; + ProfileBlock *completed_head; + ProfileBlock *completed_tail; + size_t buffered_bytes; + uint32_t next_sequence; + + // registered states (for shutdown flushing) + Mutex thread_mutex; + std::vector thread_states; + }; + + // global instance + extern BgWorkProfileManager bgwork_profiler; class BackgroundWorkManager { public: @@ -101,6 +323,7 @@ namespace Realm { BackgroundWorkItem *work_items[MAX_WORK_ITEMS]; friend class BackgroundWorkThread; + friend class BgWorkProfileManager; // to manage sleeping workers, we need to stuff three things into a // single atomically-updatable state variable: @@ -140,8 +363,12 @@ namespace Realm { // completed (or if 'make_active' has already been called) virtual bool do_work(TimeLimit work_until) = 0; + // returns the slot index assigned by the background work manager + unsigned get_slot() const { return index; } + protected: friend class BackgroundWorkManager::Worker; + friend class BgWorkProfileManager; // mark this work item as active (i.e. having work to do) void make_active(void); @@ -173,4 +400,6 @@ namespace Realm { }; // namespace Realm +#include "realm/bgwork.inl" + #endif diff --git a/src/realm/bgwork.inl b/src/realm/bgwork.inl new file mode 100644 index 00000000000..1dda2f1cfbe --- /dev/null +++ b/src/realm/bgwork.inl @@ -0,0 +1,259 @@ +/* + * Copyright 2026 Stanford University, NVIDIA Corporation + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// inline recording methods for BgWorkProfileState + +#ifndef REALM_BGWORK_INL +#define REALM_BGWORK_INL + +namespace Realm { + + // Timestamp delta encoding: + // 15-bit: 2 bytes, MSB=0 + // 30-bit: 4 bytes, MSB=10 + // 64-bit: 8 bytes, MSB=11 (stores absolute timestamp) + inline size_t BgWorkProfileState::encode_timestamp(uint8_t *buf, int64_t delta, + int64_t absolute) + { + if(delta >= 0 && delta < (1 << 15)) { + uint16_t val = static_cast(delta); + buf[0] = (val >> 8) & 0x7F; + buf[1] = val & 0xFF; + return 2; + } else if(delta >= 0 && delta < (1LL << 30)) { + uint32_t val = static_cast(delta) | 0x80000000U; + buf[0] = (val >> 24) & 0xFF; + buf[1] = (val >> 16) & 0xFF; + buf[2] = (val >> 8) & 0xFF; + buf[3] = val & 0xFF; + return 4; + } else { + // 8-byte encoding: store absolute timestamp + uint64_t val = static_cast(absolute); + buf[0] = 0xC0 | ((val >> 56) & 0x3F); + buf[1] = (val >> 48) & 0xFF; + buf[2] = (val >> 40) & 0xFF; + buf[3] = (val >> 32) & 0xFF; + buf[4] = (val >> 24) & 0xFF; + buf[5] = (val >> 16) & 0xFF; + buf[6] = (val >> 8) & 0xFF; + buf[7] = val & 0xFF; + return 8; + } + } + + // ensures enough space in the current block, rotating if needed + // returns pointer to write position, or nullptr on failure + inline uint8_t *BgWorkProfileState::ensure_space(size_t needed) + { + ProfileBlock *block = current_block; + if(block && (block->used + needed <= ProfileBlock::BLOCK_SIZE)) + return block->data + block->used; + + // need a new block - complete old one and get fresh + if(block) + bgwork_profiler.complete_block(block); + + block = bgwork_profiler.alloc_block(thread_id); + current_block = block; + if(!block) + return nullptr; + + // reset delta encoding for new block + last_timestamp = 0; + return block->data; + } + + inline void BgWorkProfileState::begin(uint8_t slot) + { + if(level == 0) + return; + REALM_ASSERT(!did_work); + did_work = true; + + int64_t now = Clock::current_time_in_nanoseconds(true /*absolute*/); + + // max record size: 8 (timestamp) + 1 (type) + 1 (slot) = 10 + uint8_t *buf = ensure_space(10); + if(!buf) + return; + + ProfileBlock *block = current_block; + if(block->num_records == 0) + block->base_timestamp = now; + + // save state for potential discard + begin_block_used = block->used; + begin_block_num_records = block->num_records; + begin_last_timestamp = last_timestamp; + + int64_t delta = now - last_timestamp; + size_t ts_size = encode_timestamp(buf, delta, now); + buf += ts_size; + + *buf++ = BGWP_COARSE_BEGIN; + *buf++ = slot; + + block->used += ts_size + 2; + block->num_records++; + last_timestamp = now; + } + + inline void BgWorkProfileState::end(const TimeLimit& time_limit) + { + if(level == 0) + return; + + if(did_work) { + did_work = false; + } else if(!time_limit.is_expired()) { + // If the timelimit expired we still record this + // because it is surprising that it took that long + discard(); + return; + } + + int64_t now = Clock::current_time_in_nanoseconds(true /*absolute*/); + + // max: 8 (timestamp) + 1 (type) = 9 + uint8_t *buf = ensure_space(9); + if(!buf) + return; + + ProfileBlock *block = current_block; + if(block->num_records == 0) + block->base_timestamp = now; + + int64_t delta = now - last_timestamp; + size_t ts_size = encode_timestamp(buf, delta, now); + buf += ts_size; + + *buf++ = BGWP_COARSE_END; + + block->used += ts_size + 1; + block->num_records++; + last_timestamp = now; + } + + inline void BgWorkProfileState::discard(void) + { + // rewind the block state to what it was before begin() + ProfileBlock *block = current_block; + if(!block) + return; + + block->used = begin_block_used; + block->num_records = begin_block_num_records; + last_timestamp = begin_last_timestamp; + } + + inline void BgWorkProfileState::fine_begin(uint16_t sub_item_id) + { + if(level < 2) + return; + + did_work = true; + + int64_t now = Clock::current_time_in_nanoseconds(true /*absolute*/); + + // max: 8 + 1 + 2 = 11 + uint8_t *buf = ensure_space(11); + if(!buf) + return; + + ProfileBlock *block = current_block; + if(block->num_records == 0) + block->base_timestamp = now; + + int64_t delta = now - last_timestamp; + size_t ts_size = encode_timestamp(buf, delta, now); + buf += ts_size; + + *buf++ = BGWP_FINE_BEGIN; + memcpy(buf, &sub_item_id, sizeof(uint16_t)); + buf += sizeof(uint16_t); + + block->used += ts_size + 3; + block->num_records++; + last_timestamp = now; + } + + inline void BgWorkProfileState::fine_end() + { + if(level < 2) + return; + + int64_t now = Clock::current_time_in_nanoseconds(true /*absolute*/); + + // max: 8 + 1 = 9 + uint8_t *buf = ensure_space(9); + if(!buf) + return; + + ProfileBlock *block = current_block; + if(block->num_records == 0) + block->base_timestamp = now; + + int64_t delta = now - last_timestamp; + size_t ts_size = encode_timestamp(buf, delta, now); + buf += ts_size; + + *buf++ = BGWP_FINE_END; + + block->used += ts_size + 1; + block->num_records++; + last_timestamp = now; + } + + inline void BgWorkProfileState::gpu_work(uint64_t proc_id, uint8_t slot, + int64_t start_time, int64_t stop_time) + { + if(level == 0) + return; + + int64_t now = Clock::current_time_in_nanoseconds(true /*absolute*/); + + // max: 8 (timestamp) + 1 (type) + 8 (proc_id) + 1 (slot) + 8 (start) + 8 (stop) = 34 + uint8_t *buf = ensure_space(34); + if(!buf) + return; + + ProfileBlock *block = current_block; + if(block->num_records == 0) + block->base_timestamp = now; + + int64_t delta = now - last_timestamp; + size_t ts_size = encode_timestamp(buf, delta, now); + buf += ts_size; + + *buf++ = BGWP_GPU_WORK; + memcpy(buf, &proc_id, sizeof(uint64_t)); + buf += sizeof(uint64_t); + *buf++ = slot; + memcpy(buf, &start_time, sizeof(int64_t)); + buf += sizeof(int64_t); + memcpy(buf, &stop_time, sizeof(int64_t)); + buf += sizeof(int64_t); + + block->used += ts_size + 26; + block->num_records++; + last_timestamp = now; + } + +}; // namespace Realm + +#endif // REALM_BGWORK_INL diff --git a/src/realm/cuda/cuda_internal.cc b/src/realm/cuda/cuda_internal.cc index 9a1ca7a475b..de8850d6310 100644 --- a/src/realm/cuda/cuda_internal.cc +++ b/src/realm/cuda/cuda_internal.cc @@ -555,6 +555,7 @@ namespace Realm { WriteSequenceCache wseqcache(this, 2 << 20); GPUStream *stream = 0; size_t total_bytes = 0; + BgWorkGpuCudaNotification *gpu_timing = nullptr; AffineCopyInfo<3> copy_infos; CUDA_MEMCPY3D cuda_copy; @@ -763,6 +764,11 @@ namespace Realm { if(in_gpu && in_gpu->can_access_peer(out_gpu) && transpose_copy.extents[0] != 0 && transpose_copy.extents[0] <= CUDA_MAX_FIELD_BYTES) { + if(!gpu_timing && bgwork_profiler.get_level() > 0) { + gpu_timing = new BgWorkGpuCudaNotification(stream->get_gpu()->proc->me.id, + channel->get_bgwork_slot()); + stream->add_notification(gpu_timing); + } stream->get_gpu()->launch_transpose_kernel(transpose_copy, min_align, stream); bytes_to_fence += transpose_copy.extents[0] * transpose_copy.extents[1] * transpose_copy.extents[2]; @@ -821,6 +827,11 @@ namespace Realm { log_gpudma.info() << "\tLaunching kernel for rects=" << copy_infos.num_rects << " bytes=" << copy_info_total << " out_is_ipc=" << out_is_ipc; + if(!gpu_timing && bgwork_profiler.get_level() > 0) { + gpu_timing = new BgWorkGpuCudaNotification(stream->get_gpu()->proc->me.id, + channel->get_bgwork_slot()); + stream->add_notification(gpu_timing); + } stream->get_gpu()->launch_batch_affine_kernel( ©_infos, 3, min_align, copy_info_total / min_align, stream); bytes_to_fence += copy_info_total; @@ -875,6 +886,11 @@ namespace Realm { } } + if(gpu_timing) { + AutoGPUContext agc(stream->get_gpu()); + stream->add_notification(gpu_timing); + } + rseqcache.flush(); wseqcache.flush(); @@ -1170,7 +1186,7 @@ namespace Realm { : SingleXDQChannel( bgwork, _kind, stringbuilder() << "cuda channel (gpu=" << _src_gpu->info->index - << " kind=" << (int)_kind << ")") + << " kind=" << _kind << ")") { src_gpu = _src_gpu; @@ -1451,7 +1467,7 @@ namespace Realm { : SingleXDQChannel( bgwork, _kind, stringbuilder() << "cuda channel (gpu=" << _src_gpu->info->index - << " kind=" << (int)_kind << ")") + << " kind=" << _kind << ")") { src_gpu = _src_gpu; @@ -1823,6 +1839,8 @@ namespace Realm { bool GPUfillXferDes::progress_xd(GPUfillChannel *channel, TimeLimit work_until) { bool did_work = false; + BgWorkGpuCudaNotification *gpu_timing = nullptr; + GPUStream *gpu_timing_stream = nullptr; ReadSequenceCache rseqcache(this, 2 << 20); WriteSequenceCache wseqcache(this, 2 << 20); @@ -1893,6 +1911,12 @@ namespace Realm { Realm::Cuda::AffineFillInfo<2, size_t>::MAX_NUM_RECTS) { // Filled the current info, time to start all over log_gpudma.info() << "pushing fill kernel"; + if(!gpu_timing && bgwork_profiler.get_level() > 0) { + gpu_timing = new BgWorkGpuCudaNotification( + stream->get_gpu()->proc->me.id, channel->get_bgwork_slot()); + stream->add_notification(gpu_timing); + gpu_timing_stream = stream; + } stream->get_gpu()->launch_batch_affine_fill_kernel( &fill_info, 2, reduced_fill_size, total_info_bytes / reduced_fill_size, stream); @@ -2071,6 +2095,12 @@ namespace Realm { if(fill_info.num_rects > 0) { log_gpudma.info() << "pushing fill kernel"; + if(!gpu_timing && bgwork_profiler.get_level() > 0) { + gpu_timing = new BgWorkGpuCudaNotification(stream->get_gpu()->proc->me.id, + channel->get_bgwork_slot()); + stream->add_notification(gpu_timing); + gpu_timing_stream = stream; + } stream->get_gpu()->launch_batch_affine_fill_kernel( &fill_info, 2, reduced_fill_size, total_info_bytes / reduced_fill_size, stream); @@ -2093,6 +2123,11 @@ namespace Realm { break; } + if(gpu_timing) { + AutoGPUContext agc(channel->gpu); + gpu_timing_stream->add_notification(gpu_timing); + } + rseqcache.flush(); return did_work; @@ -2303,6 +2338,7 @@ namespace Realm { bool GPUreduceXferDes::progress_xd(GPUreduceChannel *channel, TimeLimit work_until) { bool did_work = false; + BgWorkGpuCudaNotification *gpu_timing = nullptr; ReadSequenceCache rseqcache(this, 2 << 20); ReadSequenceCache wseqcache(this, 2 << 20); @@ -2453,6 +2489,12 @@ namespace Realm { { AutoGPUContext agc(channel->gpu); + if(!gpu_timing && bgwork_profiler.get_level() > 0) { + gpu_timing = new BgWorkGpuCudaNotification( + stream->get_gpu()->proc->me.id, channel->get_bgwork_slot()); + stream->add_notification(gpu_timing); + } + if(kernel != 0) { // Use params array to pass kernel arguments (pointers to each // parameter) instead of CU_LAUNCH_PARAM_BUFFER_POINTER (packed buffer). @@ -2538,6 +2580,11 @@ namespace Realm { break; } + if(gpu_timing) { + AutoGPUContext agc(channel->gpu); + stream->add_notification(gpu_timing); + } + rseqcache.flush(); wseqcache.flush(); diff --git a/src/realm/cuda/cuda_internal.h b/src/realm/cuda/cuda_internal.h index 614710bfe16..0400a8c92de 100644 --- a/src/realm/cuda/cuda_internal.h +++ b/src/realm/cuda/cuda_internal.h @@ -200,6 +200,22 @@ namespace Realm { virtual void request_completed(void) = 0; }; + // Profiling notification for GPU kernel timing in background work items. + // Registered as a GPUCompletionNotification on a stream twice: once before + // GPU kernel submissions (start marker) and once after (end marker). + // Uses host-side timestamps taken when the existing events are reaped. + class BgWorkGpuCudaNotification : public GPUCompletionNotification { + public: + BgWorkGpuCudaNotification(uint64_t _proc_id, uint8_t _slot); + void request_completed(void) override; + + private: + uint64_t proc_id; + uint8_t slot; + int64_t start_time; + bool started; + }; + class GPUWorkFence : public Realm::Operation::AsyncWorkItem { public: GPUWorkFence(GPU *gpu, Realm::Operation *op); diff --git a/src/realm/cuda/cuda_module.cc b/src/realm/cuda/cuda_module.cc index 0147bc2b0d2..72c46bfe5c2 100644 --- a/src/realm/cuda/cuda_module.cc +++ b/src/realm/cuda/cuda_module.cc @@ -20,6 +20,7 @@ #include "realm/cuda/cuda_internal.h" #include "realm/cuda/cuda_memcpy.h" +#include "realm/bgwork.h" #include "realm/tasks.h" #include "realm/logging.h" #include "realm/cmdline.h" @@ -299,12 +300,18 @@ namespace Realm { } // we'll keep looking at events until we find one that hasn't triggered + bool first = true; bool work_left = true; while(event_valid) { CUresult res = CUDA_DRIVER_FNPTR(cuEventQuery)(event); - if(res == CUDA_ERROR_NOT_READY) + if(res == CUDA_ERROR_NOT_READY) { return true; // oldest event hasn't triggered - check again later + } else if(first) { + // As long as we did at least one event we did work + Realm::ThreadLocal::bgwork_profstate->set_worked(true); + first = false; + } // no other kind of error is expected if(res != CUDA_SUCCESS) { @@ -365,6 +372,30 @@ namespace Realm { return work_left; } + //////////////////////////////////////////////////////////////////////// + // + // class BgWorkGpuCudaNotification + + BgWorkGpuCudaNotification::BgWorkGpuCudaNotification(uint64_t _proc_id, uint8_t _slot) + : proc_id(_proc_id) + , slot(_slot) + , start_time(0) + , started(false) + {} + + void BgWorkGpuCudaNotification::request_completed(void) + { + if(!started) { + start_time = Clock::current_time_in_nanoseconds(true /*absolute*/); + started = true; + } else { + int64_t stop_time = Clock::current_time_in_nanoseconds(true /*absolute*/); + Realm::ThreadLocal::bgwork_profstate->gpu_work(proc_id, slot, start_time, + stop_time); + delete this; + } + } + //////////////////////////////////////////////////////////////////////// // // class GPUWorkFence @@ -1303,7 +1334,7 @@ namespace Realm { } GPUWorker::GPUWorker(void) - : BackgroundWorkItem("gpu worker") + : BackgroundWorkItem("cuda poll") , condvar(lock) , core_rsrv(0) , worker_thread(0) @@ -1380,6 +1411,8 @@ namespace Realm { bool GPUWorker::do_work(TimeLimit work_until) { + // This is a polling background work item so flip work polarity + Realm::ThreadLocal::bgwork_profstate->set_worked(false); // pop the first stream off the list and immediately become re-active // if more streams remain GPUStream *stream = 0; @@ -1464,9 +1497,18 @@ namespace Realm { void GPUWorker::thread_main(void) { + // Create a background worker profiling state for this thread + BgWorkProfileState profstate; + const TimeLimit unlimited; + const uint8_t slot = get_slot(); // TODO: consider busy-waiting in some cases to reduce latency? while(!worker_shutdown_requested.load()) { + // This is a kind of background work item we're processing so time it + profstate.begin(slot); + // We're polling so set worked to false + profstate.set_worked(false); bool work_left = process_streams(true); + profstate.end(unlimited); // if there was work left, yield our thread for now to avoid a tight spin loop // TODO: enqueue a callback so we can go to sleep and wake up sooner than a kernel @@ -1476,50 +1518,6 @@ namespace Realm { } } - //////////////////////////////////////////////////////////////////////// - // - // class BlockingCompletionNotification - - class BlockingCompletionNotification : public GPUCompletionNotification { - public: - BlockingCompletionNotification(void); - virtual ~BlockingCompletionNotification(void); - - virtual void request_completed(void); - - virtual void wait(void); - - public: - atomic completed; - }; - - BlockingCompletionNotification::BlockingCompletionNotification(void) - : completed(false) - {} - - BlockingCompletionNotification::~BlockingCompletionNotification(void) {} - - void BlockingCompletionNotification::request_completed(void) - { - // no condition variable needed - the waiter is spinning - completed.store(true); - } - - void BlockingCompletionNotification::wait(void) - { - // blocking completion is horrible and should die as soon as possible - // in the mean time, we need to assist with background work to avoid - // the risk of deadlock - // note that this means you can get NESTED blocking completion - // notifications, which is just one of the ways this is horrible - BackgroundWorkManager::Worker worker; - - worker.set_manager(&(get_runtime()->bgwork)); - - while(!completed.load()) - worker.do_work(-1 /* as long as it takes */, &completed /* until this is set */); - } - //////////////////////////////////////////////////////////////////////// // // class GPUFBMemory diff --git a/src/realm/deppart/partitions.cc b/src/realm/deppart/partitions.cc index b023f468fce..042aadd7e83 100644 --- a/src/realm/deppart/partitions.cc +++ b/src/realm/deppart/partitions.cc @@ -839,6 +839,7 @@ namespace Realm { : BackgroundWorkItem("deppart op queue") , shutdown_flag(false), rsrv(_rsrv), condvar(mutex) , work_advertised(false) + , profile_sub_item_id(0), profile_id_registered(false) { if(_bgwork) add_to_manager(_bgwork); @@ -969,6 +970,13 @@ namespace Realm { make_active(); } + // lazily register for fine-grained profiling + if(bgwork_profiler.get_level() >= 2 && !profile_id_registered) { + profile_sub_item_id = + bgwork_profiler.register_sub_item(BGWP_SUB_DEPPART_OP, "deppart op"); + profile_id_registered = true; + } + // now we can work on the op we got in parallel with everybody else // (neither branch will be taken if there are dedicated workers and they // already got to the queued operations) @@ -976,7 +984,11 @@ namespace Realm { bool ok_to_run = op->mark_started(); if(ok_to_run) { log_part.info() << "worker " << this << " starting op " << op; + if(profile_id_registered) + ThreadLocal::bgwork_profstate->fine_begin(profile_sub_item_id); op->execute(); + if(profile_id_registered) + ThreadLocal::bgwork_profstate->fine_end(); log_part.info() << "worker " << this << " finished op " << op; op->mark_finished(true /*successful*/); } else { @@ -988,7 +1000,11 @@ namespace Realm { if(uop != 0) { log_part.info() << "worker " << this << " starting uop " << uop; uop->mark_started(); + if(profile_id_registered) + ThreadLocal::bgwork_profstate->fine_begin(profile_sub_item_id); uop->execute(); + if(profile_id_registered) + ThreadLocal::bgwork_profstate->fine_end(); log_part.info() << "worker " << this << " finished uop " << uop; uop->mark_finished(); } diff --git a/src/realm/deppart/partitions.h b/src/realm/deppart/partitions.h index 7bb68c3630c..4b3529bacdf 100644 --- a/src/realm/deppart/partitions.h +++ b/src/realm/deppart/partitions.h @@ -218,6 +218,8 @@ namespace Realm { Mutex::CondVar condvar; std::vector workers; bool work_advertised; + uint16_t profile_sub_item_id; + bool profile_id_registered; }; diff --git a/src/realm/gasnet1/gasnetmsg.cc b/src/realm/gasnet1/gasnetmsg.cc index d0bcd2cccb2..4f59f0af3cc 100644 --- a/src/realm/gasnet1/gasnetmsg.cc +++ b/src/realm/gasnet1/gasnetmsg.cc @@ -33,6 +33,7 @@ #include "realm/threads.h" #include "realm/timers.h" #include "realm/logging.h" +#include "realm/bgwork.h" // so OpenMPI borrowed gasnet's platform-detection code and didn't change // the define names - work around it by undef'ing anything set via mpi.h @@ -2613,6 +2614,9 @@ static void handle_new_activemsg(gasnet_token_t token, void *buf, size_t nbytes, } } else record_message(src, false); + if(ThreadLocal::bgwork_profstate) { + ThreadLocal::bgwork_profstate->set_worked(true); + } } void gasnet_parse_command_line(std::vector &cmdline) @@ -2798,6 +2802,10 @@ bool EndpointManager::do_work(TimeLimit work_until) // make sure nested active mesage calls respect the time limit ThreadLocal::gasnet_work_until = &work_until; + // This is a polling background work item so make it look like + // we did no work unless we actually do + ThreadLocal::bgwork_profstate->set_worked(false); + push_messages(max_msgs_to_send, false /*!wait*/, work_until); // poll if we're not out of time diff --git a/src/realm/gasnetex/gasnetex_internal.cc b/src/realm/gasnetex/gasnetex_internal.cc index 7060d3e4ec4..c3dc4b389ee 100644 --- a/src/realm/gasnetex/gasnetex_internal.cc +++ b/src/realm/gasnetex/gasnetex_internal.cc @@ -2807,6 +2807,8 @@ namespace Realm { bool GASNetEXPoller::do_work(TimeLimit work_until) { ThreadLocal::gex_work_until = &work_until; + // This is a poller so only count this as doing work if we handle messages + ThreadLocal::bgwork_profstate->set_worked(false); // we're going to try to be frugal about acquiring mutexes here, so peek // ahead in the critical xpair list to avoid the extra mutex acquire that @@ -5294,6 +5296,9 @@ namespace Realm { } ThreadLocal::in_am_handler = false; + if(ThreadLocal::bgwork_profstate) { + ThreadLocal::bgwork_profstate->set_worked(true); + } } }; // namespace Realm diff --git a/src/realm/hip/hip_internal.cc b/src/realm/hip/hip_internal.cc index c3d3ea07e64..a2978b53bf8 100644 --- a/src/realm/hip/hip_internal.cc +++ b/src/realm/hip/hip_internal.cc @@ -514,7 +514,7 @@ namespace Realm { : SingleXDQChannel( bgwork, _kind, stringbuilder() << "hip channel (gpu=" << _src_gpu->info->index - << " kind=" << (int)_kind << ")") + << " kind=" << _kind << ")") { src_gpu = _src_gpu; @@ -1125,6 +1125,7 @@ namespace Realm { bool GPUreduceXferDes::progress_xd(GPUreduceChannel *channel, TimeLimit work_until) { bool did_work = false; + BgWorkGpuHipNotification *gpu_timing = nullptr; ReadSequenceCache rseqcache(this, 2 << 20); ReadSequenceCache wseqcache(this, 2 << 20); @@ -1260,6 +1261,12 @@ namespace Realm { { AutoGPUContext agc(channel->gpu); + if(!gpu_timing && bgwork_profiler.get_level() > 0) { + gpu_timing = new BgWorkGpuHipNotification( + stream->get_gpu()->proc->me.id, channel->get_bgwork_slot()); + stream->add_notification(gpu_timing); + } + void *src_ptr = (void *)args->src_base; void *src_device = src_ptr; #ifndef __HIP_PLATFORM_NVIDIA__ @@ -1350,6 +1357,11 @@ namespace Realm { break; } + if(gpu_timing) { + AutoGPUContext agc(channel->gpu); + stream->add_notification(gpu_timing); + } + rseqcache.flush(); wseqcache.flush(); diff --git a/src/realm/hip/hip_internal.h b/src/realm/hip/hip_internal.h index af9fde98d16..45569e3ff1c 100644 --- a/src/realm/hip/hip_internal.h +++ b/src/realm/hip/hip_internal.h @@ -110,6 +110,22 @@ namespace Realm { virtual void request_completed(void) = 0; }; + // Profiling notification for GPU kernel timing in background work items. + // Registered as a GPUCompletionNotification on a stream twice: once before + // GPU kernel submissions (start marker) and once after (end marker). + // Uses host-side timestamps taken when the existing events are reaped. + class BgWorkGpuHipNotification : public GPUCompletionNotification { + public: + BgWorkGpuHipNotification(uint64_t _proc_id, uint8_t _slot); + void request_completed(void) override; + + private: + uint64_t proc_id; + uint8_t slot; + int64_t start_time; + bool started; + }; + class GPUPreemptionWaiter : public GPUCompletionNotification { public: GPUPreemptionWaiter(GPU *gpu); diff --git a/src/realm/hip/hip_module.cc b/src/realm/hip/hip_module.cc index 194d8844553..22683ce7610 100644 --- a/src/realm/hip/hip_module.cc +++ b/src/realm/hip/hip_module.cc @@ -19,6 +19,7 @@ #include "realm/hip/hip_internal.h" #include "realm/hip/hip_access.h" +#include "realm/bgwork.h" #include "realm/tasks.h" #include "realm/logging.h" #include "realm/cmdline.h" @@ -200,12 +201,18 @@ namespace Realm { } // we'll keep looking at events until we find one that hasn't triggered + bool first = true; bool work_left = true; while(event_valid) { hipError_t res = hipEventQuery(event); - if(res == hipErrorNotReady) + if(res == hipErrorNotReady) { return true; // oldest event hasn't triggered - check again later + } else if(first) { + // As long as we did at least one event we did work + Realm::ThreadLocal::bgwork_profstate->set_worked(true); + first = false; + } // no other kind of error is expected if(res != hipSuccess) { @@ -291,6 +298,30 @@ namespace Realm { } } + //////////////////////////////////////////////////////////////////////// + // + // class BgWorkGpuHipNotification + + BgWorkGpuHipNotification::BgWorkGpuHipNotification(uint64_t _proc_id, uint8_t _slot) + : proc_id(_proc_id) + , slot(_slot) + , start_time(0) + , started(false) + {} + + void BgWorkGpuHipNotification::request_completed(void) + { + if(!started) { + start_time = Clock::current_time_in_nanoseconds(true /*absolute*/); + started = true; + } else { + int64_t stop_time = Clock::current_time_in_nanoseconds(true /*absolute*/); + Realm::ThreadLocal::bgwork_profstate->gpu_work(proc_id, slot, start_time, + stop_time); + delete this; + } + } + //////////////////////////////////////////////////////////////////////// // // class GPUWorkFence @@ -1044,7 +1075,7 @@ namespace Realm { } GPUWorker::GPUWorker(void) - : BackgroundWorkItem("gpu worker") + : BackgroundWorkItem("hip poll") , condvar(lock) , core_rsrv(0) , worker_thread(0) @@ -1121,6 +1152,8 @@ namespace Realm { bool GPUWorker::do_work(TimeLimit work_until) { + // This is a polling background work item so flip work polarity + Realm::ThreadLocal::bgwork_profstate->set_worked(false); // pop the first stream off the list and immediately become re-active // if more streams remain GPUStream *stream = 0; @@ -1204,9 +1237,18 @@ namespace Realm { void GPUWorker::thread_main(void) { + // Create a background worker profiling state for this thread + BgWorkProfileState profstate; + const TimeLimit unlimited; + const uint8_t slot = get_slot(); // TODO: consider busy-waiting in some cases to reduce latency? while(!worker_shutdown_requested.load()) { + // This is a kind of background work item we're processing so time it + profstate.begin(slot); + // We're polling so set worked to false + profstate.set_worked(false); bool work_left = process_streams(true); + profstate.end(unlimited); // if there was work left, yield our thread for now to avoid a tight spin loop // TODO: enqueue a callback so we can go to sleep and wake up sooner than a kernel @@ -1216,50 +1258,6 @@ namespace Realm { } } - //////////////////////////////////////////////////////////////////////// - // - // class BlockingCompletionNotification - - class BlockingCompletionNotification : public GPUCompletionNotification { - public: - BlockingCompletionNotification(void); - virtual ~BlockingCompletionNotification(void); - - virtual void request_completed(void); - - virtual void wait(void); - - public: - atomic completed; - }; - - BlockingCompletionNotification::BlockingCompletionNotification(void) - : completed(false) - {} - - BlockingCompletionNotification::~BlockingCompletionNotification(void) {} - - void BlockingCompletionNotification::request_completed(void) - { - // no condition variable needed - the waiter is spinning - completed.store(true); - } - - void BlockingCompletionNotification::wait(void) - { - // blocking completion is horrible and should die as soon as possible - // in the mean time, we need to assist with background work to avoid - // the risk of deadlock - // note that this means you can get NESTED blocking completion - // notifications, which is just one of the ways this is horrible - BackgroundWorkManager::Worker worker; - - worker.set_manager(&(get_runtime()->bgwork)); - - while(!completed.load()) - worker.do_work(-1 /* as long as it takes */, &completed /* until this is set */); - } - //////////////////////////////////////////////////////////////////////// // // class GPUFBMemory diff --git a/src/realm/runtime_impl.cc b/src/realm/runtime_impl.cc index 50de5eaed09..7758541e766 100644 --- a/src/realm/runtime_impl.cc +++ b/src/realm/runtime_impl.cc @@ -841,6 +841,10 @@ namespace Realm { // enabled. cp.add_option_int("-ll:path_cache_size", Config::path_cache_lru_size); + cp.add_option_int("-ll:bgworkprofile", bgwork_profile_level); + cp.add_option_string("-ll:bgworkprofile_logfile", bgwork_profile_logfile); + cp.add_option_int("-ll:bgworkprofile_bufsize", bgwork_profile_bufsize); + bool cmdline_ok = cp.parse_command_line(cmdline); if(!cmdline_ok) { @@ -868,6 +872,12 @@ namespace Realm { "WARNING: prefix set, but NODE_LOGGING not enabled at compile time!\n"); } #endif + + // configure background work profiler + bgwork_profiler.set_level(bgwork_profile_level); + if(!bgwork_profile_logfile.empty()) + bgwork_profiler.set_logfile(bgwork_profile_logfile); + bgwork_profiler.set_bufsize(bgwork_profile_bufsize); } CoreModule::CoreModule(void) @@ -2148,6 +2158,9 @@ namespace Realm { bgwork.start_dedicated_workers(*core_reservations); + bgwork_profiler.initialize(Network::my_node_id); + bgwork_profiler.register_existing_items(bgwork); + PartitioningOpQueue::start_worker_threads(*core_reservations, &bgwork); #ifdef EVENT_TRACING @@ -2946,6 +2959,7 @@ namespace Realm { event_triggerer.shutdown_work_item(); #endif bgwork.stop_dedicated_workers(); + bgwork_profiler.shutdown(); // tear down the active message manager message_manager->shutdown(); diff --git a/src/realm/runtime_impl.h b/src/realm/runtime_impl.h index 72b72c75feb..bd92b30b7c4 100644 --- a/src/realm/runtime_impl.h +++ b/src/realm/runtime_impl.h @@ -183,6 +183,11 @@ namespace Realm { // barriers int barrier_broadcast_radix = 4; + // background work profiling + int bgwork_profile_level = 0; + std::string bgwork_profile_logfile; + int bgwork_profile_bufsize = 1024; // MB + // topology of the host const HardwareTopology *host_topology = nullptr; }; diff --git a/src/realm/tasks.cc b/src/realm/tasks.cc index 308a0a6ed03..ced7bf4fee6 100644 --- a/src/realm/tasks.cc +++ b/src/realm/tasks.cc @@ -17,8 +17,8 @@ // tasks and task scheduling for Realm +#include "realm/bgwork.h" #include "realm/tasks.h" - #include "realm/runtime_impl.h" #include "realm/proc_impl.h" @@ -1320,6 +1320,8 @@ namespace Realm { lock.unlock(); if(max_bgwork_timeslice > 0) { + // If we're going to go off and do background work then we need to profile it + BgWorkProfileState profstate; // try to be productive while we're waiting bgworker.do_work(max_bgwork_timeslice, &bgworker_interrupt); } else { diff --git a/src/realm/tasks.h b/src/realm/tasks.h index de5e6be1614..5efc3b1ffee 100644 --- a/src/realm/tasks.h +++ b/src/realm/tasks.h @@ -362,6 +362,7 @@ namespace Realm { WorkCounterUpdater wcu_resume_queue; BackgroundWorkManager::Worker bgworker; + BgWorkProfileState bgwork_profstate; atomic bgworker_interrupt; long long max_bgwork_timeslice; diff --git a/src/realm/transfer/channel.h b/src/realm/transfer/channel.h index e693d0ff01a..d3641ee9bcd 100644 --- a/src/realm/transfer/channel.h +++ b/src/realm/transfer/channel.h @@ -1008,6 +1008,8 @@ namespace Realm { bool ordered_mode, in_ordered_worker; Mutex mutex; XferDes::XferDesList ready_xds; + uint16_t profile_sub_item_id; + bool profile_id_registered; }; template @@ -1016,6 +1018,8 @@ namespace Realm { SingleXDQChannel(BackgroundWorkManager *bgwork, XferDesKind _kind, const std::string &_name, int _numa_domain = -1); + unsigned get_bgwork_slot() const { return xdq.get_slot(); } + virtual void shutdown(); virtual void enqueue_ready_xd(XferDes *xd); diff --git a/src/realm/transfer/channel.inl b/src/realm/transfer/channel.inl index 0632025de07..99851b224a8 100644 --- a/src/realm/transfer/channel.inl +++ b/src/realm/transfer/channel.inl @@ -99,6 +99,8 @@ namespace Realm { , channel(_channel) , ordered_mode(_ordered) , in_ordered_worker(false) + , profile_sub_item_id(0) + , profile_id_registered(false) {} template @@ -151,6 +153,13 @@ namespace Realm { if(still_more && !ordered_mode) make_active(); + // lazily register for fine-grained profiling + if(bgwork_profiler.get_level() >= 2 && !profile_id_registered) { + profile_sub_item_id = + bgwork_profiler.register_sub_item(BGWP_SUB_XFER_CHANNEL, name); + profile_id_registered = true; + } + // now process this transfer request, paying attention to our deadline while(true) { @@ -159,7 +168,11 @@ namespace Realm { // on it unsigned progress = xd->current_progress(); + if(profile_id_registered) + ThreadLocal::bgwork_profstate->fine_begin(profile_sub_item_id); bool did_work = xd->progress_xd(static_cast(channel), work_until); + if(profile_id_registered) + ThreadLocal::bgwork_profstate->fine_end(); // if we didn't do any work, and we're not done (i.e. by // concluding there wasn't any work to actually do), re-check diff --git a/src/realm/ucx/ucp_internal.cc b/src/realm/ucx/ucp_internal.cc index e66276b35f4..d017d88fa4a 100644 --- a/src/realm/ucx/ucp_internal.cc +++ b/src/realm/ucx/ucp_internal.cc @@ -331,6 +331,10 @@ namespace Realm { { ThreadLocal::ucp_work_until = &work_until; + // This is a polling background work item, so clear the + // worked bit and only set it if we do real work + Realm::ThreadLocal::bgwork_profstate->set_worked(false); + for(auto worker : workers) { (void)worker->progress(); } diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index a6213d8b46f..da0dbf4e388 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -295,6 +295,8 @@ set(sparse_construct_ARGS -verbose) add_integration_test(sparse_construct "${REALM_TEST_DIR}/sparse_construct.cc") add_integration_test(extres_alias "${REALM_TEST_DIR}/extres_alias.cc") add_integration_test(reservations "${REALM_TEST_DIR}/reservations.cc") +set(bgwork_profile_test_ARGS -ll:bgworkprofile 2 -ll:bgworkprofile_logfile bgwork_profile_test.bin -copies 32 -profile_file bgwork_profile_test.bin) +add_integration_test(bgwork_profile_test "${REALM_TEST_DIR}/bgwork_profile.cc") set(machine_queries_ARGS -ll:cpu 4 -ll:util 2) add_integration_test(machine_queries "${REALM_TEST_DIR}/machine_queries.cc") set(machine_config_test_ARGS diff --git a/tests/bgwork_profile.cc b/tests/bgwork_profile.cc new file mode 100644 index 00000000000..9f39179dac6 --- /dev/null +++ b/tests/bgwork_profile.cc @@ -0,0 +1,464 @@ +/* + * Copyright 2026 Stanford University, NVIDIA Corporation + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// Test program for background work profiling. +// Creates instances in available memories, runs a batch of copies to exercise +// DMA background work items, then optionally validates the output file. + +#include "realm.h" +#include "realm/cmdline.h" + +#include +#include +#include +#include +#include + +#ifdef REALM_ON_WINDOWS +#include +#include +#else +#include +#include +#include +#endif + +using namespace Realm; + +Logger log_app("app"); + +enum +{ + TOP_LEVEL_TASK = Processor::TASK_ID_FIRST_AVAILABLE + 0, +}; + +namespace TestConfig { + size_t copy_size = 1 << 20; // 1MB per copy + int num_copies = 16; + bool validate = false; + std::string profile_file = "bgwork_profile_test.bin"; +}; // namespace TestConfig + +// File format constants (duplicated from bgwork_profile.h for standalone validation) +static const char EXPECTED_MAGIC[4] = {'R', 'B', 'W', 'P'}; + +// Note: file header and block header fields are read individually +// to avoid C struct padding issues with the binary file format. + +// Read helpers +static bool read_exact(int fd, void *buf, size_t count) +{ + uint8_t *p = static_cast(buf); + while(count > 0) { + ssize_t n = read(fd, p, count); + if(n <= 0) + return false; + p += n; + count -= n; + } + return true; +} + +// Decode a timestamp delta, returns the absolute timestamp +// Updates pos to point past the consumed bytes +static bool decode_timestamp(const uint8_t *data, size_t data_size, size_t &pos, + int64_t &last_ts, int64_t &out_ts) +{ + if(pos >= data_size) + return false; + + uint8_t first = data[pos]; + if((first & 0x80) == 0) { + // 2-byte encoding, 15-bit delta + if(pos + 2 > data_size) + return false; + int64_t delta = ((int64_t)(first & 0x7F) << 8) | data[pos + 1]; + out_ts = last_ts + delta; + pos += 2; + } else if((first & 0xC0) == 0x80) { + // 4-byte encoding, 30-bit delta + if(pos + 4 > data_size) + return false; + int64_t delta = ((int64_t)(first & 0x3F) << 24) | ((int64_t)data[pos + 1] << 16) | + ((int64_t)data[pos + 2] << 8) | data[pos + 3]; + out_ts = last_ts + delta; + pos += 4; + } else { + // 8-byte encoding, absolute timestamp + if(pos + 8 > data_size) + return false; + uint64_t val = ((uint64_t)(first & 0x3F) << 56) | ((uint64_t)data[pos + 1] << 48) | + ((uint64_t)data[pos + 2] << 40) | ((uint64_t)data[pos + 3] << 32) | + ((uint64_t)data[pos + 4] << 24) | ((uint64_t)data[pos + 5] << 16) | + ((uint64_t)data[pos + 6] << 8) | (uint64_t)data[pos + 7]; + out_ts = static_cast(val); + pos += 8; + } + last_ts = out_ts; + return true; +} + +static bool validate_profile_file(const std::string &filename) +{ +#ifdef REALM_ON_WINDOWS + int fd = _open(filename.c_str(), _O_RDONLY | _O_BINARY); +#else + int fd = open(filename.c_str(), O_RDONLY); +#endif + if(fd < 0) { + fprintf(stderr, "VALIDATE: cannot open file: %s\n", filename.c_str()); + return false; + } + + // read header fields individually to avoid struct padding issues + // header: magic(4) + version(2) + flags(2) + node_id(4) + zero_time(8) + + // work_item_count(4) + sub_item_count(4) + descriptor_offset(8) = 36 bytes + char magic[4]; + uint16_t version, flags; + uint32_t node_id; + int64_t zero_time; + uint32_t work_item_count, sub_item_count; + uint64_t descriptor_offset; + + if(!read_exact(fd, magic, 4) || !read_exact(fd, &version, 2) || + !read_exact(fd, &flags, 2) || !read_exact(fd, &node_id, 4) || + !read_exact(fd, &zero_time, 8) || !read_exact(fd, &work_item_count, 4) || + !read_exact(fd, &sub_item_count, 4) || !read_exact(fd, &descriptor_offset, 8)) { + fprintf(stderr, "VALIDATE: failed to read file header\n"); + close(fd); + return false; + } + + // check magic + if(memcmp(magic, EXPECTED_MAGIC, 4) != 0) { + fprintf(stderr, "VALIDATE: bad magic: %c%c%c%c\n", magic[0], magic[1], magic[2], + magic[3]); + close(fd); + return false; + } + + fprintf(stdout, "VALIDATE: magic OK, version=%u, flags=0x%04x, node_id=%u\n", version, + flags, node_id); + fprintf(stdout, + "VALIDATE: zero_time=%lld, work_items=%u, sub_items=%u, desc_offset=%llu\n", + (long long)zero_time, work_item_count, sub_item_count, + (unsigned long long)descriptor_offset); + + // seek to descriptor tables (at end of file, after data blocks) + if(lseek(fd, descriptor_offset, SEEK_SET) < 0) { + fprintf(stderr, "VALIDATE: failed to seek to descriptor table at offset %llu\n", + (unsigned long long)descriptor_offset); + close(fd); + return false; + } + + // read work item descriptors + for(uint32_t i = 0; i < work_item_count; i++) { + uint16_t slot, name_len; + if(!read_exact(fd, &slot, 2) || !read_exact(fd, &name_len, 2)) { + fprintf(stderr, "VALIDATE: failed reading work item descriptor %u\n", i); + close(fd); + return false; + } + std::vector name(name_len); + if(name_len > 0 && !read_exact(fd, name.data(), name_len)) { + fprintf(stderr, "VALIDATE: failed reading work item name %u\n", i); + close(fd); + return false; + } + fprintf(stdout, "VALIDATE: work item: slot=%u name='%.*s'\n", slot, (int)name_len, + name.data()); + } + + // read sub-item descriptors + for(uint32_t i = 0; i < sub_item_count; i++) { + uint16_t id, name_len; + uint8_t type; + if(!read_exact(fd, &id, 2) || !read_exact(fd, &type, 1) || + !read_exact(fd, &name_len, 2)) { + fprintf(stderr, "VALIDATE: failed reading sub-item descriptor %u\n", i); + close(fd); + return false; + } + std::vector name(name_len); + if(name_len > 0 && !read_exact(fd, name.data(), name_len)) { + fprintf(stderr, "VALIDATE: failed reading sub-item name %u\n", i); + close(fd); + return false; + } + fprintf(stdout, "VALIDATE: sub-item: id=%u type=%u name='%.*s'\n", id, type, + (int)name_len, name.data()); + } + + // seek back to read data blocks (from offset 36 to descriptor_offset) + if(lseek(fd, 36, SEEK_SET) < 0) { + fprintf(stderr, "VALIDATE: failed to seek to data blocks\n"); + close(fd); + return false; + } + + // read data blocks + uint32_t total_blocks = 0; + uint32_t total_records = 0; + while(lseek(fd, 0, SEEK_CUR) < (off_t)descriptor_offset) { + // read block header fields individually to avoid padding + uint64_t blk_thread_id; + uint32_t blk_sequence, blk_record_count; + int64_t blk_base_timestamp; + uint32_t blk_data_size, blk_compressed_size; + + if(!read_exact(fd, &blk_thread_id, 8) || !read_exact(fd, &blk_sequence, 4) || + !read_exact(fd, &blk_record_count, 4) || !read_exact(fd, &blk_base_timestamp, 8) || + !read_exact(fd, &blk_data_size, 4) || !read_exact(fd, &blk_compressed_size, 4)) + break; // end of file or incomplete header + + total_blocks++; + total_records += blk_record_count; + + size_t read_size = (blk_compressed_size > 0) ? blk_compressed_size : blk_data_size; + std::vector data(read_size); + if(!read_exact(fd, data.data(), read_size)) { + fprintf(stderr, "VALIDATE: failed reading block data (block %u)\n", total_blocks); + close(fd); + return false; + } + + // if compressed, we'd need to decompress - skip detailed validation for compressed + // blocks + if(blk_compressed_size > 0) { + fprintf(stdout, + "VALIDATE: block %u: thread=%llu seq=%u records=%u compressed=%u->%u\n", + total_blocks, (unsigned long long)blk_thread_id, blk_sequence, + blk_record_count, blk_data_size, blk_compressed_size); + continue; + } + + // parse records and check timestamp monotonicity + size_t pos = 0; + int64_t last_ts = 0; + int64_t prev_ts = 0; + bool ts_monotonic = true; + + for(uint32_t r = 0; r < blk_record_count && pos < blk_data_size; r++) { + int64_t ts; + if(!decode_timestamp(data.data(), blk_data_size, pos, last_ts, ts)) { + fprintf(stderr, "VALIDATE: failed decoding timestamp in block %u record %u\n", + total_blocks, r); + close(fd); + return false; + } + + if(r > 0 && ts < prev_ts) { + ts_monotonic = false; + fprintf(stderr, + "VALIDATE: non-monotonic timestamp in block %u record %u: " + "%lld < %lld\n", + total_blocks, r, (long long)ts, (long long)prev_ts); + } + prev_ts = ts; + + // read record type + if(pos >= blk_data_size) { + fprintf(stderr, "VALIDATE: truncated record in block %u record %u\n", + total_blocks, r); + close(fd); + return false; + } + uint8_t rec_type = data[pos++]; + + // skip payload based on type + switch(rec_type) { + case 0x01: // COARSE_BEGIN: 1 byte slot + if(pos + 1 > blk_data_size) { + close(fd); + return false; + } + pos += 1; + break; + case 0x02: // COARSE_END: no payload + break; + case 0x11: // FINE_BEGIN: 2 byte sub_item_id + if(pos + 2 > blk_data_size) { + close(fd); + return false; + } + pos += 2; + break; + case 0x12: // FINE_END: no payload + break; + case 0x21: // GPU_WORK: 8 byte proc_id + 1 byte slot + 8 byte start + 8 byte stop + if(pos + 25 > blk_data_size) { + close(fd); + return false; + } + pos += 25; + break; + default: + fprintf(stderr, "VALIDATE: unknown record type 0x%02x in block %u record %u\n", + rec_type, total_blocks, r); + close(fd); + return false; + } + } + + fprintf(stdout, + "VALIDATE: block %u: thread=%llu seq=%u records=%u bytes=%u ts_mono=%s\n", + total_blocks, (unsigned long long)blk_thread_id, blk_sequence, + blk_record_count, blk_data_size, ts_monotonic ? "yes" : "NO"); + } + + close(fd); + + fprintf(stdout, "VALIDATE: total blocks=%u, total records=%u\n", total_blocks, + total_records); + + if(total_records == 0) { + fprintf(stderr, "VALIDATE: WARNING - no records found in profile\n"); + // not necessarily a failure - short tests may not generate records + } + + fprintf(stdout, "VALIDATE: PASSED\n"); + return true; +} + +void top_level_task(const void *args, size_t arglen, const void *userdata, size_t userlen, + Processor p) +{ + log_app.info() << "bgwork profile test starting"; + + // find system memory + Machine machine = Machine::get_machine(); + Machine::MemoryQuery mq(machine); + mq.only_kind(Memory::SYSTEM_MEM).local_address_space().has_capacity(1); + + std::vector sys_mems; + for(Machine::MemoryQuery::iterator it = mq.begin(); it; ++it) + sys_mems.push_back(*it); + + if(sys_mems.empty()) { + log_app.fatal() << "no system memories found!"; + abort(); + } + + log_app.info() << "found " << sys_mems.size() << " system memories"; + + // create instances in available memories + size_t num_elems = TestConfig::copy_size; + IndexSpace<1> is = Rect<1>(0, num_elems - 1); + std::vector field_sizes(1, 1); // 1 byte per element + + std::vector instances; + for(size_t i = 0; i < sys_mems.size() && i < 2; i++) { + RegionInstance inst; + RegionInstance::create_instance(inst, sys_mems[i], is, field_sizes, 0, + ProfilingRequestSet()) + .wait(); + assert(inst.exists()); + instances.push_back(inst); + log_app.info() << "created instance in memory " << sys_mems[i]; + } + + // if only one memory, create two instances in the same memory + if(instances.size() < 2) { + RegionInstance inst; + RegionInstance::create_instance(inst, sys_mems[0], is, field_sizes, 0, + ProfilingRequestSet()) + .wait(); + assert(inst.exists()); + instances.push_back(inst); + } + + // run a batch of copies between the instances + std::vector srcs(1), dsts(1); + Event prev = Event::NO_EVENT; + + for(int i = 0; i < TestConfig::num_copies; i++) { + int src_idx = i % instances.size(); + int dst_idx = (i + 1) % instances.size(); + + srcs[0].set_field(instances[src_idx], 0, 1); + dsts[0].set_field(instances[dst_idx], 0, 1); + + prev = is.copy(srcs, dsts, ProfilingRequestSet(), prev); + } + + // wait for all copies to finish + prev.wait(); + log_app.info() << "all " << TestConfig::num_copies << " copies completed"; + + // clean up instances + for(auto &inst : instances) + inst.destroy(); + + log_app.info() << "bgwork profile test done"; +} + +int main(int argc, char **argv) +{ + // pre-scan for validate-only mode + for(int i = 1; i < argc; i++) { + if(strcmp(argv[i], "-validate") == 0) { + TestConfig::validate = true; + } else if(strcmp(argv[i], "-profile_file") == 0 && i + 1 < argc) { + TestConfig::profile_file = argv[i + 1]; + i++; + } + } + + if(TestConfig::validate) { + // validate-only mode - no Realm runtime needed + if(!validate_profile_file(TestConfig::profile_file)) + return 1; + return 0; + } + + Runtime rt; + rt.init(&argc, &argv); + + CommandLineParser cp; + cp.add_option_int_units("-size", TestConfig::copy_size, 'M') + .add_option_int("-copies", TestConfig::num_copies) + .add_option_string("-profile_file", TestConfig::profile_file); + bool ok = cp.parse_command_line(argc, const_cast(argv)); + assert(ok); + + rt.register_task(TOP_LEVEL_TASK, top_level_task); + + // select a processor to run the top level task on + Processor p = Machine::ProcessorQuery(Machine::get_machine()) + .only_kind(Processor::LOC_PROC) + .first(); + assert(p.exists()); + + Event e = rt.collective_spawn(p, TOP_LEVEL_TASK, 0, 0); + rt.shutdown(e); + int ret = rt.wait_for_shutdown(); + + // validate after shutdown + if(ret == 0) { + struct stat st; + if(stat(TestConfig::profile_file.c_str(), &st) == 0) { + if(!validate_profile_file(TestConfig::profile_file)) + return 1; + } else { + fprintf(stdout, + "NOTE: profile file '%s' not found (profiling may not have been enabled)\n", + TestConfig::profile_file.c_str()); + } + } + return ret; +}