/**********************************************************************
 * This file was generated by coNCePTuaL on Mon Jul 24 17:05:46 2006
 * using the c_udgram backend (C + Unix-domain datagram sockets).
 * Do not modify this file; modify /tmp/helloworld.ncptl instead.
 *
 * Entire source program
 * ---------------------
 *   All tasks mytask output "Hello from task " and mytask and "!".
 **********************************************************************/

/*****************
 * Include files *
 *****************/

/* Header files needed by all C-based backends */
#include <stdio.h>
#include <string.h>
#include <ncptl/ncptl.h>

/* Header files specific to the c_udgram backend */
#include <errno.h>
#include <signal.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <sys/ioctl.h>
#include <sys/wait.h>
#include <sys/select.h>

/**********
 * Macros *
 **********/

/* Define the maximum loop trip count that we're willing to unroll fully. */
#define CONC_MAX_UNROLL 5

/* Specify the number of trial iterations in each FOR <time> loop. */
#define CONC_FOR_TIME_TRIALS 10

/* Define a macro that rounds a double to a ncptl_int. */
#define CONC_DBL2INT(D) ((ncptl_int)((D)+0.5))

/* Define a filename template for this program's sockets. */
#define SOCKET_TEMPLATE "c_udgram_XXXXXX"

/* Define a wrapper for ncptl_fatal() after a system call failure. */
#define CONC_SYSTEM_ERROR(MSG)                                     \
  do {                                                             \
    if (strerror(errno))                                           \
      ncptl_fatal (MSG " (%s, errno=%d)", strerror(errno), errno); \
    else                                                           \
      ncptl_fatal (MSG " (errno=%d)", errno);                      \
  }                                                                \
  while (0)

/* Define a macro that sets or resets a socket's blocking state. */
#define CONC_SET_BLOCKING(CSTATE, PEER, BLOCKING)                         \
  do {                                                                    \
    int nonblocking = 1 - (BLOCKING);                                     \
    if (ioctl ((CSTATE)->channels[PEER], FIONBIO, &nonblocking) == -1)    \
      CONC_SYSTEM_ERROR ("Failed to toggle a socket's blocking state");  \
  }                                                                       \
  while (0)

/*********************
 * Type declarations *
 *********************/

/* Define the type of event to perform. */
typedef enum {
  EV_SEND,                      /* Synchronous send */
  EV_ASEND,                     /* Asynchronous send */
  EV_RECV,                      /* Synchronous receive */
  EV_ARECV,                     /* Asynchronous receive */
  EV_WAIT,                      /* Wait for all asynchronous sends/receives to complete */
  EV_DELAY,                     /* Spin or sleep */
  EV_TOUCH,                     /* Touch a region of memory */
  EV_SYNC,                      /* Barrier synchronization */
  EV_RESET,                     /* Reset counters */
  EV_FLUSH,                     /* Compute aggregate functions for log-file columns */
  EV_MCAST,                     /* Synchronous multicast */
  EV_REDUCE,                    /* Reduction with or without a subsequent multicast */
  EV_BTIME,                     /* Beginning of a timed loop */
  EV_ETIME,                     /* Ending of a timed loop */
  EV_REPEAT,                    /* Repeatedly process the next N events */
  EV_SUPPRESS,                  /* Suppress writing to the log and standard output */
  EV_NEWSTMT,                   /* Beginning of a new top-level statement */
  EV_CODE,                      /* None of the above */
  NUM_EVS                       /* Number of event types in CONC_EVENT_TYPE */
} CONC_EVENT_TYPE;

/* Describe a synchronous or asynchronous send event. */
typedef struct {
  ncptl_int dest;               /* Destination task */
  ncptl_int size;               /* Number of bytes to send */
  ncptl_int alignment;          /* Message alignment (in bytes) */
  ncptl_int pendingsends;       /* # of outstanding sends */
  ncptl_int pendingrecvs;       /* # of outstanding receives */
  ncptl_int buffernum;          /* Buffer # to send from */
  int misaligned;               /* 1=misaligned from a page; 0=align as specified */
  int touching;                 /* 1=touch every word before sending */
  int verification;             /* 1=fill message buffer with known contents */
  void *buffer;                 /* Pointer to message memory */
} CONC_SEND_EVENT;

/* Describe a synchronous or asynchronous receive event. */
typedef struct {
  ncptl_int source;             /* Source task */
  ncptl_int size;               /* Number of bytes to receive */
  ncptl_int alignment;          /* Message alignment (in bytes) */
  ncptl_int pendingsends;       /* # of outstanding sends */
  ncptl_int pendingrecvs;       /* # of outstanding receives */
  ncptl_int buffernum;          /* Buffer # to receive into */
  int misaligned;               /* 1=misaligned from a page; 0=align as specified */
  int touching;                 /* 1=touch every word after reception */
  int verification;             /* 1=verify that all bits are correct */
  void *buffer;                 /* Pointer to message memory */
} CONC_RECV_EVENT;

/* Describe a wait-for-asynchronous-completions event. */
typedef struct {
  ncptl_int numsends;           /* # of sends we expect to complete. */
  ncptl_int numrecvs;           /* # of receives we expect to complete. */
  ncptl_int numrecvbytes;       /* # of bytes we expect to receive-complete */
  ncptl_int *touchedlist;       /* List of receives that need to be touched */
  ncptl_int numtouches;         /* # of elements in the above */
} CONC_WAIT_EVENT;

