root/dmtcp_coordinator.cpp

DEFINITIONS

1. readProcessInfo
2. getNewVirtualPid
3. handleUserCommand
4. printStatus
5. updateMinimumState
6. onData
7. removeStaleSharedAreaFile
8. preExitCleanup
9. onDisconnect
10. initializeComputation
11. onConnect
12. processDmtUserCmd
13. validateRestartingWorkerProcess
14. validateNewWorkerProcess
15. startCheckpoint
16. broadcastMessage
17. getStatus
18. writeRestartScript
19. signalHandler
20. setupSignalHandlers
21. calcLocalAddr
22. resetCkptTimer
23. updateCheckpointInterval
24. eventLoop
25. addDataSocket
26. main

   1 /****************************************************************************
   2  *   Copyright (C) 2006-2013 by Jason Ansel, Kapil Arya, and Gene Cooperman *
   3  *   jansel@csail.mit.edu, kapil@ccs.neu.edu, gene@ccs.neu.edu              *
   4  *                                                                          *
   5  *  This file is part of DMTCP.                                             *
   6  *                                                                          *
   7  *  DMTCP is free software: you can redistribute it and/or                  *
   8  *  modify it under the terms of the GNU Lesser General Public License as   *
   9  *  published by the Free Software Foundation, either version 3 of the      *
  10  *  License, or (at your option) any later version.                         *
  11  *                                                                          *
  12  *  DMTCP is distributed in the hope that it will be useful,                *
  13  *  but WITHOUT ANY WARRANTY; without even the implied warranty of          *
  14  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           *
  15  *  GNU Lesser General Public License for more details.                     *
  16  *                                                                          *
  17  *  You should have received a copy of the GNU Lesser General Public        *
  18  *  License along with DMTCP:dmtcp/src.  If not, see                        *
  19  *  <http://www.gnu.org/licenses/>.                                         *
  20  ****************************************************************************/
  22 /****************************************************************************
  23  * Coordinator code logic:                                                  *
  24  * main calls eventLoop, a top-level event loop.                            *
  25  * eventLoop calls:  onConnect, onData, onDisconnect, startCheckpoint       *
  26  *   when client or dmtcp_command talks to coordinator.                     *
  27  * onConnect called on msg at listener port.  It passes control to:         *
  28  *   handleUserCommand, which takes single char arg ('s', 'c', 'k', 'q', ...)*
  29  * handleUserCommand calls broadcastMessage to send data back               *
  30  * any message sent by broadcastMessage takes effect only on returning      *
  31  *   back up to top level monitorSockets                                    *
  32  * Hence, even for checkpoint, handleUserCommand just changes state,        *
  33  *   broadcasts an initial checkpoint command, and then returns to top      *
  34  *   level.  Replies from clients then drive further state changes.         *
  35  * The prefix command 'b' (blocking) from dmtcp_command modifies behavior   *
  36  *   of 'c' so that the reply to dmtcp_command happens only when clients    *
  37  *   are back in RUNNING state.                                             *
  38  * onData called when a message arrives at a client's port.  It either      *
  39  *   processes a per-client special request, or continues the protocol      *
  40  *   for a checkpoint or restart sequence (see below).                      *
  41  *                                                                          *
  42  * updateMinimumState() is responsible for keeping track of states.         *
  43  * The coordinator keeps a ComputationStatus, with minimumState and         *
  44  *   maximumState for states of all workers, accessed through getStatus()   *
  45  *   or through minimumState()                                              *
  46  * The states for a worker (client) are:                                    *
  47  * Checkpoint: RUNNING -> SUSPENDED -> FD_LEADER_ELECTION -> DRAINED        *
  48  *                -> CHECKPOINTED -> NAME_SERVICE_DATA_REGISTERED           *
  49  *                -> DONE_QUERYING -> REFILLED -> RUNNING                   *
  50  * Restart:    RESTARTING -> CHECKPOINTED -> NAME_SERVICE_DATA_REGISTERED   *
  51  *                -> DONE_QUERYING -> REFILLED -> RUNNING                   *
  52  * If debugging, set gdb breakpoint on:                                     *
  53  *   DmtcpCoordinator::onConnect                                            *
  54  *   DmtcpCoordinator::onData                                               *
  55  *   DmtcpCoordinator::handleUserCommand                                    *
  56  *   DmtcpCoordinator::broadcastMessage                                     *
  57  ****************************************************************************/
  59 #include <stdlib.h>
  60 #include "dmtcp_coordinator.h"
  61 #include "constants.h"
  62 #include "protectedfds.h"
  63 #include "dmtcpmessagetypes.h"
  64 #include "lookup_service.h"
  65 #include "syscallwrappers.h"
  66 #include "util.h"
  67 #include "../jalib/jconvert.h"
  68 #include "../jalib/jtimer.h"
  69 #include "../jalib/jfilesystem.h"
  70 #include <stdio.h>
  71 #include <unistd.h>
  72 #include <sys/stat.h>
  73 #include <algorithm>
  74 #include <iomanip>
  75 #include <sys/wait.h>
  76 #include <sys/types.h>
  77 #include <sys/stat.h>
  78 #include <netdb.h>
  79 #include <arpa/inet.h>
  80 #include <fcntl.h>
  81 #undef min
  82 #undef max
  84 #define BINARY_NAME "dmtcp_coordinator"
  86 using namespace dmtcp;
  88 static const char* theHelpMessage =
  89   "COMMANDS:\n"
  90   "  l : List connected nodes\n"
  91   "  s : Print status message\n"
  92   "  c : Checkpoint all nodes\n"
  93   "  i : Print current checkpoint interval\n"
  94   "      (To change checkpoint interval, use dmtcp_command)\n"
  95   "  k : Kill all nodes\n"
  96   "  q : Kill all nodes and quit\n"
  97   "  ? : Show this message\n"
  98   "\n"
  99 ;
 101 static const char* theUsage =
 102   "Usage: dmtcp_coordinator [OPTIONS] [port]\n"
 103   "Coordinates checkpoints between multiple processes.\n\n"
 104   "Options:\n"
 105   "  -p, --coord-port PORT_NUM (environment variable DMTCP_COORD_PORT)\n"
 106   "      Port to listen on (default: 7779)\n"
 107   "  --port-file filename\n"
 108   "      File to write listener port number.\n"
 109   "      (Useful with '--port 0', which is used to assign a random port)\n"
 110   "  --ckptdir (environment variable DMTCP_CHECKPOINT_DIR):\n"
 111   "      Directory to store dmtcp_restart_script.sh (default: ./)\n"
 112   "  --tmpdir (environment variable DMTCP_TMPDIR):\n"
 113   "      Directory to store temporary files (default: env var TMDPIR or /tmp)\n"
 114   "  --exit-on-last\n"
 115   "      Exit automatically when last client disconnects\n"
 116   "  --exit-after-ckpt\n"
 117   "      Exit automatically after checkpoint is created\n"
 118   "  --daemon\n"
 119   "      Run silently in the background after detaching from the parent process.\n"
 120   "  -i, --interval (environment variable DMTCP_CHECKPOINT_INTERVAL):\n"
 121   "      Time in seconds between automatic checkpoints\n"
 122   "      (default: 0, disabled)\n"
 123   "  --help:\n"
 124   "      Print this message and exit.\n"
 125   "  --version:\n"
 126   "      Print version information and exit.\n"
 127   "\n"
 128   "COMMANDS:\n"
 129   "      type '?<return>' at runtime for list\n"
 130   "\n"
