Write spike output to out.dat with parallel i/o when MPI is enabled (#71)

Author: pramodk

README.md

@@ -211,11 +211,10 @@ Note that celsius and dt, if not specified, will get their values from the model
 
 # Results
 
-Currently CoreNEURON only outputs spike data. When running the simulation, each MPI rank writes spike information
-into a file `out.#mpi_rank`. These files should be combined and sorted to compare with NEURON spike output.
+Currently CoreNEURON only outputs spike data. Spike output file need to be sorted to compare with NEURON:
 
 ```
-cat out[0-9]*.dat | sort -k 1n,1n -k 2n,2n > out.spk
+sort -k 1n,1n -k 2n,2n out.dat > out.spk
 ```
 
 # Running tests

coreneuron/nrniv/output_spikes.cpp

@@ -27,14 +27,17 @@ THE POSSIBILITY OF SUCH DAMAGE.
 */
 
 #include <iostream>
+#include <sstream>
+#include <string.h>
 #include <stdexcept> // std::lenght_error
 #include <vector>
 #include "coreneuron/nrnconf.h"
 #include "coreneuron/nrniv/nrniv_decl.h"
 #include "coreneuron/nrniv/output_spikes.h"
 #include "coreneuron/nrnmpi/nrnmpi.h"
 #include "coreneuron/nrniv/nrnmutdec.h"
-#include "coreneuron/utils/sdprintf.h"
+#include "coreneuron/nrnmpi/nrnmpi_impl.h"
+#include "coreneuron/nrnmpi/nrnmpidec.h"
 
 std::vector<double> spikevec_time;
 std::vector<int> spikevec_gid;
@@ -55,28 +58,113 @@ static MUTDEC
 void spikevec_lock() {
     MUTLOCK
 }
+
 void spikevec_unlock() {
     MUTUNLOCK
 }
 
-void output_spikes(const char* outpath) {
-    char fnamebuf[100];
-    sd_ptr fname = sdprintf(fnamebuf, sizeof(fnamebuf), "%s/out%d.dat", outpath, nrnmpi_myid);
-    FILE* f = fopen(fname, "w");
+#if NRNMPI
+/** Write generated spikes to out.dat using mpi parallel i/o.
+ *  \todo : MPI related code should be factored into nrnmpi.c
+ *          Check spike record length which is set to 64 chars
+ */
+void output_spikes_parallel(const char* outpath) {
+    std::stringstream ss;
+    ss << outpath << "/out.dat";
+    std::string fname = ss.str();
+
+    // remove if file already exist
+    if(nrnmpi_myid == 0) {
+        remove(fname.c_str());
+    }
+    nrnmpi_barrier();
+
+    // each spike record in the file is time + gid (64 chars sufficient)
+    const int SPIKE_RECORD_LEN = 64;
+    unsigned num_spikes = spikevec_gid.size();
+    unsigned num_bytes = (sizeof(char) * num_spikes * SPIKE_RECORD_LEN);
+    char *spike_data = (char*) malloc(num_bytes);
+
+    if(spike_data == NULL) {
+        printf("Error while writing spikes due to memory allocation\n");
+        return;
+    }
+
+    // empty if no spikes
+    strcpy(spike_data, "");
+
+    // populate buffer with all spike entries
+    char spike_entry[SPIKE_RECORD_LEN];
+    for(unsigned i = 0; i < num_spikes; i++) {
+        snprintf(spike_entry, 64, "%.8g\t%d\n", spikevec_time[i], spikevec_gid[i]);
+        strcat(spike_data, spike_entry);
+    }
+
+    // calculate offset into global file. note that we don't write
+    // all num_bytes but only "populated" buffer
+    unsigned long num_chars = strlen(spike_data);
+    unsigned long offset = 0;
+
+    // global offset into file
+    MPI_Exscan(&num_chars, &offset, 1, MPI_UNSIGNED_LONG, MPI_SUM, MPI_COMM_WORLD);
+
+    // write to file using parallel mpi i/o
+    MPI_File fh;
+    MPI_Status status;
+
+    // ibm mpi (bg-q) expects char* instead of const char* (even though it's standard)
+    int op_status = MPI_File_open(MPI_COMM_WORLD, (char*) fname.c_str(), MPI_MODE_CREATE | MPI_MODE_WRONLY, MPI_INFO_NULL, &fh);
+    if(op_status != MPI_SUCCESS && nrnmpi_myid == 0) {
+        std::cerr << "Error while opening spike output file " << fname << std::endl;
+        abort();
+    }
+
+    op_status = MPI_File_write_at_all(fh, offset, spike_data, num_chars, MPI_BYTE, &status);
+    if(op_status != MPI_SUCCESS && nrnmpi_myid == 0) {
+        std::cerr << "Error while writing spike output " << std::endl;
+        abort();
+    }
+
+    MPI_File_close(&fh);
+}
+#endif
+
+void output_spikes_serial(const char* outpath) {
+    std::stringstream ss;
+    ss << outpath << "/out.dat";
+    std::string fname = ss.str();
+
+    // remove if file already exist
+    remove(fname.c_str());
+
+    FILE* f = fopen(fname.c_str(), "w");
     if (!f && nrnmpi_myid == 0) {
         std::cout << "WARNING: Could not open file for writing spikes." << std::endl;
         return;
     }
 
-    for (int i = 0; i < spikevec_gid.size(); ++i)
+    for (unsigned i = 0; i < spikevec_gid.size(); ++i)
         if (spikevec_gid[i] > -1)
             fprintf(f, "%.8g\t%d\n", spikevec_time[i], spikevec_gid[i]);
 
     fclose(f);
 }
 
+void output_spikes(const char* outpath) {
+#if NRNMPI
+    if(nrnmpi_initialized()) {
+        output_spikes_parallel(outpath);
+    } else {
+        output_spikes_serial(outpath);
+    }
+#else
+    output_spikes_serial(outpath);
+#endif
+}
+
+
 void validation(std::vector<std::pair<double, int> >& res) {
-    for (int i = 0; i < spikevec_gid.size(); ++i)
+    for (unsigned i = 0; i < spikevec_gid.size(); ++i)
         if (spikevec_gid[i] > -1)
             res.push_back(std::make_pair(spikevec_time[i], spikevec_gid[i]));
 }

coreneuron/nrnmpi/nrnmpi.c

@@ -207,3 +207,11 @@ void nrn_fatal_error(const char* msg) {
     }
     nrn_abort(-1);
 }
+
+int nrnmpi_initialized() {
+    int flag = 0;
+#if NRNMPI
+    MPI_Initialized(&flag);
+#endif
+    return flag;
+}

coreneuron/nrnmpi/nrnmpidec.h

@@ -125,7 +125,7 @@ extern long nrnmpi_long_allreduce(long x, int type);
 extern void nrnmpi_dbl_allreduce_vec(double* src, double* dest, int cnt, int type);
 extern void nrnmpi_long_allreduce_vec(long* src, long* dest, int cnt, int type);
 extern void nrnmpi_dbl_allgather(double* s, double* r, int n);
-
+extern int nrnmpi_initialized();
 #if NRN_MULTISEND
 extern void nrnmpi_multisend_comm();
 extern void nrnmpi_multisend(NRNMPI_Spike* spk, int n, int* hosts);

tests/integration/integration_test.sh.in

@@ -24,14 +24,13 @@ fi
 # diff outputed files with reference
 cd @CMAKE_CURRENT_BINARY_DIR@/@SIM_NAME@
 
-if [ ! -f out0.dat ]
+if [ ! -f out.dat ]
 then
   echo "No output files. Test failed!"
   exit 1
 fi
 
-cat out[0-9]*.dat > out_cn.dat
-sort -k 1n,1n -k 2n,2n out_cn.dat > sort_out.dat
+sort -k 1n,1n -k 2n,2n out.dat > sort_out.dat
 diff -w sort_out.dat @CMAKE_CURRENT_SOURCE_DIR@/@SIM_NAME@/out.dat.ref > diff.dat 2>&1
 
 if [ -s diff.dat ]