/* Describe a spin or sleep delay. */
typedef struct {
  uint64_t microseconds;        /* Length of delay in microseconds */
  int spin0sleep1;              /* 0=spin; 1=sleep */
} CONC_DELAY_EVENT;

/* Describe a barrier synchronization event. */
typedef struct {
  NCPTL_QUEUE *peerqueue;       /* Map from barrier rank to physical rank */
  int syncrank;                 /* This task's rank in the barrier */
} CONC_SYNC_EVENT;

/* Describe a walk over a memory-region. */
typedef struct {
  ncptl_int regionbytes;        /* Size in bytes of the region to touch */
  ncptl_int bytestride;         /* Stride in bytes to touch */
  ncptl_int numaccesses;        /* Number of words to touch */
  ncptl_int wordsize;           /* Size in bytes of each touch */
  ncptl_int firstbyte;          /* Byte offset of the first byte to touch */
} CONC_TOUCH_EVENT;

/* Describe a synchronous multicast event. */
typedef struct {
  ncptl_int source;             /* Source task */
  ncptl_int size;               /* Number of bytes to send */
  ncptl_int alignment;          /* Message alignment (in bytes) */
  ncptl_int pendingsends;       /* # of outstanding sends */
  ncptl_int pendingrecvs;       /* # of outstanding receives */
  ncptl_int buffernum;          /* Buffer # to send/receive from */
  int misaligned;               /* 1=misaligned from a page; 0=align as specified */
  int touching;                 /* 1=touch every word before sending */
  int verification;             /* 1=verify that all bits are correct */
  void *buffer;                 /* Pointer to message memory */
  NCPTL_QUEUE *peerqueue;       /* Map from multicast rank to physical rank */
  int mcastrank;                /* This task's rank in the multicast */
} CONC_MCAST_EVENT;

/* Describe a reduction event. */
typedef struct {
  ncptl_int numitems;           /* # of items to reduce */
  ncptl_int itemsize;           /* # of bytes per item */
  ncptl_int alignment;          /* Message alignment (in bytes) */
  ncptl_int pendingsends;       /* # of outstanding sends */
  ncptl_int pendingrecvs;       /* # of outstanding receives */
  ncptl_int buffernum;          /* Buffer # to send/receive from */
  int misaligned;               /* 1=misaligned from a page; 0=align as specified */
  int touching;                 /* 1=touch every word before sending/after receiving */
  int sending;                  /* 1=we're a sender */
  int receiving;                /* 1=we're a receiver */
  void *buffer;                 /* Pointer to message memory */
  NCPTL_QUEUE *all_senders;     /* Map from reduction rank to physical rank (reduce step) */
  NCPTL_QUEUE *all_receivers;   /* Map from reduction rank to physical rank (multicast step) */
  int sendrank;                 /* This task's rank in the reduction (reduce step) */
  int recvrank;                 /* This task's rank in the reduction (multicast step) */
} CONC_REDUCE_EVENT;

/* Describe an event representing the beginning of a timed loop. */
typedef struct {
  uint64_t usecs;               /* Requested loop duration */
  uint64_t starttime;           /* Time at which the loop state last changed */
  uint64_t itersleft;           /* # of iterations remaining */
  uint64_t previters;           /* # of iterations we performed last time */
  int prev_quiet;               /* Previous value of suppress_output */
  int timing_trial;             /* 1=performing a timing trial; 0=running for real */
  volatile int finished;        /* 1=time has expired; 0=still ticking */
} CONC_BTIME_EVENT;

/* Describe an event representing the end of a timed loop. */
typedef struct {
  ncptl_int begin_event;        /* Index into eventlist[] of the corresponding BTIME event */
} CONC_ETIME_EVENT;

/* Describe an event representing repetitions of subsequent events. */
typedef struct {
  ncptl_int end_event;          /* Index into eventlist[] of the last event to repeat */
  ncptl_int numreps;            /* # of repetitions to perform */
} CONC_REPEAT_EVENT;

/* Describe an event representing output suppression (either on or off). */
typedef struct conc_suppress_event {
  int quiet;                    /* 0=allow output; 1=suppress it */
  int prev_quiet;               /* Previous value of suppress_output */
  ncptl_int matching_event;     /* Event ID of the "suppression on" event */
} CONC_SUPPRESS_EVENT;

/* Describe an event representing arbitrary code to execute at run time. */
typedef struct {
  ncptl_int number;             /* Unique number corresponding to a specific piece of code */
  ncptl_int var_mytask;         /* Copy of var_mytask to use within a piece of code */
} CONC_CODE_EVENT;

/* Describe an arbitrary coNCePTuaL event. */
typedef struct {
  CONC_EVENT_TYPE type;         /* Type of event */
  union {
    CONC_SEND_EVENT send;       /* Send state */
    CONC_RECV_EVENT recv;       /* Receive state */
    CONC_WAIT_EVENT wait;       /* Wait-for-completions state */
    CONC_DELAY_EVENT delay;     /* State for spins and sleeps */
    CONC_TOUCH_EVENT touch;     /* State for memory touching */
    CONC_SYNC_EVENT sync;       /* Synchronization state */
    CONC_MCAST_EVENT mcast;     /* Multicast state */
    CONC_REDUCE_EVENT reduce;   /* Reduction state */
    CONC_BTIME_EVENT btime;     /* Timed-loop state */
    CONC_ETIME_EVENT etime;     /* Additional timed-loop state */
    CONC_REPEAT_EVENT rep;      /* Repeated-events state */
    CONC_SUPPRESS_EVENT suppress;       /* State for suppressing output */
    CONC_CODE_EVENT code;       /* State for arbitrary code */
  } s;
} CONC_EVENT;