HELP_AND_CONTACT_INFO
 132   "\n"
 133 ;
 136 static const char* theRestartScriptHeader =
 137   "#!/bin/bash\n\n"
 138   "set -m # turn on job control\n\n"
 139   "#This script launches all the restarts in the background.\n"
 140   "#Suggestions for editing:\n"
 141   "#  1. For those processes executing on the localhost, remove\n"
 142   "#     'ssh <hostname> from the start of the line.\n"
 143   "#  2. If using ssh, verify that ssh does not require passwords or other\n"
 144   "#     prompts.\n"
 145   "#  3. Verify that the dmtcp_restart command is in your path on all hosts,\n"
 146   "#     otherwise set the dmt_rstr_cmd appropriately.\n"
 147   "#  4. Verify DMTCP_COORD_HOST and DMTCP_COORD_PORT match the location of\n"
 148   "#     the dmtcp_coordinator. If necessary, add\n"
 149   "#     'DMTCP_COORD_PORT=<dmtcp_coordinator port>' after\n"
 150   "#     'DMTCP_COORD_HOST=<...>'.\n"
 151   "#  5. Remove the '&' from a line if that process reads STDIN.\n"
 152   "#     If multiple processes read STDIN then prefix the line with\n"
 153   "#     'xterm -hold -e' and put '&' at the end of the line.\n"
 154   "#  6. Processes on same host can be restarted with single dmtcp_restart\n"
 155   "#     command.\n\n\n"
 156 ;
 159 static const char* theRestartScriptCheckLocal =
 160   "check_local()\n"
 161   "{\n"
 162   "  worker_host=$1\n"
 163   "  unset is_local_node\n"
 164   "  worker_ip=$(gethostip -d $worker_host 2> /dev/null)\n"
 165   "  if [ -z \"$worker_ip\" ]; then\n"
 166   "    worker_ip=$(nslookup $worker_host | grep -A1 'Name:' | grep 'Address:' | sed -e 's/Address://' -e 's/ //' -e 's/ //')\n"
 167   "  fi\n"
 168   "  if [ -z \"$worker_ip\" ]; then\n"
 169   "    worker_ip=$(getent ahosts $worker_host |grep \"^[0-9]\\+\\.[0-9]\\+\\.[0-9]\\+\\.[0-9]\\+ *STREAM\" | cut -d' ' -f1)\n"
 170   "  fi\n"
 171   "  if [ -z \"$worker_ip\" ]; then\n"
 172   "    echo Could not find ip-address for $worker_host. Exiting...\n"
 173   "    exit 1\n"
 174   "  fi\n"
 175   "  ifconfig_path=$(which ifconfig)\n"
 176   "  if [ -z \"$ifconfig_path\" ]; then\n"
 177   "    ifconfig_path=\"/sbin/ifconfig\"\n"
 178   "  fi\n"
 179   "  output=$($ifconfig_path -a | grep \"inet addr:.*${worker_ip} .*Bcast\")\n"
 180   "  if [ -n \"$output\" ]; then\n"
 181   "    is_local_node=1\n"
 182   "  else\n"
 183   "    is_local_node=0\n"
 184   "  fi\n"
 185   "}\n\n\n";
 187 static const char *slurmHelperContactFunction =
 188 "pass_slurm_helper_contact()\n"
 189 "{\n"
 190 "  LOCAL_FILES=\"$1\"\n"
 191 "  # Create temp directory if needed\n"
 192 "  if [ -n \"$DMTCP_TMPDIR\" ]; then\n"
 193 "    CURRENT_TMPDIR=$DMTCP_TMPDIR/dmtcp-`whoami`@`hostname`\n"
 194 "  elif [ -n \"$TMPDIR\" ]; then\n"
 195 "    CURRENT_TMPDIR=$TMPDIR/dmtcp-`whoami`@`hostname`\n"
 196 "  else\n"
 197 "    CURRENT_TMPDIR=/tmp/dmtcp-`whoami`@`hostname`\n"
 198 "  fi\n"
 199 "  if [ ! -d \"$CURRENT_TMPDIR\" ]; then\n"
 200 "    mkdir -p $CURRENT_TMPDIR\n"
 201 "  fi\n"
 202 "  # Create files with SLURM environment\n"
 203 "  for CKPT_FILE in $LOCAL_FILES; do\n"
 204 "    SUFFIX=${CKPT_FILE%%.dmtcp}\n"
 205 "    SLURM_ENV_FILE=$CURRENT_TMPDIR/slurm_env_${SUFFIX##*_}\n"
 206 "    echo \"DMTCP_SRUN_HELPER_ADDR=$DMTCP_SRUN_HELPER_ADDR\" >> $SLURM_ENV_FILE\n"
 207 "  done\n"
 208 "}\n\n\n";
 210 static const char* theRestartScriptUsage =
 211   "usage_str='USAGE:\n"
 212   "  dmtcp_restart_script.sh [OPTIONS]\n\n"
 213   "OPTIONS:\n"
 214   "  --coord-host, -h, (environment variable DMTCP_COORD_HOST):\n"
 215   "      Hostname where dmtcp_coordinator is running\n"
 216   "  --coord-port, -p, (environment variable DMTCP_COORD_PORT):\n"
 217   "      Port where dmtcp_coordinator is running\n"
 218   "  --hostfile <arg0> :\n"
 219   "      Provide a hostfile (One host per line, \"#\" indicates comments)\n"
 220   "  --ckptdir, -d, (environment variable DMTCP_CHECKPOINT_DIR):\n"
 221   "      Directory to store checkpoint images\n"
 222   "      (default: use the same directory used in previous checkpoint)\n"
 223   "  --restartdir, -d, (environment variable DMTCP_RESTART_DIR):\n"
 224   "      Directory to read checkpoint images from\n"
 225   "  --tmpdir, -t, (environment variable DMTCP_TMPDIR):\n"
 226   "      Directory to store temporary files (default: $TMDPIR or /tmp)\n"
 227   "  --no-strict-uid-checking:\n"
 228   "      Disable uid checking for the checkpoint image.  This allows the\n"
 229   "        checkpoint image to be restarted by a different user than the one\n"
 230   "        that create it. (environment variable DMTCP_DISABLE_UID_CHECKING)\n"
 231   "  --interval, -i, (environment variable DMTCP_CHECKPOINT_INTERVAL):\n"
 232   "      Time in seconds between automatic checkpoints\n"
 233   "      (Default: Use pre-checkpoint value)\n"
 234   "  --help:\n"
 235   "      Print this message and exit.\'\n"
 236   "\n\n"
 237 ;
 239 static const char* theRestartScriptCmdlineArgHandler =
 240   "if [ $# -gt 0 ]; then\n"
 241   "  while [ $# -gt 0 ]\n"
 242   "  do\n"
 243   "    if [ $1 = \"--help\" ]; then\n"
 244   "      echo \"$usage_str\"\n"
 245   "      exit\n"
 246   "    elif [ $# -ge 1 ]; then\n"
 247   "      case \"$1\" in\n"
 248   "        --coord-host|--host|-h)\n"
 249   "          coord_host=\"$2\"\n"
 250   "          shift; shift;;\n"
 251   "        --coord-port|--port|-p)\n"
 252   "          coord_port=\"$2\"\n"
 253   "          shift; shift;;\n"
 254   "        --hostfile)\n"
 255   "          hostfile=\"$2\"\n"
 256   "          if [ ! -f \"$hostfile\" ]; then\n"
 257   "            echo \"ERROR: hostfile $hostfile not found\"\n"
 258   "            exit\n"
 259   "          fi\n"
 260   "          shift; shift;;\n"
 261   "        --restartdir|-d)\n"
 262   "          DMTCP_RESTART_DIR=$2\n"
 263   "          shift; shift;;\n"
 264   "        --ckptdir|-d)\n"
 265   "          DMTCP_CKPT_DIR=$2\n"
 266   "          shift; shift;;\n"
 267   "        --tmpdir|-t)\n"
 268   "          DMTCP_TMPDIR=$2\n"
 269   "          shift; shift;;\n"
 270   "        --no-strict-uid-checking)\n"
 271   "          noStrictUidChecking=\"--no-strict-uid-checking\"\n"
 272   "          shift;;\n"
 273   "        --interval|-i)\n"
 274   "          checkpoint_interval=$2\n"
 275   "          shift; shift;;\n"
 276   "        *)\n"
 277   "          echo \"$0: unrecognized option \'$1\'. See correct usage below\"\n"
 278   "          echo \"$usage_str\"\n"
 279   "          exit;;\n"
 280   "      esac\n"
 281   "    elif [ $1 = \"--help\" ]; then\n"
 282   "      echo \"$usage_str\"\n"
 283   "      exit\n"
 284   "    else\n"
 285   "      echo \"$0: Incorrect usage.  See correct usage below\"\n"
 286   "      echo\n"
 287   "      echo \"$usage_str\"\n"
 288   "      exit\n"
 289   "    fi\n"
 290   "  done\n"
 291   "fi\n\n"
 292 ;
 294 static const char* theRestartScriptSingleHostProcessing =
 295   "ckpt_files=\"\"\n"
 296   "if [ ! -z \"$DMTCP_RESTART_DIR\" ]; then\n"
 297   "  for tmp in $given_ckpt_files; do\n"
 298   "    ckpt_files=\"$DMTCP_RESTART_DIR/$(basename $tmp) $ckpt_files\"\n"
 299   "  done\n"
 300   "else\n"
 301   "  ckpt_files=$given_ckpt_files\n"
 302   "fi\n\n"
 304   "coordinator_info=\"--coord-host $coord_host --coord-port $coord_port\"\n"
 306   "tmpdir=\n"
 307   "if [ ! -z \"$DMTCP_TMPDIR\" ]; then\n"
 308   "  tmpdir=\"--tmpdir $DMTCP_TMPDIR\"\n"
 309   "fi\n\n"
 311   "ckpt_dir=\n"
 312   "if [ ! -z \"$DMTCP_CKPT_DIR\" ]; then\n"
 313   "  ckpt_dir=\"--ckptdir $DMTCP_CKPT_DIR\"\n"
 314   "fi\n\n"
 316   "exec $dmt_rstr_cmd $coordinator_info $ckpt_dir \\\n"
 317   "  $maybejoin --interval \"$checkpoint_interval\" $tmpdir $noStrictUidChecking \\\n"
 318   "  $ckpt_files\n"
 319 ;
 321 static const char* theRestartScriptMultiHostProcessing =
 322   "worker_ckpts_regexp=\\\n"
 323   "\'[^:]*::[ \\t\\n]*\\([^ \\t\\n]\\+\\)[ \\t\\n]*:\\([a-z]\\+\\):[ \\t\\n]*\\([^:]\\+\\)\'\n\n"
 325   "worker_hosts=$(\\\n"
 326   "  echo $worker_ckpts | sed -e \'s/\'\"$worker_ckpts_regexp\"\'/\\1 /g\')\n"
 327   "restart_modes=$(\\\n"
 328   "  echo $worker_ckpts | sed -e \'s/\'\"$worker_ckpts_regexp\"\'/: \\2/g\')\n"
 329   "ckpt_files_groups=$(\\\n"
 330   "  echo $worker_ckpts | sed -e \'s/\'\"$worker_ckpts_regexp\"\'/: \\3/g\')\n"
 331   "\n"
 332   "if [ ! -z \"$hostfile\" ]; then\n"
 333   "  worker_hosts=$(\\\n"
 334   "    cat \"$hostfile\" | sed -e \'s/#.*//\' -e \'s/[ \\t\\r]*//\' -e \'/^$/ d\')\n"
 335   "fi\n\n"
 337   "localhost_ckpt_files_group=\n\n"
 339   "num_worker_hosts=$(echo $worker_hosts | wc -w)\n\n"
 341   "maybejoin=\n"
 342   "if [ \"$num_worker_hosts\" != \"1\" ]; then\n"
 343   "  maybejoin='--join'\n"
 344   "fi\n\n"
 346   "for worker_host in $worker_hosts\n"
 347   "do\n\n"
 348   "  ckpt_files_group=$(\\\n"
 349   "    echo $ckpt_files_groups | sed -e \'s/[^:]*:[ \\t\\n]*\\([^:]*\\).*/\\1/\')\n"
 350   "  ckpt_files_groups=$(echo $ckpt_files_groups | sed -e \'s/[^:]*:[^:]*//\')\n"
 351   "\n"
 352   "  mode=$(echo $restart_modes | sed -e \'s/[^:]*:[ \\t\\n]*\\([^:]*\\).*/\\1/\')\n"
 353   "  restart_modes=$(echo $restart_modes | sed -e \'s/[^:]*:[^:]*//\')\n\n"
 354   "  maybexterm=\n"
 355   "  maybebg=\n"
 356   "  case $mode in\n"
 357   "    bg) maybebg=\'bg\';;\n"
 358   "    xterm) maybexterm=xterm;;\n"
 359   "    fg) ;;\n"
 360   "    *) echo \"WARNING: Unknown Mode\";;\n"
 361   "  esac\n\n"
 362   "  if [ -z \"$ckpt_files_group\" ]; then\n"
 363   "    break;\n"
 364   "  fi\n\n"
 366   "  new_ckpt_files_group=\"\"\n"
 367   "  for tmp in $ckpt_files_group\n"
 368   "  do\n"
 369   "      if  [ ! -z \"$DMTCP_RESTART_DIR\" ]; then\n"
 370   "        tmp=$DMTCP_RESTART_DIR/$(basename $tmp)\n"
 371   "      fi\n"
 372   "      new_ckpt_files_group=\"$new_ckpt_files_group $tmp\"\n"
 373   "  done\n\n"
 375   "tmpdir=\n"
 376   "if [ ! -z \"$DMTCP_TMPDIR\" ]; then\n"
 377   "  tmpdir=\"--tmpdir $DMTCP_TMPDIR\"\n"
 378   "fi\n\n"
 380   "  check_local $worker_host\n"
 381   "  if [ \"$is_local_node\" -eq 1 -o \"$num_worker_hosts\" == \"1\" ]; then\n"
 382   "    localhost_ckpt_files_group=\"$new_ckpt_files_group\"\n"
 383   "    continue\n"
 384   "  fi\n"
 386   "  if [ -z $maybebg ]; then\n"
 387   "    $maybexterm /usr/bin/ssh -t \"$worker_host\" \\\n"
 388   "      $dmt_rstr_cmd --coord-host \"$coord_host\""
 389                                              " --cord-port \"$coord_port\"\\\n"
 390   "      $ckpt_dir --join --interval \"$checkpoint_interval\" $tmpdir \\\n"
 391   "      $new_ckpt_files_group\n"
 392   "  else\n"
 393   "    $maybexterm /usr/bin/ssh \"$worker_host\" \\\n"
 394   // In Open MPI 1.4, without this (sh -c ...), orterun hangs at the
 395   // end of the computation until user presses enter key.
 396   "      \"/bin/sh -c \'$dmt_rstr_cmd --coord-host $coord_host"
 397                                                 " --coord-port $coord_port\\\n"
 398   "      $ckpt_dir --join --interval \"$checkpoint_interval\" $tmpdir \\\n"
 399   "      $new_ckpt_files_group\'\" &\n"
 400   "  fi\n\n"
 401   "done\n\n"
 402   "if [ -n \"$localhost_ckpt_files_group\" ]; then\n"
 403   "exec $dmt_rstr_cmd --coord-host \"$coord_host\""
 404                                            " --coord-port \"$coord_port\" \\\n"
 405   "  $ckpt_dir $maybejoin --interval \"$checkpoint_interval\" $tmpdir $noStrictUidChecking $localhost_ckpt_files_group\n"
 406   "fi\n\n"
 408   "#wait for them all to finish\n"
 409   "wait\n"
 410 ;
 412 static int thePort = -1;
 413 static string thePortFile;
 415 static bool exitOnLast = false;
 416 static bool blockUntilDone = false;
 417 static bool exitAfterCkpt = false;
 418 static int blockUntilDoneRemote = -1;
 420 static DmtcpCoordinator prog;
 422 /* The coordinator can receive a second checkpoint request while processing the
 423  * first one.  If the second request comes at a point where the coordinator has
 424  * broadcast DMT_DO_SUSPEND message but the workers haven't replied, the
 425  * coordinator sends another DMT_DO_SUSPEND message.  The workers, having
 426  * replied to the first DMTCP_DO_SUSPEND message (by suspending all the user
 427  * threads), are waiting for the next message (DMT_DO_FD_LEADER_ELECTION or
 428  * DMT_KILL_PEER), however they receive DMT_DO_SUSPEND message and thus exit()
 429  * indicating an error.
 430  * The fix to this problem is to introduce a global
 431  * variable "workersRunningAndSuspendMsgSent" which, as the name implies,
 432  * indicates that the DMT_DO_SUSPEND message has been sent and the coordinator
 433  * is waiting for replies from the workers.  If this variable is set, the
 434  * coordinator will not process another checkpoint request.
 435 */
 436 static bool workersRunningAndSuspendMsgSent = false;
 438 static bool killInProgress = false;
 439 static bool uniqueCkptFilenames = false;
 441 /* If dmtcp_launch/dmtcp_restart specifies '-i', theCheckpointInterval
 442  * will be reset accordingly (valid for current computation).  If dmtcp_command
 443  * specifies '-i' (or if user interactively invokes 'i' in coordinator),
 444  * then both theCheckpointInterval and theDefaultCheckpointInterval are set.
 445  * A value of '0' means:  never checkpoint (manual checkpoint only).
 446  */
 447 static uint32_t theCheckpointInterval = 0; /* Current checkpoint interval */
 448 static uint32_t theDefaultCheckpointInterval = 0; /* Reset to this on new comp. */
 449 static bool isRestarting = false;
 450 static bool timerExpired = false;
 452 static void resetCkptTimer();
 454 const int STDIN_FD = fileno ( stdin );
 456 JTIMER ( checkpoint );
 457 JTIMER ( restart );
 459 static UniquePid compId;
 460 static int numPeers = -1;
 461 static time_t curTimeStamp = -1;
 462 static time_t ckptTimeStamp = -1;
 464 static LookupService lookupService;
 466 static string coordHostname;
 467 static struct in_addr localhostIPAddr;
 469 static string tmpDir;
 470 static string ckptDir;
 472 #define MAX_EVENTS 10000
 473 struct epoll_event events[MAX_EVENTS];
 474 int epollFd;
 475 static jalib::JSocket *listenSock = NULL;
 477 static void removeStaleSharedAreaFile();
 478 static void preExitCleanup();
 480 static pid_t _nextVirtualPid = INITIAL_VIRTUAL_PID;
 482 static int theNextClientNumber = 1;
