<commit_before>//
//  SCAF Client library implementation
//
//  <NAME> <<EMAIL>> - University of Maryland, 2012
//
#include "../config.h"
#include <sys/time.h>
#include <sys/resource.h>
#include <fcntl.h>
#include <time.h>
#include <zmq.h>
#include "scaf.h"
#if(HAVE_LIBPAPI)
#include <papi.h>
#endif
#include <stdio.h>
#include <signal.h>
#include <unistd.h>
#include <sys/reg.h>
#include <sys/syscall.h>
#include <sys/wait.h>
#include <assert.h>
#include <pthread.h>

#if defined(__linux__)
#include <sys/ptrace.h>
#include <linux/ptrace.h>
#include <sys/mman.h>

#if defined(__i386__)
#define ORIG_ACCUM	(4 * ORIG_EAX)
#define ARGREG	(4 * EBX)
#define ARG2REG	(4 * ECX)
#define ARG3REG	(4 * EDX)
#elif defined(__x86_64__)
#define ORIG_ACCUM	(8 * ORIG_RAX)
#define ARGREG	(8 * RDI)
#define ARG2REG	(8 * RSI)
#define ARG3REG	(8 * RDX)
#elif defined(__tilegx__)
// This is for TileGx, which is 64-bit. Guessing this stuff mostly.
#define ORIG_ACCUM   (8 * TREG_SYSCALL_NR)
#define ARGREG      (8 * 0)
//TODO: These two are just guesses. Verify.
#define ARG2REG      (8 * 1)
#define ARG3REG      (8 * 2)
#else
#error unsupported architecture
#endif

#endif //__linux__

#if defined(__sun)
#include "solaris_trace_utils.h"
// Solaris has actual APIs for tracing syscalls and peeking at arguments, so
// none of the above hackery required for Linux is necessary.
#endif //__sun

#define SCAFD_TIMEOUT_SECONDS 1

// Experiment-related options
// Set this to 1 if parallel execution should wait for experiment to finish;
// otherwise it will be forced to terminate early.
#define SCAF_PARALLEL_WAIT_FOR_EXPERIMENT 0
// Set to 1 if SCAF should enforce a hard limit on the amount of time spent
// running a serial experiment.
#define SCAF_ENFORCE_EXPERIMENT_TIME_LIMIT 0
// The maximum amount of time that a experiment fork will run for.
#define SCAF_EXPERIMENT_TIME_LIMIT_SECONDS 10

// Limit measured efficiency to this value. This can restrict the effects of
// timing glitches resulting in crazy values.
#ifdef __KNC__
#define SCAF_MEASURED_EFF_LIMIT 5.0
#else
#define SCAF_MEASURED_EFF_LIMIT 2.0
#endif //__KNC__

#if defined(__sun)
#define SCAF_LOWPASS_TIME_CONSTANT (15.0)
#elif defined(__KNC__)
#define SCAF_LOWPASS_TIME_CONSTANT (4.0)
#else //__sun
#define SCAF_LOWPASS_TIME_CONSTANT (2.0)
#endif //__sun

// PAPI high-level event to measure scalability by
//#define PAPI_HL_MEASURE PAPI_flops
//#define PAPI_HL_MEASURE PAPI_flips
#define PAPI_HL_MEASURE PAPI_ipc

#ifdef __KNC__
int scaf_last_core_offset;
int scaf_last_threads_per_core;
#endif //__KNC__

static void* scaf_gomp_experiment_control(void *unused);
static scaf_client_experiment_description scaf_experiment_desc;
static inline void scaf_experiment_start(void);
static inline void scaf_experiment_end(int);
void scaf_advise_experiment_start(void);
void scaf_advise_experiment_stop(void);
static int inline scaf_will_create_experiment(void);

static int did_scaf_startup;
static void *scafd;
static void *scafd_context;

static int scafd_available;
static int scaf_disable_experiments = 0;
static int scaf_enable_firsttouch = 0;
static int scaf_experiment_process = 0;
volatile int scaf_experiment_starting = 0;

static int scaf_experiment_running = 0;
static int scaf_lazy_experiments;
static int scaf_mypid;
static int scaf_num_online_hardware_threads;
static int scaf_nullfd;
static int notified_not_malleable = 0;
static int scaf_section_dumps = 0;
static FILE* scaf_sd;

static double scaf_init_rtclock;
static float scaf_section_duration;
static float scaf_section_ipc;
static float scaf_section_start_time;
static float scaf_section_start_process_time;
static float scaf_section_end_time;
static float scaf_section_end_process_time;
static float scaf_serial_duration;
static float scaf_section_efficiency;
static pthread_t scaf_master_thread;

// For rate-limiting communications
// Maximum will be max_comms/per
static double scaf_rate_limit_per;
static double scaf_rate_limit_max_comms;
static double scaf_rate_limit_allowance;
static double scaf_rate_limit_last_check;
static int scaf_skip_communication_for_section  = 0;

static void* current_section_id;
static int current_threads;
static int current_num_clients = 1;
static scaf_client_section *current_section = NULL;
static scaf_client_section *sections = NULL;
static int scaf_in_parallel_section = 0;

#ifdef __KNC__
#define LOWPASS_INITIAL (2.0)
#else
#define LOWPASS_INITIAL (0.5)
#endif //__KNC__