/* Fully specify an arbitrary for() loop (used by FOR EACH). */
typedef struct {
  int integral;                 /* 1=integral values; 0=floating-point values */
  enum {                        /* Comparison of loop variable to end variable */
    CONC_LEQ,                   /* Increasing progression */
    CONC_GEQ                    /* Decreasing progression */
  } comparator;
  enum {                        /* How to increment the loop variable */
    CONC_ADD,                   /* Arithmetically */
    CONC_MULT,                  /* Geometrically increasing */
    CONC_DIV                    /* Geometrically decreasing */
  } increment;
  union {
    struct {
      ncptl_int loopvar;        /* Loop variable */
      ncptl_int prev_loopvar;   /* Previous value of loop variable */
      ncptl_int startval;       /* Initial value of loop variable */
      ncptl_int endval;         /* Value not to exceed */
      ncptl_int incval;         /* Loop-variable increment */
    } i;
    struct {
      double loopvar;           /* Loop variable */
      double prev_loopvar;      /* Previous value of loop variable */
      double startval;          /* Initial value of loop variable */
      double endval;            /* Value not to exceed */
      double incval;            /* Loop-variable increment */
    } d;
  } u;
} LOOPBOUNDS;

/* Declare a type that encapsulates much of the communication state. */
typedef struct {
  int *channels;                /* Socket connections to each other task */
  NCPTL_QUEUE **blockedsendQ;   /* Queue of events corresponding to blocked sends to each destination */
  NCPTL_QUEUE **blockedrecvQ;   /* Queue of events corresponding to blocked receives from each source */
  int sendsblocked;             /* 0=all blockedsendQ[] are empty; 1=nonempty */
  int recvsblocked;             /* 0=all blockedrecvQ[] are empty; 1=nonempty */
} COMMSTATE;

/* Create symbolic names for each set of communication state we plan to use. */
typedef enum {
  COMMSTATE_PT2PT,              /* State for point-to-point communication */
  COMMSTATE_COLL,               /* State for collective communication */
  NUMCHANNELSETS                /* Sentinel describing the number of entries in COMMSTATE_TYPES */
} COMMSTATE_TYPES;

/********************
 * Global variables *
 ********************/

/* Variables exported to coNCePTuaL programs */
ncptl_int var_bytes_received = 0;       /* Total number of bytes received */
ncptl_int var_msgs_received = 0;        /* Total number of messages received */
ncptl_int var_bit_errors = 0;   /* Total number of bit errors observed */
ncptl_int var_total_msgs = 0;   /* Sum of messages sent and messages received */
ncptl_int var_msgs_sent = 0;    /* Total number of messages sent */
ncptl_int var_bytes_sent = 0;   /* Total number of bytes sent */
ncptl_int var_num_tasks = 1;    /* Number of tasks running the program */
ncptl_int var_elapsed_usecs = 0;        /* Elapsed time in microseconds */
ncptl_int var_total_bytes = 0;  /* Sum of bytes sent and bytes received */

/* Variables used internally by boilerplate code */
uint64_t starttime;             /* Time the clock was last reset (microseconds) */
ncptl_int pendingrecvs = 0;     /* Current # of outstanding receives */
ncptl_int pendingrecvbytes = 0; /* Current # of bytes in outstanding receives */
NCPTL_QUEUE *touchedqueue;      /* Queue of asynchronous receives to touch */
ncptl_int pendingsends = 0;     /* Current # of outstanding sends */
NCPTL_QUEUE *eventqueue;        /* List of coNCePTuaL events to perform */
int within_time_loop = 0;       /* 1=we're within a FOR <time> loop */
int suppress_output = 0;        /* 1=suppress output to stdout and the log file */
void *touch_region = NULL;      /* Memory region to touch */
ncptl_int touch_region_size = 0;        /* # of bytes in the above */
int virtrank;                   /* This task's virtual rank in the computation */
int physrank;                   /* This task's physical rank in the computation */
NCPTL_VIRT_PHYS_MAP *procmap;   /* Virtual to physical rank mapping */
char *logfiletmpl;              /* Template for the log file's name */
char *logfiletmpl_default;      /* Default value of the above */

/* Global variables specific to the c_udgram backend */
COMMSTATE commstate[NUMCHANNELSETS];    /* Communication state for multiple independent channels */
int abnormal_exit = 1;          /* 1=exit handler invoked after an abnormal exit; 0=invoked after a normal one */
ncptl_int maxpacketlen;         /* Maximum message length that can be transmitted whole */
NCPTL_QUEUE *alltasksQ;         /* List of 0, ..., var_num_tasks-1 for all-task synchronization */

/*************************
 * Function declarations *
 *************************/

/* Attempt to send data.  Abort on error.  Return 1 on success,
 * 0 if the send blocked. */
static inline int conc_send_packet (COMMSTATE * cstate, int dest,
                                    void *buffer, int packetsize)
{
  int bytessent;                /* Number of bytes actually sent */

  do
    bytessent = send (cstate->channels[dest], buffer, (size_t) packetsize, 0);
  while (bytessent == -1 && errno == EINTR);
  if (bytessent == -1)
    if (errno == EAGAIN)
      return 0;                 /* send() blocked. */
    else
      CONC_SYSTEM_ERROR ("Failed to send a message");
  else
    /* send() did not block. */
  if (bytessent != packetsize)
    ncptl_fatal ("Expected to send %d bytes but actually sent %d bytes",
                 packetsize, bytessent);
  return 1;
}

