I'm not an OMPI person, just one of the users, but you can have the code I use
to verify that my build is using the Infiniband network. You may need to
change the path to the Infiniband counters. It uses OpenSHMEM but you can
convert it MPI or whatever you use for parallel codes.
// Compile with -std=c++11
// All in one file to make it easier to share
//
#include <sys/types.h>
#include <sys/time.h>
#include <stdio.h>
#include <unistd.h>
#include <shmem.h>
#include <stdlib.h>
#include <string.h>
#include <string>
#include <iostream>
#include <fstream>
#include <array>
#include <memory>
#include <sstream>
#include <vector>
#include <map>
using namespace std;
using std::string;
using std::cout;
using std::cerr;
class NetworkCounters {
public:
NetworkCounters() {}
~NetworkCounters() {}
virtual void reset() = 0;
// Values are -1 on error.
// Keys are "xmit_Mb", "xmit_pkts", "recv_Mb", "recv_pkts"
virtual void get_counter_values(std::map<string, int64_t>& values) = 0;
};
class IbCounters : public NetworkCounters {
// To build for Infiniband/Neptune set the environment:
// module unload sos
// source /export/share/esp/ib_openshmem/VARS.sh
// At runtime:
// export PTL_IFACE_NAME=ib1
public:
IbCounters() {
ib_path = string("/sys/class/infiniband/mlx4_0/ports/1/counters/");
}
~IbCounters() {}
void reset() {
// Need to validate that this user-level command works
system("/usr/sbin/perfquery -C mlx4_0 -R -a");
}
void get_counter_values(std::map<string, int64_t>& values) {
values.clear();
values["xmit_Mb"] = open_and_read(string("port_xmit_data"));
values["xmit_pkts"] = open_and_read(string("port_xmit_packets"));
values["recv_Mb"] = open_and_read(string("port_rcv_data"));
values["recv_pkts"] = open_and_read(string("port_rcv_packets"));
}
private:
string ib_path;
int64_t open_and_read(string fname) {
std::ifstream fs(ib_path + fname);
if (!fs.is_open()) {
return -1;
}
int64_t val;
fs >> val;
fs.close();
return val;
}
}; // end class IbCounters
class OpaCounters : public NetworkCounters {
// To build for OmniPath, set the environment: module load sos
// Problem: We can call opainfo at most 25 times per second. In tests
// the time to parse the opainfo output was insignificant. It's the call
// alone that is the cost.
public:
OpaCounters() {
opainfo = string("/usr/sbin/opainfo");
}
~OpaCounters() {}
void reset() {
// Sorry - haven't figured out how to reset. Joe Robichaux on the case.
}
void get_counter_values(std::map<string, int64_t>& values) {
values.clear();
values["xmit_Mb"] = -1;
values["xmit_pkts"] = -1;
values["recv_Mb"] = -1;
values["recv_Mb"] = -1;
std::vector<string> lines;
if (get_opa_info(lines) != 0) {
return;
}
// We want these lines:
// Xmit Data: 34844887 MB Pkts: 363712809088
// Recv Data: 34832523 MB Pkts: 363595900175
string xmit("Xmit");
string recv("Recv");
string data("Data:");
std::vector<long> val;
int count(0);
for (int i=0; i < static_cast<int>(lines.size()); i++) {
string w1, w2;
istringstream iss(lines[i]);
iss >> w1 >> w2;
if (w2 == data) {
if (w1 == xmit) {
if (get_counters(lines[i], val) != 0) {
return;
}
values["xmit_Mb"] = val[0];
values["xmit_pkts"] = val[1];
count += 1;
}
else if (w1 == recv) {
if (get_counters(lines[i], val) != 0) {
return;
}
values["recv_Mb"] = val[0];
values["recv_pkts"] = val[1];
count += 1;
}
if (count == 2) {
break;
}
}
}
return;
}
private:
string opainfo;
void to_lines(string const& in, std::vector<string> &out) {
char newline('\n'); // assume Unix-style line termination
size_t from(0), to(0), last(in.size());
out.clear();
while (from < last) {
to = in.find_first_of(newline, from);
if (to == string::npos) {
if (last - from > 0) {
to = last; // last line has no newline
}
else {
break;
}
}
out.push_back(in.substr(from, to-from));
from = to + 1;
while ((from < last) && (in[from] == newline)) {
from++; // skip blank lines
}
}
}
int get_counters(string const& line, std::vector<long> &counters) {
long x, y;
string w1, w2, w3, w4;
std::istringstream iss(line);
try {
iss >> w1 >> w2 >> x >> w3 >> w4 >> y;
} catch (...) {
ostringstream msg("get_counter can not parse: ");
msg << line;
cerr << msg.str() << endl;
return 1;
}
counters.clear();
counters.push_back(x);
counters.push_back(y);
return 0;
}
int get_opa_info(std::vector<string>& lines) {
std::array<char, 256> buf;
string output;
std::shared_ptr<FILE> pipe(popen(opainfo.c_str(), "r"), pclose);
if (!pipe) {
ostringstream msg("get_opa_info can not find ");
msg << opainfo;
cerr << msg.str() << endl;
return 1;
}
while (!feof(pipe.get())) {
if (fgets(buf.data(), 256, pipe.get()) != nullptr) {
output += buf.data();
}
}
lines.clear();
to_lines(output, lines);
return 0;
}
}; // end class OpaCounters
void do_comm(int n, int *buf, size_t num_ints, int pe, int src_pe, int dest_pe)
{
if (pe == src_pe) {
for (int i = 0; i < n; i++) {
shmem_int_put(buf, buf, num_ints, dest_pe);
buf[0] = buf[num_ints - 1] = i;
shmem_quiet(); // returns when data available at receiver
}
}
}
void get_ifconfig(string fname) {
string cmd("ifconfig &> ");
cmd += fname;
system(cmd.c_str());
}
double mbytes_per_second(int nMb, struct timeval &tv) {
double seconds = tv.tv_sec + (static_cast<double>(tv.tv_usec) / 1000000.0);
return (static_cast<double>(nMb) / seconds);
}
void usage(int pe) {
if (pe == 0) {
cout << "Arguments:" << endl;
cout << " Number of bursts" << endl;
cout << " Burst size in Mbytes" << endl;
cout << " \"ib\" or \"opa\"" << endl;
cout << "PE 0 will send {Number of bursts} shmem_puts of size {Burst size
in Mbytes}" << endl;
cout << " Mbytes to PE 1." << endl;
cout << " Interface will be Infiniband (ib) or OmniPath/STL1 (opa)." <<
endl;
}
shmem_finalize();
exit(1);
}
int get_int(char *c, int pe) {
int val;
std::istringstream iss1(c);
try {
iss1 >> val;
}
catch (...) {
usage(pe);
exit(1);
}
return val;
}
int main(int argc, char *argv[]) {
shmem_init();
int pe = shmem_my_pe();
int npes = shmem_n_pes();
if (npes < 2) {
cerr << "need at least 2 PEs" << endl;
shmem_finalize();
return 1;
}
if (argc < 4) {
usage(pe);
return 1;
}
unsigned rounds = get_int(argv[1], pe);
unsigned nMb = get_int(argv[2], pe);
bool use_ib = false;
NetworkCounters *counters = nullptr;
string net_type(argv[3]);
if (net_type == string("ib")) {
use_ib=true;
counters = new IbCounters();
}
else if (net_type != string("opa")) {
usage(pe);
}
else {
counters = new OpaCounters();
}
int src_pe = 0;
int dest_pe = 1;
char nm[128];
gethostname(nm, 128);
shmem_barrier_all();
if (pe == src_pe) {
cerr << rounds << " sends of size " << nMb << " Mbytes, " << endl;
cerr << "PE: " << pe << " node: " << nm << " role: Sender" << endl;
shmem_barrier_all();
}
else if (pe == dest_pe) {
shmem_barrier_all();
cerr << "PE: " << pe << " node: " << nm << " role: Receiver" << endl;
}
//counters->reset();
struct timeval tv0;
gettimeofday(&tv0, nullptr);
std::map<string, int64_t> initial;
counters->get_counter_values(initial);
#if 0
if (pe == src_pe) {
get_ifconfig("before.txt"); // because counters make no sense
}
#endif
size_t megabyte = 1024 * 1024;
size_t xmit_size = megabyte * nMb;
int *buf = reinterpret_cast<int *>(shmem_malloc(xmit_size));
size_t num_ints = xmit_size / sizeof(int);
do_comm(rounds, buf, num_ints, pe, src_pe, dest_pe);
shmem_barrier_all();
struct timeval tv1;
gettimeofday(&tv1, nullptr);
std::map<string, int64_t> final;
counters->get_counter_values(final);
#if 0
if (pe == src_pe) {
get_ifconfig("after.txt"); // because counters make no sense
}
#endif
long xmit_Mb = final["xmit_Mb"] - initial["xmit_Mb"];
long xmit_pkts = final["xmit_pkts"] - initial["xmit_pkts"];
long recv_Mb = final["recv_Mb"] - initial["recv_Mb"];
long recv_pkts = final["recv_pkts"] - initial["recv_pkts"];
struct timeval diff;
timersub(&tv1, &tv0, &diff);
long total_mbytes = nMb * rounds;
if (pe == src_pe) {
shmem_barrier_all();
shmem_barrier_all();
cerr << "SENDER" << endl;
cerr << "Total bytes sent: " << total_mbytes << " Mbytes" << endl;
cerr << "Counter recv MB: " << recv_Mb*4 << " (scaled by 4)" << endl;
cerr << "Counter xmit MB: " << xmit_Mb*4 << " (scaled by 4)" << endl;
cerr << "Time: " << diff.tv_sec << " sec " << diff.tv_usec << " usec" <<
endl;
cerr << "Roughly " << mbytes_per_second(total_mbytes, diff);
cerr << " Mbytes / second" << endl;
}
else if (pe == dest_pe) {
cerr << "RECEIVER" << endl;
cerr << "Total bytes received: " << total_mbytes << " Mbytes" << endl;
cerr << "Counter recv MB: " << recv_Mb*4 << " (scaled by 4)" << endl;
cerr << "Counter xmit MB: " << xmit_Mb*4 << " (scaled by 4)" << endl;
shmem_barrier_all();
shmem_barrier_all();
}
shmem_free(buf);
shmem_finalize();
delete counters;
return 0;
}
Date: Wed, 15 Jan 2020 21:22:03 +0000
From: SOPORTE MODEMAT
<soporte.mode...@epn.edu.ec<mailto:soporte.mode...@epn.edu.ec>>
To: "users@lists.open-mpi.org<mailto:users@lists.open-mpi.org>"
<users@lists.open-mpi.org<mailto:users@lists.open-mpi.org>>
Subject: [OMPI users] need a tool and its use to verify use of
infiniband network
Message-ID:
<dm6pr17mb28093bde314d78e71c646e93c6...@dm6pr17mb2809.namprd17.prod.outlook.com<mailto:dm6pr17mb28093bde314d78e71c646e93c6...@dm6pr17mb2809.namprd17.prod.outlook.com>>
Content-Type: text/plain; charset="iso-8859-1"
Hello guys.
I would like you to help me with one tool or manner to verify the use of
infiniband network interface when I run the command:
/opt/mpi/openmpi_intel-2.1.1/bin/mpirun --mca btl self,openib,vader python
mpi_hola.py Is there a way to verify that the infiniband interface is being
used? If so, how can I do it?
Thank you in advance for your help
Saludos cordiales.
Msc. Mercy Anchundia Ruiz.
-----
Lee Ann Riesen, Intel Corp, Hillsboro, OR, USA