// Discrete IIR, single-pole lowpass filter.
// Time constant rc is expected to be the same across calls. Inputs x and dt
// are the data and time interval, respectively.
#if defined(__GNUC__)
inline float lowpass(float x, float dt, float rc){
#else
float lowpass(float x, float dt, float rc){
#endif //__GNUC__
   static float yp = LOWPASS_INITIAL;
   float alpha = dt / (rc + dt);
   yp = alpha * x + (1.0-alpha) * yp;
   return yp;
}

// Reset the lowpass filter to its initial state.
#if defined(__GNUC__)
inline float lowpass_reset(void){
#else
float lowpass_reset(void){
#endif //__GNUC__
   // By feeding the initial value back in with a zero time-constant, the
   // filter value is set to exactly the initial value.
   return lowpass(LOWPASS_INITIAL, 1.0, 0.0);
}

// Return 1 if we have been communicating too quickly. Implements a token
// bucket. Our convention will be that rate limiting is disabled if
// scaf_rate_limit_max_comms is negative.
static inline int scaf_communication_rate_limit(double current_time){
   if(scaf_rate_limit_max_comms < 0)
      return 0;

   double time_passed = current_time - scaf_rate_limit_last_check;
   scaf_rate_limit_last_check = current_time;
   scaf_rate_limit_allowance += time_passed * (scaf_rate_limit_max_comms / scaf_rate_limit_per);
   if(scaf_rate_limit_allowance > scaf_rate_limit_max_comms)
      scaf_rate_limit_allowance = scaf_rate_limit_max_comms;
   if(scaf_rate_limit_allowance < 1.0){
      return 1; // do not allow the communication
   }
   else{
      scaf_rate_limit_allowance -= 1.0;
      return 0; // allow the communication.
   }
}

static void scaf_feedback_requested(int sig){
   // Don't mask/queue the signal until we're done. Simply ignore it.
   signal(SIGCONT, SIG_IGN);

   // Only respond if we are in a parallel section. If for some reason the
   // experiment process gets this signal, do nothing.
   if(scaf_in_parallel_section && !scaf_experiment_process){
      if(scaf_master_thread == pthread_self()){
         // Disable section dumps for these section start/stops since they are
         // not in the code.
         int scaf_section_dumps_save = scaf_section_dumps;
         scaf_section_dumps = 0;
         // If we're the master thread, fake an empty serial section by ending
         // the parallel section as far as bookkeeping is concerned.
         scaf_section_end();
         // Next, terminate any experiment and collect its results.
         scaf_gomp_experiment_destroy();
         // Finally, continue bookkeeping for the ongoing parallel section.
         scaf_section_start(current_section_id);
         // Re-enable scaf_section_dumps.
         scaf_section_dumps = scaf_section_dumps_save;
      }else{
         // If we're not the master thread, explicitly send the signal to the
         // master thread, but otherwise do nothing.
         pthread_kill(scaf_master_thread, sig);
      }
   }

   // In any case, set up the handler again.
   signal(SIGCONT, scaf_feedback_requested);
}

// If possible, set up a signal handler for SIGCONT to interpret SIGCONT as a
// request to send an update to scafd. This is best-effort, and is allowed to
// fail. TODO: if there was already a handler, call it too.
static void scaf_init_signal_handler(void){
   void *oldsh = NULL;
   oldsh = (void*)signal(SIGCONT, scaf_feedback_requested);
   if(oldsh != (void*)SIG_DFL){
      always_print(BOLDRED "The signal handler for SIGCONT was not SIG_DFL. SCAF has replaced it!" RESET "\n");
   }
}

inline static void scaf_dump_section_header(void){
   fprintf(scaf_sd, "time, section, instance, event\n");
}
inline static void scaf_dump_section_start(void *section_id){
   static unsigned instance = 0;
   fprintf(scaf_sd, "%3.6f, %p, %d, start\n", rtclock(), section_id, instance++);
}
inline static void scaf_dump_section_stop(void *section_id){
   static unsigned instance = 0;
   fprintf(scaf_sd, "%3.6f, %p, %d, stop\n", rtclock(), section_id, instance++);
}

static void* scaf_init(void **context_p);
static int scaf_connect(void *scafd);
static scaf_client_section*  scaf_add_client_section(void *section_id);
static scaf_client_section* scaf_find_client_section(void *section_id);

static scaf_client_section inline *scaf_add_client_section(void *section_id){
   scaf_client_section *new_section = malloc(sizeof(scaf_client_section));
   new_section->section_id = section_id;
   new_section->last_time = 0;
   new_section->last_ipc = 1;
   new_section->experiment_complete = 0;
   new_section->first_touch_complete = !scaf_enable_firsttouch;
   new_section->experiment_serial_ipc = 0.5;
   HASH_ADD_PTR(sections, section_id, new_section);
   return new_section;
}

static scaf_client_section inline *scaf_find_client_section(void *section_id){
   scaf_client_section *found = NULL;
   HASH_FIND_PTR(sections, &section_id, found);
   return found;
}

static int scaf_connect(void *scafd){
   scafd_available = 1;
   zmq_pollitem_t pi;
   pi.socket = scafd;
   pi.events = ZMQ_POLLIN;

   assert(zmq_connect(scafd, SCAF_CONNECT_STRING) == 0);
   // send new client request and get initial num threads
   zmq_msg_t request;
   int retv = zmq_msg_init_size(&request, sizeof(scaf_client_message));
   assert(retv==0);
   scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
   scaf_message->message = SCAF_NEW_CLIENT;
   scaf_message->pid = scaf_mypid;
   scaf_message->section = current_section_id;
#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &request, 0);
#else
   zmq_send(scafd, &request, 0);
#endif
   zmq_msg_close(&request);

   // Stop and poll just to see if we timeout. If no reply, then assume there
   // is no scafd for the rest of execution.
#if ZMQ_VERSION_MAJOR > 2
   int rc = zmq_poll(&pi, 1, SCAFD_TIMEOUT_SECONDS*1000);
#else
   int rc = zmq_poll(&pi, 1, SCAFD_TIMEOUT_SECONDS*1000000);
#endif
   if(rc == 1){
      zmq_msg_t reply;
      zmq_msg_init(&reply);
#if ZMQ_VERSION_MAJOR > 2
      zmq_recvmsg(scafd, &reply, 0);
#else
      zmq_recv(scafd, &reply, 0);
#endif
      scaf_daemon_message response = *((scaf_daemon_message*)(zmq_msg_data(&reply)));
      assert(response.message == SCAF_DAEMON_FEEDBACK);
      zmq_msg_close(&reply);

      // Register scaf_retire() to be called upon exit. This just a courtesy
      // effort to notify scafd; if your platform for some reason never calls
      // this scafd will still figure out that you're gone on its own after a
      // few seconds.
      atexit(scaf_retire);

      scaf_num_online_hardware_threads = response.threads;

      // If a new client has arrived, reset our reported value so that
      // equipartitioning will be effected briefly.
      if(response.num_clients > current_num_clients)
         lowpass_reset();
      current_num_clients = response.num_clients;
      return response.threads;
   } else {
      // No response.
      scafd_available = 0;
      always_print(RED "WARNING: This SCAF client could not communicate with scafd. Using %d threads." RESET "\n", scaf_num_online_hardware_threads);
      return scaf_num_online_hardware_threads;
   }
}


static void* scaf_init(void **context_p){
   scaf_mypid = getpid();
   scaf_init_rtclock = rtclock();
   scaf_nullfd = open("/dev/null", O_WRONLY | O_NONBLOCK);

   scaf_num_online_hardware_threads = scaf_get_num_cpus();

   scaf_init_signal_handler();

   // Initialize stuff for rate limiting
   char *ratelimit = getenv("SCAF_COMM_RATE_LIMIT");
   if(ratelimit)
      scaf_rate_limit_max_comms = atof(ratelimit);
   else
      scaf_rate_limit_max_comms = 5.0;
   scaf_rate_limit_per = 1.0;
   scaf_rate_limit_allowance = scaf_rate_limit_max_comms;
   scaf_rate_limit_last_check = 0;

   // We used to call experiments "training." Obey the old environment variable
   // for now.
   char *noexperiments = getenv("SCAF_DISABLE_EXPERIMENTS");
   char *notraining = getenv("SCAF_DISABLE_TRAINING");
   if(noexperiments)
     scaf_disable_experiments = atoi(noexperiments);
   else if(notraining)
     scaf_disable_experiments = atoi(notraining);
   else
     scaf_disable_experiments = 0;

   char *lazyexperiments = getenv("SCAF_LAZY_EXPERIMENTS");
   if(lazyexperiments)
      scaf_lazy_experiments = atoi(lazyexperiments);
   else
      scaf_lazy_experiments = 1;

   char *firsttouch = getenv("SCAF_ENABLE_FIRSTTOUCH");
   if(firsttouch)
     scaf_enable_firsttouch = atoi(firsttouch);
   else
     scaf_enable_firsttouch = 0;

   char *sectiondumps = getenv("SCAF_SECTION_DUMPS");
   if(sectiondumps)
     scaf_section_dumps = atoi(sectiondumps);
   else
     scaf_section_dumps = 0;

   if(scaf_section_dumps){
      char section_dump_filename[128];
      sprintf(section_dump_filename, "/tmp/scaf-sectiondump.%d.csv", scaf_mypid);
      scaf_sd = fopen(section_dump_filename, "w");
      scaf_dump_section_header();
   }

   void *context = zmq_init(1);
   *context_p = context;
   void *requester = zmq_socket (context, ZMQ_REQ);
#if HAVE_LIBPAPI
   int initval = PAPI_library_init(PAPI_VER_CURRENT);
   PAPI_thread_init((unsigned long (*)(void) )pthread_self);
   assert(initval == PAPI_VER_CURRENT || initval == PAPI_OK);
   assert(PAPI_multiplex_init() == PAPI_OK);
#endif
   scaf_master_thread = pthread_self();

   return requester;
}

void scaf_not_malleable(void){
   // This can be known before a parallel section starts. Initialize
   // communication with scafd if necessary.
   if(!did_scaf_startup){
      scafd = scaf_init(&scafd_context);
      did_scaf_startup=1;

      // scaf_connect gives a reply, but we just ignore it.
      scaf_connect(scafd);
   }

   if(!scafd_available)
      return;

   // send notice that we are not malleable.
   debug_print(BOLDRED "Notifying that we are NOT malleable." RESET "\n");
   zmq_msg_t request;
   zmq_msg_init_size(&request, sizeof(scaf_client_message));
   scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
   scaf_message->message = SCAF_NOT_MALLEABLE;
   scaf_message->pid = scaf_mypid;
#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &request, 0);
#else
   zmq_send(scafd, &request, 0);
#endif
   zmq_msg_close(&request);

   zmq_msg_t reply;
   zmq_msg_init(&reply);
#if ZMQ_VERSION_MAJOR > 2
   zmq_recvmsg(scafd, &reply, 0);
#else
   zmq_recv(scafd, &reply, 0);
#endif
   scaf_daemon_message response = *((scaf_daemon_message*)(zmq_msg_data(&reply)));
   assert(response.message == SCAF_DAEMON_FEEDBACK);
   zmq_msg_close(&reply);

   notified_not_malleable = 1;

   return;
}

void scaf_advise_experiment_start(void){
   debug_print(BOLDRED "Notifying scafd of experiment start." RESET "\n");
   zmq_msg_t request;
   zmq_msg_init_size(&request, sizeof(scaf_client_message));
   scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
   scaf_message->message = SCAF_EXPT_START;
   scaf_message->pid = scaf_mypid;
   scaf_message->message_value.experiment_pid = scaf_experiment_desc.experiment_pid;
#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &request, 0);
#else
   zmq_send(scafd, &request, 0);
#endif
   zmq_msg_close(&request);

   zmq_msg_t reply;
   zmq_msg_init(&reply);
#if ZMQ_VERSION_MAJOR > 2
   zmq_recvmsg(scafd, &reply, 0);
#else
   zmq_recv(scafd, &reply, 0);
#endif
   scaf_daemon_message response = *((scaf_daemon_message*)(zmq_msg_data(&reply)));
   assert(response.message == SCAF_DAEMON_FEEDBACK);
   zmq_msg_close(&reply);
   return;
}

void scaf_advise_experiment_stop(void){
   debug_print(BOLDRED "Notifying scafd of experiment stop." RESET "\n");
   zmq_msg_t request;
   zmq_msg_init_size(&request, sizeof(scaf_client_message));
   scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
   scaf_message->message = SCAF_EXPT_STOP;
   scaf_message->pid = scaf_mypid;
#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &request, 0);
#else
   zmq_send(scafd, &request, 0);
#endif
   zmq_msg_close(&request);

   zmq_msg_t reply;
   zmq_msg_init(&reply);
#if ZMQ_VERSION_MAJOR > 2
   zmq_recvmsg(scafd, &reply, 0);
#else
   zmq_recv(scafd, &reply, 0);
#endif
   scaf_daemon_message response = *((scaf_daemon_message*)(zmq_msg_data(&reply)));
   assert(response.message == SCAF_DAEMON_FEEDBACK);
   zmq_msg_close(&reply);
   return;
}

void scaf_retire(void){
   // If we're just an experiment (with the atexit called before the fork),
   // then don't tell anyone anything.
   if(scaf_experiment_process)
      return;

   // send retire request
   zmq_msg_t request;
   zmq_msg_init_size(&request, sizeof(scaf_client_message));
   scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
   scaf_message->message = SCAF_FORMER_CLIENT;
   scaf_message->pid = scaf_mypid;
#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &request, 0);
#else
   zmq_send(scafd, &request, 0);
#endif
   zmq_msg_close(&request);
   zmq_close (scafd);
   zmq_term (scafd_context);

   if(scaf_section_dumps)
      fclose(scaf_sd);

   return;
}

int scaf_section_start(void* section){
   current_section_id = section;
   if(current_section == NULL || current_section->section_id != section){
      current_section = scaf_find_client_section(current_section_id);
      if(current_section == NULL){
         current_section = scaf_add_client_section(current_section_id);
      }

      if(!did_scaf_startup){
         scafd = scaf_init(&scafd_context);
         did_scaf_startup=1;

         // scaf_connect gives a reply, but we just ignore it.
         scaf_connect(scafd);
      }
   }

   if(scaf_section_dumps)
      scaf_dump_section_start(current_section_id);

   if(!scafd_available){
      current_threads = scaf_num_online_hardware_threads;
      scaf_in_parallel_section = 1;
      return scaf_num_online_hardware_threads;
   }

   if(current_threads < 1)
      current_threads=1;
   if(scaf_section_duration <= 0.000001){
      scaf_section_duration = 0.01;
      scaf_section_efficiency = 0.5;
   }

   // Compute results for reporting before this section
   float scaf_serial_efficiency = 1.0 / current_threads;
   float scaf_latest_efficiency_duration = (scaf_section_duration + scaf_serial_duration);
   float scaf_latest_efficiency = (scaf_section_efficiency * scaf_section_duration + scaf_serial_efficiency * scaf_serial_duration) / scaf_latest_efficiency_duration;
   float scaf_latest_efficiency_smooth = lowpass(scaf_latest_efficiency, scaf_latest_efficiency_duration, SCAF_LOWPASS_TIME_CONSTANT);

   // Collect data for future reporting
#if(HAVE_LIBPAPI)
   {
      float ipc, ptime;
      long long int ins;
      int ret = PAPI_HL_MEASURE(&scaf_section_start_time, &ptime, &ins, &ipc);
      if(ret != PAPI_OK) always_print(RED "WARNING: Bad PAPI things happening. (%s)" RESET "\n", PAPI_strerror(ret));
      scaf_section_start_process_time = pclock();
      scaf_serial_duration = scaf_section_start_time - scaf_section_end_time;

   }
#else
   {
      scaf_section_start_time = (float)(rtclock() - scaf_init_rtclock);
      scaf_serial_duration = scaf_section_start_time - scaf_section_end_time;
   }
#endif

   scaf_skip_communication_for_section = scaf_communication_rate_limit(scaf_section_start_time);
   if(!scaf_skip_communication_for_section){
      // Get num threads
      zmq_msg_t request;
      zmq_msg_init_size(&request, sizeof(scaf_client_message));
      scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
      scaf_message->message = SCAF_SECTION_START;
      scaf_message->pid = scaf_mypid;
      scaf_message->section = section;

      scaf_message->message_value.efficiency = scaf_latest_efficiency_smooth;

#if ZMQ_VERSION_MAJOR > 2
      zmq_sendmsg(scafd, &request, 0);
#else
      zmq_send(scafd, &request, 0);
#endif
      zmq_msg_close(&request);

      zmq_msg_t reply;
      zmq_msg_init(&reply);
#if ZMQ_VERSION_MAJOR > 2
      zmq_recvmsg(scafd, &reply, 0);
#else
      zmq_recv(scafd, &reply, 0);
#endif
      scaf_daemon_message response = *((scaf_daemon_message*)(zmq_msg_data(&reply)));
      assert(response.message == SCAF_DAEMON_FEEDBACK);
      zmq_msg_close(&reply);
      current_num_clients = response.num_clients;
      current_threads = response.threads;
#ifdef __KNC__
      scaf_last_core_offset = response.core_offset;
      scaf_last_threads_per_core = response.threads_per_core;
#endif //__KNC__
   }

   scaf_section_ipc = 0.0;
   scaf_in_parallel_section = 1;

   if(notified_not_malleable)
      return scaf_num_online_hardware_threads;

   if(scaf_will_create_experiment()){
      return current_threads-1;
   }

   return current_threads;
}

void scaf_section_end(void){

   if(scaf_section_dumps)
      scaf_dump_section_stop(current_section_id);

   if(!scafd_available){
      scaf_in_parallel_section = 0;
      return;
   }

#if(HAVE_LIBPAPI)
   {
      float rtime, ptime, ipc;
      long long int ins;
      int ret = PAPI_HL_MEASURE(&scaf_section_end_time, &ptime, &ins, &ipc);
      if(ret != PAPI_OK) always_print(RED "WARNING: Bad PAPI things happening. (%s)" RESET "\n", PAPI_strerror(ret));
      scaf_section_duration = (scaf_section_end_time - scaf_section_start_time);
      scaf_section_end_process_time = pclock();
      // The HL_MEASURE rate we get from PAPI is just for this thread, while it
      // was running. This may not account for all of the wall time. Estimate
      // the effective average rate over all threads by assuming that all
      // threads had similar rates while they were running, and 0 otherwise.
      float oldipc = ipc;
      if(!notified_not_malleable)
#ifdef __KNC__
         ipc *=
            min(1.0,
            (scaf_section_end_process_time-scaf_section_start_process_time) /
            (current_threads*scaf_last_threads_per_core*scaf_section_duration));
#else
         ipc *=
            min(1.0,
            (scaf_section_end_process_time-scaf_section_start_process_time) /
            (current_threads*scaf_section_duration));
#endif //__KNC__
      scaf_section_ipc += ipc;
   }
#else
   {
      scaf_section_ipc += 0.5;
      scaf_section_end_time = (float)(float)(rtclock() - scaf_init_rtclock);
      scaf_section_duration = (scaf_section_end_time - scaf_section_start_time);
   }
#endif

   if(notified_not_malleable)
#ifdef __KNC__
      scaf_section_ipc = (scaf_section_ipc * scaf_num_online_hardware_threads) / current_threads;
#else
      scaf_section_ipc = (scaf_section_ipc * scaf_num_online_hardware_threads) /
         (current_threads * scaf_last_threads_per_core);
#endif //__KNC__

   current_section->last_time = scaf_section_duration;
   current_section->last_ipc  = scaf_section_ipc;
   scaf_section_efficiency = min(SCAF_MEASURED_EFF_LIMIT, scaf_section_ipc / current_section->experiment_serial_ipc);
#ifdef __KNC__
   debug_print(CYAN "Section (%p): @(%d){%f}{sIPC: %f; pIPC: %f} -> {EFF: %f; SPU: %f}" RESET "\n", current_section->section_id, current_threads, scaf_section_duration, current_section->experiment_serial_ipc, scaf_section_ipc, scaf_section_efficiency, scaf_section_efficiency*current_threads);
#else
   debug_print(CYAN "Section (%p): @(%d){%f}{sIPC: %f; pIPC: %f} -> {EFF: %f; SPU: %f}" RESET "\n", current_section->section_id, current_threads, scaf_section_duration, current_section->experiment_serial_ipc, scaf_section_ipc, scaf_section_efficiency, scaf_section_efficiency*current_threads*scaf_last_threads_per_core);
#endif //__KNC__

#ifndef __KNC__
   // If our "parallel" section was actually run on only 1 thread, also store
   // the results as the result of an experiment. (Even if an experiment had
   // already been run.)
   if(current_threads == 1 && !notified_not_malleable){
      current_section->experiment_threads = 1;
      current_section->experiment_serial_ipc = scaf_section_ipc;
      current_section->experiment_parallel_ipc = scaf_section_ipc;
      current_section->experiment_ipc_eff = 1;
      current_section->experiment_ipc_speedup = 1;
      current_section->experiment_complete = 1;
   }

   scaf_in_parallel_section = 0;
   return;
}

static inline void scaf_experiment_start(void){

#if(HAVE_LIBPAPI)
   {
      // Begin gathering information with PAPI.
      PAPI_shutdown();
      int initval = PAPI_library_init(PAPI_VER_CURRENT);
      PAPI_thread_init((unsigned long (*)(void) )pthread_self);
      assert(initval == PAPI_VER_CURRENT || initval == PAPI_OK);
      assert(PAPI_multiplex_init() == PAPI_OK);

      float ipc, ptime;
      long long int ins;
      PAPI_HL_MEASURE(&scaf_section_start_time, &ptime, &ins, &ipc);
      int ret = PAPI_HL_MEASURE(&scaf_section_start_time, &ptime, &ins, &ipc);
      if(ret != PAPI_OK) always_print(RED "WARNING: Bad PAPI things happening. (%s)" RESET "\n", PAPI_strerror(ret));
      scaf_section_start_process_time = pclock();
   }
#else
   {
      scaf_section_start_time = 1.0;
   }
#endif

   // Install the end of the experiment as the SIGINT handler.
   signal(SIGINT, scaf_experiment_end);
   // Also install the end of the experiment as the SIGALRM handler.
   signal(SIGALRM, scaf_experiment_end);

   if(SCAF_ENFORCE_EXPERIMENT_TIME_LIMIT)
      alarm(SCAF_EXPERIMENT_TIME_LIMIT_SECONDS);
}

static inline void scaf_experiment_end(int sig){

   syscall(__NR_scaf_experiment_done);

   // Ignore all the signals which we might still get.
   signal(SIGALRM, SIG_IGN);
   signal(SIGINT, SIG_IGN);

   debug_print(BLUE "SCAF experiment ending.");
   if(sig == SIGALRM){
     debug_print(" (took too long.)\n");
   }
   else if(sig == 0){
     debug_print(" (finished.)\n");
   }
   else if(sig == SIGINT){
     debug_print(" (killed.)\n");
   }
   else {
     debug_print(" (not sure why?)\n");
   }
   debug_print(RESET);

#if(HAVE_LIBPAPI)
   {
      // Get the results from PAPI.
      float ipc, ptime;
      long long int ins;
      int ret = PAPI_HL_MEASURE(&scaf_section_end_time, &ptime, &ins, &ipc);
      if(ret != PAPI_OK) always_print(RED "WARNING: Bad PAPI things happening. (%s)" RESET "\n", PAPI_strerror(ret));
      scaf_section_duration = (scaf_section_end_time - scaf_section_start_time);
      scaf_section_end_process_time = pclock();
      // The HL_MEASURE rate we get from PAPI is just for this thread, while it
      // was running. This may not account for all of the wall time. Estimate
      // the effective average rate over all threads by assuming that all
      // threads had similar rates while they were running, and 0 otherwise.
      float oldipc = ipc;
      if(!notified_not_malleable &&
            scaf_section_end_process_time != scaf_section_start_process_time)
         ipc = ipc *
            (scaf_section_end_process_time-scaf_section_start_process_time) /
            (scaf_section_duration);

      scaf_section_ipc = ipc;
   }
#else
   {
      scaf_section_ipc = 1.0;
      scaf_section_duration = 1.0;
   }
#endif

   void *context = zmq_init(1);
   scafd = zmq_socket (context, ZMQ_REQ);
   char parent_connect_string[64];
   sprintf(parent_connect_string, "ipc:///tmp/scaf-ipc-%d", scaf_mypid);
   assert(0==zmq_connect(scafd, parent_connect_string));
   zmq_msg_t request;
   assert(0==zmq_msg_init_data(&request, &scaf_section_ipc, sizeof(float), NULL, NULL));
   
#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &request, 0);
#else
   zmq_send(scafd, &request, 0);
#endif
   zmq_msg_close(&request);

   zmq_msg_t reply;
   zmq_msg_init(&reply);
#if ZMQ_VERSION_MAJOR > 2
      zmq_recvmsg(scafd, &reply, 0);
#else
      zmq_recv(scafd, &reply, 0);
#endif
   //int response = *((int*)(zmq_msg_data(&reply)));
   zmq_msg_close(&reply);

   zmq_close(scafd);
   _Exit(0);
}

// Just decide whether or not we need to and can make an experiment.
static int inline scaf_will_create_experiment(void){
   // First of all, only experiment if necessary.
   if(scaf_disable_experiments || !scafd_available || current_section->experiment_complete)
      return 0;

   // Do not launch an experiment if we can only use one processor right now.
   if(current_threads < 2)
      return 0;

   // Also do not experiment if we are doing lazy experiments and there is no
   // multiprogramming right now.
   if(scaf_lazy_experiments && current_num_clients < 2)
      return 0;

#if(! HAVE_LIBPAPI)
   {
      return 0;
   }
#endif

   if(!current_section->first_touch_complete)
      return 0;
<commit_msg>libscaf: set up experiment zmq socket first. (Not during signal handler.)<commit_after>//
//  SCAF Client library implementation
//
//  <NAME> <<EMAIL>> - University of Maryland, 2012
//
#include "../config.h"
#include <sys/time.h>
#include <sys/resource.h>
#include <fcntl.h>
#include <time.h>
#include <zmq.h>
#include "scaf.h"
#if(HAVE_LIBPAPI)
#include <papi.h>
#endif
#include <stdio.h>
#include <signal.h>
#include <unistd.h>
#include <sys/reg.h>
#include <sys/syscall.h>
#include <sys/wait.h>
#include <assert.h>
#include <pthread.h>

#if defined(__linux__)
#include <sys/ptrace.h>
#include <linux/ptrace.h>
#include <sys/mman.h>

#if defined(__i386__)
#define ORIG_ACCUM	(4 * ORIG_EAX)
#define ARGREG	(4 * EBX)
#define ARG2REG	(4 * ECX)
#define ARG3REG	(4 * EDX)
#elif defined(__x86_64__)
#define ORIG_ACCUM	(8 * ORIG_RAX)
#define ARGREG	(8 * RDI)
#define ARG2REG	(8 * RSI)
#define ARG3REG	(8 * RDX)
#elif defined(__tilegx__)
// This is for TileGx, which is 64-bit. Guessing this stuff mostly.
#define ORIG_ACCUM   (8 * TREG_SYSCALL_NR)
#define ARGREG      (8 * 0)
//TODO: These two are just guesses. Verify.
#define ARG2REG      (8 * 1)
#define ARG3REG      (8 * 2)
#else
#error unsupported architecture
#endif

#endif //__linux__

#if defined(__sun)
#include "solaris_trace_utils.h"
// Solaris has actual APIs for tracing syscalls and peeking at arguments, so
// none of the above hackery required for Linux is necessary.
#endif //__sun

#define SCAFD_TIMEOUT_SECONDS 1

// Experiment-related options
// Set this to 1 if parallel execution should wait for experiment to finish;
// otherwise it will be forced to terminate early.
#define SCAF_PARALLEL_WAIT_FOR_EXPERIMENT 0
// Set to 1 if SCAF should enforce a hard limit on the amount of time spent
// running a serial experiment.
#define SCAF_ENFORCE_EXPERIMENT_TIME_LIMIT 0
// The maximum amount of time that a experiment fork will run for.
#define SCAF_EXPERIMENT_TIME_LIMIT_SECONDS 10

// Limit measured efficiency to this value. This can restrict the effects of
// timing glitches resulting in crazy values.
#ifdef __KNC__
#define SCAF_MEASURED_EFF_LIMIT 5.0
#else
#define SCAF_MEASURED_EFF_LIMIT 2.0
#endif //__KNC__

#if defined(__sun)
#define SCAF_LOWPASS_TIME_CONSTANT (15.0)
#elif defined(__KNC__)
#define SCAF_LOWPASS_TIME_CONSTANT (4.0)
#else //__sun
#define SCAF_LOWPASS_TIME_CONSTANT (2.0)
#endif //__sun

// PAPI high-level event to measure scalability by
//#define PAPI_HL_MEASURE PAPI_flops
//#define PAPI_HL_MEASURE PAPI_flips
#define PAPI_HL_MEASURE PAPI_ipc

#ifdef __KNC__
int scaf_last_core_offset;
int scaf_last_threads_per_core;
#endif //__KNC__

static void* scaf_gomp_experiment_control(void *unused);
static scaf_client_experiment_description scaf_experiment_desc;
static inline void scaf_experiment_start(void);
static inline void scaf_experiment_end(int);
void scaf_advise_experiment_start(void);
void scaf_advise_experiment_stop(void);
static int inline scaf_will_create_experiment(void);

static int did_scaf_startup;
static void *scafd;
static void *scafd_context;

static int scafd_available;
static int scaf_disable_experiments = 0;
static int scaf_enable_firsttouch = 0;
static int scaf_experiment_process = 0;
volatile int scaf_experiment_starting = 0;
static zmq_msg_t scaf_experiment_request;
static zmq_msg_t scaf_experiment_reply;

static int scaf_experiment_running = 0;
static int scaf_lazy_experiments;
static int scaf_mypid;
static int scaf_num_online_hardware_threads;
static int scaf_nullfd;
static int notified_not_malleable = 0;
static int scaf_section_dumps = 0;
static FILE* scaf_sd;

static double scaf_init_rtclock;
static float scaf_section_duration;
static float scaf_section_ipc;
static float scaf_section_start_time;
static float scaf_section_start_process_time;
static float scaf_section_end_time;
static float scaf_section_end_process_time;
static float scaf_serial_duration;
static float scaf_section_efficiency;
static pthread_t scaf_master_thread;

// For rate-limiting communications
// Maximum will be max_comms/per
static double scaf_rate_limit_per;
static double scaf_rate_limit_max_comms;
static double scaf_rate_limit_allowance;
static double scaf_rate_limit_last_check;
static int scaf_skip_communication_for_section  = 0;

static void* current_section_id;
static int current_threads;
static int current_num_clients = 1;
static scaf_client_section *current_section = NULL;
static scaf_client_section *sections = NULL;
static int scaf_in_parallel_section = 0;

#ifdef __KNC__
#define LOWPASS_INITIAL (2.0)
#else
#define LOWPASS_INITIAL (0.5)
#endif //__KNC__

// Discrete IIR, single-pole lowpass filter.
// Time constant rc is expected to be the same across calls. Inputs x and dt
// are the data and time interval, respectively.
#if defined(__GNUC__)
inline float lowpass(float x, float dt, float rc){
#else
float lowpass(float x, float dt, float rc){
#endif //__GNUC__
   static float yp = LOWPASS_INITIAL;
   float alpha = dt / (rc + dt);
   yp = alpha * x + (1.0-alpha) * yp;
   return yp;
}

// Reset the lowpass filter to its initial state.
#if defined(__GNUC__)
inline float lowpass_reset(void){
#else
float lowpass_reset(void){
#endif //__GNUC__
   // By feeding the initial value back in with a zero time-constant, the
   // filter value is set to exactly the initial value.
   return lowpass(LOWPASS_INITIAL, 1.0, 0.0);
}

// Return 1 if we have been communicating too quickly. Implements a token
// bucket. Our convention will be that rate limiting is disabled if
// scaf_rate_limit_max_comms is negative.
static inline int scaf_communication_rate_limit(double current_time){
   if(scaf_rate_limit_max_comms < 0)
      return 0;

   double time_passed = current_time - scaf_rate_limit_last_check;
   scaf_rate_limit_last_check = current_time;
   scaf_rate_limit_allowance += time_passed * (scaf_rate_limit_max_comms / scaf_rate_limit_per);
   if(scaf_rate_limit_allowance > scaf_rate_limit_max_comms)
      scaf_rate_limit_allowance = scaf_rate_limit_max_comms;
   if(scaf_rate_limit_allowance < 1.0){
      return 1; // do not allow the communication
   }
   else{
      scaf_rate_limit_allowance -= 1.0;
      return 0; // allow the communication.
   }
}

static void scaf_feedback_requested(int sig){
   // Don't mask/queue the signal until we're done. Simply ignore it.
   signal(SIGCONT, SIG_IGN);

   // Only respond if we are in a parallel section. If for some reason the
   // experiment process gets this signal, do nothing.
   if(scaf_in_parallel_section && !scaf_experiment_process){
      if(scaf_master_thread == pthread_self()){
         // Disable section dumps for these section start/stops since they are
         // not in the code.
         int scaf_section_dumps_save = scaf_section_dumps;
         scaf_section_dumps = 0;
         // If we're the master thread, fake an empty serial section by ending
         // the parallel section as far as bookkeeping is concerned.
         scaf_section_end();
         // Next, terminate any experiment and collect its results.
         scaf_gomp_experiment_destroy();
         // Finally, continue bookkeeping for the ongoing parallel section.
         scaf_section_start(current_section_id);
         // Re-enable scaf_section_dumps.
         scaf_section_dumps = scaf_section_dumps_save;
      }else{
         // If we're not the master thread, explicitly send the signal to the
         // master thread, but otherwise do nothing.
         pthread_kill(scaf_master_thread, sig);
      }
   }

   // In any case, set up the handler again.
   signal(SIGCONT, scaf_feedback_requested);
}

// If possible, set up a signal handler for SIGCONT to interpret SIGCONT as a
// request to send an update to scafd. This is best-effort, and is allowed to
// fail. TODO: if there was already a handler, call it too.
static void scaf_init_signal_handler(void){
   void *oldsh = NULL;
   oldsh = (void*)signal(SIGCONT, scaf_feedback_requested);
   if(oldsh != (void*)SIG_DFL){
      always_print(BOLDRED "The signal handler for SIGCONT was not SIG_DFL. SCAF has replaced it!" RESET "\n");
   }
}

inline static void scaf_dump_section_header(void){
   fprintf(scaf_sd, "time, section, instance, event\n");
}
inline static void scaf_dump_section_start(void *section_id){
   static unsigned instance = 0;
   fprintf(scaf_sd, "%3.6f, %p, %d, start\n", rtclock(), section_id, instance++);
}
inline static void scaf_dump_section_stop(void *section_id){
   static unsigned instance = 0;
   fprintf(scaf_sd, "%3.6f, %p, %d, stop\n", rtclock(), section_id, instance++);
}

static void* scaf_init(void **context_p);
static int scaf_connect(void *scafd);
static scaf_client_section*  scaf_add_client_section(void *section_id);
static scaf_client_section* scaf_find_client_section(void *section_id);

static scaf_client_section inline *scaf_add_client_section(void *section_id){
   scaf_client_section *new_section = malloc(sizeof(scaf_client_section));
   new_section->section_id = section_id;
   new_section->last_time = 0;
   new_section->last_ipc = 1;
   new_section->experiment_complete = 0;
   new_section->first_touch_complete = !scaf_enable_firsttouch;
   new_section->experiment_serial_ipc = 0.5;
   HASH_ADD_PTR(sections, section_id, new_section);
   return new_section;
}

static scaf_client_section inline *scaf_find_client_section(void *section_id){
   scaf_client_section *found = NULL;
   HASH_FIND_PTR(sections, &section_id, found);
   return found;
}

static int scaf_connect(void *scafd){
   scafd_available = 1;
   zmq_pollitem_t pi;
   pi.socket = scafd;
   pi.events = ZMQ_POLLIN;

   assert(zmq_connect(scafd, SCAF_CONNECT_STRING) == 0);
   // send new client request and get initial num threads
   zmq_msg_t request;
   int retv = zmq_msg_init_size(&request, sizeof(scaf_client_message));
   assert(retv==0);
   scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
   scaf_message->message = SCAF_NEW_CLIENT;
   scaf_message->pid = scaf_mypid;
   scaf_message->section = current_section_id;
#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &request, 0);
#else
   zmq_send(scafd, &request, 0);
#endif
   zmq_msg_close(&request);

   // Stop and poll just to see if we timeout. If no reply, then assume there
   // is no scafd for the rest of execution.
#if ZMQ_VERSION_MAJOR > 2
   int rc = zmq_poll(&pi, 1, SCAFD_TIMEOUT_SECONDS*1000);
#else
   int rc = zmq_poll(&pi, 1, SCAFD_TIMEOUT_SECONDS*1000000);
#endif
   if(rc == 1){
      zmq_msg_t reply;
      zmq_msg_init(&reply);
#if ZMQ_VERSION_MAJOR > 2
      zmq_recvmsg(scafd, &reply, 0);
#else
      zmq_recv(scafd, &reply, 0);
#endif
      scaf_daemon_message response = *((scaf_daemon_message*)(zmq_msg_data(&reply)));
      assert(response.message == SCAF_DAEMON_FEEDBACK);
      zmq_msg_close(&reply);

      // Register scaf_retire() to be called upon exit. This just a courtesy
      // effort to notify scafd; if your platform for some reason never calls
      // this scafd will still figure out that you're gone on its own after a
      // few seconds.
      atexit(scaf_retire);

      scaf_num_online_hardware_threads = response.threads;

      // If a new client has arrived, reset our reported value so that
      // equipartitioning will be effected briefly.
      if(response.num_clients > current_num_clients)
         lowpass_reset();
      current_num_clients = response.num_clients;
      return response.threads;
   } else {
      // No response.
      scafd_available = 0;
      always_print(RED "WARNING: This SCAF client could not communicate with scafd. Using %d threads." RESET "\n", scaf_num_online_hardware_threads);
      return scaf_num_online_hardware_threads;
   }
}


static void* scaf_init(void **context_p){
   scaf_mypid = getpid();
   scaf_init_rtclock = rtclock();
   scaf_nullfd = open("/dev/null", O_WRONLY | O_NONBLOCK);

   scaf_num_online_hardware_threads = scaf_get_num_cpus();

   scaf_init_signal_handler();

   // Initialize stuff for rate limiting
   char *ratelimit = getenv("SCAF_COMM_RATE_LIMIT");
   if(ratelimit)
      scaf_rate_limit_max_comms = atof(ratelimit);
   else
      scaf_rate_limit_max_comms = 5.0;
   scaf_rate_limit_per = 1.0;
   scaf_rate_limit_allowance = scaf_rate_limit_max_comms;
   scaf_rate_limit_last_check = 0;

   // We used to call experiments "training." Obey the old environment variable
   // for now.
   char *noexperiments = getenv("SCAF_DISABLE_EXPERIMENTS");
   char *notraining = getenv("SCAF_DISABLE_TRAINING");
   if(noexperiments)
     scaf_disable_experiments = atoi(noexperiments);
   else if(notraining)
     scaf_disable_experiments = atoi(notraining);
   else
     scaf_disable_experiments = 0;

   char *lazyexperiments = getenv("SCAF_LAZY_EXPERIMENTS");
   if(lazyexperiments)
      scaf_lazy_experiments = atoi(lazyexperiments);
   else
      scaf_lazy_experiments = 1;

   char *firsttouch = getenv("SCAF_ENABLE_FIRSTTOUCH");
   if(firsttouch)
     scaf_enable_firsttouch = atoi(firsttouch);
   else
     scaf_enable_firsttouch = 0;

   char *sectiondumps = getenv("SCAF_SECTION_DUMPS");
   if(sectiondumps)
     scaf_section_dumps = atoi(sectiondumps);
   else
     scaf_section_dumps = 0;

   if(scaf_section_dumps){
      char section_dump_filename[128];
      sprintf(section_dump_filename, "/tmp/scaf-sectiondump.%d.csv", scaf_mypid);
      scaf_sd = fopen(section_dump_filename, "w");
      scaf_dump_section_header();
   }

   void *context = zmq_init(1);
   *context_p = context;
   void *requester = zmq_socket (context, ZMQ_REQ);
#if HAVE_LIBPAPI
   int initval = PAPI_library_init(PAPI_VER_CURRENT);
   PAPI_thread_init((unsigned long (*)(void) )pthread_self);
   assert(initval == PAPI_VER_CURRENT || initval == PAPI_OK);
   assert(PAPI_multiplex_init() == PAPI_OK);
#endif
   scaf_master_thread = pthread_self();

   return requester;
}

void scaf_not_malleable(void){
   // This can be known before a parallel section starts. Initialize
   // communication with scafd if necessary.
   if(!did_scaf_startup){
      scafd = scaf_init(&scafd_context);
      did_scaf_startup=1;

      // scaf_connect gives a reply, but we just ignore it.
      scaf_connect(scafd);
   }

   if(!scafd_available)
      return;

   // send notice that we are not malleable.
   debug_print(BOLDRED "Notifying that we are NOT malleable." RESET "\n");
   zmq_msg_t request;
   zmq_msg_init_size(&request, sizeof(scaf_client_message));
   scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
   scaf_message->message = SCAF_NOT_MALLEABLE;
   scaf_message->pid = scaf_mypid;
#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &request, 0);
#else
   zmq_send(scafd, &request, 0);
#endif
   zmq_msg_close(&request);

   zmq_msg_t reply;
   zmq_msg_init(&reply);
#if ZMQ_VERSION_MAJOR > 2
   zmq_recvmsg(scafd, &reply, 0);
#else
   zmq_recv(scafd, &reply, 0);
#endif
   scaf_daemon_message response = *((scaf_daemon_message*)(zmq_msg_data(&reply)));
   assert(response.message == SCAF_DAEMON_FEEDBACK);
   zmq_msg_close(&reply);

   notified_not_malleable = 1;

   return;
}

void scaf_advise_experiment_start(void){
   debug_print(BOLDRED "Notifying scafd of experiment start." RESET "\n");
   zmq_msg_t request;
   zmq_msg_init_size(&request, sizeof(scaf_client_message));
   scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
   scaf_message->message = SCAF_EXPT_START;
   scaf_message->pid = scaf_mypid;
   scaf_message->message_value.experiment_pid = scaf_experiment_desc.experiment_pid;
#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &request, 0);
#else
   zmq_send(scafd, &request, 0);
#endif
   zmq_msg_close(&request);

   zmq_msg_t reply;
   zmq_msg_init(&reply);
#if ZMQ_VERSION_MAJOR > 2
   zmq_recvmsg(scafd, &reply, 0);
#else
   zmq_recv(scafd, &reply, 0);
#endif
   scaf_daemon_message response = *((scaf_daemon_message*)(zmq_msg_data(&reply)));
   assert(response.message == SCAF_DAEMON_FEEDBACK);
   zmq_msg_close(&reply);
   return;
}

void scaf_advise_experiment_stop(void){
   debug_print(BOLDRED "Notifying scafd of experiment stop." RESET "\n");
   zmq_msg_t request;
   zmq_msg_init_size(&request, sizeof(scaf_client_message));
   scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
   scaf_message->message = SCAF_EXPT_STOP;
   scaf_message->pid = scaf_mypid;
#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &request, 0);
#else
   zmq_send(scafd, &request, 0);
#endif
   zmq_msg_close(&request);

   zmq_msg_t reply;
   zmq_msg_init(&reply);
#if ZMQ_VERSION_MAJOR > 2
   zmq_recvmsg(scafd, &reply, 0);
#else
   zmq_recv(scafd, &reply, 0);
#endif
   scaf_daemon_message response = *((scaf_daemon_message*)(zmq_msg_data(&reply)));
   assert(response.message == SCAF_DAEMON_FEEDBACK);
   zmq_msg_close(&reply);
   return;
}

void scaf_retire(void){
   // If we're just an experiment (with the atexit called before the fork),
   // then don't tell anyone anything.
   if(scaf_experiment_process)
      return;

   // send retire request
   zmq_msg_t request;
   zmq_msg_init_size(&request, sizeof(scaf_client_message));
   scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
   scaf_message->message = SCAF_FORMER_CLIENT;
   scaf_message->pid = scaf_mypid;
#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &request, 0);
#else
   zmq_send(scafd, &request, 0);
#endif
   zmq_msg_close(&request);
   zmq_close (scafd);
   zmq_term (scafd_context);

   if(scaf_section_dumps)
      fclose(scaf_sd);

   return;
}

int scaf_section_start(void* section){
   current_section_id = section;
   if(current_section == NULL || current_section->section_id != section){
      current_section = scaf_find_client_section(current_section_id);
      if(current_section == NULL){
         current_section = scaf_add_client_section(current_section_id);
      }

      if(!did_scaf_startup){
         scafd = scaf_init(&scafd_context);
         did_scaf_startup=1;

         // scaf_connect gives a reply, but we just ignore it.
         scaf_connect(scafd);
      }
   }

   if(scaf_section_dumps)
      scaf_dump_section_start(current_section_id);

   if(!scafd_available){
      current_threads = scaf_num_online_hardware_threads;
      scaf_in_parallel_section = 1;
      return scaf_num_online_hardware_threads;
   }

   if(current_threads < 1)
      current_threads=1;
   if(scaf_section_duration <= 0.000001){
      scaf_section_duration = 0.01;
      scaf_section_efficiency = 0.5;
   }

   // Compute results for reporting before this section
   float scaf_serial_efficiency = 1.0 / current_threads;
   float scaf_latest_efficiency_duration = (scaf_section_duration + scaf_serial_duration);
   float scaf_latest_efficiency = (scaf_section_efficiency * scaf_section_duration + scaf_serial_efficiency * scaf_serial_duration) / scaf_latest_efficiency_duration;
   float scaf_latest_efficiency_smooth = lowpass(scaf_latest_efficiency, scaf_latest_efficiency_duration, SCAF_LOWPASS_TIME_CONSTANT);

   // Collect data for future reporting
#if(HAVE_LIBPAPI)
   {
      float ipc, ptime;
      long long int ins;
      int ret = PAPI_HL_MEASURE(&scaf_section_start_time, &ptime, &ins, &ipc);
      if(ret != PAPI_OK) always_print(RED "WARNING: Bad PAPI things happening. (%s)" RESET "\n", PAPI_strerror(ret));
      scaf_section_start_process_time = pclock();
      scaf_serial_duration = scaf_section_start_time - scaf_section_end_time;

   }
#else
   {
      scaf_section_start_time = (float)(rtclock() - scaf_init_rtclock);
      scaf_serial_duration = scaf_section_start_time - scaf_section_end_time;
   }
#endif

   scaf_skip_communication_for_section = scaf_communication_rate_limit(scaf_section_start_time);
   if(!scaf_skip_communication_for_section){
      // Get num threads
      zmq_msg_t request;
      zmq_msg_init_size(&request, sizeof(scaf_client_message));
      scaf_client_message *scaf_message = (scaf_client_message*)(zmq_msg_data(&request));
      scaf_message->message = SCAF_SECTION_START;
      scaf_message->pid = scaf_mypid;
      scaf_message->section = section;

      scaf_message->message_value.efficiency = scaf_latest_efficiency_smooth;

#if ZMQ_VERSION_MAJOR > 2
      zmq_sendmsg(scafd, &request, 0);
#else
      zmq_send(scafd, &request, 0);
#endif
      zmq_msg_close(&request);

      zmq_msg_t reply;
      zmq_msg_init(&reply);
#if ZMQ_VERSION_MAJOR > 2
      zmq_recvmsg(scafd, &reply, 0);
#else
      zmq_recv(scafd, &reply, 0);
#endif
      scaf_daemon_message response = *((scaf_daemon_message*)(zmq_msg_data(&reply)));
      assert(response.message == SCAF_DAEMON_FEEDBACK);
      zmq_msg_close(&reply);
      current_num_clients = response.num_clients;
      current_threads = response.threads;
#ifdef __KNC__
      scaf_last_core_offset = response.core_offset;
      scaf_last_threads_per_core = response.threads_per_core;
#endif //__KNC__
   }

   scaf_section_ipc = 0.0;
   scaf_in_parallel_section = 1;

   if(notified_not_malleable)
      return scaf_num_online_hardware_threads;

   if(scaf_will_create_experiment()){
      return current_threads-1;
   }

   return current_threads;
}

void scaf_section_end(void){

   if(scaf_section_dumps)
      scaf_dump_section_stop(current_section_id);

   if(!scafd_available){
      scaf_in_parallel_section = 0;
      return;
   }

#if(HAVE_LIBPAPI)
   {
      float rtime, ptime, ipc;
      long long int ins;
      int ret = PAPI_HL_MEASURE(&scaf_section_end_time, &ptime, &ins, &ipc);
      if(ret != PAPI_OK) always_print(RED "WARNING: Bad PAPI things happening. (%s)" RESET "\n", PAPI_strerror(ret));
      scaf_section_duration = (scaf_section_end_time - scaf_section_start_time);
      scaf_section_end_process_time = pclock();
      // The HL_MEASURE rate we get from PAPI is just for this thread, while it
      // was running. This may not account for all of the wall time. Estimate
      // the effective average rate over all threads by assuming that all
      // threads had similar rates while they were running, and 0 otherwise.
      float oldipc = ipc;
      if(!notified_not_malleable)
#ifdef __KNC__
         ipc *=
            min(1.0,
            (scaf_section_end_process_time-scaf_section_start_process_time) /
            (current_threads*scaf_last_threads_per_core*scaf_section_duration));
#else
         ipc *=
            min(1.0,
            (scaf_section_end_process_time-scaf_section_start_process_time) /
            (current_threads*scaf_section_duration));
#endif //__KNC__
      scaf_section_ipc += ipc;
   }
#else
   {
      scaf_section_ipc += 0.5;
      scaf_section_end_time = (float)(float)(rtclock() - scaf_init_rtclock);
      scaf_section_duration = (scaf_section_end_time - scaf_section_start_time);
   }
#endif

   if(notified_not_malleable)
#ifdef __KNC__
      scaf_section_ipc = (scaf_section_ipc * scaf_num_online_hardware_threads) / current_threads;
#else
      scaf_section_ipc = (scaf_section_ipc * scaf_num_online_hardware_threads) /
         (current_threads * scaf_last_threads_per_core);
#endif //__KNC__

   current_section->last_time = scaf_section_duration;
   current_section->last_ipc  = scaf_section_ipc;
   scaf_section_efficiency = min(SCAF_MEASURED_EFF_LIMIT, scaf_section_ipc / current_section->experiment_serial_ipc);
#ifdef __KNC__
   debug_print(CYAN "Section (%p): @(%d){%f}{sIPC: %f; pIPC: %f} -> {EFF: %f; SPU: %f}" RESET "\n", current_section->section_id, current_threads, scaf_section_duration, current_section->experiment_serial_ipc, scaf_section_ipc, scaf_section_efficiency, scaf_section_efficiency*current_threads);
#else
   debug_print(CYAN "Section (%p): @(%d){%f}{sIPC: %f; pIPC: %f} -> {EFF: %f; SPU: %f}" RESET "\n", current_section->section_id, current_threads, scaf_section_duration, current_section->experiment_serial_ipc, scaf_section_ipc, scaf_section_efficiency, scaf_section_efficiency*current_threads*scaf_last_threads_per_core);
#endif //__KNC__

#ifndef __KNC__
   // If our "parallel" section was actually run on only 1 thread, also store
   // the results as the result of an experiment. (Even if an experiment had
   // already been run.)
   if(current_threads == 1 && !notified_not_malleable){
      current_section->experiment_threads = 1;
      current_section->experiment_serial_ipc = scaf_section_ipc;
      current_section->experiment_parallel_ipc = scaf_section_ipc;
      current_section->experiment_ipc_eff = 1;
      current_section->experiment_ipc_speedup = 1;
      current_section->experiment_complete = 1;
   }
#endif // !__KNC__

skip_math:
   scaf_in_parallel_section = 0;
   return;
}

static inline void scaf_experiment_start(void){

   // Set up a new ZMQ context of our own.
   void *context = zmq_init(1);
   scafd = zmq_socket (context, ZMQ_REQ);
   char parent_connect_string[64];
   sprintf(parent_connect_string, "ipc:///tmp/scaf-ipc-%d", scaf_mypid);
   assert(0==zmq_connect(scafd, parent_connect_string));
   // Pre-allocate messages which will be used to send the experiment results.
   assert(0==zmq_msg_init_data(&scaf_experiment_request, &scaf_section_ipc, sizeof(float), NULL, NULL));
   zmq_msg_init(&scaf_experiment_reply);


#if(HAVE_LIBPAPI)
   {
      // Begin gathering information with PAPI.
      PAPI_shutdown();
      int initval = PAPI_library_init(PAPI_VER_CURRENT);
      PAPI_thread_init((unsigned long (*)(void) )pthread_self);
      assert(initval == PAPI_VER_CURRENT || initval == PAPI_OK);
      assert(PAPI_multiplex_init() == PAPI_OK);

      float ipc, ptime;
      long long int ins;
      PAPI_HL_MEASURE(&scaf_section_start_time, &ptime, &ins, &ipc);
      int ret = PAPI_HL_MEASURE(&scaf_section_start_time, &ptime, &ins, &ipc);
      if(ret != PAPI_OK) always_print(RED "WARNING: Bad PAPI things happening. (%s)" RESET "\n", PAPI_strerror(ret));
      scaf_section_start_process_time = pclock();
   }
#else
   {
      scaf_section_start_time = 1.0;
   }
#endif

   // Install the end of the experiment as the SIGINT handler.
   signal(SIGINT, scaf_experiment_end);
   // Also install the end of the experiment as the SIGALRM handler.
   signal(SIGALRM, scaf_experiment_end);

   if(SCAF_ENFORCE_EXPERIMENT_TIME_LIMIT)
      alarm(SCAF_EXPERIMENT_TIME_LIMIT_SECONDS);
}

static inline void scaf_experiment_end(int sig){

   syscall(__NR_scaf_experiment_done);

   // Ignore all the signals which we might still get.
   signal(SIGALRM, SIG_IGN);
   signal(SIGINT, SIG_IGN);

   debug_print(BLUE "SCAF experiment ending.");
   if(sig == SIGALRM){
     debug_print(" (took too long.)\n");
   }
   else if(sig == 0){
     debug_print(" (finished.)\n");
   }
   else if(sig == SIGINT){
     debug_print(" (killed.)\n");
   }
   else {
     debug_print(" (not sure why?)\n");
   }
   debug_print(RESET);

#if(HAVE_LIBPAPI)
   {
      // Get the results from PAPI.
      float ipc, ptime;
      long long int ins;
      int ret = PAPI_HL_MEASURE(&scaf_section_end_time, &ptime, &ins, &ipc);
      if(ret != PAPI_OK) always_print(RED "WARNING: Bad PAPI things happening. (%s)" RESET "\n", PAPI_strerror(ret));
      scaf_section_duration = (scaf_section_end_time - scaf_section_start_time);
      scaf_section_end_process_time = pclock();
      // The HL_MEASURE rate we get from PAPI is just for this thread, while it
      // was running. This may not account for all of the wall time. Estimate
      // the effective average rate over all threads by assuming that all
      // threads had similar rates while they were running, and 0 otherwise.
      float oldipc = ipc;
      if(!notified_not_malleable &&
            scaf_section_end_process_time != scaf_section_start_process_time)
         ipc = ipc *
            (scaf_section_end_process_time-scaf_section_start_process_time) /
            (scaf_section_duration);

      scaf_section_ipc = ipc;
   }
#else
   {
      scaf_section_ipc = 1.0;
      scaf_section_duration = 1.0;
   }
#endif

#if ZMQ_VERSION_MAJOR > 2
   zmq_sendmsg(scafd, &scaf_experiment_request, 0);
#else
   zmq_send(scafd, &scaf_experiment_request, 0);
#endif
   zmq_msg_close(&scaf_experiment_request);

#if ZMQ_VERSION_MAJOR > 2
      zmq_recvmsg(scafd, &scaf_experiment_reply, 0);
#else
      zmq_recv(scafd, &scaf_experiment_reply, 0);
#endif
   //int response = *((int*)(zmq_msg_data(&reply)));
   zmq_msg_close(&scaf_experiment_reply);

   zmq_close(scafd);
   _Exit(0);
}

// Just decide whether or not we need to and can make an experiment.
static int inline scaf_will_create_experiment(void){
   // First of all, only experiment if necessary.
   if(scaf_disable_experiments || !scafd_available || current_section->experiment_complete)
      return 0;

   // Do not launch an experiment if we can only use one processor right now.
   if(current_threads < 2)
      return 0;

   // Also do not experiment if we are doing lazy experiments and there is no
   // multiprogramming right now.
   if(scaf_lazy_experiments && current_num_clients < 2)
      return 0;

#if(! HAVE_LIBPAPI)
   {
      return 0;
   }
#endif

   if(!current_section->first_touch_complete)
      return 0;

{'fraction_non_alphanumeric': 0.09706769262763856, 'fraction_numerical': 0.007764028976465287, 'mean_word_length': 4.205972573078311, 'pattern_counts': {'":': 0, '<': 55, '<?xml version=': 0, '>': 155, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1401348', 'n_tokens_mistral': 21047, 'n_tokens_neox': 20463, 'n_words': 4665}