/* Attempt to send data.  Abort on error.  Return 1 on success,
 * 0 if the send blocked. */
static inline int conc_receive_packet (COMMSTATE * cstate, int source,
                                       void *buffer, int packetsize)
{
  int bytesreceived;            /* Number of bytes actually received */

  do
    bytesreceived = recv (cstate->channels[source],
                          buffer, packetsize, MSG_WAITALL | MSG_TRUNC);
  while (bytesreceived == -1 && errno == EINTR);
  if (bytesreceived == -1)
    if (errno == EAGAIN)
      return 0;                 /* recv() blocked. */
    else
      CONC_SYSTEM_ERROR ("Failed to receive a message");
  else
    /* recv() did not block. */
  if (bytesreceived != packetsize)
    ncptl_fatal
      ("Expected to receive %d bytes but actually received %d bytes",
       packetsize, bytesreceived);
  return 1;
}

/* Binary search for the largest valid packet length. */
static ncptl_int conc_find_max_packet_len (COMMSTATE * cstate)
{
  int maxsize;                  /* Maximum valid message size */
  int delta;                    /* Change in message size to try next */
  socklen_t intsize = sizeof (int);     /* Number of bytes in maxsize */
  char *msgbuffer;              /* Buffer used for sending and receiving */
  int bytessent;                /* Number of bytes actually sent */
  int bytesreceived;            /* Number of bytes actually received */

  CONC_SET_BLOCKING (cstate, physrank, 1);
  if (getsockopt
      (cstate->channels[physrank], SOL_SOCKET, SO_SNDBUF, &maxsize,
       &intsize) == -1 || !maxsize)
    CONC_SYSTEM_ERROR ("Failed to get a socket parameter");
  msgbuffer = (char *) ncptl_malloc (maxsize, 0);

  for (delta = maxsize / 2; delta; delta /= 2) {
    do
      bytessent =
        send (cstate->channels[physrank], msgbuffer, (size_t) maxsize, 0);
    while (bytessent == -1 && errno == EINTR);
    if (bytessent != -1) {
      do
        bytesreceived = recv (cstate->channels[physrank], msgbuffer,
                              (size_t) maxsize, MSG_WAITALL | MSG_TRUNC);
      while (bytesreceived == -1 && errno == EINTR);
      if (bytesreceived != (int) maxsize)
        ncptl_fatal
          ("Expected to receive %d bytes but actually received %d bytes",
           maxsize, bytesreceived);
      maxsize += delta;
    }
    else
      maxsize -= delta;
  }
  ncptl_free (msgbuffer);
  CONC_SET_BLOCKING (cstate, physrank, 0);
  return (maxsize / sizeof (ncptl_int)) * sizeof (ncptl_int) - 1;
}

 /* Wait until a specific send/receive request completes.
  * ASSUMPTION: target_req has not yet completed. */
static inline void conc_wait_one (COMMSTATE * cstate, void *target_req)
{
  struct timeval polltime;      /* Time structure corresponding to a minimal poll time */

  /* Alternately complete receives and sends until target_req
   * is satisfied. */
  while (1) {
    ncptl_int task_ofs;         /* Offset from our task ID */

    /* Process each task ID in turn, starting from our own. */
    for (task_ofs = 0; task_ofs < var_num_tasks; task_ofs++) {
      ncptl_int taskID = (physrank + task_ofs) % var_num_tasks; /* Task ID to process */
      ncptl_int pendingsends = ncptl_queue_length (cstate->blockedsendQ[taskID]);       /* Number of blocked sends */
      ncptl_int pendingrecvs = ncptl_queue_length (cstate->blockedrecvQ[taskID]);       /* Number of blocked receives */
      int making_progress;      /* 1=a packet was sent/received; 0=we idled */

      /* Keep sending from task taskID and/or receiving from task
       * taskID until we can no longer do so. */
      making_progress = pendingrecvs || pendingsends;
      while (making_progress) {
        int recvfd = 0;         /* File descriptor for a receive channel */
        int sendfd = 0;         /* File descriptor for a send channel */
        fd_set firstrecv;       /* Set containing either recvfd or nothing */
        fd_set firstsend;       /* Set containing either sendfd or nothing */
        int last_fd = -1;       /* 1 + highest-numbered descriptor on which to select() */
        CONC_RECV_EVENT *recvev = NULL; /* List of blockedrecvQ's contents */
        CONC_SEND_EVENT *sendev = NULL; /* List of blockedsendQ's contents */
        int recv_available;     /* 1=recv() won't block; 0=it will */
        int send_available;     /* 1=send() won't block; 0=it will */

        FD_ZERO (&firstrecv);
        if (pendingrecvs) {
          recvev = ncptl_queue_contents (cstate->blockedrecvQ[taskID], 0);
          recvfd = cstate->channels[recvev->source];
          FD_SET (recvfd, &firstrecv);
          last_fd = recvfd > last_fd ? recvfd : last_fd;
        }
        FD_ZERO (&firstsend);
        if (pendingsends) {
          sendev = ncptl_queue_contents (cstate->blockedsendQ[taskID], 0);
          sendfd = cstate->channels[sendev->dest];
          FD_SET (sendfd, &firstsend);
          last_fd = sendfd > last_fd ? sendfd : last_fd;
        }
        polltime.tv_sec = 0;
        polltime.tv_usec = 1;   /* Setting to 0 tends to hog the CPU. */
        select (last_fd + 1, &firstrecv, &firstsend, NULL, &polltime);
        send_available = sendfd && FD_ISSET (sendfd, &firstsend);
        recv_available = recvfd && FD_ISSET (recvfd, &firstrecv);
        if (!send_available && !recv_available)
          /* Give up if we're blocked on both the send and receive sides. */
          break;
        making_progress = 0;    /* Assume no progress. */

        /* Process a pending receive, if any. */
        if (pendingrecvs && recv_available) {
          int packetsize = (int) (recvev->size > maxpacketlen ? maxpacketlen : recvev->size);   /* Number of bytes to receive per call */

          if (conc_receive_packet
              (cstate, recvev->source, recvev->buffer, packetsize)) {
            /* The receive went through. */
            making_progress = 1;
            recvev->buffer = (void *) ((char *) recvev->buffer + packetsize);
            recvev->size -= packetsize;
            if (!recvev->size) {
              /* We received a complete message. */
              (void) ncptl_queue_pop (cstate->blockedrecvQ[taskID]);
              if (!--pendingrecvs)
                ncptl_queue_empty (cstate->blockedrecvQ[taskID]);
              if (recvev == target_req)
                /* We finished the one message we care about. */
                return;
            }
          }
        }

        /* Process a pending send, if any. */
        if (pendingsends && send_available) {
          int packetsize = (int) (sendev->size > maxpacketlen ? maxpacketlen : sendev->size);   /* Number of bytes to send per call */

          if (conc_send_packet
              (cstate, sendev->dest, sendev->buffer, packetsize)) {
            /* The send went through. */
            making_progress = 1;
            sendev->buffer = (void *) ((char *) sendev->buffer + packetsize);
            sendev->size -= packetsize;
            if (!sendev->size) {
              /* We sent a complete message. */
              (void) ncptl_queue_pop (cstate->blockedsendQ[taskID]);
              if (!--pendingsends)
                ncptl_queue_empty (cstate->blockedsendQ[taskID]);
              if (sendev == target_req)
                /* We finished the one message we care about. */
                return;
            }
          }
        }
      }
    }
  }
}