vector<CoordClient*> clients;
CoordClient::CoordClient(const jalib::JSocket& sock,
 486                          const struct sockaddr_storage *addr,
socklen_t len,
DmtcpMessage &hello_remote,
 489                          int isNSWorker)
 490   : _sock(sock)
 491 {
_isNSWorker = isNSWorker;
_realPid = hello_remote.realPid;
_clientNumber = theNextClientNumber++;
_identity = hello_remote.from;
_state = hello_remote.state;
 497   struct sockaddr_in *in = (struct sockaddr_in*) addr;
_ip = inet_ntoa(in->sin_addr);
 499 }
 501 void CoordClient::readProcessInfo(DmtcpMessage& msg)
 502 {
 503   if (msg.extraBytes > 0) {
 504     char* extraData = new char[msg.extraBytes];
_sock.readAll(extraData, msg.extraBytes);
_hostname = extraData;
_progname = extraData + _hostname.length() + 1;
 508     delete [] extraData;
 509   }
 510 }
pid_t DmtcpCoordinator::getNewVirtualPid()
 513 {
pid_t pid = -1;
JASSERT(_virtualPidToClientMap.size() < MAX_VIRTUAL_PID/1000)
 516     .Text("Exceeded maximum number of processes allowed");
 517   while (1) {
pid = _nextVirtualPid;
_nextVirtualPid += 1000;
 520     if (_nextVirtualPid > MAX_VIRTUAL_PID) {
_nextVirtualPid = INITIAL_VIRTUAL_PID;
 522     }
 523     if (_virtualPidToClientMap.find(pid) == _virtualPidToClientMap.end()) {
 524       break;
 525     }
 526   }
JASSERT(pid != -1) .Text("Not Reachable");
 528   return pid;
 529 }
 531 void DmtcpCoordinator::handleUserCommand(char cmd, DmtcpMessage* reply /*= NULL*/)
 532 {
 533   if (reply != NULL) reply->coordCmdStatus = CoordCmdStatus::NOERROR;
 535   switch ( cmd ){
 536   case 'b': case 'B':  // prefix blocking command, prior to checkpoint command
JTRACE ( "blocking checkpoint beginning..." );
blockUntilDone = true;
 539     break;
 540   case 'x': case 'X':  // prefix exit command, prior to checkpoint command
JTRACE ( "Will exit after creating the checkpoint..." );
exitAfterCkpt = true;
 543     break;
 544   case 'c': case 'C':
JTRACE ( "checkpointing..." );
 546     if(startCheckpoint()){
 547       if (reply != NULL) reply->numPeers = getStatus().numPeers;
 548     }else{
 549       if (reply != NULL) reply->coordCmdStatus = CoordCmdStatus::ERROR_NOT_RUNNING_STATE;
 550     }
 551     break;
 552   case 'i': case 'I':
JTRACE("setting checkpoint interval...");
updateCheckpointInterval ( theCheckpointInterval );
 555     if (theCheckpointInterval == 0)
printf("Current Checkpoint Interval:"
 557              " Disabled (checkpoint manually instead)\n");
 558     else
printf("Current Checkpoint Interval: %d\n", theCheckpointInterval);
 560     if (theDefaultCheckpointInterval == 0)
printf("Default Checkpoint Interval:"
 562              " Disabled (checkpoint manually instead)\n");
 563     else
printf("Default Checkpoint Interval: %d\n", theDefaultCheckpointInterval);
 565     break;
 566   case 'l': case 'L':
 567   case 't': case 'T':
JASSERT_STDERR << "Client List:\n";
JASSERT_STDERR << "#, PROG[virtPID:realPID]@HOST, DMTCP-UNIQUEPID, STATE\n";
 570     for (size_t i = 0; i < clients.size(); i++) {
JASSERT_STDERR << clients[i]->clientNumber()
 572         << ", " << clients[i]->progname()
 573         << "[" << clients[i]->identity().pid() << ":" << clients[i]->realPid()
 574         << "]@" << clients[i]->hostname()
 575 #ifdef PRINT_REMOTE_IP
 576         << "(" << clients[i]->ip() << ")"
 577 #endif
 578         << ", " << clients[i]->identity()
 579         << ", " << clients[i]->state().toString()
 580         << '\n';
 581     }
 582     break;
 583   case 'q': case 'Q':
 584   {
JNOTE ( "killing all connected peers and quitting ..." );
broadcastMessage ( DMT_KILL_PEER );
JASSERT_STDERR << "DMTCP coordinator exiting... (per request)\n";
 588     for (size_t i = 0; i < clients.size(); i++) {
clients[i]->sock().close();
 590     }
listenSock->close();
preExitCleanup();
JTRACE ("Exiting ...");
exit ( 0 );
 595     break;
 596   }
 597   case 'k': case 'K':
JNOTE ( "Killing all connected Peers..." );
 599     //FIXME: What happens if a 'k' command is followed by a 'c' command before
 600     //       the *real* broadcast takes place?         --Kapil
broadcastMessage ( DMT_KILL_PEER );
 602     break;
 603   case 'h': case 'H': case '?':
JASSERT_STDERR << theHelpMessage;
 605     break;
 606   case 's': case 'S':
 607   {
ComputationStatus s = getStatus();
 609     bool running = (s.minimumStateUnanimous &&
s.minimumState==WorkerState::RUNNING);
 611     if (reply != NULL) {
reply->numPeers = s.numPeers;
reply->isRunning = running;
reply->theCheckpointInterval = theCheckpointInterval;
 615     } else {
printStatus(s.numPeers, running);
 617     }
 618     break;
 619   }
 620   case ' ': case '\t': case '\n': case '\r':
 621     //ignore whitespace
 622     break;
 623   default:
JTRACE("unhandled user command")(cmd);
 625     if (reply != NULL){
reply->coordCmdStatus = CoordCmdStatus::ERROR_INVALID_COMMAND;
 627     }
 628   }
 629   return;
 630 }
 632 void DmtcpCoordinator::printStatus(size_t numPeers, bool isRunning)
 633 {
ostringstream o;
o << "Status..." << std::endl
 636     << "Host: " << coordHostname
 637     << " (" << inet_ntoa(localhostIPAddr) << ")" << std::endl
 638     << "Port: " << thePort << std::endl
 639     << "Checkpoint Interval: ";
 641   if (theCheckpointInterval == 0) {
o << "disabled (checkpoint manually instead)" << std::endl;
 643   } else {
o << theCheckpointInterval << std::endl;
 645   }
o << "Exit on last client: " << exitOnLast << std::endl
 648     << "Exit after checkpoint: " << exitAfterCkpt << std::endl
 649     << "Computation Id: " << compId << std::endl
 650     << "Checkpoint Dir: " << ckptDir << std::endl
 651     << "NUM_PEERS=" << numPeers << std::endl
 652     << "RUNNING=" << (isRunning ? "yes" : "no") << std::endl;
printf("%s", o.str().c_str());
 654   fflush(stdout);
 655 }
 657 void DmtcpCoordinator::updateMinimumState(WorkerState oldState)
 658 {
WorkerState newState = minimumState();
 661   if ( oldState == WorkerState::RUNNING
 662        && newState == WorkerState::SUSPENDED )
 663   {
JNOTE ( "locking all nodes" );
broadcastMessage(DMT_DO_FD_LEADER_ELECTION, getStatus().numPeers );
 666   }
 667   if ( oldState == WorkerState::SUSPENDED
 668        && newState == WorkerState::FD_LEADER_ELECTION )
 669   {
JNOTE ( "draining all nodes" );
broadcastMessage ( DMT_DO_DRAIN );
 672   }
 673   if ( oldState == WorkerState::FD_LEADER_ELECTION
 674        && newState == WorkerState::DRAINED )
 675   {
JNOTE ( "checkpointing all nodes" );
broadcastMessage ( DMT_DO_CHECKPOINT );
 678   }
 680 #ifdef COORD_NAMESERVICE
 681   if ( oldState == WorkerState::DRAINED
 682        && newState == WorkerState::CHECKPOINTED )
 683   {
writeRestartScript();
 685     if (exitAfterCkpt) {
JNOTE("Checkpoint Done. Killing all peers.");
broadcastMessage(DMT_KILL_PEER);
exitAfterCkpt = false;
 689     } else {
JNOTE ( "building name service database" );
lookupService.reset();
broadcastMessage ( DMT_DO_REGISTER_NAME_SERVICE_DATA );
 693     }
 694   }
 695   if ( oldState == WorkerState::RESTARTING
 696        && newState == WorkerState::CHECKPOINTED )
 697   {
JTIMER_STOP ( restart );
lookupService.reset();
JNOTE ( "building name service database (after restart)" );
broadcastMessage ( DMT_DO_REGISTER_NAME_SERVICE_DATA );
 703   }
 704   if ( oldState == WorkerState::CHECKPOINTED
 705        && newState == WorkerState::NAME_SERVICE_DATA_REGISTERED ){
JNOTE ( "entertaining queries now" );
broadcastMessage ( DMT_DO_SEND_QUERIES );
 708   }
 709   if ( oldState == WorkerState::NAME_SERVICE_DATA_REGISTERED
 710        && newState == WorkerState::DONE_QUERYING ){
JNOTE ( "refilling all nodes" );
broadcastMessage ( DMT_DO_REFILL );
 713   }
 714   if ( oldState == WorkerState::DONE_QUERYING
 715        && newState == WorkerState::REFILLED )
 716     /* then broadcastMessage(DMT_DO_RESUME) after the #endif, below */
 717 #else
 718   if ( oldState == WorkerState::DRAINED
 719        && newState == WorkerState::CHECKPOINTED )
 720   {
writeRestartScript();
 722     if (exitAfterCkpt) {
JNOTE("Checkpoint Done. Killing all peers.");
broadcastMessage(DMT_KILL_PEER);
exitAfterCkpt = false;
 726     } else {
JNOTE ( "refilling all nodes" );
broadcastMessage ( DMT_DO_REFILL );
 729     }
 730   }
 731   if ( oldState == WorkerState::RESTARTING
 732        && newState == WorkerState::CHECKPOINTED )
 733   {
JTIMER_STOP ( restart );
JNOTE ( "refilling all nodes (after checkpoint)" );
broadcastMessage ( DMT_DO_REFILL );
 738   }
 739   if ( oldState == WorkerState::CHECKPOINTED
 740        && newState == WorkerState::REFILLED )
 741 #endif
 742   {
JNOTE ( "restarting all nodes" );
broadcastMessage ( DMT_DO_RESUME );
JTIMER_STOP ( checkpoint );
isRestarting = false;
resetCkptTimer();
 751     if (blockUntilDone) {
DmtcpMessage blockUntilDoneReply(DMT_USER_CMD_RESULT);
JNOTE ( "replying to dmtcp_command:  we're done" );
 754       // These were set in DmtcpCoordinator::onConnect in this file
jalib::JSocket remote ( blockUntilDoneRemote );
remote << blockUntilDoneReply;
remote.close();
blockUntilDone = false;
blockUntilDoneRemote = -1;
 760     }
 761   }
 762 }
 764 void DmtcpCoordinator::onData(CoordClient *client)
 765 {
DmtcpMessage msg;
JASSERT(client != NULL);
client->sock() >> msg;
msg.assertValid();
 771   char *extraData = 0;
 772   if (msg.extraBytes > 0) {
extraData = new char[msg.extraBytes];
client->sock().readAll(extraData, msg.extraBytes);
 775   }
 777   switch ( msg.type )
 778   {
 779     case DMT_OK:
 780     {
WorkerState oldState = client->state();
client->setState ( msg.state );
ComputationStatus s = getStatus();
WorkerState newState = s.minimumState;
JTRACE ("got DMT_OK message")
 787         ( oldState )( msg.from )( msg.state )( newState );
updateMinimumState(oldState);
 790       break;
 791     }
 792     case DMT_UNIQUE_CKPT_FILENAME:
uniqueCkptFilenames = true;
 794       // Fall though
 795     case DMT_CKPT_FILENAME:
 796     {
JASSERT ( extraData!=0 )
 798         .Text ( "extra data expected with DMT_CKPT_FILENAME message" );
string ckptFilename;
string hostname;
ckptFilename = extraData;
hostname = extraData + ckptFilename.length() + 1;
JTRACE ( "recording restart info" ) ( ckptFilename ) ( hostname );
_restartFilenames[hostname].push_back ( ckptFilename );
 806     }
 807     break;
 808     case DMT_GET_CKPT_DIR:
 809     {
DmtcpMessage reply(DMT_GET_CKPT_DIR_RESULT);
reply.extraBytes = ckptDir.length() + 1;
client->sock() << reply;
client->sock().writeAll(ckptDir.c_str(), reply.extraBytes);
 814     }
 815     break;
 816     case DMT_UPDATE_CKPT_DIR:
 817     {
JASSERT(extraData != 0)
 819         .Text("extra data expected with DMT_UPDATE_CKPT_DIR message");
 820       if (strcmp(ckptDir.c_str(), extraData) != 0) {
ckptDir = extraData;
JNOTE("Updated ckptDir") (ckptDir);
 823       }
 824     }
 825     break;
 827 #ifdef COORD_NAMESERVICE
 828     case DMT_REGISTER_NAME_SERVICE_DATA:
 829     {
JTRACE ("received REGISTER_NAME_SERVICE_DATA msg") (client->identity());
lookupService.registerData(msg, (const void*) extraData);
 832     }
 833     break;
 835     case DMT_REGISTER_NAME_SERVICE_DATA_SYNC:
 836     {
JTRACE ("received REGISTER_NAME_SERVICE_DATA_SYNC msg") (client->identity());
lookupService.registerData(msg, (const void*) extraData);
DmtcpMessage response(DMT_REGISTER_NAME_SERVICE_DATA_SYNC_RESPONSE);
JTRACE("Sending NS response to the client...");
client->sock() << response;
 842     }
 843     break;
 844     case DMT_NAME_SERVICE_QUERY:
 845     {
JTRACE ("received NAME_SERVICE_QUERY msg") (client->identity());
lookupService.respondToQuery(client->sock(), msg,
 848                                    (const void*) extraData);
 849     }
 850     break;
 851 #endif
 853     case DMT_UPDATE_PROCESS_INFO_AFTER_FORK:
 854     {
JNOTE("Updating process Information after fork()")
 856           (client->hostname()) (client->progname()) (msg.from) (client->identity());
client->identity(msg.from);
client->realPid(msg.realPid);
 859     }
 860     break;
 861     case DMT_UPDATE_PROCESS_INFO_AFTER_INIT_OR_EXEC:
 862     {
string progname = extraData;
JNOTE("Updating process Information after exec()")
 865         (progname) (msg.from) (client->identity());
client->progname(progname);
client->identity(msg.from);
 868     }
 869     break;
 871     case DMT_NULL:
JWARNING(false) (msg.type) .Text("unexpected message from worker. Closing connection");
onDisconnect(client);
 874       break;
 875     default:
JASSERT ( false ) ( msg.from ) ( msg.type )
 877         .Text ( "unexpected message from worker" );
 878   }
 880   delete[] extraData;
 881 }
 884 static void removeStaleSharedAreaFile()
 885 {
ostringstream o;
o << tmpDir
 888     << "/dmtcpSharedArea." << compId << "." << std::hex << curTimeStamp;
JTRACE("Removing sharedArea file.") (o.str());
unlink(o.str().c_str());
 891 }
 893 static void preExitCleanup()
 894 {
removeStaleSharedAreaFile();
JTRACE("Removing port-file") (thePortFile);
unlink(thePortFile.c_str());
 898 }
 900 void DmtcpCoordinator::onDisconnect(CoordClient *client)
 901 {
 902   if (client->isNSWorker()) {
client->sock().close();
 904     delete client;
 905     return;
 906   }
 907   for (size_t i = 0; i < clients.size(); i++) {
 908     if (clients[i] == client) {
clients.erase(clients.begin() + i);
 910       break;
 911     }
 912   }
client->sock().close();
JNOTE ( "client disconnected" ) ( client->identity() );
_virtualPidToClientMap.erase(client->virtualPid());
ComputationStatus s = getStatus();
 918   if (s.numPeers < 1) {
 919     if (exitOnLast) {
JNOTE ("last client exited, shutting down..");
handleUserCommand('q');
 922     } else {
removeStaleSharedAreaFile();
 924     }
 925     // If a kill in is progress, the coordinator refuses any new connections,
 926     // thus we need to reset it to false once all the processes in the
 927     // computations have disconnected.
killInProgress = false;
 929     if (theCheckpointInterval != theDefaultCheckpointInterval) {
updateCheckpointInterval(theDefaultCheckpointInterval);
JNOTE ( "CheckpointInterval reset on end of current computation" )
 932         ( theCheckpointInterval );
 933     }
 934   } else {
updateMinimumState(client->state());
 936   }
 937 }
 939 void DmtcpCoordinator::initializeComputation()
 940 {
 941   //this is the first connection, do some initializations
workersRunningAndSuspendMsgSent = false;
killInProgress = false;
 944   //_nextVirtualPid = INITIAL_VIRTUAL_PID;
 946   // theCheckpointInterval can be overridden later by msg from this client.
updateCheckpointInterval( theDefaultCheckpointInterval );
 949   // drop current computation group to 0
compId = UniquePid(0,0,0);
curTimeStamp = 0; // Drop timestamp to 0
numPeers = -1; // Drop number of peers to unknown
blockUntilDone = false;
 954   //exitAfterCkpt = false;
 955 }
 957 void DmtcpCoordinator::onConnect()
 958 {
 959   struct sockaddr_storage remoteAddr;
socklen_t remoteLen = sizeof(remoteAddr);
jalib::JSocket remote = listenSock->accept(&remoteAddr, &remoteLen);
JTRACE("accepting new connection") (remote.sockfd()) (JASSERT_ERRNO);
 964   if (!remote.isValid()) {
remote.close();
 966     return;
 967   }
DmtcpMessage hello_remote;
hello_remote.poison();
JTRACE("Reading from incoming connection...");
remote >> hello_remote;
 973   if (!remote.isValid()) {
remote.close();
 975     return;
 976   }
 978 #ifdef COORD_NAMESERVICE
 979   if (hello_remote.type == DMT_NAME_SERVICE_WORKER) {
CoordClient *client = new CoordClient(remote, &remoteAddr, remoteLen,
hello_remote);
addDataSocket(client);
 984     return;
 985   }
 986   if (hello_remote.type == DMT_NAME_SERVICE_QUERY) {
JASSERT(hello_remote.extraBytes > 0) (hello_remote.extraBytes);
 988     char *extraData = new char[hello_remote.extraBytes];
remote.readAll(extraData, hello_remote.extraBytes);
JTRACE ("received NAME_SERVICE_QUERY msg on running") (hello_remote.from);
lookupService.respondToQuery(remote, hello_remote, extraData);
 993     delete [] extraData;
remote.close();
 995     return;
 996   }
 997   if (hello_remote.type == DMT_REGISTER_NAME_SERVICE_DATA) {
JASSERT(hello_remote.extraBytes > 0) (hello_remote.extraBytes);
 999     char *extraData = new char[hello_remote.extraBytes];
remote.readAll(extraData, hello_remote.extraBytes);
JTRACE ("received REGISTER_NAME_SERVICE_DATA msg on running") (hello_remote.from);
lookupService.registerData(hello_remote, (const void*) extraData);
1004     delete [] extraData;
remote.close();
1006     return;
1007   }
1008   if (hello_remote.type == DMT_REGISTER_NAME_SERVICE_DATA_SYNC) {
JASSERT(hello_remote.extraBytes > 0) (hello_remote.extraBytes);
1010     char *extraData = new char[hello_remote.extraBytes];
remote.readAll(extraData, hello_remote.extraBytes);
JTRACE ("received REGISTER_NAME_SERVICE_DATA msg on running") (hello_remote.from);
lookupService.registerData(hello_remote, (const void*) extraData);
1015     delete [] extraData;
DmtcpMessage response(DMT_REGISTER_NAME_SERVICE_DATA_SYNC_RESPONSE);
JTRACE("Reading from incoming connection...");
remote << response;
remote.close();
1020     return;
1021   }
1022 #endif
1024   if (hello_remote.type == DMT_USER_CMD) {
1025     // TODO(kapil): Update ckpt interval only if a valid one was supplied to
1026     // dmtcp_command.
updateCheckpointInterval(hello_remote.theCheckpointInterval);
processDmtUserCmd(hello_remote, remote);
1029     return;
1030   }
1032   if (killInProgress) {
JNOTE("Connection request received in the middle of killing computation. "
1034           "Sending it the kill message.");
DmtcpMessage msg;
msg.type = DMT_KILL_PEER;
remote << msg;
remote.close();
1039     return;
1040   }
1042   // If no client is connected to Coordinator, then there can be only zero data
1043   // sockets OR there can be one data socket and that should be STDIN.
1044   if (clients.size() == 0) {
initializeComputation();
1046   }
CoordClient *client = new CoordClient(remote, &remoteAddr, remoteLen,
hello_remote);
1051   if( hello_remote.extraBytes > 0 ){
client->readProcessInfo(hello_remote);
1053   }
1055   if (hello_remote.type == DMT_RESTART_WORKER) {
1056     if (!validateRestartingWorkerProcess(hello_remote, remote,
1057                                          &remoteAddr, remoteLen)) {
1058       return;
1059     }
client->virtualPid(hello_remote.from.pid());
_virtualPidToClientMap[client->virtualPid()] = client;
isRestarting = true;
1063   } else if (hello_remote.type == DMT_NEW_WORKER) {
JASSERT(hello_remote.state == WorkerState::RUNNING ||
hello_remote.state == WorkerState::UNKNOWN);
JASSERT(hello_remote.virtualPid == -1);
client->virtualPid(getNewVirtualPid());
1068     if (!validateNewWorkerProcess(hello_remote, remote, client,
1069                                   &remoteAddr, remoteLen)) {
1070       return;
1071     }
_virtualPidToClientMap[client->virtualPid()] = client;
1073   } else {
JASSERT(false) (hello_remote.type)
1075       .Text("Connect request from Unknown Remote Process Type");
1076   }
updateCheckpointInterval(hello_remote.theCheckpointInterval);
JNOTE ( "worker connected" ) ( hello_remote.from );
clients.push_back(client);
addDataSocket(client);
JTRACE("END") (clients.size());
1085 }
1087 void DmtcpCoordinator::processDmtUserCmd(DmtcpMessage& hello_remote,
jalib::JSocket& remote )
1089 {
1090   //dmtcp_command doesn't handshake (it is antisocial)
JTRACE("got user command from dmtcp_command")(hello_remote.coordCmd);
DmtcpMessage reply;
reply.type = DMT_USER_CMD_RESULT;
1094   // if previous 'b' blocking prefix command had set blockUntilDone
1095   if (blockUntilDone && blockUntilDoneRemote == -1  &&
hello_remote.coordCmd == 'c') {
1097     // Reply will be done in DmtcpCoordinator::onData in this file.
blockUntilDoneRemote = remote.sockfd();
handleUserCommand( hello_remote.coordCmd, &reply );
1100   } else if ( (hello_remote.coordCmd == 'i')
1101                && hello_remote.theCheckpointInterval >= 0 ) {
1102 //    theDefaultCheckpointInterval = hello_remote.theCheckpointInterval;
1103 //    theCheckpointInterval = theDefaultCheckpointInterval;
handleUserCommand( hello_remote.coordCmd, &reply );
remote << reply;
remote.close();
1107   } else {
handleUserCommand( hello_remote.coordCmd, &reply );
remote << reply;
remote.close();
1111   }
1112   return;
1113 }
1115 bool DmtcpCoordinator::validateRestartingWorkerProcess(
DmtcpMessage& hello_remote,
jalib::JSocket& remote,
1118     const struct sockaddr_storage* remoteAddr,
socklen_t remoteLen)
1120 {
1121   struct timeval tv;
1122   const struct sockaddr_in *sin = (const struct sockaddr_in*) remoteAddr;
string remoteIP = inet_ntoa(sin->sin_addr);
DmtcpMessage hello_local ( DMT_ACCEPT );
JASSERT(hello_remote.state == WorkerState::RESTARTING) (hello_remote.state);
1128   if (compId == UniquePid(0,0,0)) {
JASSERT ( minimumState() == WorkerState::UNKNOWN )
1130       .Text ( "Coordinator should be idle at this moment" );
1131     // Coordinator is free at this moment - set up all the things
compId = hello_remote.compGroup;
numPeers = hello_remote.numPeers;
JASSERT(gettimeofday(&tv, NULL) == 0);
1135     // Get the resolution down to 100 mili seconds.
curTimeStamp = (tv.tv_sec << 4) | (tv.tv_usec / (100*1000));
JNOTE ( "FIRST dmtcp_restart connection.  Set numPeers. Generate timestamp" )
1138       ( numPeers ) ( curTimeStamp ) ( compId );
JTIMER_START(restart);
1140   } else if (minimumState() != WorkerState::RESTARTING &&
minimumState() != WorkerState::CHECKPOINTED) {
JNOTE ("Computation not in RESTARTING or CHECKPOINTED state."
1143            "  Reject incoming computation process requesting restart.")
1144       (compId) (hello_remote.compGroup) (minimumState());
hello_local.type = DMT_REJECT_NOT_RESTARTING;
remote << hello_local;
remote.close();
1148     return false;
1149   } else if ( hello_remote.compGroup != compId) {
JNOTE ("Reject incoming computation process requesting restart,"
1151            " since it is not from current computation.")
1152       ( compId ) ( hello_remote.compGroup );
hello_local.type = DMT_REJECT_WRONG_COMP;
remote << hello_local;
remote.close();
1156     return false;
1157   }
1158   // dmtcp_restart already connected and compGroup created.
1159   // Computation process connection
JASSERT ( curTimeStamp != 0 );
JTRACE("Connection from (restarting) computation process")
1163     ( compId ) ( hello_remote.compGroup ) ( minimumState() );
hello_local.coordTimeStamp = curTimeStamp;
1166   if (Util::strStartsWith(remoteIP, "127.")) {
memcpy(&hello_local.ipAddr, &localhostIPAddr, sizeof localhostIPAddr);
1168   } else {
memcpy(&hello_local.ipAddr, &sin->sin_addr, sizeof localhostIPAddr);
1170   }
remote << hello_local;
1173   // NOTE: Sending the same message twice. We want to make sure that the
1174   // worker process receives/processes the first messages as soon as it
1175   // connects to the coordinator. The second message will be processed in
1176   // postRestart routine in DmtcpWorker.
1177   //
1178   // The reason to do this is the following. The dmtcp_restart process
1179   // connects to the coordinator at a very early stage. Later on, before
1180   // exec()'ing into mtcp_restart, it reconnects to the coordinator using
1181   // it's original UniquiePid and closes the earlier socket connection.
1182   // However, the coordinator might process the disconnect() before it
1183   // processes the connect() which would lead to a situation where the
1184   // coordinator is not connected to any worker processes. The coordinator
1185   // would now process the connect() and may reject the worker because the
1186   // worker state is RESTARTING, but the minimumState() is UNKNOWN.
1187   //remote << hello_local;
1189   return true;
1190 }
1192 bool DmtcpCoordinator::validateNewWorkerProcess(
DmtcpMessage& hello_remote,
jalib::JSocket& remote,
CoordClient *client,
1196     const struct sockaddr_storage* remoteAddr,
socklen_t remoteLen)
1198 {
1199   const struct sockaddr_in *sin = (const struct sockaddr_in*) remoteAddr;
string remoteIP = inet_ntoa(sin->sin_addr);
DmtcpMessage hello_local(DMT_ACCEPT);
hello_local.virtualPid = client->virtualPid();
ComputationStatus s = getStatus();
JASSERT(hello_remote.state == WorkerState::RUNNING ||
hello_remote.state == WorkerState::UNKNOWN) (hello_remote.state);
1208   if (workersRunningAndSuspendMsgSent == true) {
1209     /* Worker trying to connect after SUSPEND message has been sent.
1210      * This happens if the worker process is executing a fork() system call
1211      * when the DMT_DO_SUSPEND is broadcast. We need to make sure that the
1212      * child process is allowed to participate in the current checkpoint.
1213      */
JASSERT(s.numPeers > 0) (s.numPeers);
JASSERT(s.minimumState != WorkerState::SUSPENDED) (s.minimumState);
1217     // Handshake
hello_local.compGroup = compId;
remote << hello_local;
1221     // Now send DMT_DO_SUSPEND message so that this process can also
1222     // participate in the current checkpoint
DmtcpMessage suspendMsg (DMT_DO_SUSPEND);
suspendMsg.compGroup = compId;
remote << suspendMsg;
1227   } else if (s.numPeers > 0 && s.minimumState != WorkerState::RUNNING &&
s.minimumState != WorkerState::UNKNOWN) {
1229     // If some of the processes are not in RUNNING state
JNOTE("Current computation not in RUNNING state."
1231           "  Refusing to accept new connections.")
1232       (compId) (hello_remote.from)
1233       (s.numPeers) (s.minimumState);
hello_local.type = DMT_REJECT_NOT_RUNNING;
remote << hello_local;
remote.close();
1237     return false;
1239   } else if (hello_remote.compGroup != UniquePid()) {
1240     // New Process trying to connect to Coordinator but already has compGroup
JNOTE ("New process not part of currently running computation group"
1242            "on this coordinator.  Rejecting.")
1243       (hello_remote.compGroup);
hello_local.type = DMT_REJECT_WRONG_COMP;
remote << hello_local;
remote.close();
1248     return false;
1250   } else {
1251     // If first process, create the new computation group
1252     if (compId == UniquePid(0,0,0)) {
1253       struct timeval tv;
1254       // Connection of new computation.
compId = UniquePid(hello_remote.from.hostid(), client->virtualPid(),
hello_remote.from.time(),
hello_remote.from.computationGeneration());
JASSERT(gettimeofday(&tv, NULL) == 0);
1260       // Get the resolution down to 100 mili seconds.
curTimeStamp = (tv.tv_sec << 4) | (tv.tv_usec / (100*1000));
numPeers = -1;
JTRACE("First process connected.  Creating new computation group.")
1264         (compId);
1265     } else {
JTRACE("New process connected")
1267         (hello_remote.from) (client->virtualPid());
1268     }
hello_local.compGroup = compId;
hello_local.coordTimeStamp = curTimeStamp;
1271     if (Util::strStartsWith(remoteIP, "127.")) {
memcpy(&hello_local.ipAddr, &localhostIPAddr, sizeof localhostIPAddr);
1273     } else {
memcpy(&hello_local.ipAddr, &sin->sin_addr, sizeof localhostIPAddr);
1275     }
remote << hello_local;
1277   }
1278   return true;
1279 }
1281 bool DmtcpCoordinator::startCheckpoint()
1282 {
uniqueCkptFilenames = false;
ComputationStatus s = getStatus();
1285   if ( s.minimumState == WorkerState::RUNNING && s.minimumStateUnanimous
1286        && !workersRunningAndSuspendMsgSent )
1287   {
time(&ckptTimeStamp);
JTIMER_START ( checkpoint );
_restartFilenames.clear();
JNOTE ( "starting checkpoint, suspending all nodes" )( s.numPeers );
compId.incrementGeneration();
JNOTE("Incremented computationGeneration") (compId.computationGeneration());
1294     // Pass number of connected peers to all clients
broadcastMessage(DMT_DO_SUSPEND);
1297     // Suspend Message has been sent but the workers are still in running
1298     // state.  If the coordinator receives another checkpoint request from user
1299     // at this point, it should fail.
workersRunningAndSuspendMsgSent = true;
1301     return true;
1302   } else {
1303     if (s.numPeers > 0) {
JTRACE ( "delaying checkpoint, workers not ready" ) ( s.minimumState )
1305              ( s.numPeers );
1306     }
1307     return false;
1308   }
1309 }
1311 void DmtcpCoordinator::broadcastMessage(DmtcpMessageType type, int numPeers)
1312 {
DmtcpMessage msg;
msg.type = type;
msg.compGroup = compId;
1316   if (numPeers > 0) {
msg.numPeers = numPeers;
1318   }
1320   if (msg.type == DMT_KILL_PEER && clients.size() > 0) {
killInProgress = true;
1322   } else if (msg.type == DMT_DO_FD_LEADER_ELECTION) {
1323     // All the workers are in SUSPENDED state, now it is safe to reset
1324     // this flag.
workersRunningAndSuspendMsgSent = false;
1326   }
1328   for (size_t i = 0; i < clients.size(); i++) {
clients[i]->sock() << msg;
1330   }
JTRACE ("sending message")( type );
1332 }
DmtcpCoordinator::ComputationStatus DmtcpCoordinator::getStatus() const
1335 {
ComputationStatus status;
1337   const static int INITIAL_MIN = WorkerState::_MAX;
1338   const static int INITIAL_MAX = WorkerState::UNKNOWN;
1339   int min = INITIAL_MIN;
1340   int max = INITIAL_MAX;
1341   int count = 0;
1342   bool unanimous = true;
1343   for (size_t i = 0; i < clients.size(); i++) {
1344     int cliState = clients[i]->state().value();
count++;
unanimous = unanimous && (min==cliState || min==INITIAL_MIN);
1347     if ( cliState < min ) min = cliState;
1348     if ( cliState > max ) max = cliState;
1349   }
status.minimumState = ( min==INITIAL_MIN ? WorkerState::UNKNOWN
1352                           : (WorkerState::eWorkerState)min );
1353   if( status.minimumState == WorkerState::CHECKPOINTED &&
isRestarting && count < numPeers ){
JTRACE("minimal state counted as CHECKPOINTED but not all processes"
1356            " are connected yet.  So we wait.") ( numPeers ) ( count );
status.minimumState = WorkerState::RESTARTING;
1358   }
status.minimumStateUnanimous = unanimous;
status.maximumState = ( max==INITIAL_MAX ? WorkerState::UNKNOWN
1362                           : (WorkerState::eWorkerState)max );
status.numPeers = count;
1364   return status;
1365 }
1367 void DmtcpCoordinator::writeRestartScript()
1368 {
ostringstream o;
string uniqueFilename;
o << string(ckptDir) << "/"
1373     << RESTART_SCRIPT_BASENAME << "_" << compId;
1374   if (uniqueCkptFilenames) {
o << "_" << std::setw(5) << std::setfill('0') << compId.computationGeneration();
1376   }
o << "." << RESTART_SCRIPT_EXT;
uniqueFilename = o.str();
1380   const bool isSingleHost = (_restartFilenames.size() == 1);
map< string, vector<string> >::const_iterator host;
vector<string>::const_iterator file;
1385   char hostname[80];
1386   char timestamp[80];
gethostname ( hostname, 80 );
JTRACE ( "writing restart script" ) ( uniqueFilename );
FILE* fp = fopen ( uniqueFilename.c_str(),"w" );
JASSERT ( fp!=0 )(JASSERT_ERRNO)( uniqueFilename )
1393     .Text ( "failed to open file" );
fprintf ( fp, "%s", theRestartScriptHeader );
fprintf ( fp, "%s", theRestartScriptCheckLocal );
fprintf ( fp, "%s", slurmHelperContactFunction );
fprintf ( fp, "%s", theRestartScriptUsage );
1400   ctime_r(&ckptTimeStamp, timestamp);
1401   // Remove the trailing '\n'
timestamp[strlen(timestamp) - 1] = '\0';
fprintf ( fp, "ckpt_timestamp=\"%s\"\n\n", timestamp );
fprintf ( fp, "coord_host=$" ENV_VAR_NAME_HOST "\n"
1406                 "if test -z \"$" ENV_VAR_NAME_HOST "\"; then\n"
1407                 "  coord_host=%s\nfi\n\n"
1408                 "coord_port=$" ENV_VAR_NAME_PORT "\n"
1409                 "if test -z \"$" ENV_VAR_NAME_PORT "\"; then\n"
1410                 "  coord_port=%d\nfi\n\n"
1411                 "checkpoint_interval=$" ENV_VAR_CKPT_INTR "\n"
1412                 "if test -z \"$" ENV_VAR_CKPT_INTR "\"; then\n"
1413                 "  checkpoint_interval=%d\nfi\n"
1414                 "export DMTCP_CHECKPOINT_INTERVAL=${checkpoint_interval}\n\n",
hostname, thePort, theCheckpointInterval );
fprintf ( fp, "%s", theRestartScriptCmdlineArgHandler );
fprintf ( fp, "dmt_rstr_cmd=%s/" DMTCP_RESTART_CMD "\n"
1420                 "which $dmt_rstr_cmd > /dev/null 2>&1"
1421                 " || dmt_rstr_cmd=" DMTCP_RESTART_CMD "\n"
1422                 "which $dmt_rstr_cmd > /dev/null 2>&1"
1423                 " || echo \"$0: $dmt_rstr_cmd not found\"\n"
1424                 "which $dmt_rstr_cmd > /dev/null 2>&1 || exit 1\n\n",
jalib::Filesystem::GetProgramDir().c_str());
fprintf ( fp, "# Number of hosts in the computation = %zd\n"
1428                 "# Number of processes in the computation = %d\n\n",
_restartFilenames.size(), getStatus().numPeers );
1431   if ( isSingleHost ) {
JTRACE ( "Single HOST" );
host=_restartFilenames.begin();
ostringstream o;
1436     for ( file=host->second.begin(); file!=host->second.end(); ++file ) {
o << " " << *file;
1438     }
fprintf ( fp, "given_ckpt_files=\"%s\"\n\n", o.str().c_str());
fprintf ( fp, "%s", theRestartScriptSingleHostProcessing );
1442   }
1443   else
1444   {
fprintf ( fp, "%s",
1446               "# SYNTAX:\n"
1447               "#  :: <HOST> :<MODE>: <CHECKPOINT_IMAGE> ...\n"
1448               "# Host names and filenames must not include \':\'\n"
1449               "# At most one fg (foreground) mode allowed; it must be last.\n"
1450               "# \'maybexterm\' and \'maybebg\' are set from <MODE>.\n");
fprintf ( fp, "%s", "worker_ckpts=\'" );
1453     for ( host=_restartFilenames.begin(); host!=_restartFilenames.end(); ++host ) {
fprintf ( fp, "\n :: %s :bg:", host->first.c_str() );
1455       for ( file=host->second.begin(); file!=host->second.end(); ++file ) {
fprintf ( fp," %s", file->c_str() );
1457       }
1458     }
fprintf ( fp, "%s", "\n\'\n\n" );
fprintf( fp,  "# Check for resource manager\n"
1462                   "ibrun_path=$(which ibrun 2> /dev/null)\n"
1463                   "if [ ! -n \"$ibrun_path\" ]; then\n"
1464                   "  discover_rm_path=$(which dmtcp_discover_rm)\n"
1465                   "  if [ -n \"$discover_rm_path\" ]; then\n"
1466                   "    eval $(dmtcp_discover_rm -t)\n"
1467                   "    srun_path=$(which srun 2> /dev/null)\n"
1468                   "    llaunch=`which dmtcp_rm_loclaunch`\n"
1469                   "    if [ $RES_MANAGER = \"SLURM\" ] && [ -n \"$srun_path\" ]; then\n"
1470                   "      eval $(dmtcp_discover_rm -n \"$worker_ckpts\")\n"
1471                   "      if [ -n \"$DMTCP_DISCOVER_RM_ERROR\" ]; then\n"
1472                   "          echo \"Restart error: $DMTCP_DISCOVER_RM_ERROR\"\n"
1473                   "          echo \"Allocated resources: $manager_resources\"\n"
1474                   "          exit 0\n"
1475                   "      fi\n"
1476                   "      export DMTCP_REMLAUNCH_NODES=$DMTCP_REMLAUNCH_NODES\n"
1477                   "      bound=$(($DMTCP_REMLAUNCH_NODES - 1))\n"
1478                   "      for i in $(seq 0 $bound); do\n"
1479                   "        eval \"val=\\${DMTCP_REMLAUNCH_${i}_SLOTS}\"\n"
1480                   "        export DMTCP_REMLAUNCH_${i}_SLOTS=\"$val\"\n"
1481                   "        bound2=$(($val - 1))\n"
1482                   "        for j in $(seq 0 $bound2); do\n"
1483                   "          eval \"ckpts=\\${DMTCP_REMLAUNCH_${i}_${j}}\"\n"
1484                   "          export DMTCP_REMLAUNCH_${i}_${j}=\"$ckpts\"\n"
1485                   "        done\n"
1486                   "      done\n"
1487                   "      if [ \"$DMTCP_DISCOVER_PM_TYPE\" = \"HYDRA\" ]; then\n"
1488                   "        export DMTCP_SRUN_HELPER_SYNCFILE=`mktemp ./tmp.XXXXXXXXXX`\n"
1489                   "        rm $DMTCP_SRUN_HELPER_SYNCFILE\n"
1490                   "        dmtcp_srun_helper -r $srun_path \"$llaunch\"\n"
1491                   "        if [ ! -f $DMTCP_SRUN_HELPER_SYNCFILE ]; then\n"
1492                   "          echo \"Error launching application\"\n"
1493                   "          exit 1\n"
1494                   "        fi\n"
1495                   "        # export helper contact info\n"
1496                   "        . $DMTCP_SRUN_HELPER_SYNCFILE\n"
1497                   "        pass_slurm_helper_contact \"$DMTCP_LAUNCH_CKPTS\"\n"
1498                   "        rm $DMTCP_SRUN_HELPER_SYNCFILE\n"
1499                   "        dmtcp_restart --join --coord-host $DMTCP_COORD_HOST"
1500                               " --coord-port $DMTCP_COORD_PORT"
1501                               " $DMTCP_LAUNCH_CKPTS\n"
1502                   "      else\n"
1503                   "        DMTCP_REMLAUNCH_0_0=\"$DMTCP_REMLAUNCH_0_0"
1504                                                      " $DMTCP_LAUNCH_CKPTS\"\n"
1505                   "        $srun_path \"$llaunch\"\n"
1506                   "      fi\n"
1507                   "      exit 0\n"
1508                   "    elif [ $RES_MANAGER = \"TORQUE\" ]; then\n"
1509                   "      #eval $(dmtcp_discover_rm \"$worker_ckpts\")\n"
1510                   "      #if [ -n \"$new_worker_ckpts\" ]; then\n"
1511                   "      #  worker_ckpts=\"$new_worker_ckpts\"\n"
1512                   "      #fi\n"
1513                   "      eval $(dmtcp_discover_rm -n \"$worker_ckpts\")\n"
1514                   "      if [ -n \"$DMTCP_DISCOVER_RM_ERROR\" ]; then\n"
1515                   "          echo \"Restart error: $DMTCP_DISCOVER_RM_ERROR\"\n"
1516                   "          echo \"Allocated resources: $manager_resources\"\n"
1517                   "          exit 0\n"
1518                   "      fi\n"
1519                   "      arguments=\"PATH=$PATH DMTCP_COORD_HOST=$DMTCP_COORD_HOST"
1520                                       " DMTCP_COORD_PORT=$DMTCP_COORD_PORT\"\n"
1521                   "      arguments=$arguments\" DMTCP_CHECKPOINT_INTERVAL=$DMTCP_CHECKPOINT_INTERVAL\"\n"
1522                   "      arguments=$arguments\" DMTCP_TMPDIR=$DMTCP_TMPDIR\"\n"
1523                   "      arguments=$arguments\" DMTCP_REMLAUNCH_NODES=$DMTCP_REMLAUNCH_NODES\"\n"
1524                   "      bound=$(($DMTCP_REMLAUNCH_NODES - 1))\n"
1525                   "      for i in $(seq 0 $bound); do\n"
1526                   "        eval \"val=\\${DMTCP_REMLAUNCH_${i}_SLOTS}\"\n"
1527                   "        arguments=$arguments\" DMTCP_REMLAUNCH_${i}_SLOTS=\\\"$val\\\"\"\n"
1528                   "        bound2=$(($val - 1))\n"
1529                   "        for j in $(seq 0 $bound2); do\n"
1530                   "          eval \"ckpts=\\${DMTCP_REMLAUNCH_${i}_${j}}\"\n"
1531                   "          arguments=$arguments\" DMTCP_REMLAUNCH_${i}_${j}=\\\"$ckpts\\\"\"\n"
1532                   "        done\n"
1533                   "      done\n"
1534                   "      pbsdsh -u \"$llaunch\" \"$arguments\"\n"
1535                   "      exit 0\n"
1536                   "    fi\n"
1537                   "  fi\n"
1538                   "fi\n"
1539                   "\n\n"
1540              );
fprintf ( fp, "%s", theRestartScriptMultiHostProcessing );
1543   }
fclose ( fp );
1546   {
string filename = RESTART_SCRIPT_BASENAME "." RESTART_SCRIPT_EXT;
string dirname = jalib::Filesystem::DirName(uniqueFilename);
1549     int dirfd = open(dirname.c_str(), O_DIRECTORY | O_RDONLY);
JASSERT(dirfd != -1) (dirname) (JASSERT_ERRNO);
1552     /* Set execute permission for user. */
1553     struct stat buf;
JASSERT(stat(uniqueFilename.c_str(), &buf) == 0);
JASSERT(chmod(uniqueFilename.c_str(), buf.st_mode | S_IXUSR) == 0);
1556     // Create a symlink from
1557     //   dmtcp_restart_script.sh -> dmtcp_restart_script_<curCompId>.sh
unlink(filename.c_str());
JTRACE("linking \"dmtcp_restart_script.sh\" filename to uniqueFilename")
1560       (filename) (dirname) (uniqueFilename);
1561     // FIXME:  Handle error case of symlink()
JWARNING(symlinkat(uniqueFilename.c_str(), dirfd, filename.c_str()) == 0);
JASSERT(close(dirfd) == 0);
1564   }
_restartFilenames.clear();
1566 }
1568 static void signalHandler(int signum)
1569 {
1570   if (signum == SIGINT) {
prog.handleUserCommand('q');
1572   } else if (signum == SIGALRM) {
timerExpired = true;
1574   } else {
JASSERT(false) .Text("Not reached");
1576   }
1577 }
1579 static void setupSignalHandlers()
1580 {
1581   struct sigaction action;
sigemptyset(&action.sa_mask);
action.sa_flags = 0;
action.sa_handler = signalHandler;
sigaction(SIGINT, &action, NULL);
sigaction(SIGALRM, &action, NULL);
1588 }
1590 // This code is also copied to ssh.cpp:updateCoordHost()
1591 static void calcLocalAddr()
1592 {
string cmd;
1594   char hostname[HOST_NAME_MAX];
JASSERT(gethostname(hostname, sizeof hostname) == 0) (JASSERT_ERRNO);
1596   struct addrinfo *result;
1597   struct addrinfo *res;
1598   int error;
1599   struct addrinfo hints;
memset(&localhostIPAddr, 0, sizeof localhostIPAddr);
memset(&hints, 0, sizeof(struct addrinfo));
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
hints.ai_flags = AI_PASSIVE;
hints.ai_protocol = 0;
hints.ai_canonname = NULL;
hints.ai_addr = NULL;
hints.ai_next = NULL;
1611   /* resolve the domain name into a list of addresses */
error = getaddrinfo(hostname, NULL, &hints, &result);
1613   if (error == 0) {
1614     /* loop over all returned results and do inverse lookup */
1615     bool success = false;
1616     for (res = result; res != NULL; res = res->ai_next) {
1617       char name[NI_MAXHOST] = "";
1618       struct sockaddr_in *s = (struct sockaddr_in*) res->ai_addr;
error = getnameinfo(res->ai_addr, res->ai_addrlen, name, NI_MAXHOST, NULL, 0, 0);
1621       if (error != 0) {
JTRACE("getnameinfo() failed.") (gai_strerror(error));
1623         continue;
1624       }
1625       if (Util::strStartsWith(name, hostname) ||
Util::strStartsWith(hostname, name)) {
JASSERT(sizeof localhostIPAddr == sizeof s->sin_addr);
success = true;
memcpy(&localhostIPAddr, &s->sin_addr, sizeof s->sin_addr);
1630       }
1631     }
1632     if (!success) {
JWARNING("Failed to find coordinator IP address.  DMTCP may fail.") (hostname) ;
1634     }
1635   } else {
1636     if (error == EAI_SYSTEM) {
perror("getaddrinfo");
1638     } else {
JTRACE("Error in getaddrinfo") (gai_strerror(error));
1640     }
inet_aton("127.0.0.1", &localhostIPAddr);
1642   }
coordHostname = hostname;
1644 }
1646 static void resetCkptTimer()
1647 {
alarm(theCheckpointInterval);
1649 }
1651 void DmtcpCoordinator::updateCheckpointInterval(uint32_t interval)
1652 {
1653   if ( interval != DMTCPMESSAGE_SAME_CKPT_INTERVAL &&
interval != theCheckpointInterval) {
1655     int oldInterval = theCheckpointInterval;
theCheckpointInterval = interval;
JNOTE ( "CheckpointInterval updated (for this computation only)" )
1658       ( oldInterval ) ( theCheckpointInterval );
resetCkptTimer();
1660   }
1661 }
1663 void DmtcpCoordinator::eventLoop(bool daemon)
1664 {
1665   struct epoll_event ev;
epollFd = epoll_create(MAX_EVENTS);
JASSERT(epollFd != -1) (JASSERT_ERRNO);
ev.events = EPOLLIN;
ev.data.ptr = listenSock;
JASSERT(epoll_ctl(epollFd, EPOLL_CTL_ADD, listenSock->sockfd(), &ev) != -1)
1672     (JASSERT_ERRNO);
1674   if (!daemon &&
1675       // epoll_ctl below fails if STDIN is pointing to /dev/null.
1676       // Not sure why.
jalib::Filesystem::GetDeviceName(0) != "/dev/null" &&
jalib::Filesystem::GetDeviceName(0) != "/dev/zero" &&
jalib::Filesystem::GetDeviceName(0) != "/dev/random") {
ev.events = EPOLLIN;
1681 #ifdef EPOLLRDHUP
ev.events |= EPOLLRDHUP;
1683 #endif
ev.data.ptr = (void*) STDIN_FILENO;
JASSERT(epoll_ctl(epollFd, EPOLL_CTL_ADD, STDIN_FILENO, &ev) != -1)
1686       (JASSERT_ERRNO);
1687   }
1689   while (true) {
1690     // Wait until either there is some activity on client sockets, or the timer
1691     // has expired.
1692     int nfds = epoll_wait(epollFd, events, MAX_EVENTS, -1);
1694     // The ckpt timer has expired; it's time to checkpoint.
1695     if (nfds == -1 && errno == EINTR && timerExpired) {
timerExpired = false;
startCheckpoint();
1698       continue;
1699     }
JASSERT(nfds != -1) (JASSERT_ERRNO);
1703     for (int n = 0; n < nfds; ++n) {
1704       void *ptr = events[n].data.ptr;
1705       if ((events[n].events & EPOLLHUP) ||
1706 #ifdef EPOLLRDHUP
1707           (events[n].events & EPOLLRDHUP) ||
1708 #endif
1709           (events[n].events & EPOLLERR)) {
JASSERT(ptr != listenSock);
1711         if (ptr == (void*) STDIN_FILENO) {
JASSERT(epoll_ctl(epollFd, EPOLL_CTL_DEL, STDIN_FILENO, &ev) != -1)
1713             (JASSERT_ERRNO);
close(STDIN_FD);
1715         } else {
onDisconnect((CoordClient*)ptr);
1717         }
1718       } else if (events[n].events & EPOLLIN) {
1719         if (ptr == (void*) listenSock) {
onConnect();
1721         } else if (ptr == (void*) STDIN_FILENO) {
1722           char buf[1];
1723           int ret = Util::readAll(STDIN_FD, buf, sizeof(buf));
JASSERT(ret != -1) (JASSERT_ERRNO);
1725           if (ret > 0) {
handleUserCommand(buf[0]);
1727           } else {
JNOTE("closing stdin");
JASSERT(epoll_ctl(epollFd, EPOLL_CTL_DEL, STDIN_FILENO, &ev) != -1)
1730               (JASSERT_ERRNO);
close(STDIN_FD);
1732           }
1733         } else {
onData((CoordClient*)ptr);
1735         }
1736       }
1737     }
1738   }
1739 }
1741 void DmtcpCoordinator::addDataSocket(CoordClient *client)
1742 {
1743   struct epoll_event ev;
1745 #ifdef EPOLLRDHUP
ev.events = EPOLLIN | EPOLLRDHUP;
1747 #else
ev.events = EPOLLIN;
1749 #endif
ev.data.ptr = client;
JASSERT(epoll_ctl(epollFd, EPOLL_CTL_ADD, client->sock().sockfd(), &ev) != -1)
1752     (JASSERT_ERRNO);
1753 }
1755 #define shift argc--; argv++
1757 int main ( int argc, char** argv )
1758 {
initializeJalib();
1761   //parse port
thePort = DEFAULT_PORT;
1763   const char* portStr = getenv( ENV_VAR_NAME_PORT );
1764   if ( portStr == NULL ) portStr = getenv("DMTCP_PORT"); // deprecated
1765   if ( portStr != NULL ) thePort = jalib::StringToInt( portStr );
1767   bool daemon = false;
1769   char * tmpdir_arg = NULL;
shift;
1772   while(argc > 0){
string s = argv[0];
1774     if(s=="-h" || s=="--help"){
printf("%s", theUsage);
1776       return 1;
1777     } else if ((s=="--version") && argc==1){
printf("%s", DMTCP_VERSION_AND_COPYRIGHT_INFO);
1779       return 1;
1780     }else if(s == "-q" || s == "--quiet"){
1781       // TODO(kapil): Ignored for now. Remove in later versions.
shift;
1783     }else if(s=="--exit-on-last"){
exitOnLast = true;
shift;
1786     }else if(s=="--exit-after-ckpt"){
exitAfterCkpt = true;
shift;
1789     }else if(s=="--daemon"){
daemon = true;
shift;
1792     } else if (s == "-i" || s == "--interval") {
setenv(ENV_VAR_CKPT_INTR, argv[1], 1);
shift; shift;
1795     } else if (argv[0][0] == '-' && argv[0][1] == 'i' &&
isdigit(argv[0][2])) { // else if -i5, for example
setenv(ENV_VAR_CKPT_INTR, argv[0]+2, 1);
shift;
1799     } else if (argc>1 && (s == "-p" || s == "--port" || s == "--coord-port")) {
thePort = jalib::StringToInt( argv[1] );
shift; shift;
1802     } else if (argv[0][0] == '-' && argv[0][1] == 'p' &&
isdigit(argv[0][2])) { // else if -p0, for example
thePort = jalib::StringToInt( argv[0]+2 );
shift;
1806     }else if(argc>1 && s == "--port-file"){
thePortFile = argv[1];
shift; shift;
1809     }else if(argc>1 && (s == "-c" || s == "--ckptdir")){
setenv(ENV_VAR_CHECKPOINT_DIR, argv[1], 1);
shift; shift;
1812     }else if(argc>1 && (s == "-t" || s == "--tmpdir")){
tmpdir_arg = argv[1];
shift; shift;
1815     }else if(argc == 1){ //last arg can be port
1816       char *endptr;
1817       long x = strtol(argv[0], &endptr, 10);
1818       if ((ssize_t)strlen(argv[0]) != endptr - argv[0]) {
fprintf(stderr, theUsage, DEFAULT_PORT);
1820         return 1;
1821       } else {
thePort = jalib::StringToInt( argv[0] );
shift;
1824       }
x++, x--; // to suppress unused variable warning
1826     }else{
fprintf(stderr, theUsage, DEFAULT_PORT);
1828       return 1;
1829     }
1830   }
tmpDir = Util::calcTmpDir(tmpdir_arg);
Util::initializeLogFile(tmpDir);
JTRACE ( "New DMTCP coordinator starting." )
1836     ( UniquePid::ThisProcess() );
1838   if ( thePort < 0 )
1839   {
fprintf(stderr, theUsage, DEFAULT_PORT);
1841     return 1;
1842   }
calcLocalAddr();
1846   if (getenv(ENV_VAR_CHECKPOINT_DIR) != NULL) {
ckptDir = getenv(ENV_VAR_CHECKPOINT_DIR);
1848   } else {
ckptDir = get_current_dir_name();
1850   }
1852   /*Test if the listener socket is already open*/
1853   if ( fcntl(PROTECTED_COORD_FD, F_GETFD) != -1 ) {
listenSock = new jalib::JServerSocket ( PROTECTED_COORD_FD );
JASSERT ( listenSock->port() != -1 ) .Text ( "Invalid listener socket" );
JTRACE ( "Using already created listener socker" ) ( listenSock->port() );
1857   } else {
errno = 0;
listenSock = new jalib::JServerSocket(jalib::JSockAddr::ANY, thePort, 128);
JASSERT ( listenSock->isValid() ) ( thePort ) ( JASSERT_ERRNO )
1862       .Text ( "Failed to create listen socket."
1863        "\nIf msg is \"Address already in use\", this may be an old coordinator."
1864        "\nKill default coordinator and try again:  dmtcp_command -q"
1865        "\nIf that fails, \"pkill -9 dmtcp_coord\","
1866        " and try again in a minute or so." );
1867   }
thePort = listenSock->port();
1870   if (!thePortFile.empty()) {
Util::writeCoordPortToFile(thePort, thePortFile.c_str());
1872   }
1874   //parse checkpoint interval
1875   const char* interval = getenv ( ENV_VAR_CKPT_INTR );
1876   if ( interval != NULL ) {
theDefaultCheckpointInterval = jalib::StringToInt ( interval );
theCheckpointInterval = theDefaultCheckpointInterval;
1879   }
1881 #if 0
JASSERT_STDERR <<
1883     "dmtcp_coordinator starting..." <<
1884     "\n    Port: " << thePort <<
1885     "\n    Checkpoint Interval: ";
1886   if(theCheckpointInterval==0)
JASSERT_STDERR << "disabled (checkpoint manually instead)";
1888   else
JASSERT_STDERR << theCheckpointInterval;
JASSERT_STDERR  <<
1891     "\n    Exit on last client: " << exitOnLast << "\n";
1892 #else
fprintf(stderr, "dmtcp_coordinator starting..."
1895           "\n    Host: %s (%s)"
1896           "\n    Port: %d"
1897           "\n    Checkpoint Interval: ",
coordHostname.c_str(), inet_ntoa(localhostIPAddr), thePort);
1899   if(theCheckpointInterval==0)
fprintf(stderr, "disabled (checkpoint manually instead)");
1901   else
fprintf(stderr, "%d", theCheckpointInterval);
fprintf(stderr, "\n    Exit on last client: %d\n", exitOnLast);
1904 #endif
1906   if (daemon) {
JASSERT_STDERR  << "Backgrounding...\n";
1908     int fd = open("/dev/null", O_RDWR);
JASSERT(dup2(fd, STDIN_FILENO) == STDIN_FILENO);
JASSERT(dup2(fd, STDOUT_FILENO) == STDOUT_FILENO);
JASSERT(dup2(fd, STDERR_FILENO) == STDERR_FILENO);
1912     JASSERT_CLOSE_STDERR();
1913     if (fd > STDERR_FILENO) {
close(fd);
1915     }
1917     if (fork() > 0) {
JTRACE ( "Parent Exiting after fork()" );
exit(0);
1920     }
1921     //pid_t sid = setsid();
1922   } else {
JASSERT_STDERR  <<
1924       "Type '?' for help." <<
1925       "\n\n";
1926   }
1928   /* We set up the signal handler for SIGINT and SIGALRM.
1929    * SIGINT is used to send DMT_KILL_PEER message to all the connected peers
1930    * before exiting.
1931    * SIGALRM is used for interval checkpointing.
1932    */
setupSignalHandlers();
1935   /* If the coordinator was started transparently by dmtcp_launch, then we
1936    * want to block signals, such as SIGINT.  To see why this is important:
1937    * % gdb dmtcp_launch a.out
1938    * (gdb) run
1939    * ^C   # Stop gdb to get its attention, and continue debugging.
1940    * # The above scenario causes the SIGINT to go to a.out and its child,
1941    * # the dmtcp_coordinator.  The coord then triggers the SIGINT handler,
1942    * # which sends DMT_KILL_PEER to kill a.out.
1943    */
1944   if ( exitOnLast && daemon ) {
sigset_t set;
sigfillset(&set);
1947     // sigprocmask is only per-thread; but the coordinator is single-threaded.
sigprocmask(SIG_BLOCK, &set, NULL);
1949   }
prog.eventLoop(daemon);
1952   return 0;
1953 }