/* Wait until all of our blocked sends and receives complete. */
static inline void conc_wait_all (COMMSTATE * cstate)
{
  ncptl_int taskID;             /* Loop over all task IDs */

  /* Ensure we have something to do. */
  if (!cstate->sendsblocked && !cstate->recvsblocked)
    return;

  /* Await completion from each task in turn -- not very efficient
   * but we expect var_num_tasks to be small. */
  for (taskID = 0; taskID < var_num_tasks; taskID++) {
    ncptl_int pendingsends;     /* Number of blocked sends */
    ncptl_int pendingrecvs;     /* Number of blocked receives */

    /* Wait for the final send to complete. */
    pendingsends = ncptl_queue_length (cstate->blockedsendQ[taskID]);
    if (pendingsends) {
      CONC_SEND_EVENT *lastsend;        /* Task taskID's final pending send */

      lastsend =
        (CONC_SEND_EVENT *) ncptl_queue_contents (cstate->
                                                  blockedsendQ[taskID], 0);
      lastsend += pendingsends - 1;
      conc_wait_one (cstate, (void *) lastsend);
    }

    /* Wait for the final receive to complete. */
    pendingrecvs = ncptl_queue_length (cstate->blockedrecvQ[taskID]);
    if (pendingrecvs) {
      CONC_RECV_EVENT *lastrecv;        /* Task taskID's final pending receive */

      lastrecv =
        (CONC_RECV_EVENT *) ncptl_queue_contents (cstate->
                                                  blockedrecvQ[taskID], 0);
      lastrecv += pendingrecvs - 1;
      conc_wait_one (cstate, (void *) lastrecv);        /* Wait for the final receive */
    }
  }

  /* Enable send/receive functions to proceed. */
  cstate->sendsblocked = 0;
  cstate->recvsblocked = 0;
}

/* Post a receive and block until it completes. */
static inline void conc_recv_msg (COMMSTATE * cstate,
                                  CONC_RECV_EVENT * recvev)
{
  cstate->recvsblocked = 1;
  conc_wait_one (cstate,
                 ncptl_queue_push (cstate->blockedrecvQ[recvev->source],
                                   (void *) recvev));
}

/* Enqueue a message on a given channel (nonblocking). */
static inline void conc_asend_msg (COMMSTATE * cstate,
                                   CONC_SEND_EVENT * sendev)
{
  cstate->sendsblocked = 1;
  ncptl_queue_push (cstate->blockedsendQ[sendev->dest], (void *) sendev);
}

/* Barrier-synchronize a list of tasks using a butterfly pattern. */
static inline void conc_synchronize (COMMSTATE * cstate, int *peerlist,
                                     ncptl_int maxrank, ncptl_int syncrank)
{
  ncptl_int stage;              /* Current stage of the butterfly pattern */
  CONC_SEND_EVENT sendev;       /* Send event to pass to conc_asend_msg() */
  CONC_RECV_EVENT recvev;       /* Receive event to pass to conc_recv_msg() */
  ncptl_int dummybuffer;        /* Dummy message buffer */

  memset ((void *) &sendev, 0, sizeof (CONC_SEND_EVENT));
  memset ((void *) &recvev, 0, sizeof (CONC_RECV_EVENT));
  for (stage = ncptl_func_bits (maxrank) - 1; stage >= 0; stage--) {
    ncptl_int peernum = syncrank ^ (1 << stage);        /* Offset into peerlist[] of our current peer */

    if (peernum <= maxrank) {
      sendev.dest = peerlist[peernum];
      sendev.buffer = (void *) &dummybuffer;
      conc_asend_msg (cstate, &sendev);
      recvev.source = peerlist[peernum];
      recvev.buffer = (void *) &dummybuffer;
      conc_recv_msg (cstate, &recvev);
      conc_wait_all (cstate);
    }
  }
}


/* Allocate a new event of a given type and return a pointer to it. */
static CONC_EVENT *conc_allocate_event (CONC_EVENT_TYPE type)
{
  CONC_EVENT *newevent = ncptl_queue_allocate (eventqueue);

  newevent->type = type;
  return newevent;
}

/* Declare an exit handler that gets called automatically when the
 * program terminates, whether successfully or not. */
static void conc_exit_handler (void)
{
  if (abnormal_exit)
    (void) killpg (0, SIGTERM);
}

/* Initialize coNCePTuaL, the messaging layer, and this program itself. */
static void conc_initialize (int argc, char *argv[])
{
  /* Variables needed by all C-based backends */
  CONC_EVENT *eventlist;        /* List of events to execute */
  ncptl_int numevents;          /* Number of entries in eventlist[] */
  int help_only = 0;            /* 1=User specified --help; save time by skipping ncptl_init() */
  char *argv0 = strrchr (argv[0], '/') ? strrchr (argv[0], '/') + 1 : argv[0];  /* Base name of the executable program */
  int i;                        /* Generic loop variable */

  /* Declare all of our command-line arguments. */
  NCPTL_CMDLINE arguments[] = {
    {NCPTL_TYPE_STRING, NULL, "logfile", 'L', "Log-file template", {0}},
    {NCPTL_TYPE_INT, NULL, "tasks", 'T', "Number of tasks to use", {0}}
  };

  /* Incorporate the complete coNCePTuaL source code as an array
   * for use by ncptl_log_write_prologue(). */
  char *sourcecode[] = {
    "All tasks mytask output \"Hello from task \" and mytask and \"!\".",
    NULL
  };

  /* Variables specific to the c_udgram backend */
  int cset;                     /* Index into commstate[] */
  ncptl_int taskID;             /* Loop over all task IDs */
  struct sockaddr_un ***socketinfo;     /* Socket information (e.g., filename) for every socket in channels[][][] */
  char maxpacketlen_str[50];    /* String equivalent of maxpacketlen */

  /* As a special case, if the command line contains --help, then skip
   * the coNCePTuaL initialization step. */
  for (i = 1; i < argc; i++)
    if (!strcmp (argv[i], "--"))
      break;
    else if (!strcmp (argv[i], "--help") || !strcmp (argv[i], "-?")) {
      argv[1] = "-?";           /* Guaranteed to work, even with getopt() */
      help_only = 1;
      break;
    }

  /* Initialize the coNCePTuaL run-time library. */
  if (!help_only)
    ncptl_init (NCPTL_RUN_TIME_VERSION, argv[0]);
  (void) atexit (conc_exit_handler);

  /* Initialize the communication routines needed by the c_udgram backend. */
  /* Prevent exiting children from killing their parent. */
  ncptl_permit_signal (SIGCHLD);

  /* Plug variables and default values into the NCPTL_CMDLINE structure. */
  arguments[0].variable = (CMDLINE_VALUE *) & logfiletmpl;
  arguments[1].variable = (CMDLINE_VALUE *) & var_num_tasks;
  arguments[1].defaultvalue.intval = 1;
  logfiletmpl_default = (char *) ncptl_malloc (strlen (argv0) + 15, 0);
  sprintf (logfiletmpl_default, "%s-%%p.log", argv0);
  arguments[0].defaultvalue.stringval = logfiletmpl_default;

  /* Parse the command line. */
  abnormal_exit = 0;            /* Let "--help" exit normally. */
  ncptl_parse_command_line (argc, argv, arguments,
                            sizeof (arguments) / sizeof (NCPTL_CMDLINE));
  if (help_only)
    ncptl_fatal
      ("Internal error in the c_generic backend: failed to exit after giving help");
  abnormal_exit = 1;

  /* Initialize the communication routines needed by the c_udgram backend. */
  /* Create names for all of the sockets every task will need. */
  socketinfo =
    (struct sockaddr_un ***) ncptl_malloc (NUMCHANNELSETS *
                                           sizeof (struct sockaddr_un **), 0);
  for (cset = 0; cset < NUMCHANNELSETS; cset++) {
    ncptl_int src;              /* Source task ID */
    ncptl_int dest;             /* Destination task ID */
    socketinfo[cset] =
      (struct sockaddr_un **) ncptl_malloc (var_num_tasks *
                                            sizeof (struct sockaddr_un *), 0);
    for (dest = 0; dest < var_num_tasks; dest++) {
      socketinfo[cset][dest] =
        (struct sockaddr_un *) ncptl_malloc (var_num_tasks *
                                             sizeof (struct sockaddr_un), 0);
      for (src = 0; src < var_num_tasks; src++) {
        int tempfd;             /* File descriptor returned by mkstemp() (not used) */

        socketinfo[cset][dest][src].sun_family = AF_UNIX;
        strcpy (socketinfo[cset][dest][src].sun_path, SOCKET_TEMPLATE);
        if ((tempfd = mkstemp (socketinfo[cset][dest][src].sun_path)) == -1)
          CONC_SYSTEM_ERROR ("Failed to create a unique filename");
        if (close (tempfd) == -1)
          CONC_SYSTEM_ERROR ("Failed to close an open file");
      }
    }
  }

  /* Spawn var_num_tasks-1 processes (one per task excluding the master). */
  physrank = 0;
  if (setpgid (0, 0) == -1)
    CONC_SYSTEM_ERROR ("Failed to start a new process group");
  for (i = 1; i < var_num_tasks; i++) {
    /* Spawn a child task and assign it a virtual task ID.  While
     * this could be done in a tree pattern instead of linearly,
     * we expect var_num_tasks to be small; plus, a simple mapping
     * from PIDs to tasks may help debug deadlocks and other
     * problems. */
    int newpid;                 /* 0=child; other=parent */

    if ((newpid = fork ()) == -1)
      CONC_SYSTEM_ERROR ("Failed to spawn a child process");
    else if (!newpid) {
      /* Child */
      physrank = (ncptl_int) i;
      break;
    }
  }

  /* Create and bind sockets for inter-task communication. */
  for (cset = 0; cset < NUMCHANNELSETS; cset++) {
    COMMSTATE *cstate = &commstate[cset];       /* Current channel set */
    cstate->channels = (int *) ncptl_malloc (var_num_tasks * sizeof (int), 0);
    for (taskID = 0; taskID < var_num_tasks; taskID++) {
      if ((cstate->channels[taskID] = socket (PF_UNIX, SOCK_DGRAM, 0)) == -1)
        CONC_SYSTEM_ERROR ("Socket creation failed");
      if (unlink (socketinfo[cset][physrank][taskID].sun_path) == -1)
        CONC_SYSTEM_ERROR ("File deletion failed");
      if (bind (cstate->channels[taskID],
                (const struct sockaddr *) &socketinfo[cset][physrank][taskID],
                SUN_LEN (&socketinfo[cset][physrank][taskID])) == -1)
        CONC_SYSTEM_ERROR ("Failed to bind a socket");
    }
  }

  /* Connect to each of our peers in turn. */
  for (cset = 0; cset < NUMCHANNELSETS; cset++) {
    COMMSTATE *cstate = &commstate[cset];       /* Current channel set */

    for (taskID = 0; taskID < var_num_tasks; taskID++) {
      /* Keep trying to connect until the socket we want
       * to connect to exists and is, in fact, a socket. */
      while (connect (cstate->channels[taskID],
                      (const struct sockaddr *)
                      &socketinfo[cset][taskID][physrank],
                      SUN_LEN (&socketinfo[cset][taskID][physrank])) == -1) {
        if (errno != ECONNREFUSED && errno != ENOENT)
          CONC_SYSTEM_ERROR ("Failed to connect a socket");
        usleep (1000);
      }

      /* Now that we're connected, put the socket in nonblocking mode. */
      CONC_SET_BLOCKING (cstate, taskID, 0);
    }
  }

  /* Free all of the resources we no longer need. */
  for (cset = 0; cset < NUMCHANNELSETS; cset++) {
    for (taskID = 0; taskID < var_num_tasks; taskID++) {
      (void) unlink (socketinfo[cset][taskID][physrank].sun_path);
      ncptl_free (socketinfo[cset][taskID]);
    }
    ncptl_free (socketinfo[cset]);
  }

  /* Determine the maximum packet length. */
  maxpacketlen = conc_find_max_packet_len (&commstate[0]);
  sprintf (maxpacketlen_str, "%" NICS " bytes", maxpacketlen);
  ncptl_log_add_comment ("Maximum payload per datagram", maxpacketlen_str);

  /* Establish a mapping from (virtual) task IDs to (physical) ranks. */
  procmap = ncptl_allocate_task_map (var_num_tasks);
  virtrank = ncptl_physical_to_virtual (procmap, physrank);

  /* Allocate a variety of dynamically growing queues. */
  eventqueue = ncptl_queue_init (sizeof (CONC_EVENT));
  touchedqueue = ncptl_queue_init (sizeof (ncptl_int));

  /* Perform initializations specific to the c_udgram backend. */
  for (cset = 0; cset < NUMCHANNELSETS; cset++) {
    COMMSTATE *cstate = &commstate[cset];       /* Current channel set */

    cstate->blockedsendQ =
      ncptl_malloc (var_num_tasks * sizeof (NCPTL_QUEUE *), 0);
    cstate->blockedrecvQ =
      ncptl_malloc (var_num_tasks * sizeof (NCPTL_QUEUE *), 0);
    for (taskID = 0; taskID < var_num_tasks; taskID++) {
      cstate->blockedsendQ[taskID] =
        ncptl_queue_init (sizeof (CONC_SEND_EVENT));
      cstate->blockedrecvQ[taskID] =
        ncptl_queue_init (sizeof (CONC_RECV_EVENT));
    }
    cstate->sendsblocked = 0;
    cstate->recvsblocked = 0;
  }

  alltasksQ = ncptl_queue_init (sizeof (int));
  for (i = 0; i < (int) var_num_tasks; i++)
    ncptl_queue_push (alltasksQ, &i);

 /****************************************************
  * Generated, program-specific initialization code. *
  ****************************************************/

  /* ALL TASKS var_mytask OUTPUT ['"Hello from task "', 'var_mytask', '"!"'] */
  {                             /* ALL TASKS var_mytask */
    ncptl_int var_mytask = virtrank;

    /* The current coNCePTuaL statement applies to our task. */
    CONC_EVENT *thisev = conc_allocate_event (EV_CODE);

    thisev->s.code.number = 0;
    thisev->s.code.var_mytask = var_mytask;
  }

 /*************************
  * More boilerplate code *
  *************************/

  /* Abort if the program will terminate with pending messages. */
  if (pendingsends && pendingrecvs)
    ncptl_fatal ("Neglected to await the completion of %" NICS
                 " asynchronous %s and %" NICS " asynchronous %s",
                 pendingsends, pendingsends == 1LL ? "send" : "sends",
                 pendingrecvs, pendingrecvs == 1LL ? "receive" : "receives");
  else if (pendingsends)
    ncptl_fatal ("Neglected to await the completion of %" NICS
                 " asynchronous %s", pendingsends,
                 pendingsends == 1LL ? "send" : "sends");
  else if (pendingrecvs)
    ncptl_fatal ("Neglected to await the completion of %" NICS
                 " asynchronous %s", pendingrecvs,
                 pendingrecvs == 1LL ? "receive" : "receives");

  /* Allocate memory for non-unique messages and asynchronous
   * message handles now that we know how much memory we need
   * to allocate. */
  eventlist = ncptl_queue_contents (eventqueue, 0);
  numevents = ncptl_queue_length (eventqueue);
  for (i = 0; i < numevents; i++) {
    CONC_EVENT *thisev = &eventlist[i]; /* Cache of the current event */

    switch (thisev->type) {
      default:
        break;
    }
  }
}

/* Process a subset of the events in a given event list. */
static void conc_process_events (CONC_EVENT * eventlist,
                                 ncptl_int firstev, ncptl_int lastev,
                                 ncptl_int numreps)
{
  CONC_EVENT *thisev;           /* Cache of the current event */
  CONC_EVENT *thisev_first = &eventlist[firstev];       /* Cache of the first event */
  ncptl_int i;                  /* Iterate over events. */
  ncptl_int j;                  /* Iterate over repetitions. */

  /* Process from event firstev to event lastev (both inclusive). */
  for (j = numreps; j > 0; j--)
    for (i = firstev, thisev = thisev_first; i <= lastev; i++, thisev++) {
      /* Declare variables needed by all C-based backends. */
      void *msgbuffer;          /* Send or receive buffer */

      /* Process a single event. */
      switch (thisev->type) {
        case EV_CODE:
          /* Execute an arbitrary piece of code. */
          switch (thisev->s.code.number) {
            case 0:
              /* ALL TASKS var_mytask OUTPUT ['"Hello from task "', 'var_mytask', '"!"'] */
              if (!suppress_output) {
                uint64_t stop_elapsed_usecs = ncptl_time ();

                var_elapsed_usecs = stop_elapsed_usecs - starttime;
                printf ("%s%.10lg%s\n", "Hello from task ",
                        (double) thisev->s.code.var_mytask, "!");
                starttime += ncptl_time () - stop_elapsed_usecs;
              }
              break;

            default:
              /* The C code generation module must be broken. */
              ncptl_fatal ("Internal error: unknown EV_CODE block %" NICS,
                           thisev->s.code.number);
              break;
          }
          break;

        default:
          /* The c_generic backend or the c_udgram backend must be broken. */
          ncptl_fatal ("Internal error: unknown event type %d", thisev->type);
          break;
      }
    }
}

/* Finish up cleanly and return a status code. */
static int conc_finalize (void)
{
  int exitcode = 0;             /* Program exit code (to pass to exit()) */

  /* Declarations specific to the c_udgram backend */
  int childstatus;              /* Exit status of a child process */
  COMMSTATE *cstate;            /* Pointer into commstate[] */
  ncptl_int i;                  /* Loop over child task IDs */

  /* Inform the run-time library that it's no longer needed. */
  ncptl_queue_empty (eventqueue);
  ncptl_free (eventqueue);
  ncptl_finalize ();

  /* Finalization code specific to the c_udgram backend */
  conc_synchronize (&commstate[COMMSTATE_COLL],
                    ncptl_queue_contents (alltasksQ, 0),
                    ncptl_queue_length (alltasksQ) - 1, physrank);
  if (physrank == 0)
    for (i = 1; i < var_num_tasks; i++)
      if (wait (&childstatus) == -1 || !WIFEXITED (childstatus))
        exitcode = 1;
  for (cstate = commstate; cstate < commstate + NUMCHANNELSETS; cstate++) {
    for (i = 0; i < var_num_tasks; i++) {
      ncptl_queue_empty (cstate->blockedsendQ[i]);
      ncptl_queue_empty (cstate->blockedrecvQ[i]);
    }
    ncptl_free (cstate->channels);
    ncptl_free (cstate->blockedsendQ);
    ncptl_free (cstate->blockedrecvQ);
  }
  abnormal_exit = 0;

  /* Return an exit status code. */
  return exitcode;
}

/*************************************************************************/
/***************************** MAIN ROUTINE ******************************/
/*************************************************************************/

/* Program execution starts here. */
int main (int argc, char *argv[])
{
  /* Declare variables needed by all C-based backends. */
  CONC_EVENT *eventlist;        /* List of events to execute */
  ncptl_int numevents;          /* Number of entries in eventlist[] */

  /* ----- Initialization ----- */
  conc_initialize (argc, argv);
  eventlist = ncptl_queue_contents (eventqueue, 0);
  numevents = ncptl_queue_length (eventqueue);
  starttime = ncptl_time ();

  /* ----- Event-list processing ----- */
  conc_process_events (eventlist, 0, numevents - 1, 1);

  /* ----- Finalization ----- */
  return conc_finalize ();
}